"""c_deploy_supervisor_monitor (03-C §3.15 / 04-P §4.6 / design doc §6): post-cutover monitoring -- four signals sampled periodically while `deployment.status='monitoring'`, called repeatedly by Deploy Supervisor's own poll loop (U-036's `poll_fn`), not a one-shot call (03-C's own text: "由 Deploy Supervisor 自身的调度循环周期性调用"). `run_monitor_cycle()` is ONE sampling pass: process_alive + http_health + validator_approval_rate every cycle, canary_task only when the caller says `run_canary=True` (real scheduling of "every CANARY_INTERVAL" is the caller's job -- this module just runs whatever it's told to run once, matching `c_deploy_supervisor_health_check`'s own "idempotent, doesn't itself decide cadence" framing). **`validator_approval_rate` operationalization (documented interpretation, not literally spelled out anywhere in 03-C/04-P/design doc -- there's no `event_type='validator_approved'` or similar in 01-M's execution_event enum, so "approved 占比" has to be derived from what IS actually persisted)**: a task counts as Validator-approved if it reached `task_status_changed( to_status='validating')` and NO `bug_record(source='validator')` was ever created for it afterward (Validator's real judgment: PASS -> straight to `close_task_done`, no bug_record at all; FAIL/INCONCLUSIVE -> `org_level_ loop.run_fix_loop` always creates one via `root_cause_trace.judge_root_cause` -> `bug_record.create_bug_record`). This is a real, always-true structural fact about how this codebase's Validator path already works (scheduler.py's `finish_task_lifecycle`), not a guess -- verified against that code before writing this, not merely assumed. **Crash-count / consecutive-fail windows are approximated by sample COUNT, not wall-clock TIME (documented simplification)**: `MAX_CRASH_COUNT`/ `CRASH_WINDOW` and `MAX_CONSECUTIVE_HTTP_FAIL` are defined in 05 §5.9 as count-within-a-time-window, but this module counts among the most recent N persisted `deploy_health_signal` rows (N = CRASH_WINDOW / MONITOR_SAMPLE_INTERVAL) instead of re-deriving a true rolling time window per sample -- correct as long as sampling stays roughly interval-paced (the whole point of a periodic monitor loop); revisit only if sampling cadence ever becomes genuinely irregular. """ import json import time import uuid from datetime import datetime, timezone from supervisor import deploy_health_check, deploy_supervisor MONITOR_SAMPLE_INTERVAL = 5 * 60 CANARY_INTERVAL = 60 * 60 MAX_CRASH_COUNT = 2 CRASH_WINDOW = 10 * 60 MAX_CONSECUTIVE_HTTP_FAIL = 3 VALIDATOR_RATE_DEVIATION_THRESHOLD = 0.30 # fraction; 05 §5.9's "±30 个百分点" MONITORING_WINDOW_HOURS = 24 MONITORING_MIN_TASK_COUNT = 10 def _parse_sqlite_timestamp(text): """Same convention as work_node_run.py's own helper of the same name -- duplicated (not imported) since it's a 2-line stdlib parse and this module has no other reason to depend on work_node_run.py.""" return datetime.strptime(text, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc).timestamp() def _format_sqlite_timestamp(epoch_seconds): return datetime.fromtimestamp(epoch_seconds, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S") def compute_validator_approval_rate(conn, project_id, since_text, until_text=None): """Returns (rate: float in [0,1], sample_size: int), or None if no task reached 'validating' in the window (not enough data to mean anything).""" query = ( "SELECT task_id, payload FROM execution_event " "WHERE project_id = ? AND event_type = 'task_status_changed' AND created_at >= ?" ) params = [project_id, since_text] if until_text is not None: query += " AND created_at < ?" params.append(until_text) rows = conn.execute(query, params).fetchall() task_ids = set() for task_id, payload_json in rows: payload = json.loads(payload_json) if payload.get("to_status") == "validating": task_ids.add(task_id) if not task_ids: return None approved = sum( 1 for task_id in task_ids if conn.execute( "SELECT 1 FROM bug_record WHERE task_id = ? AND source = 'validator' LIMIT 1", (task_id,) ).fetchone() is None ) return approved / len(task_ids), len(task_ids) def compute_baseline_validator_approval_rate(conn, project_id, before_text, lookback_hours=MONITORING_WINDOW_HOURS): """Historical rate over the `lookback_hours` immediately preceding `before_text` (typically `cutover_at`) -- "与该 deployment 之前的历史基线 比较" (design doc §6), computed from the same persisted data rather than requiring a separately-stored baseline field. Returns None (same as compute_validator_approval_rate) if there's no prior data -- callers should treat a None baseline as "can't judge deviation yet", not as 0.""" since_text = _format_sqlite_timestamp(_parse_sqlite_timestamp(before_text) - lookback_hours * 3600) return compute_validator_approval_rate(conn, project_id, since_text, until_text=before_text) def _record_signal(conn, deployment_id, signal): conn.execute( "INSERT INTO deploy_health_signal (signal_id, deployment_id, sampled_at, signal_type, status, detail) " "VALUES (?, ?, CURRENT_TIMESTAMP, ?, ?, ?)", (str(uuid.uuid4()), deployment_id, signal["signal_type"], signal["status"], signal["detail"]), ) conn.commit() def _recent_statuses(conn, deployment_id, signal_type, limit): rows = conn.execute( "SELECT status FROM deploy_health_signal WHERE deployment_id = ? AND signal_type = ? " "ORDER BY rowid DESC LIMIT ?", (deployment_id, signal_type, limit), ).fetchall() return [r[0] for r in rows] def _sample_process_alive(pid): alive = not deploy_supervisor.reap_if_exited(pid) return {"signal_type": "process_alive", "status": "ok" if alive else "fail", "detail": None} def _sample_validator_approval_rate(conn, project_id, baseline_rate, window_start_text, threshold): if baseline_rate is None: return { "signal_type": "validator_approval_rate", "status": "ok", "detail": "no baseline yet -- insufficient history to judge deviation", } result = compute_validator_approval_rate(conn, project_id, window_start_text) if result is None: return { "signal_type": "validator_approval_rate", "status": "ok", "detail": "no validated tasks in window yet", } rate, sample_size = result deviation = abs(rate - baseline_rate) status = "fail" if deviation > threshold else "ok" return { "signal_type": "validator_approval_rate", "status": status, "detail": "rate={:.2f} n={} baseline={:.2f} deviation={:.2f}".format(rate, sample_size, baseline_rate, deviation), } def _real_task_count_since(conn, project_id, since_text): return conn.execute( "SELECT COUNT(*) FROM task WHERE project_id = ? AND created_at >= ?", (project_id, since_text) ).fetchone()[0] def run_monitor_cycle( conn, deployment_id, pid, port, project_id, baseline_validator_rate, window_start_text, run_canary=True, crash_window_samples=None, consecutive_http_fail_limit=MAX_CONSECUTIVE_HTTP_FAIL, validator_threshold=VALIDATOR_RATE_DEVIATION_THRESHOLD, min_task_count=MONITORING_MIN_TASK_COUNT, window_hours=MONITORING_WINDOW_HOURS, now=time.time, ): """One sampling pass. Returns {status, latest_signal}. `window_start_text` is `deployment.cutover_at` (a sqlite TEXT timestamp) -- used both for the elapsed-time-since-cutover math and as the SQL lower bound for "real tasks since the window opened"/"validator judgments since the window opened". Stops sampling further signals the moment one comes back fail (same fail-closed, don't-pile-on-more-checks pattern as `c_deploy_supervisor_health_check`) and transitions `deployment.status='anomaly_detected'` immediately -- 04-P's own state machine has no intermediate "still checking" state distinct from 'monitoring' itself. """ if crash_window_samples is None: crash_window_samples = max(1, CRASH_WINDOW // MONITOR_SAMPLE_INTERVAL) process_signal = _sample_process_alive(pid) _record_signal(conn, deployment_id, process_signal) latest_signal = process_signal anomaly = _recent_statuses(conn, deployment_id, "process_alive", crash_window_samples).count("fail") >= MAX_CRASH_COUNT if not anomaly: http_signal = deploy_health_check.check_http_health(port) _record_signal(conn, deployment_id, http_signal) latest_signal = http_signal recent_http = _recent_statuses(conn, deployment_id, "http_health", consecutive_http_fail_limit) anomaly = len(recent_http) >= consecutive_http_fail_limit and all(s == "fail" for s in recent_http) if not anomaly and run_canary: canary_signal = deploy_health_check.check_canary_task(port) _record_signal(conn, deployment_id, canary_signal) latest_signal = canary_signal anomaly = canary_signal["status"] == "fail" # single failure -- no grey area (design doc §6) if not anomaly: rate_signal = _sample_validator_approval_rate( conn, project_id, baseline_validator_rate, window_start_text, validator_threshold ) _record_signal(conn, deployment_id, rate_signal) latest_signal = rate_signal anomaly = rate_signal["status"] == "fail" if anomaly: conn.execute( "UPDATE deployment SET status = 'anomaly_detected', rollback_reason = ? WHERE deployment_id = ?", (json.dumps(latest_signal), deployment_id), ) conn.commit() return {"status": "anomaly_detected", "latest_signal": latest_signal} elapsed_hours = (now() - _parse_sqlite_timestamp(window_start_text)) / 3600.0 real_task_count = _real_task_count_since(conn, project_id, window_start_text) if elapsed_hours >= window_hours or real_task_count >= min_task_count: conn.execute( "UPDATE deployment SET status = 'monitoring_window_clean', monitoring_closed_at = CURRENT_TIMESTAMP " "WHERE deployment_id = ?", (deployment_id,), ) conn.commit() return {"status": "monitoring_window_clean", "latest_signal": latest_signal} conn.execute("UPDATE deployment SET status = 'monitoring' WHERE deployment_id = ?", (deployment_id,)) conn.commit() return {"status": "monitoring", "latest_signal": latest_signal}