telemetry: CI emitter for admin.windyword.ai (inert until token)
ci.run (one row per finished job, exactly once via a high-water mark; branch_kind default|pr|other so the dashboard can show "main is red") and service.health (interval counts: finished/failed/cancelled, waiting, running, runners online, oldest wait). Shapes declared with Windy Telemetry 40; sends nothing until WINDYGIT_TELEMETRY_TOKEN exists. State is only advanced after a 2xx, so a failed post retries. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
196
scripts/telemetry_emit.py
Normal file
196
scripts/telemetry_emit.py
Normal file
@@ -0,0 +1,196 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Emit Windy Git CI telemetry to admin.windyword.ai (Windy Telemetry 40's ledger).
|
||||
|
||||
Runs on Veron after every sync (root; reads the gitea DB via `docker exec`).
|
||||
Shapes are declared with Telemetry 40 (2026-09-23) — do not add keys or enum
|
||||
values without re-declaring: a declared family quarantines any row that
|
||||
doesn't match.
|
||||
|
||||
ci.run one row per FINISHED job, exactly once (high-water mark on
|
||||
action_run_job.id in STATE)
|
||||
service.health one row per invocation: CI plane counts for the interval
|
||||
|
||||
Privacy: ids, names of repos/jobs, codes, counts, durations. No commit
|
||||
messages, no logs, no author names.
|
||||
|
||||
--dry-run print the batch instead of posting (and don't advance STATE)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import datetime, timezone
|
||||
|
||||
INGEST = os.environ.get("TELEMETRY_INGEST_URL", "https://admin.windyword.ai/v1/events")
|
||||
TOKEN = os.environ.get("WINDYGIT_TELEMETRY_TOKEN", "")
|
||||
STATE = os.environ.get("TELEMETRY_STATE", "/var/lib/windy-git/telemetry-state.json")
|
||||
PLATFORM, SERVICE = "windy-git", "ci"
|
||||
OUTCOME = {1: "success", 2: "failure", 3: "cancelled", 4: "skipped"}
|
||||
EVENTS = {"push", "pull_request", "pull_request_sync", "schedule", "workflow_dispatch"}
|
||||
RUNNERS_EXPECTED = 6
|
||||
|
||||
|
||||
def sql(query: str) -> list[dict]:
|
||||
"""Rows as dicts, via psql's json_agg — no driver needed on the host."""
|
||||
wrapped = f"select coalesce(json_agg(t), '[]'::json) from ({query}) t;"
|
||||
out = subprocess.run(
|
||||
[
|
||||
"docker",
|
||||
"exec",
|
||||
"-i",
|
||||
"windy-git-db-1",
|
||||
"sh",
|
||||
"-c",
|
||||
'psql -U "$POSTGRES_USER" -d gitea -At -v ON_ERROR_STOP=1',
|
||||
],
|
||||
input=wrapped,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
).stdout.strip()
|
||||
return json.loads(out or "[]")
|
||||
|
||||
|
||||
def load_state() -> dict:
|
||||
try:
|
||||
with open(STATE) as f:
|
||||
return json.load(f)
|
||||
except (OSError, ValueError):
|
||||
return {}
|
||||
|
||||
|
||||
def iso(epoch: float) -> str:
|
||||
return datetime.fromtimestamp(epoch, timezone.utc).isoformat().replace("+00:00", "Z")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
dry = "--dry-run" in sys.argv
|
||||
state = load_state()
|
||||
now = time.time()
|
||||
last_job = int(state.get("last_job_id", 0))
|
||||
since = float(state.get("last_ts", now - 300))
|
||||
|
||||
if not last_job:
|
||||
# First run: start at the current high-water mark rather than replaying
|
||||
# a month of history into the ledger as if it happened now.
|
||||
last_job = int(sql("select coalesce(max(id),0) as m from action_run_job")[0]["m"])
|
||||
|
||||
jobs = sql(f"""
|
||||
select j.id, j.name as job, j.status, j.started, j.stopped,
|
||||
p.lower_name as repo, p.default_branch, r.workflow_id, r.event,
|
||||
r.ref, r.index as run, left(r.commit_sha, 7) as sha
|
||||
from action_run_job j
|
||||
join action_run r on r.id = j.run_id
|
||||
join repository p on p.id = r.repo_id
|
||||
where j.id > {last_job} and j.status in (1, 2, 3, 4) and j.stopped > 0
|
||||
order by j.id
|
||||
limit 2000""")
|
||||
|
||||
events = []
|
||||
for j in jobs:
|
||||
ref = j["ref"] or ""
|
||||
if ref.startswith("refs/pull/"):
|
||||
kind = "pr"
|
||||
elif ref == f"refs/heads/{j['default_branch']}":
|
||||
kind = "default"
|
||||
else:
|
||||
kind = "other"
|
||||
dur = (j["stopped"] - j["started"]) * 1000 if j["started"] else None
|
||||
ev = {
|
||||
"ts": iso(j["stopped"]),
|
||||
"platform": PLATFORM,
|
||||
"service": SERVICE,
|
||||
"event_type": "ci.run",
|
||||
"actor_type": "system",
|
||||
"metadata": {
|
||||
"repo": j["repo"],
|
||||
"workflow": (j["workflow_id"] or "").removesuffix(".yml").removesuffix(".yaml"),
|
||||
"job": j["job"],
|
||||
"outcome": OUTCOME[j["status"]],
|
||||
"event": j["event"] if j["event"] in EVENTS else "other",
|
||||
"branch_kind": kind,
|
||||
"run": j["run"],
|
||||
"sha": j["sha"],
|
||||
},
|
||||
}
|
||||
if dur is not None and dur >= 0:
|
||||
ev["duration_ms"] = int(dur)
|
||||
events.append(ev)
|
||||
|
||||
# --- heartbeat: counts since the previous invocation --------------------
|
||||
h = sql(f"""
|
||||
select
|
||||
(select count(*) from action_run_job where stopped >= {int(since)} and status in (1,2,3,4)) as jobs_finished,
|
||||
(select count(*) from action_run_job where stopped >= {int(since)} and status = 2) as jobs_failed,
|
||||
(select count(*) from action_run_job where stopped >= {int(since)} and status = 3) as jobs_cancelled,
|
||||
(select count(*) from action_run_job where status in (5, 7)) as jobs_waiting,
|
||||
(select count(*) from action_run_job where status = 6) as jobs_running,
|
||||
(select count(*) from action_runner where deleted is null and last_online >= {int(now) - 120}) as runners_online,
|
||||
(select coalesce(extract(epoch from now())::bigint - min(created), 0)
|
||||
from action_run_job where status in (5, 7)) as oldest_waiting_s""")[0]
|
||||
meta = {k: int(v) for k, v in h.items()}
|
||||
for k in ("repos_synced", "repos_sync_failed", "statuses_posted", "bridge_errors"):
|
||||
v = os.environ.get(f"TELEMETRY_{k.upper()}")
|
||||
if v is not None and v.isdigit(): # absent = couldn't count; never invent 0
|
||||
meta[k] = int(v)
|
||||
events.append(
|
||||
{
|
||||
"ts": iso(now),
|
||||
"platform": PLATFORM,
|
||||
"service": SERVICE,
|
||||
"event_type": "service.health",
|
||||
"actor_type": "system",
|
||||
"metadata": meta,
|
||||
}
|
||||
)
|
||||
|
||||
if dry:
|
||||
print(json.dumps({"events": events}, indent=1)[:4000])
|
||||
print(f"[telemetry] DRY RUN: {len(events)} events ({len(jobs)} ci.run)")
|
||||
return 0
|
||||
if not TOKEN:
|
||||
print("[telemetry] WINDYGIT_TELEMETRY_TOKEN unset — not sending (not a failure)")
|
||||
return 0
|
||||
|
||||
for i in range(0, len(events), 500):
|
||||
req = urllib.request.Request(
|
||||
INGEST,
|
||||
data=json.dumps({"events": events[i : i + 500]}).encode(),
|
||||
method="POST",
|
||||
headers={
|
||||
"Authorization": f"Bearer {TOKEN}",
|
||||
"Content-Type": "application/json",
|
||||
"User-Agent": "windy-git-telemetry/1",
|
||||
},
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30) as r:
|
||||
body = r.read()[:300]
|
||||
if r.status >= 300:
|
||||
raise urllib.error.HTTPError(
|
||||
INGEST, r.status, body.decode(errors="replace"), None, None
|
||||
)
|
||||
except urllib.error.HTTPError as e:
|
||||
print(f"[telemetry] FAILED ingest HTTP {e.code}: {e.read()[:200]!r}")
|
||||
return 1 # state NOT advanced: the same rows retry next run
|
||||
except urllib.error.URLError as e:
|
||||
print(f"[telemetry] FAILED ingest: {e.reason}")
|
||||
return 1
|
||||
|
||||
os.makedirs(os.path.dirname(STATE), exist_ok=True)
|
||||
new_last = max([j["id"] for j in jobs], default=last_job)
|
||||
with open(STATE + ".tmp", "w") as f:
|
||||
json.dump({"last_job_id": new_last, "last_ts": now}, f)
|
||||
os.replace(STATE + ".tmp", STATE)
|
||||
print(f"[telemetry] sent {len(events)} events ({len(jobs)} ci.run)")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user