File size: 2,654 Bytes
970c0d3 92d50ea 970c0d3 92d50ea 970c0d3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 | """Azure timeline snapshots shared by the dispatcher and queue UI.
Only completed pipeline steps are counted; this is not iteration/sample progress.
Network reads happen in the dispatcher, never in a browser refresh callback.
"""
from html import escape
from threading import Lock
from time import monotonic
import requests
_snapshots = {}
_lock = Lock()
_MAX_AGE = 180
def refresh_progress(run_id, url, headers):
"""Fetch a timeline without allowing an unavailable API to affect dispatch."""
try:
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status()
records = response.json().get("records", [])
tasks = [r for r in records if r.get("type") == "Task"]
if not tasks:
return
# A retried task's latest attempt is the authoritative one.
latest = {}
for task in tasks:
key = task.get("identifier") or task.get("id")
if key not in latest or task.get("attempt", 1) >= latest[key].get("attempt", 1):
latest[key] = task
tasks = list(latest.values())
completed = sum(t.get("state") == "completed" for t in tasks)
active = [str(t.get("name", "Running")) for t in tasks if t.get("state") == "inProgress"]
snapshot = (monotonic(), completed, len(tasks), ", ".join(active))
with _lock:
for key in list(_snapshots):
if monotonic() - _snapshots[key][0] > _MAX_AGE:
del _snapshots[key]
_snapshots[str(run_id)] = snapshot
except (requests.RequestException, ValueError, TypeError, AttributeError):
return
def render_progress(entry):
"""Render escaped, accessible progress; never invent a percentage."""
status = entry.get("status", "Pending")
if status.startswith("Finished"):
return '<progress max="100" value="100" aria-label="Completed"></progress> Completed'
if status not in {"Running", "Triggered", "Waiting"}:
return escape(status)
with _lock:
snapshot = _snapshots.get(str(entry.get("ci_run_id", "")))
if not snapshot or monotonic() - snapshot[0] > _MAX_AGE:
return '<progress aria-label="Waiting for progress information"></progress> Waiting for progress information'
_, completed, total, active = snapshot
label = f"Pipeline steps: {completed}/{total} completed"
detail = active or ("Waiting for results" if completed == total else "Waiting for next step")
return (
f'<progress max="{total}" value="{completed}" aria-label="{label}"></progress>'
f' <span>{label} — {escape(detail)}</span>'
)
|