from __future__ import annotations
|
|
import time
|
from dataclasses import dataclass, field
|
from datetime import datetime
|
from typing import Any
|
|
|
def aware_now() -> datetime:
|
return datetime.now().astimezone()
|
|
|
def iso(value: datetime) -> str:
|
return value.isoformat(timespec="milliseconds")
|
|
|
def parse_task_start(raw: str | None, process_start: datetime) -> tuple[datetime, str]:
|
if raw is None:
|
return process_start, "process_start"
|
try:
|
value = datetime.fromisoformat(raw)
|
except ValueError as exc:
|
raise ValueError("--task-start 必须是 ISO-8601 时间") from exc
|
if value.tzinfo is None or value.utcoffset() is None:
|
raise ValueError("--task-start 必须带时区")
|
if value > process_start:
|
raise ValueError("--task-start 不得晚于进程开始时间")
|
return value, "arg"
|
|
|
@dataclass
|
class RunClock:
|
process_start_wall: datetime = field(default_factory=aware_now)
|
process_start_mono: float = field(default_factory=time.monotonic)
|
task_start_wall: datetime | None = None
|
task_start_source: str | None = None
|
phases: dict[str, dict[str, Any]] = field(default_factory=dict)
|
providers: list[dict[str, Any]] = field(default_factory=list)
|
|
def set_task_start(self, raw: str | None) -> None:
|
self.task_start_wall, self.task_start_source = parse_task_start(
|
raw, self.process_start_wall
|
)
|
|
def begin(self, name: str) -> None:
|
self.phases[name] = {
|
"started_at": iso(aware_now()),
|
"started_mono": time.monotonic(),
|
"status": "RUNNING",
|
}
|
|
def end(self, name: str, status: str = "OK") -> None:
|
phase = self.phases[name]
|
end_mono = time.monotonic()
|
phase.update(
|
{
|
"finished_at": iso(aware_now()),
|
"elapsed_seconds": max(0.0, end_mono - phase.pop("started_mono")),
|
"status": status,
|
}
|
)
|
|
def walls(self, end_wall: datetime | None = None, end_mono: float | None = None) -> tuple[float, float]:
|
end_wall = end_wall or aware_now()
|
end_mono = end_mono if end_mono is not None else time.monotonic()
|
process = max(0.0, end_mono - self.process_start_mono)
|
assert self.task_start_wall is not None
|
task = max(process, (end_wall - self.task_start_wall).total_seconds())
|
return task, process
|
|
def metrics(self, run_id: str, status: str, commit_ready_at: datetime) -> dict[str, Any]:
|
task, process = self.walls(commit_ready_at)
|
return {
|
"schema_version": 1,
|
"run_id": run_id,
|
"status": status,
|
"task_start": iso(self.task_start_wall),
|
"task_start_source": self.task_start_source,
|
"process_start": iso(self.process_start_wall),
|
"commit_ready_at": iso(commit_ready_at),
|
"task_wall_seconds": task,
|
"process_wall_seconds": process,
|
"wall_scope": "through_commit_ready",
|
"phases": self.phases,
|
"providers": self.providers,
|
}
|
