diff --git a/frontend/app/api/ops/observation-collector-status/route.ts b/frontend/app/api/ops/observation-collector-status/route.ts new file mode 100644 index 00000000..1e3ee9d3 --- /dev/null +++ b/frontend/app/api/ops/observation-collector-status/route.ts @@ -0,0 +1,47 @@ +import { NextRequest, NextResponse } from "next/server"; +import { buildProxyExceptionResponse } from "@/lib/api-proxy"; +import { + applyAuthResponseCookies, + buildBackendRequestHeaders, +} from "@/lib/backend-auth"; +import { requireOpsProxyAuth } from "@/lib/ops-proxy-auth"; + +const API_BASE = process.env.POLYWEATHER_API_BASE_URL; + +export async function GET(req: NextRequest) { + if (!API_BASE) { + return NextResponse.json( + { error: "POLYWEATHER_API_BASE_URL is not configured" }, + { status: 500 }, + ); + } + + try { + const auth = await buildBackendRequestHeaders(req); + const authError = requireOpsProxyAuth(req, auth); + if (authError) return authError; + + const upstream = new URL(`${API_BASE}/api/ops/observation-collector-status`); + req.nextUrl.searchParams.forEach((value, key) => { + upstream.searchParams.set(key, value); + }); + + const res = await fetch(upstream.toString(), { + cache: "no-store", + headers: auth.headers, + }); + const raw = await res.text(); + const response = new NextResponse(raw, { + headers: { + "Cache-Control": "no-store", + "Content-Type": "application/json", + }, + status: res.status, + }); + return applyAuthResponseCookies(response, auth.response); + } catch (error) { + return buildProxyExceptionResponse(error, { + publicMessage: "Observation collector status check failed", + }); + } +} diff --git a/frontend/components/ops/__tests__/opsSourceHealth.test.ts b/frontend/components/ops/__tests__/opsSourceHealth.test.ts index a2f772bf..481edf4b 100644 --- a/frontend/components/ops/__tests__/opsSourceHealth.test.ts +++ b/frontend/components/ops/__tests__/opsSourceHealth.test.ts @@ -16,6 +16,10 @@ export function runTests() { path.join(projectRoot, "app", "api", "ops", "source-health", "route.ts"), "utf8", ); + const collectorRoute = fs.readFileSync( + path.join(projectRoot, "app", "api", "ops", "observation-collector-status", "route.ts"), + "utf8", + ); assert( opsApi.includes("sourceHealth") && @@ -40,4 +44,23 @@ export function runTests() { nextRoute.includes("no-store"), "source health proxy must stay ops-admin protected and uncached", ); + assert( + opsApi.includes("observationCollectorStatus") && + opsApi.includes("/api/ops/observation-collector-status"), + "ops client must expose observation collector status endpoint", + ); + assert( + systemPage.includes("观测采集器") && + systemPage.includes("collectorStatus") && + systemPage.includes("failure_count") && + systemPage.includes("last_latency_ms") && + systemPage.includes("冷却"), + "ops system page must show observation collector failures, latency, and cooldown status", + ); + assert( + collectorRoute.includes("requireOpsProxyAuth") && + collectorRoute.includes("/api/ops/observation-collector-status") && + collectorRoute.includes("no-store"), + "observation collector proxy must stay ops-admin protected and uncached", + ); } diff --git a/frontend/components/ops/system/SystemPageClient.tsx b/frontend/components/ops/system/SystemPageClient.tsx index adc949eb..d0ed5071 100644 --- a/frontend/components/ops/system/SystemPageClient.tsx +++ b/frontend/components/ops/system/SystemPageClient.tsx @@ -1,12 +1,17 @@ "use client"; import { useEffect, useState } from "react"; -import { AlertTriangle, RefreshCcw, ShieldCheck, Database, Cpu, HardDrive, RadioTower } from "lucide-react"; +import { Activity, AlertTriangle, RefreshCcw, ShieldCheck, Database, Cpu, HardDrive, RadioTower } from "lucide-react"; import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card"; import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; import { opsApi } from "@/lib/ops-api"; -import type { SourceHealthPayload, SystemStatusPayload, HealthPayload } from "@/types/ops"; +import type { + ObservationCollectorStatusPayload, + SourceHealthPayload, + SystemStatusPayload, + HealthPayload, +} from "@/types/ops"; function sourceStatusTone(status?: string) { if (status === "fresh") return "text-emerald-500"; @@ -25,12 +30,48 @@ function sourceStatusLabel(status?: string) { return "未知"; } +function collectorStatusTone(status?: string) { + if (status === "ok") return "text-emerald-500"; + if (status === "due") return "text-blue-500"; + if (status === "cooldown") return "text-amber-500"; + if (status === "failed" || status === "never_run") return "text-red-500"; + return "text-slate-500"; +} + +function collectorStatusLabel(status?: string) { + if (status === "ok") return "正常"; + if (status === "due") return "到期"; + if (status === "cooldown") return "冷却"; + if (status === "failed") return "失败"; + if (status === "never_run") return "未采集"; + return "未知"; +} + function formatAge(ageMin?: number | null) { if (ageMin == null) return "—"; if (ageMin < 60) return `${Math.round(ageMin)}m`; return `${(ageMin / 60).toFixed(1)}h`; } +function formatSeconds(seconds?: number | null) { + if (seconds == null) return "—"; + if (seconds <= 0) return "已到期"; + if (seconds < 60) return `${Math.round(seconds)}s`; + if (seconds < 3600) return `${Math.round(seconds / 60)}m`; + return `${(seconds / 3600).toFixed(1)}h`; +} + +function formatLatency(ms?: number | null) { + if (ms == null) return "—"; + if (ms < 1000) return `${Math.round(ms)}ms`; + return `${(ms / 1000).toFixed(1)}s`; +} + +function formatTimestamp(value?: string | null) { + if (!value) return "—"; + return value.replace("T", " ").replace("Z", "").slice(0, 19); +} + function sourceReasonLabel(reason?: string | null) { const key = String(reason || "").trim().toLowerCase(); if (key === "observation_time_missing") return "观测时间缺失"; @@ -65,20 +106,23 @@ export function SystemPageClient() { const [health, setHealth] = useState(null); const [status, setStatus] = useState(null); const [sourceHealth, setSourceHealth] = useState(null); + const [collectorStatus, setCollectorStatus] = useState(null); const [error, setError] = useState(""); const load = async () => { setLoading(true); setError(""); try { - const [h, s, sh] = await Promise.all([ + const [h, s, sh, cs] = await Promise.all([ opsApi.health(), opsApi.systemStatus() as Promise, opsApi.sourceHealth(80) as Promise, + opsApi.observationCollectorStatus(200) as Promise, ]); setHealth(h); setStatus(s); setSourceHealth(sh); + setCollectorStatus(cs); } catch (e) { setError(String(e).slice(0, 200)); } finally { @@ -100,6 +144,12 @@ export function SystemPageClient() { const dbOk = status?.db?.ok ?? health?.db?.ok; const cacheAnalysis = status?.cache?.analysis; + const collectorIssues = (collectorStatus?.entries || []) + .filter((entry) => { + const state = String(entry.status || ""); + return ["failed", "cooldown", "never_run", "due"].includes(state) || (entry.failure_count ?? 0) > 0; + }) + .slice(0, 12); const sourceIssues = (sourceHealth?.cities || []) .flatMap((city) => (city.sources || []) @@ -240,6 +290,96 @@ export function SystemPageClient() { ) : null} + + + + + 观测采集器 + + + +
+ {["ok", "due", "cooldown", "failed", "never_run"].map((key) => ( +
+
{collectorStatusLabel(key)}
+
+ {collectorStatus?.status_counts?.[key] ?? 0} +
+
+ ))} +
+ + {(collectorStatus?.sources || []).length ? ( +
+ {(collectorStatus?.sources || []).map((source) => ( +
+
+ {source.source} + + {collectorStatusLabel(source.worst_status)} + +
+
+ 城市 {source.city_count ?? 0} + 间隔 {source.min_interval_sec ?? source.interval_sec ?? "—"}s + 失败 {source.failure_count ?? 0} + 冷却 {source.cooldown_count ?? 0} + 延迟 {formatLatency(source.avg_latency_ms)} + 成功 {formatTimestamp(source.last_success_at)} +
+
+ ))} +
+ ) : ( +
+ 暂无后台采集状态；collector 首次写入后会显示每个 source/city 的最近采集时间、失败次数、延迟和冷却状态。 +
+ )} + + {collectorIssues.length ? ( +
+ + + + + + + + + + + + + + + {collectorIssues.map((entry) => ( + + + + + + + + + + + ))} + +
Source城市状态最近成功失败次数延迟下次错误
{entry.source}{entry.city} + {collectorStatusLabel(entry.status)} + {formatTimestamp(entry.last_success_at)}{entry.failure_count ?? 0}{formatLatency(entry.last_latency_ms)}{formatSeconds(entry.due_in_sec)} + {entry.last_error || "—"} +
+
+ ) : ( +
+ + 当前后台采集器没有失败、冷却或到期积压。 +
+ )} + + + diff --git a/frontend/lib/ops-api.ts b/frontend/lib/ops-api.ts index 790b84b8..662f4735 100644 --- a/frontend/lib/ops-api.ts +++ b/frontend/lib/ops-api.ts @@ -19,6 +19,9 @@ export const opsApi = { sourceHealth(limit = 80) { return opsFetch>(`/api/ops/source-health?limit=${limit}`); }, + observationCollectorStatus(limit = 200) { + return opsFetch>(`/api/ops/observation-collector-status?limit=${limit}`); + }, paymentRuntime() { return opsFetch>("/api/payments/runtime"); }, diff --git a/frontend/types/ops.ts b/frontend/types/ops.ts index 6a8543f1..2b1ba005 100644 --- a/frontend/types/ops.ts +++ b/frontend/types/ops.ts @@ -84,6 +84,57 @@ export type SourceHealthPayload = { total_cities?: number; }; +export type ObservationCollectorStatus = + | "ok" + | "due" + | "cooldown" + | "failed" + | "never_run" + | string; + +export type ObservationCollectorEntry = { + source: string; + city: string; + interval_sec?: number; + last_due_at?: string | null; + last_started_at?: string | null; + last_success_at?: string | null; + last_failure_at?: string | null; + last_latency_ms?: number | null; + failure_count?: number; + last_error?: string | null; + updated_at?: string | null; + next_due_at?: string | null; + due_in_sec?: number | null; + age_sec?: number | null; + in_cooldown?: boolean; + cooldown_until_at?: string | null; + status?: ObservationCollectorStatus; +}; + +export type ObservationCollectorSourceSummary = { + source: string; + city_count?: number; + interval_sec?: number; + min_interval_sec?: number; + max_interval_sec?: number; + failure_count?: number; + cooldown_count?: number; + status_counts?: Record; + avg_latency_ms?: number | null; + last_success_at?: string | null; + last_failure_at?: string | null; + worst_status?: ObservationCollectorStatus; +}; + +export type ObservationCollectorStatusPayload = { + checked_at?: string; + entries?: ObservationCollectorEntry[]; + sources?: ObservationCollectorSourceSummary[]; + status_counts?: Record; + total_entries?: number; +}; + export type PaymentRuntimePayload = { rpc?: Record | string; chain_id?: number; diff --git a/src/database/runtime_state.py b/src/database/runtime_state.py index 970ade83..f206c3ba 100644 --- a/src/database/runtime_state.py +++ b/src/database/runtime_state.py @@ -6,6 +6,7 @@ import os import sqlite3 import threading import time +from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional @@ -239,9 +240,264 @@ class RuntimeStateDB: conn.execute( "CREATE INDEX IF NOT EXISTS idx_intraday_path_snapshots_city_date ON intraday_path_snapshots_store(city, target_date, id DESC)" ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS observation_collector_status_store ( + source TEXT NOT NULL, + city TEXT NOT NULL, + interval_sec INTEGER NOT NULL, + last_due_ts REAL, + last_started_ts REAL, + last_success_ts REAL, + last_failure_ts REAL, + last_latency_ms REAL, + failure_count INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + updated_at REAL NOT NULL, + payload_json TEXT NOT NULL, + PRIMARY KEY (source, city) + ) + """ + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_observation_collector_status_source ON observation_collector_status_store(source, updated_at DESC)" + ) conn.commit() +def _ts_to_utc_iso(value: Any) -> Optional[str]: + try: + ts = float(value) + except (TypeError, ValueError): + return None + if ts <= 0: + return None + return datetime.fromtimestamp(ts, timezone.utc).isoformat().replace("+00:00", "Z") + + +class ObservationCollectorStatusRepository: + def __init__(self, db: Optional[RuntimeStateDB] = None): + self.db = db or RuntimeStateDB.instance() + + def record_result( + self, + *, + source: str, + city: str, + interval_sec: int, + due_ts: float, + started_ts: float, + completed_ts: float, + ok: bool, + error: Optional[str] = None, + ) -> None: + source_key = str(source or "").strip().lower() + city_key = str(city or "").strip().lower() + if not source_key or not city_key: + return + safe_interval = max(1, int(interval_sec or 60)) + due = float(due_ts or completed_ts) + started = float(started_ts or due) + completed = float(completed_ts or time.time()) + latency_ms = round(max(0.0, completed - started) * 1000.0, 1) + error_text = None if ok else str(error or "no_results").strip()[:500] + payload = { + "source": source_key, + "city": city_key, + "ok": bool(ok), + "error": error_text, + "interval_sec": safe_interval, + "due_ts": due, + "started_ts": started, + "completed_ts": completed, + } + with self.db.connect() as conn: + conn.execute( + """ + INSERT INTO observation_collector_status_store ( + source, city, interval_sec, last_due_ts, last_started_ts, + last_success_ts, last_failure_ts, last_latency_ms, + failure_count, last_error, updated_at, payload_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(source, city) DO UPDATE SET + interval_sec = excluded.interval_sec, + last_due_ts = excluded.last_due_ts, + last_started_ts = excluded.last_started_ts, + last_success_ts = COALESCE(excluded.last_success_ts, observation_collector_status_store.last_success_ts), + last_failure_ts = COALESCE(excluded.last_failure_ts, observation_collector_status_store.last_failure_ts), + last_latency_ms = excluded.last_latency_ms, + failure_count = CASE + WHEN excluded.last_failure_ts IS NOT NULL + THEN observation_collector_status_store.failure_count + 1 + ELSE observation_collector_status_store.failure_count + END, + last_error = excluded.last_error, + updated_at = excluded.updated_at, + payload_json = excluded.payload_json + """, + ( + source_key, + city_key, + safe_interval, + due, + started, + completed if ok else None, + completed if not ok else None, + latency_ms, + 0 if ok else 1, + error_text, + completed, + json.dumps(payload, ensure_ascii=False), + ), + ) + conn.commit() + + def load_snapshot(self, *, now_ts: Optional[float] = None, limit: int = 500) -> Dict[str, Any]: + now = float(time.time() if now_ts is None else now_ts) + safe_limit = max(1, min(int(limit or 500), 1000)) + with self.db.connect() as conn: + rows = conn.execute( + """ + SELECT source, city, interval_sec, last_due_ts, last_started_ts, + last_success_ts, last_failure_ts, last_latency_ms, + failure_count, last_error, updated_at + FROM observation_collector_status_store + ORDER BY source ASC, city ASC + """ + ).fetchall() + + entries: List[Dict[str, Any]] = [] + status_counts: Dict[str, int] = {} + source_summary: Dict[str, Dict[str, Any]] = {} + + for row in rows: + source = str(row["source"] or "") + city = str(row["city"] or "") + interval = max(1, int(row["interval_sec"] or 60)) + last_due_ts = _float_or_none(row["last_due_ts"]) + last_started_ts = _float_or_none(row["last_started_ts"]) + last_success_ts = _float_or_none(row["last_success_ts"]) + last_failure_ts = _float_or_none(row["last_failure_ts"]) + updated_at_ts = _float_or_none(row["updated_at"]) + next_due_ts = last_due_ts + interval if last_due_ts is not None else None + due_in_sec = round(next_due_ts - now, 1) if next_due_ts is not None else None + latest_failure = ( + last_failure_ts is not None + and (last_success_ts is None or last_failure_ts >= last_success_ts) + ) + in_cooldown = bool(latest_failure and next_due_ts is not None and next_due_ts > now) + if last_started_ts is None: + status = "never_run" + elif latest_failure: + status = "cooldown" if in_cooldown else "failed" + elif next_due_ts is not None and next_due_ts <= now: + status = "due" + else: + status = "ok" + + latency = _float_or_none(row["last_latency_ms"]) + failure_count = int(row["failure_count"] or 0) + entry = { + "source": source, + "city": city, + "interval_sec": interval, + "last_due_ts": last_due_ts, + "last_due_at": _ts_to_utc_iso(last_due_ts), + "last_started_ts": last_started_ts, + "last_started_at": _ts_to_utc_iso(last_started_ts), + "last_success_ts": last_success_ts, + "last_success_at": _ts_to_utc_iso(last_success_ts), + "last_failure_ts": last_failure_ts, + "last_failure_at": _ts_to_utc_iso(last_failure_ts), + "last_latency_ms": latency, + "failure_count": failure_count, + "last_error": row["last_error"], + "updated_at_ts": updated_at_ts, + "updated_at": _ts_to_utc_iso(updated_at_ts), + "next_due_ts": next_due_ts, + "next_due_at": _ts_to_utc_iso(next_due_ts), + "due_in_sec": due_in_sec, + "age_sec": round(now - last_success_ts, 1) if last_success_ts is not None else None, + "in_cooldown": in_cooldown, + "cooldown_until_ts": next_due_ts if in_cooldown else None, + "cooldown_until_at": _ts_to_utc_iso(next_due_ts) if in_cooldown else None, + "status": status, + } + entries.append(entry) + status_counts[status] = status_counts.get(status, 0) + 1 + + summary = source_summary.setdefault( + source, + { + "source": source, + "city_count": 0, + "interval_sec": interval, + "min_interval_sec": interval, + "max_interval_sec": interval, + "failure_count": 0, + "cooldown_count": 0, + "status_counts": {}, + "_latencies": [], + "_last_success_ts": None, + "_last_failure_ts": None, + }, + ) + summary["city_count"] += 1 + summary["min_interval_sec"] = min(int(summary["min_interval_sec"]), interval) + summary["max_interval_sec"] = max(int(summary["max_interval_sec"]), interval) + summary["failure_count"] += failure_count + if status == "cooldown": + summary["cooldown_count"] += 1 + summary["status_counts"][status] = summary["status_counts"].get(status, 0) + 1 + if latency is not None: + summary["_latencies"].append(latency) + if last_success_ts is not None: + current_success = summary["_last_success_ts"] + summary["_last_success_ts"] = ( + last_success_ts if current_success is None else max(current_success, last_success_ts) + ) + if last_failure_ts is not None: + current_failure = summary["_last_failure_ts"] + summary["_last_failure_ts"] = ( + last_failure_ts if current_failure is None else max(current_failure, last_failure_ts) + ) + + source_priority = {"failed": 5, "cooldown": 4, "never_run": 3, "due": 2, "ok": 1} + sources: List[Dict[str, Any]] = [] + for summary in source_summary.values(): + latencies = summary.pop("_latencies") + last_success_ts = summary.pop("_last_success_ts") + last_failure_ts = summary.pop("_last_failure_ts") + summary["avg_latency_ms"] = ( + round(sum(latencies) / len(latencies), 1) if latencies else None + ) + summary["last_success_at"] = _ts_to_utc_iso(last_success_ts) + summary["last_failure_at"] = _ts_to_utc_iso(last_failure_ts) + summary["worst_status"] = max( + summary["status_counts"], + key=lambda key: source_priority.get(str(key), 0), + default="unknown", + ) + sources.append(summary) + + return { + "checked_at": _ts_to_utc_iso(now), + "entries": entries[:safe_limit], + "sources": sorted(sources, key=lambda item: str(item.get("source") or "")), + "status_counts": status_counts, + "total_entries": len(entries), + } + + +def _float_or_none(value: Any) -> Optional[float]: + try: + if value is None or value == "": + return None + return float(value) + except (TypeError, ValueError): + return None + + class DailyRecordRepository: def __init__(self, db: Optional[RuntimeStateDB] = None): self.db = db or RuntimeStateDB.instance() diff --git a/tests/test_observation_collector.py b/tests/test_observation_collector.py index 3e3e72cb..0eb94a47 100644 --- a/tests/test_observation_collector.py +++ b/tests/test_observation_collector.py @@ -133,6 +133,132 @@ def test_observation_collector_run_due_once_refreshes_panel_cache(): assert refreshed == ["qingdao", "qingdao"] +def test_observation_collector_records_source_status_to_runtime_state(tmp_path): + from src.database.runtime_state import ObservationCollectorStatusRepository, RuntimeStateDB + from web.observation_collector_service import ( + ObservationCollector, + ObservationSourceProfile, + ) + + class FakeWeather: + def _uses_fahrenheit(self, city): + return False + + def _attach_china_amsc_awos_data(self, results, city, use_fahrenheit): + results["amsc_awos"] = {"source": "amsc_awos", "temp_c": 22.8} + + db = RuntimeStateDB(str(tmp_path / "polyweather.db")) + status_repo = ObservationCollectorStatusRepository(db) + collector = ObservationCollector( + weather=FakeWeather(), + profiles=[ObservationSourceProfile("amsc_awos", ("beijing",), 180)], + status_recorder=status_repo, + ) + + assert collector.run_due_once(now_ts=1000.0) == 1 + + payload = status_repo.load_snapshot(now_ts=1001.0) + assert payload["total_entries"] == 1 + assert payload["status_counts"] == {"ok": 1} + + entry = payload["entries"][0] + assert entry["source"] == "amsc_awos" + assert entry["city"] == "beijing" + assert entry["interval_sec"] == 180 + assert entry["failure_count"] == 0 + assert entry["last_error"] is None + assert entry["last_success_at"] is not None + assert entry["last_failure_at"] is None + assert entry["last_latency_ms"] is not None + assert entry["next_due_ts"] == 1180.0 + assert entry["in_cooldown"] is False + assert entry["status"] == "ok" + + source = payload["sources"][0] + assert source["source"] == "amsc_awos" + assert source["city_count"] == 1 + assert source["failure_count"] == 0 + assert source["avg_latency_ms"] is not None + assert source["status_counts"] == {"ok": 1} + + +def test_observation_collector_records_failure_and_cooldown(tmp_path): + from src.database.runtime_state import ObservationCollectorStatusRepository, RuntimeStateDB + from web.observation_collector_service import ( + ObservationCollector, + ObservationSourceProfile, + ) + + class FakeWeather: + def _uses_fahrenheit(self, city): + return False + + def _attach_china_amsc_awos_data(self, results, city, use_fahrenheit): + raise RuntimeError("upstream timeout") + + db = RuntimeStateDB(str(tmp_path / "polyweather.db")) + status_repo = ObservationCollectorStatusRepository(db) + collector = ObservationCollector( + weather=FakeWeather(), + profiles=[ObservationSourceProfile("amsc_awos", ("seoul",), 180)], + status_recorder=status_repo, + ) + + assert collector.run_due_once(now_ts=2000.0) == 0 + + payload = status_repo.load_snapshot(now_ts=2010.0) + assert payload["total_entries"] == 1 + assert payload["status_counts"] == {"cooldown": 1} + + entry = payload["entries"][0] + assert entry["source"] == "amsc_awos" + assert entry["city"] == "seoul" + assert entry["failure_count"] == 1 + assert entry["last_success_at"] is None + assert entry["last_failure_at"] is not None + assert entry["last_error"] == "upstream timeout" + assert entry["next_due_ts"] == 2180.0 + assert entry["in_cooldown"] is True + assert entry["status"] == "cooldown" + + source = payload["sources"][0] + assert source["failure_count"] == 1 + assert source["cooldown_count"] == 1 + + +def test_ops_observation_collector_status_returns_runtime_snapshot(monkeypatch, tmp_path): + from src.database.runtime_state import ObservationCollectorStatusRepository, RuntimeStateDB + from web.services import ops_api + + db = RuntimeStateDB(str(tmp_path / "polyweather.db")) + status_repo = ObservationCollectorStatusRepository(db) + now = time.time() + status_repo.record_result( + source="cowin_obs", + city="hong kong", + interval_sec=60, + due_ts=now, + started_ts=now, + completed_ts=now + 0.25, + ok=True, + ) + + monkeypatch.setattr( + ops_api.legacy_routes, + "_require_ops_admin", + lambda request: {"email": "ops@example.com"}, + ) + monkeypatch.setattr(ops_api, "ObservationCollectorStatusRepository", lambda: status_repo) + + payload = ops_api.get_ops_observation_collector_status(object(), limit=10) + + assert payload["total_entries"] == 1 + assert payload["entries"][0]["source"] == "cowin_obs" + assert payload["entries"][0]["city"] == "hong kong" + assert payload["sources"][0]["source"] == "cowin_obs" + assert payload["status_counts"] == {"ok": 1} + + def test_observation_collector_worker_entrypoint_exists(): from web import observation_collector_worker diff --git a/web/observation_collector_service.py b/web/observation_collector_service.py index 357e18a7..dc0ffa6b 100644 --- a/web/observation_collector_service.py +++ b/web/observation_collector_service.py @@ -14,6 +14,7 @@ from src.data_collection.amos_station_sources import AMOS_AIRPORT_CODES from src.data_collection.amsc_awos_sources import AMSC_AWOS_AIRPORTS from src.data_collection.city_registry import CITY_REGISTRY from src.data_collection.hko_obs_sources import HKO_STATIONS +from src.database.runtime_state import ObservationCollectorStatusRepository def _env_bool(name: str, default: bool) -> bool: @@ -51,10 +52,12 @@ class ObservationCollector: weather: Any, profiles: Sequence[ObservationSourceProfile], cache_refresher: Optional[Callable[[str], Any]] = None, + status_recorder: Optional[ObservationCollectorStatusRepository] = None, ) -> None: self.weather = weather self.profiles = list(profiles) self.cache_refresher = cache_refresher + self.status_recorder = status_recorder self._last_run_ts: dict[tuple[str, str], float] = {} self._lock = threading.Lock() @@ -73,19 +76,70 @@ class ObservationCollector: completed = 0 for profile, city in due: + started_wall = time.time() + started_ts = now if now_ts is not None else started_wall + ok = False + error: Optional[str] = None try: - if self._collect_city_source(profile.source, city): + ok = self._collect_city_source(profile.source, city) + if ok: completed += 1 self._refresh_city_cache(city) + else: + error = "no_results" except Exception as exc: + error = str(exc) or exc.__class__.__name__ logger.warning( "observation collector source failed source={} city={}: {}", profile.source, city, exc, ) + finally: + completed_ts = started_ts + max(0.0, time.time() - started_wall) + self._record_source_status( + profile=profile, + city=city, + due_ts=now, + started_ts=started_ts, + completed_ts=completed_ts, + ok=ok, + error=error, + ) return completed + def _record_source_status( + self, + *, + profile: ObservationSourceProfile, + city: str, + due_ts: float, + started_ts: float, + completed_ts: float, + ok: bool, + error: Optional[str], + ) -> None: + if not self.status_recorder: + return + try: + self.status_recorder.record_result( + source=profile.source, + city=city, + interval_sec=profile.interval_sec, + due_ts=due_ts, + started_ts=started_ts, + completed_ts=completed_ts, + ok=ok, + error=error, + ) + except Exception as exc: + logger.warning( + "observation collector status write failed source={} city={}: {}", + profile.source, + city, + exc, + ) + def _collect_city_source(self, source: str, city: str) -> bool: normalized_source = str(source or "").strip().lower() normalized_city = str(city or "").strip().lower() @@ -162,6 +216,7 @@ def start_observation_collector_loop( weather: Any, cache_refresher: Optional[Callable[[str], Any]] = None, profiles: Optional[Sequence[ObservationSourceProfile]] = None, + status_recorder: Optional[ObservationCollectorStatusRepository] = None, ) -> Optional[threading.Thread]: if not _env_bool("POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED", True): return None @@ -175,6 +230,7 @@ def start_observation_collector_loop( weather=weather, profiles=selected_profiles, cache_refresher=cache_refresher, + status_recorder=status_recorder or ObservationCollectorStatusRepository(), ) global _COLLECTOR_THREAD diff --git a/web/routers/ops.py b/web/routers/ops.py index 9040c332..7f32a192 100644 --- a/web/routers/ops.py +++ b/web/routers/ops.py @@ -13,6 +13,7 @@ from web.services.ops_api import ( get_ops_memberships_overview, get_ops_health_check, get_ops_logs, + get_ops_observation_collector_status, get_ops_source_health, get_ops_truth_history, get_ops_weekly_leaderboard, @@ -266,6 +267,14 @@ async def ops_source_health( return get_ops_source_health(request, cities=cities, limit=limit) +@router.get("/api/ops/observation-collector-status") +async def ops_observation_collector_status( + request: Request, + limit: int = 200, +): + return get_ops_observation_collector_status(request, limit=limit) + + @router.get("/api/ops/training/accuracy") async def ops_training_accuracy(request: Request): return get_ops_training_accuracy(request) diff --git a/web/services/ops_api.py b/web/services/ops_api.py index e54e5e27..97e1d226 100644 --- a/web/services/ops_api.py +++ b/web/services/ops_api.py @@ -10,6 +10,7 @@ from fastapi import HTTPException, Request import requests as _requests from src.database.db_manager import DBManager +from src.database.runtime_state import ObservationCollectorStatusRepository from src.utils.runtime_secrets import get_runtime_secret, get_runtime_secret_status from web.services.observation_freshness import ( build_observation_freshness, @@ -1960,6 +1961,15 @@ def get_ops_source_health( } +def get_ops_observation_collector_status( + request: Request, + limit: int = 200, +) -> dict[str, Any]: + _require_ops(request) + safe_limit = max(1, min(int(limit or 200), 500)) + return ObservationCollectorStatusRepository().load_snapshot(limit=safe_limit) + + def get_ops_health_check(request: Request) -> dict[str, Any]: _require_ops(request) import os