182 lines
7.6 KiB
Python
182 lines
7.6 KiB
Python
"""Bounded read-only CI statistics; absent measurements remain unknown."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import re
|
|
import threading
|
|
import time
|
|
from datetime import UTC, datetime
|
|
from typing import Literal
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, ValidationError
|
|
|
|
from .deployment_broker import DeploymentBrokerClient
|
|
|
|
|
|
class _StatisticsModel(BaseModel):
|
|
model_config = ConfigDict(extra="ignore", allow_inf_nan=False, strict=True)
|
|
|
|
|
|
class _Budget(_StatisticsModel):
|
|
cpuLimit: float | int | None = Field(default=None, ge=0, le=20)
|
|
memoryLimitBytes: int | None = Field(default=None, ge=0)
|
|
activeJobLimit: int | None = Field(default=None, ge=0, le=6)
|
|
heavyJobLimit: int | None = Field(default=None, ge=0, le=6)
|
|
reservedCpu: float | int | None = Field(default=None, ge=0)
|
|
reservedMemoryBytes: int | None = Field(default=None, ge=0)
|
|
activeJobs: int | None = Field(default=None, ge=0)
|
|
heavyJobs: int | None = Field(default=None, ge=0)
|
|
|
|
|
|
class _Host(_StatisticsModel):
|
|
cpuBusyPercent: float | int | None = Field(default=None, ge=0, le=100)
|
|
loadRatio: float | int | None = Field(default=None, ge=0)
|
|
availableMemoryBytes: int | None = Field(default=None, ge=0)
|
|
cacheFreeBytes: int | None = Field(default=None, ge=0)
|
|
|
|
|
|
class _Profile(_StatisticsModel):
|
|
id: Literal["checks", "backend", "frontend", "security", "browser", "build"]
|
|
displayName: str = Field(min_length=1, max_length=80)
|
|
labels: list[str] = Field(max_length=8)
|
|
runnerName: str = Field(min_length=1, max_length=80)
|
|
status: str = Field(min_length=1, max_length=80)
|
|
queueDepth: int | None = Field(default=None, ge=0)
|
|
activeJobs: int | None = Field(default=None, ge=0)
|
|
oldestQueuedSeconds: float | int | None = Field(default=None, ge=0)
|
|
reason: str | None = Field(default=None, max_length=120)
|
|
cpuLimit: float | int | None = Field(default=None, ge=0, le=8)
|
|
memoryLimitBytes: int | None = Field(default=None, ge=0)
|
|
cpuUsed: float | int | None = Field(default=None, ge=0)
|
|
memoryUsedBytes: int | None = Field(default=None, ge=0)
|
|
completedJobs: int = Field(default=0, ge=0)
|
|
p50WaitSeconds: float | int | None = Field(default=None, ge=0)
|
|
p95WaitSeconds: float | int | None = Field(default=None, ge=0)
|
|
p50DurationSeconds: float | int | None = Field(default=None, ge=0)
|
|
p95DurationSeconds: float | int | None = Field(default=None, ge=0)
|
|
lastDurationSeconds: float | int | None = Field(default=None, ge=0)
|
|
|
|
|
|
class _Window(_StatisticsModel):
|
|
startAt: str | None = Field(default=None, max_length=40)
|
|
endAt: str | None = Field(default=None, max_length=40)
|
|
completedJobs: int = Field(default=0, ge=0)
|
|
samples: int = Field(default=0, ge=0)
|
|
coverage: Literal["bounded", "complete", "unavailable"] = "unavailable"
|
|
|
|
|
|
class _History(_StatisticsModel):
|
|
at: str = Field(min_length=1, max_length=40)
|
|
activeJobs: int | None = Field(default=None, ge=0)
|
|
queuedJobs: int | None = Field(default=None, ge=0)
|
|
cpuUsed: float | int | None = Field(default=None, ge=0)
|
|
memoryUsedBytes: int | None = Field(default=None, ge=0)
|
|
|
|
|
|
class RunnerStatistics(_StatisticsModel):
|
|
schemaVersion: Literal[1]
|
|
status: Literal["healthy", "degraded", "unavailable", "not_configured"]
|
|
scope: Literal["ci-fleet"]
|
|
generatedAt: str = Field(min_length=1, max_length=40)
|
|
budget: _Budget
|
|
host: _Host
|
|
profiles: list[_Profile] = Field(max_length=6)
|
|
window: _Window
|
|
history: list[_History] = Field(max_length=288)
|
|
|
|
|
|
def _valid_timestamp(value: str) -> bool:
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return False
|
|
return parsed.tzinfo is not None and 0 < parsed.timestamp() <= time.time() + 15
|
|
|
|
|
|
def unavailable_statistics(*, configured: bool = True) -> dict[str, object]:
|
|
return RunnerStatistics(
|
|
schemaVersion=1,
|
|
status="unavailable" if configured else "not_configured",
|
|
scope="ci-fleet",
|
|
generatedAt=datetime.now(UTC).isoformat(),
|
|
budget=_Budget(),
|
|
host=_Host(),
|
|
profiles=[],
|
|
window=_Window(),
|
|
history=[],
|
|
).model_dump(mode="json")
|
|
|
|
|
|
def sanitize_statistics(payload: object) -> dict[str, object]:
|
|
"""Accept only the public contract, never raw paths, credentials or jobs."""
|
|
try:
|
|
snapshot = RunnerStatistics.model_validate(payload)
|
|
except (ValidationError, TypeError, ValueError):
|
|
return unavailable_statistics()
|
|
timestamps = [snapshot.generatedAt, *(sample.at for sample in snapshot.history)]
|
|
timestamps.extend(value for value in (snapshot.window.startAt, snapshot.window.endAt) if value)
|
|
if any(not _valid_timestamp(value) for value in timestamps):
|
|
return unavailable_statistics()
|
|
if len({profile.id for profile in snapshot.profiles}) != len(snapshot.profiles):
|
|
return unavailable_statistics()
|
|
for profile in snapshot.profiles:
|
|
allowed_labels = {f"itworx-ci-{profile.id}"}
|
|
if profile.id in {"backend", "frontend"}:
|
|
allowed_labels.update({"linux-validation", "ubuntu-latest"})
|
|
elif profile.id == "build":
|
|
allowed_labels.add("docker-validation")
|
|
if (
|
|
profile.displayName != f"ITWorx CI · {profile.id.title()}"
|
|
or profile.runnerName != f"itworx-ci-{profile.id}"
|
|
or not profile.labels
|
|
or len(set(profile.labels)) != len(profile.labels)
|
|
or not set(profile.labels).issubset(allowed_labels)
|
|
or re.fullmatch(r"[a-z][a-z0-9_-]{0,79}", profile.status) is None
|
|
or (profile.reason is not None and re.fullmatch(r"[a-z][a-z0-9_-]{0,119}", profile.reason) is None)
|
|
):
|
|
return unavailable_statistics()
|
|
sample_time = datetime.fromisoformat(snapshot.generatedAt.replace("Z", "+00:00")).timestamp()
|
|
history_times = [datetime.fromisoformat(sample.at.replace("Z", "+00:00")).timestamp() for sample in snapshot.history]
|
|
if history_times != sorted(set(history_times)) or any(at > sample_time + 15 for at in history_times):
|
|
return unavailable_statistics()
|
|
window = snapshot.window
|
|
if (window.startAt is None) != (window.endAt is None):
|
|
return unavailable_statistics()
|
|
if window.startAt is not None and window.endAt is not None:
|
|
start = datetime.fromisoformat(window.startAt.replace("Z", "+00:00")).timestamp()
|
|
end = datetime.fromisoformat(window.endAt.replace("Z", "+00:00")).timestamp()
|
|
if not start <= end <= sample_time + 15:
|
|
return unavailable_statistics()
|
|
elif window.coverage != "unavailable":
|
|
return unavailable_statistics()
|
|
if time.time() - sample_time > 90 and snapshot.status == "healthy":
|
|
snapshot.status = "degraded"
|
|
return snapshot.model_dump(mode="json")
|
|
|
|
|
|
_lock = threading.Lock()
|
|
_cached: dict[str, object] | None = None
|
|
_cached_at = 0.0
|
|
_CACHE_SECONDS = 30.0
|
|
|
|
|
|
class RunnerStatisticsService:
|
|
def snapshot(self) -> dict[str, object]:
|
|
global _cached, _cached_at
|
|
with _lock:
|
|
now = time.monotonic()
|
|
if _cached is not None and now - _cached_at < _CACHE_SECONDS:
|
|
return copy.deepcopy(_cached)
|
|
broker = DeploymentBrokerClient()
|
|
if not broker.is_configured():
|
|
snapshot = unavailable_statistics(configured=False)
|
|
else:
|
|
try:
|
|
snapshot = sanitize_statistics(broker.runner_fleet_statistics())
|
|
except Exception:
|
|
snapshot = unavailable_statistics()
|
|
_cached = snapshot
|
|
_cached_at = time.monotonic()
|
|
return copy.deepcopy(snapshot)
|