import json import os import socket import sqlite3 import threading import time from collections import Counter, defaultdict from contextlib import contextmanager from datetime import datetime, timedelta from typing import Any, Dict, Iterable, List, Optional, Tuple from zoneinfo import ZoneInfo from config import ( MANAGED_WORKER_IDS, MONITORING_DB_PATH, MONITORING_TIMEZONE, MONITORING_TIMELINE_BUCKET_MINUTES, normalize_worker_id, ) from log_manager import logger from result_codes import ( APP_ERROR_DESC, BUSINESS_ERROR_DESC, DOWNLOAD_ERROR_DESC, INFRA_ERROR_DESC, AppError, BusinessError, DownloadError, InfraError, ) SHANGHAI_TZ = ZoneInfo(MONITORING_TIMEZONE) RUNNING_STATES = {"running_download", "running_collect"} ONLINE_STATES = {"idle_waiting_task", "running_download", "running_collect", "disabled"} DISPLAY_STATE_LABELS = { "idle_waiting_task": "idle", "running_download": "download", "running_collect": "collect", "disabled": "disabled", "offline": "offline", } RUNTIME_STATE_ORDER = [ "running_download", "running_collect", "idle_waiting_task", "disabled", "offline", ] RUNTIME_STATE_LABELS = { "running_download": "下载中", "running_collect": "采集中", "idle_waiting_task": "空闲等待", "disabled": "已禁用", "offline": "离线", } STAGE_LABELS = { "download": "下载", "collect": "采集", } STAGE_ORDER = ["download", "collect"] FAILURE_DOMAIN_LABELS = { "download_network": "下载网络", "download_other": "下载其他", "infra": "基础设施", "app": "应用问题", "business": "业务限制", "other": "其他", } DOWNLOAD_NETWORK_ERROR_CODES = { int(DownloadError.DOWNLOAD_TIMEOUT), int(DownloadError.NETWORK_ERROR), } DOWNLOAD_NETWORK_MESSAGE_HINTS = ( "timeout", "overtime", "jump failed", "app page load failed", "page load failed", "network error", "connection error", "连接超时", "网络错误", "超时", ) def _safe_json_dumps(payload: Dict[str, Any]) -> str: return json.dumps(payload, ensure_ascii=False, sort_keys=True) def _safe_json_loads(payload: Optional[str], fallback: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: if not payload: return fallback or {} try: value = json.loads(payload) return value if isinstance(value, dict) else (fallback or {}) except (TypeError, ValueError): return fallback or {} def _date_str_from_ts(ts: float) -> str: return datetime.fromtimestamp(ts, SHANGHAI_TZ).strftime("%Y-%m-%d") def _day_bounds(date_str: str, now_ts: Optional[float] = None) -> Tuple[float, float, bool]: local_day = datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=SHANGHAI_TZ) day_start = local_day.timestamp() day_end = (local_day + timedelta(days=1)).timestamp() now_ts = now_ts or time.time() is_today = _date_str_from_ts(now_ts) == date_str return day_start, min(day_end, now_ts) if is_today else day_end, is_today def _interval_overlap(start: float, end: float, bucket_start: float, bucket_end: float) -> bool: return max(start, bucket_start) < min(end, bucket_end) def _build_histogram(values: List[float], edges: List[float]) -> List[Dict[str, Any]]: buckets: List[Dict[str, Any]] = [] previous = 0.0 for edge in edges: buckets.append({"label": _format_bucket_label(previous, edge), "start": previous, "end": edge, "count": 0}) previous = edge buckets.append({"label": _format_bucket_label(previous, None), "start": previous, "end": None, "count": 0}) for value in values: if value is None: continue placed = False for bucket in buckets: bucket_end = bucket["end"] if bucket_end is None: bucket["count"] += 1 placed = True break if bucket["start"] <= value < bucket_end: bucket["count"] += 1 placed = True break if not placed and buckets: buckets[-1]["count"] += 1 return buckets def _percentile(values: List[float], ratio: float) -> Optional[float]: if not values: return None ordered = sorted(values) index = int(round((len(ordered) - 1) * ratio)) return round(ordered[index], 2) def _format_minutes(seconds: Optional[float]) -> str: seconds = float(seconds or 0.0) minutes = seconds / 60.0 if minutes >= 10: return f"{round(minutes):.0f}m" if minutes >= 1: return f"{minutes:.1f}m" if seconds <= 0: return "0m" return f"{minutes:.1f}m" def _format_bucket_label(start_seconds: float, end_seconds: Optional[float]) -> str: if end_seconds is None: return f"{_format_minutes(start_seconds)}+" return f"{_format_minutes(start_seconds)}-{_format_minutes(end_seconds)}" def _humanize_failure_reason(error_type: Optional[str], error_message: Optional[str]) -> str: if error_message: return str(error_message) if not error_type: return "未知原因" try: category_name, code_str = str(error_type).split("/", 1) code = int(code_str) except (ValueError, TypeError): return str(error_type) mapping = { "INFRA_ERROR": INFRA_ERROR_DESC, "DOWNLOAD_ERROR": DOWNLOAD_ERROR_DESC, "APP_ERROR": APP_ERROR_DESC, "BUSINESS_ERROR": BUSINESS_ERROR_DESC, }.get(category_name) if not mapping: return str(error_type) return mapping.get(code, str(error_type)) def _task_row_total_duration(row: sqlite3.Row) -> float: return float(row["total_duration_seconds"] or row["failed_total_duration_seconds"] or 0.0) def _build_failure_task_sample(row: sqlite3.Row) -> Dict[str, Any]: task_started_at = float(row["task_started_at"]) if row["task_started_at"] is not None else None task_ended_at = float(row["task_ended_at"]) if row["task_ended_at"] is not None else None failure_domain = row["failure_domain"] failure_subtype = row["failure_subtype"] diagnostics = _extract_task_diagnostics(row) if (row["status"] or "unknown") != "success" and (not failure_domain or not failure_subtype): failure_domain, failure_subtype = _classify_failure(row["error_type"], row["error_message"]) return { "execution_id": row["execution_id"], "task_key": row["task_key"], "app_name": row["app_name"] or "--", "package_name": row["package_name"] or "--", "worker_id": row["worker_id"], "status": row["status"] or "unknown", "failure_domain": failure_domain or "", "failure_domain_label": _humanize_failure_domain(failure_domain), "failure_subtype": failure_subtype or "", "failure_subtype_label": _humanize_failure_subtype( failure_domain, failure_subtype, row["error_type"], row["error_message"], ), "failure_reason": _humanize_failure_reason(row["error_type"], row["error_message"]), "error_type": row["error_type"] or "", "error_message": row["error_message"] or "", "failed_stage": row["failed_stage"] or "--", "download_duration_seconds": row["download_duration_seconds"] or 0.0, "collect_duration_seconds": row["collect_duration_seconds"] or 0.0, "total_duration_seconds": row["total_duration_seconds"] or 0.0, "failed_total_duration_seconds": row["failed_total_duration_seconds"] or 0.0, "diagnostics": diagnostics, "diagnostics_note": _format_diagnostics_note(diagnostics, int(row["retry_count"] or 0)), "task_started_at": datetime.fromtimestamp(task_started_at, SHANGHAI_TZ).strftime("%Y-%m-%d %H:%M:%S") if task_started_at else "--", "task_ended_at": datetime.fromtimestamp(task_ended_at, SHANGHAI_TZ).strftime("%Y-%m-%d %H:%M:%S") if task_ended_at else "--", } def _parse_error_type(error_type: Optional[str]) -> Tuple[Optional[str], Optional[int]]: if not error_type: return None, None try: category_name, code_str = str(error_type).split("/", 1) return category_name, int(code_str) except (TypeError, ValueError): return str(error_type), None def _download_error_message_is_network(message: Optional[str]) -> bool: if not message: return False lowered = str(message).lower() return any(token in lowered for token in DOWNLOAD_NETWORK_MESSAGE_HINTS) def _normalize_download_errors(download_errors: Any) -> Dict[str, Dict[str, Any]]: if isinstance(download_errors, dict): return download_errors return {} def _all_download_errors_are_network(download_errors: Any) -> bool: errors = _normalize_download_errors(download_errors) if not errors: return False for detail in errors.values(): if not isinstance(detail, dict): return False code = detail.get("code") try: if code is not None and int(code) in DOWNLOAD_NETWORK_ERROR_CODES: continue except (TypeError, ValueError): pass if _download_error_message_is_network(detail.get("message")): continue return False return True def _error_code_name(category_name: Optional[str], code: Optional[int]) -> str: if category_name == "DOWNLOAD_ERROR" and code is not None: try: return DownloadError(int(code)).name except ValueError: return f"DOWNLOAD_{code}" if category_name == "INFRA_ERROR" and code is not None: try: return InfraError(int(code)).name except ValueError: return f"INFRA_{code}" if category_name == "APP_ERROR" and code is not None: try: return AppError(int(code)).name except ValueError: return f"APP_{code}" if category_name == "BUSINESS_ERROR" and code is not None: try: return BusinessError(int(code)).name except ValueError: return f"BUSINESS_{code}" if category_name: return str(category_name) return "OTHER" def _classify_failure( error_type: Optional[str], error_message: Optional[str], download_errors: Any = None, ) -> Tuple[str, str]: category_name, code = _parse_error_type(error_type) subtype = _error_code_name(category_name, code) if category_name == "DOWNLOAD_ERROR": if code in DOWNLOAD_NETWORK_ERROR_CODES: return "download_network", subtype if code == int(DownloadError.ALL_SOURCES_FAILED) and _all_download_errors_are_network(download_errors): return "download_network", "ALL_SOURCES_NETWORK_FAILED" if _download_error_message_is_network(error_message): return "download_network", subtype return "download_other", subtype if category_name == "INFRA_ERROR": return "infra", subtype if category_name == "APP_ERROR": return "app", subtype if category_name == "BUSINESS_ERROR": return "business", subtype return "other", subtype def _humanize_failure_domain(domain: Optional[str]) -> str: return FAILURE_DOMAIN_LABELS.get(domain or "", "其他") def _humanize_failure_subtype( domain: Optional[str], subtype: Optional[str], error_type: Optional[str], error_message: Optional[str], ) -> str: if domain == "download_network" and subtype == "ALL_SOURCES_NETWORK_FAILED": return "所有下载源网络失败" reason = _humanize_failure_reason(error_type, None) if reason and reason != str(error_type): return reason if error_message and not subtype: return str(error_message) if subtype: return str(subtype) return "未知原因" def _extract_task_diagnostics(row: sqlite3.Row) -> Dict[str, Any]: trace_json = row["trace_json"] if "trace_json" in row.keys() else None trace = _safe_json_loads(trace_json) diagnostics = trace.get("diagnostics") return diagnostics if isinstance(diagnostics, dict) else {} def _format_diagnostics_note(diagnostics: Dict[str, Any], retry_count: int) -> str: labels = [] if retry_count > 0: labels.append(f"重试{retry_count}次") if diagnostics.get("emulator_restarted"): labels.append("模拟器重启") if diagnostics.get("airtest_reinitialized"): labels.append("Airtest重连") if diagnostics.get("download_recovered"): labels.append("下载恢复") if diagnostics.get("collect_recovered"): labels.append("采集恢复") return " / ".join(labels) if labels else "--" class MonitoringRepository: def __init__(self, db_path: str = MONITORING_DB_PATH, initialize_schema: bool = True): self.db_path = db_path self._write_lock = threading.Lock() if initialize_schema: self._initialize_schema() @contextmanager def _connect(self): """每次操作使用独立连接，避免跨线程共享连接导致游标状态串扰。""" connection = sqlite3.connect(self.db_path, timeout=30) connection.row_factory = sqlite3.Row connection.execute("PRAGMA busy_timeout=30000") connection.execute("PRAGMA synchronous=NORMAL") connection.execute("PRAGMA wal_autocheckpoint=1000") connection.execute("PRAGMA cache_size=-8192") connection.execute("PRAGMA temp_store=MEMORY") try: yield connection connection.commit() except Exception: try: connection.rollback() except Exception: pass raise finally: connection.close() def _table_exists(self, connection: sqlite3.Connection, table_name: str) -> bool: row = connection.execute( """ SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ? """, (table_name,), ).fetchone() return bool(row) def close(self): """保留接口兼容；独立连接模式下无需额外关闭。""" return def _initialize_schema(self): os.makedirs(os.path.dirname(self.db_path) or ".", exist_ok=True) with self._connect() as connection: connection.execute("PRAGMA journal_mode=WAL") connection.executescript( """ CREATE TABLE IF NOT EXISTS controller_session ( id INTEGER PRIMARY KEY AUTOINCREMENT, started_at REAL NOT NULL, ended_at REAL, host TEXT, version TEXT, stop_reason TEXT, last_heartbeat_at REAL ); CREATE TABLE IF NOT EXISTS worker_state_event ( id INTEGER PRIMARY KEY AUTOINCREMENT, worker_id TEXT NOT NULL, session_id INTEGER, event_time REAL NOT NULL, state TEXT NOT NULL, reason TEXT, task_key TEXT ); CREATE INDEX IF NOT EXISTS idx_worker_state_event_worker_time ON worker_state_event(worker_id, event_time); CREATE TABLE IF NOT EXISTS task_stage_event ( id INTEGER PRIMARY KEY AUTOINCREMENT, execution_id TEXT NOT NULL, task_key TEXT NOT NULL, worker_id TEXT NOT NULL, session_id INTEGER, event_time REAL NOT NULL, event_type TEXT NOT NULL, stage TEXT, status TEXT, reason TEXT, metrics_json TEXT ); CREATE INDEX IF NOT EXISTS idx_task_stage_event_exec_time ON task_stage_event(execution_id, event_time); CREATE TABLE IF NOT EXISTS task_execution ( execution_id TEXT PRIMARY KEY, task_key TEXT NOT NULL, app_name TEXT, package_name TEXT, last_updated TEXT, collection_task_type TEXT, worker_id TEXT NOT NULL, session_id INTEGER, stat_date TEXT NOT NULL, task_started_at REAL, task_ended_at REAL, status TEXT, failed_stage TEXT, failure_bucket TEXT, retry_count INTEGER DEFAULT 0, download_duration_seconds REAL, collect_duration_seconds REAL, failed_total_duration_seconds REAL, total_duration_seconds REAL, download_source TEXT, error_type TEXT, error_message TEXT, result_detail TEXT, droidbot_steps INTEGER, guiagent_steps INTEGER, total_steps INTEGER, num_nodes INTEGER, num_reached_activities INTEGER, app_num_total_activities INTEGER, failure_domain TEXT, failure_subtype TEXT, trace_json TEXT, last_updated_at REAL ); CREATE INDEX IF NOT EXISTS idx_task_execution_stat_date ON task_execution(stat_date, worker_id, task_started_at); """ ) columns = { row["name"] for row in connection.execute("PRAGMA table_info(task_execution)").fetchall() } if "failure_bucket" not in columns: connection.execute("ALTER TABLE task_execution ADD COLUMN failure_bucket TEXT") if "failure_domain" not in columns: connection.execute("ALTER TABLE task_execution ADD COLUMN failure_domain TEXT") if "failure_subtype" not in columns: connection.execute("ALTER TABLE task_execution ADD COLUMN failure_subtype TEXT") if "trace_json" not in columns: connection.execute("ALTER TABLE task_execution ADD COLUMN trace_json TEXT") if "result_detail" not in columns: connection.execute("ALTER TABLE task_execution ADD COLUMN result_detail TEXT") if "droidbot_steps" not in columns: connection.execute("ALTER TABLE task_execution ADD COLUMN droidbot_steps INTEGER") if "guiagent_steps" not in columns: connection.execute("ALTER TABLE task_execution ADD COLUMN guiagent_steps INTEGER") if "total_steps" not in columns: connection.execute("ALTER TABLE task_execution ADD COLUMN total_steps INTEGER") if "num_nodes" not in columns: connection.execute("ALTER TABLE task_execution ADD COLUMN num_nodes INTEGER") if "num_reached_activities" not in columns: connection.execute("ALTER TABLE task_execution ADD COLUMN num_reached_activities INTEGER") if "app_num_total_activities" not in columns: connection.execute("ALTER TABLE task_execution ADD COLUMN app_num_total_activities INTEGER") if "last_updated" not in columns: connection.execute("ALTER TABLE task_execution ADD COLUMN last_updated TEXT") if "collection_task_type" not in columns: connection.execute("ALTER TABLE task_execution ADD COLUMN collection_task_type TEXT") # ── APK 注册表（由 apk_cloud.registry.ApkRegistry 使用）── connection.executescript( """ CREATE TABLE IF NOT EXISTS apk_registry ( package_name TEXT PRIMARY KEY, download_date TEXT NOT NULL, download_time REAL NOT NULL, local_dir TEXT, smb_dir TEXT, version_name TEXT, source TEXT, apk_files_json TEXT, updated_at REAL NOT NULL ); CREATE TABLE IF NOT EXISTS apk_pending_download ( package_name TEXT PRIMARY KEY, created_at REAL NOT NULL ); CREATE TABLE IF NOT EXISTS app_download_record ( id INTEGER PRIMARY KEY AUTOINCREMENT, package_name TEXT NOT NULL, country_code TEXT NOT NULL DEFAULT '', app_name TEXT DEFAULT '', last_updated TEXT NOT NULL DEFAULT '', downloads INTEGER DEFAULT 0, worker_play_status TEXT NOT NULL DEFAULT 'pending', worker_play_error TEXT, worker_play_date TEXT, us_play_status TEXT NOT NULL DEFAULT 'pending', us_play_error TEXT, us_play_date TEXT, us_aurora_status TEXT NOT NULL DEFAULT 'pending', us_aurora_error TEXT, us_aurora_date TEXT, overall_status TEXT NOT NULL DEFAULT 'pending', overall_error TEXT, apk_version_name TEXT DEFAULT '', local_apk_dir TEXT DEFAULT '', local_apk_file_count INTEGER DEFAULT 0, local_apk_is_stale INTEGER DEFAULT 1, created_at REAL NOT NULL, updated_at REAL NOT NULL, UNIQUE(package_name, country_code) ); CREATE INDEX IF NOT EXISTS idx_download_record_pkg ON app_download_record(package_name); CREATE INDEX IF NOT EXISTS idx_download_record_country ON app_download_record(country_code); CREATE INDEX IF NOT EXISTS idx_download_record_overall ON app_download_record(overall_status); CREATE INDEX IF NOT EXISTS idx_download_record_stale ON app_download_record(local_apk_is_stale); """ ) def clear_monitoring_tables(self): table_names = [ "controller_session", "worker_state_event", "task_stage_event", "task_execution", ] with self._write_lock, self._connect() as connection: for table_name in table_names: if self._table_exists(connection, table_name): connection.execute(f"DELETE FROM {table_name}") if self._table_exists(connection, "sqlite_sequence"): for table_name in table_names: connection.execute("DELETE FROM sqlite_sequence WHERE name = ?", (table_name,)) def create_controller_session(self, host: str, version: str, started_at: float) -> int: with self._write_lock, self._connect() as connection: cursor = connection.execute( """ INSERT INTO controller_session ( started_at, host, version, last_heartbeat_at ) VALUES (?, ?, ?, ?) """, (started_at, host, version, started_at), ) return int(cursor.lastrowid) def heartbeat_controller_session(self, session_id: int, heartbeat_at: float): with self._write_lock, self._connect() as connection: connection.execute( """ UPDATE controller_session SET last_heartbeat_at = ? WHERE id = ? """, (heartbeat_at, session_id), ) def close_controller_session(self, session_id: int, ended_at: float, stop_reason: str): with self._write_lock, self._connect() as connection: connection.execute( """ UPDATE controller_session SET ended_at = ?, stop_reason = ?, last_heartbeat_at = COALESCE(last_heartbeat_at, ?) WHERE id = ? """, (ended_at, stop_reason, ended_at, session_id), ) def repair_open_sessions(self, repaired_at: float): with self._write_lock, self._connect() as connection: rows = connection.execute( """ SELECT id, last_heartbeat_at FROM controller_session WHERE ended_at IS NULL """ ).fetchall() for row in rows: ended_at = float(row["last_heartbeat_at"] or repaired_at) connection.execute( """ UPDATE controller_session SET ended_at = ?, stop_reason = COALESCE(stop_reason, 'recovered_after_restart') WHERE id = ? """, (ended_at, row["id"]), ) def insert_worker_state_event( self, worker_id: str, session_id: Optional[int], event_time: float, state: str, reason: Optional[str], task_key: Optional[str], ): with self._write_lock, self._connect() as connection: connection.execute( """ INSERT INTO worker_state_event ( worker_id, session_id, event_time, state, reason, task_key ) VALUES (?, ?, ?, ?, ?, ?) """, (worker_id, session_id, event_time, state, reason, task_key), ) def insert_task_stage_event( self, execution_id: str, task_key: str, worker_id: str, session_id: Optional[int], event_time: float, event_type: str, stage: Optional[str], status: Optional[str], reason: Optional[str], metrics: Optional[Dict[str, Any]], ): with self._write_lock, self._connect() as connection: connection.execute( """ INSERT INTO task_stage_event ( execution_id, task_key, worker_id, session_id, event_time, event_type, stage, status, reason, metrics_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( execution_id, task_key, worker_id, session_id, event_time, event_type, stage, status, reason, _safe_json_dumps(metrics or {}), ), ) def upsert_task_execution( self, execution_id: str, task_key: str, worker_id: str, session_id: Optional[int], stat_date: str, task_started_at: float, app_name: Optional[str] = None, package_name: Optional[str] = None, last_updated: Optional[str] = None, collection_task_type: Optional[str] = None, retry_count: int = 0, ): with self._write_lock, self._connect() as connection: connection.execute( """ INSERT INTO task_execution ( execution_id, task_key, app_name, package_name, last_updated, collection_task_type, worker_id, session_id, stat_date, task_started_at, retry_count, last_updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(execution_id) DO UPDATE SET app_name = COALESCE(excluded.app_name, task_execution.app_name), package_name = COALESCE(excluded.package_name, task_execution.package_name), last_updated = COALESCE(excluded.last_updated, task_execution.last_updated), collection_task_type = COALESCE(excluded.collection_task_type, task_execution.collection_task_type), retry_count = MAX(task_execution.retry_count, excluded.retry_count), worker_id = excluded.worker_id, session_id = excluded.session_id, last_updated_at = excluded.last_updated_at """, ( execution_id, task_key, app_name, package_name, last_updated, collection_task_type, worker_id, session_id, stat_date, task_started_at, retry_count, time.time(), ), ) def update_task_execution_stage( self, execution_id: str, stage: str, duration_seconds: Optional[float], download_source: Optional[str] = None, ): if duration_seconds is None: return with self._write_lock, self._connect() as connection: if stage == "download": connection.execute( """ UPDATE task_execution SET download_duration_seconds = ?, download_source = COALESCE(?, download_source), last_updated_at = ? WHERE execution_id = ? """, (duration_seconds, download_source, time.time(), execution_id), ) elif stage == "collect": connection.execute( """ UPDATE task_execution SET collect_duration_seconds = ?, last_updated_at = ? WHERE execution_id = ? """, (duration_seconds, time.time(), execution_id), ) def finish_task_execution( self, execution_id: str, task_ended_at: float, status: str, failed_stage: Optional[str], failure_bucket: Optional[str], total_duration_seconds: Optional[float], failed_total_duration_seconds: Optional[float], error_type: Optional[str], error_message: Optional[str], result_detail: Optional[str], droidbot_steps: Optional[int], guiagent_steps: Optional[int], total_steps: Optional[int], num_nodes: Optional[int], num_reached_activities: Optional[int], app_num_total_activities: Optional[int], failure_domain: Optional[str], failure_subtype: Optional[str], trace_payload: Optional[Dict[str, Any]], retry_count: Optional[int] = None, ): with self._write_lock, self._connect() as connection: connection.execute( """ UPDATE task_execution SET task_ended_at = ?, status = ?, failed_stage = ?, failure_bucket = COALESCE(?, failure_bucket), total_duration_seconds = COALESCE(?, total_duration_seconds), failed_total_duration_seconds = COALESCE(?, failed_total_duration_seconds), error_type = COALESCE(?, error_type), error_message = COALESCE(?, error_message), result_detail = COALESCE(?, result_detail), droidbot_steps = COALESCE(?, droidbot_steps), guiagent_steps = COALESCE(?, guiagent_steps), total_steps = COALESCE(?, total_steps), num_nodes = COALESCE(?, num_nodes), num_reached_activities = COALESCE(?, num_reached_activities), app_num_total_activities = COALESCE(?, app_num_total_activities), failure_domain = COALESCE(?, failure_domain), failure_subtype = COALESCE(?, failure_subtype), trace_json = COALESCE(?, trace_json), retry_count = COALESCE(?, retry_count), last_updated_at = ? WHERE execution_id = ? """, ( task_ended_at, status, failed_stage, failure_bucket, total_duration_seconds, failed_total_duration_seconds, error_type, error_message, result_detail, droidbot_steps, guiagent_steps, total_steps, num_nodes, num_reached_activities, app_num_total_activities, failure_domain, failure_subtype, _safe_json_dumps(trace_payload) if trace_payload else None, retry_count, time.time(), execution_id, ), ) def list_task_executions(self, stat_date: str, worker_id: Optional[str] = None, limit: int = 200) -> List[sqlite3.Row]: with self._connect() as connection: if worker_id: return connection.execute( """ SELECT * FROM task_execution WHERE stat_date = ? AND worker_id = ? ORDER BY COALESCE(task_started_at, last_updated_at, 0) DESC LIMIT ? """, (stat_date, worker_id, limit), ).fetchall() return connection.execute( """ SELECT * FROM task_execution WHERE stat_date = ? ORDER BY COALESCE(task_started_at, last_updated_at, 0) DESC LIMIT ? """, (stat_date, limit), ).fetchall() def list_app_collect_statuses(self, package_names: Iterable[str]) -> Dict[str, Dict[str, str]]: """ 查询应用的采集状态分类（新架构）。 直接从 collection_task 查询最新执行记录，基于 execution_status, error_category, error_code, num_nodes 和流量比例动态计算状态。 """ normalized = [str(item or "").strip() for item in package_names if str(item or "").strip()] if not normalized: return {} with self._connect() as connection: rows = [] for package_name in normalized: row = connection.execute( """ SELECT package_name, execution_status, error_category, error_code, num_nodes, CASE WHEN total_traffic_bytes > 0 THEN self_traffic_bytes * 100.0 / total_traffic_bytes ELSE 0 END AS self_ratio FROM collection_task WHERE package_name = ? ORDER BY datetime(COALESCE(NULLIF(completed_at, ''), NULLIF(started_at, ''), created_at)) DESC, attempt DESC, batch_tag DESC, run_kind DESC LIMIT 1 """, (package_name,), ).fetchone() if row: rows.append(row) result = {} # 不可重试的 DOWNLOAD_ERROR 错误码 NON_RETRYABLE_DOWNLOAD_CODES = {1, 5, 6, 10, 403, 404} for row in rows: pkg = row["package_name"] if not pkg: continue exec_status = (row["execution_status"] or "").strip() error_cat = (row["error_category"] or "").strip() error_code = row["error_code"] nodes = row["num_nodes"] or 0 self_ratio = row["self_ratio"] or 0 # 动态计算 restriction_status if exec_status == "success": # 成功任务：根据质量指标判断 if nodes >= 5 and self_ratio >= 30: restriction_status = "success" else: restriction_status = "light_restricted" else: # 失败任务：严重受限 restriction_status = "severe_restricted" # 动态计算 retryability if exec_status == "success": retryability = "retryable" elif error_cat == "DOWNLOAD_ERROR" and error_code in NON_RETRYABLE_DOWNLOAD_CODES: retryability = "non_retryable" else: retryability = "retryable" # 动态计算 collection_status（业务层分类） if restriction_status == "success": collection_status = "qualified" else: collection_status = "restricted" result[pkg] = { "collection_status": collection_status, "restriction_status": restriction_status, "retryability": retryability, } return result def get_task_execution_totals(self, worker_id: Optional[str] = None) -> Dict[str, int]: with self._connect() as connection: if worker_id: row = connection.execute( """ SELECT COUNT(*) AS task_total, COALESCE(SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END), 0) AS task_success, COALESCE(SUM(CASE WHEN status != 'success' THEN 1 ELSE 0 END), 0) AS task_non_success, COALESCE(SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END), 0) AS task_failed FROM task_execution WHERE worker_id = ? """, (worker_id,), ).fetchone() else: row = connection.execute( """ SELECT COUNT(*) AS task_total, COALESCE(SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END), 0) AS task_success, COALESCE(SUM(CASE WHEN status != 'success' THEN 1 ELSE 0 END), 0) AS task_non_success, COALESCE(SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END), 0) AS task_failed FROM task_execution """ ).fetchone() return { "task_total": int(row["task_total"] or 0), "task_success": int(row["task_success"] or 0), "task_non_success": int(row["task_non_success"] or 0), "task_failed": int(row["task_failed"] or 0), } def get_task_overview_metrics(self, stat_date: str, worker_id: Optional[str] = None) -> Dict[str, Any]: with self._connect() as connection: where_clause = "WHERE stat_date = ?" params: List[Any] = [stat_date] if worker_id: where_clause += " AND worker_id = ?" params.append(worker_id) summary_row = connection.execute( f""" SELECT COUNT(*) AS task_total, COALESCE(SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END), 0) AS task_success, COALESCE(SUM(CASE WHEN status != 'success' THEN 1 ELSE 0 END), 0) AS task_non_success, COALESCE(SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END), 0) AS task_failed, COALESCE(AVG(download_duration_seconds), 0) AS avg_download_duration_seconds, COALESCE(AVG(collect_duration_seconds), 0) AS avg_collect_duration_seconds, COALESCE(SUM(CASE WHEN download_duration_seconds IS NOT NULL THEN 1 ELSE 0 END), 0) AS download_value_count, COALESCE(SUM(CASE WHEN collect_duration_seconds IS NOT NULL THEN 1 ELSE 0 END), 0) AS collect_value_count FROM task_execution {where_clause} """, params, ).fetchone() def percentile_for(column: str, count: int, ratio: float) -> float: if count <= 0: return 0.0 offset = int(round((count - 1) * ratio)) row = connection.execute( f""" SELECT {column} AS value FROM task_execution {where_clause} AND {column} IS NOT NULL ORDER BY {column} LIMIT 1 OFFSET ? """, [*params, offset], ).fetchone() if not row or row["value"] is None: return 0.0 return round(float(row["value"]), 2) download_count = int(summary_row["download_value_count"] or 0) collect_count = int(summary_row["collect_value_count"] or 0) return { "task_total": int(summary_row["task_total"] or 0), "task_success": int(summary_row["task_success"] or 0), "task_non_success": int(summary_row["task_non_success"] or 0), "task_failed": int(summary_row["task_failed"] or 0), "avg_download_duration_seconds": round(float(summary_row["avg_download_duration_seconds"] or 0.0), 2), "avg_collect_duration_seconds": round(float(summary_row["avg_collect_duration_seconds"] or 0.0), 2), "p50_download_duration_seconds": percentile_for("download_duration_seconds", download_count, 0.5), "p90_download_duration_seconds": percentile_for("download_duration_seconds", download_count, 0.9), "p50_collect_duration_seconds": percentile_for("collect_duration_seconds", collect_count, 0.5), "p90_collect_duration_seconds": percentile_for("collect_duration_seconds", collect_count, 0.9), } def list_state_events_before(self, day_end: float, worker_ids: Optional[Iterable[str]] = None) -> List[sqlite3.Row]: with self._connect() as connection: if worker_ids: worker_ids = list(worker_ids) placeholders = ",".join("?" for _ in worker_ids) query = f""" SELECT worker_id, event_time, state, reason, task_key FROM worker_state_event WHERE event_time < ? AND worker_id IN ({placeholders}) ORDER BY worker_id, event_time """ return connection.execute(query, [day_end, *worker_ids]).fetchall() return connection.execute( """ SELECT worker_id, event_time, state, reason, task_key FROM worker_state_event WHERE event_time < ? ORDER BY worker_id, event_time """, (day_end,), ).fetchall() def list_worker_ids_up_to(self, day_end: float) -> List[str]: with self._connect() as connection: rows = connection.execute( """ SELECT DISTINCT worker_id FROM worker_state_event WHERE event_time < ? """, (day_end,), ).fetchall() return [row["worker_id"] for row in rows] def list_sessions_overlapping(self, day_start: float, day_end: float) -> List[sqlite3.Row]: with self._connect() as connection: return connection.execute( """ SELECT * FROM controller_session WHERE started_at < ? AND COALESCE(ended_at, last_heartbeat_at, ?) > ? ORDER BY started_at """, (day_end, day_end, day_start), ).fetchall() def update_failure_bucket(self, execution_id: str, failure_bucket: str): with self._write_lock, self._connect() as connection: connection.execute( """ UPDATE task_execution SET failure_bucket = ?, last_updated_at = ? WHERE execution_id = ? """, (failure_bucket, time.time(), execution_id), ) def update_failure_bucket_by_task_key(self, task_key: str, failure_bucket: str): with self._write_lock, self._connect() as connection: connection.execute( """ UPDATE task_execution SET failure_bucket = ?, last_updated_at = ? WHERE execution_id = ( SELECT execution_id FROM task_execution WHERE task_key = ? ORDER BY COALESCE(task_ended_at, last_updated_at, task_started_at) DESC LIMIT 1 ) """, (failure_bucket, time.time(), task_key), ) def get_latest_execution_id(self, task_key: str) -> Optional[str]: with self._connect() as connection: row = connection.execute( """ SELECT execution_id FROM task_execution WHERE task_key = ? ORDER BY COALESCE(task_ended_at, last_updated_at, task_started_at) DESC LIMIT 1 """, (task_key,), ).fetchone() return row["execution_id"] if row else None def list_stage_finish_events(self, stat_date: str) -> List[sqlite3.Row]: with self._connect() as connection: return connection.execute( """ SELECT te.execution_id, te.status, te.failure_bucket, te.total_duration_seconds, te.failed_total_duration_seconds, te.error_type, te.error_message, e.stage, e.metrics_json FROM task_stage_event e JOIN task_execution te ON te.execution_id = e.execution_id WHERE te.stat_date = ? AND e.event_type = 'stage_finished' ORDER BY e.event_time """, (stat_date,), ).fetchall() class MonitoringService: def __init__(self, redis_client, managed_worker_ids: Optional[List[str]] = None): self.redis = redis_client self.repo = MonitoringRepository() self.managed_worker_ids = list(managed_worker_ids or MANAGED_WORKER_IDS or []) self._lock = threading.RLock() self._session_id: Optional[int] = None self._version = "unknown" self.repo.repair_open_sessions(time.time()) def start_controller_session(self, version: str = "unknown") -> int: with self._lock: if self._session_id is not None: return self._session_id started_at = time.time() self._version = version self._session_id = self.repo.create_controller_session( host=socket.gethostname(), version=version, started_at=started_at, ) self.redis.set("controller:current_session", str(self._session_id)) return self._session_id def heartbeat_controller_session(self): with self._lock: session_id = self.current_session_id if session_id is None: return self.repo.heartbeat_controller_session(session_id, time.time()) def close_controller_session(self, stop_reason: str = "stopped"): with self._lock: session_id = self._session_id if session_id is None: return self.repo.close_controller_session(session_id, time.time(), stop_reason) self.redis.delete("controller:current_session") self._session_id = None def _classify_rows_by_collection_quality( self, rows: List[sqlite3.Row], ) -> Tuple[List[sqlite3.Row], List[sqlite3.Row], Dict[str, int]]: package_map = self.repo.list_app_collect_statuses( row["package_name"] for row in rows if row["package_name"] ) qualified_items: List[sqlite3.Row] = [] severe_items: List[sqlite3.Row] = [] counters = { "qualified_task_count": 0, "qualified_success_task_count": 0, "qualified_light_task_count": 0, "non_success_task_count": 0, "non_retryable_task_count": 0, } for row in rows: package_name = str(row["package_name"] or "").strip() row_state = package_map.get(package_name, {}) restriction_status = row_state.get("restriction_status", "") retryability = row_state.get("retryability", "") if restriction_status == "severe_restricted": severe_items.append(row) counters["non_success_task_count"] += 1 if retryability == "non_retryable": counters["non_retryable_task_count"] += 1 else: qualified_items.append(row) counters["qualified_task_count"] += 1 if restriction_status == "light_restricted": counters["qualified_light_task_count"] += 1 else: counters["qualified_success_task_count"] += 1 return qualified_items, severe_items, counters def _split_rows_by_quality(self, rows: List[sqlite3.Row]) -> Tuple[List[sqlite3.Row], List[sqlite3.Row]]: qualified_items, severe_items, _ = self._classify_rows_by_collection_quality(rows) return qualified_items, severe_items @property def current_session_id(self) -> Optional[int]: if self._session_id is not None: return self._session_id raw = self.redis.get("controller:current_session") if raw: try: self._session_id = int(raw) except (TypeError, ValueError): self._session_id = None return self._session_id def _execution_key(self, task_key: str) -> str: return f"monitor:execution:{task_key}" def _current_state_key(self, worker_id: str) -> str: return f"worker:current_state:{worker_id}" def _get_current_state(self, worker_id: str) -> Dict[str, Any]: return _safe_json_loads(self.redis.get(self._current_state_key(worker_id))) def _set_current_state(self, worker_id: str, payload: Dict[str, Any]): self.redis.set(self._current_state_key(worker_id), _safe_json_dumps(payload)) def _delete_monitoring_redis_state(self): self.redis.delete("controller:current_session") patterns = [ "worker:current_state:*", "worker:last_event:*", "monitor:execution:*", ] keys_to_delete: List[str] = [] for pattern in patterns: keys_to_delete.extend(list(self.redis.scan_iter(match=pattern))) if keys_to_delete: self.redis.delete(*keys_to_delete) def _normalize_reseed_state(self, worker: Dict[str, Any]) -> Optional[str]: if not worker.get("online", True): return None state = str(worker.get("state") or "").strip() if state in {"idle_waiting_task", "running_download", "running_collect", "disabled"}: return state if worker.get("status") == "busy": return "running_collect" if worker.get("status") == "idle": return "idle_waiting_task" return "idle_waiting_task" def reset_monitoring_data(self, worker_snapshots: Optional[List[Dict[str, Any]]] = None) -> Dict[str, Any]: with self._lock: reset_time = time.time() self.close_controller_session(stop_reason="dashboard_reset") self.repo.clear_monitoring_tables() self._delete_monitoring_redis_state() self._session_id = None session_id = self.start_controller_session(version=self._version) reseeded_workers = 0 for worker in worker_snapshots or []: worker_id = worker.get("worker_id") if not worker_id: continue state = self._normalize_reseed_state(worker) if not state: continue self.set_worker_state( worker_id, state, reason="dashboard_reset_reseed", task_key=worker.get("current_task"), event_time=reset_time, ) current_task = worker.get("current_task") if current_task: task_metadata = self._resolve_task_metadata(current_task) retry_count = 0 status_json = self.redis.hget("task:status", current_task) if status_json: status_payload = _safe_json_loads(status_json) retry_count = int(status_payload.get("retry_count") or 0) execution_id = f"{current_task}:{int(reset_time * 1000)}" self.repo.upsert_task_execution( execution_id=execution_id, task_key=current_task, worker_id=worker_id, session_id=self.current_session_id, stat_date=_date_str_from_ts(reset_time), task_started_at=reset_time, app_name=task_metadata.get("app_name"), package_name=task_metadata.get("package_name"), last_updated=task_metadata.get("last_updated"), collection_task_type=task_metadata.get("collection_task_type"), retry_count=retry_count, ) self.redis.set(self._execution_key(current_task), execution_id) reseeded_workers += 1 return { "ok": True, "reset_at": datetime.fromtimestamp(reset_time, SHANGHAI_TZ).strftime("%Y-%m-%d %H:%M:%S"), "session_id": session_id, "reseeded_workers": reseeded_workers, } def _update_worker_info_state( self, worker_id: str, state: str, task_key: Optional[str], reason: Optional[str], event_time: float, ): info_json = self.redis.hget("workers:info", worker_id) if not info_json: return info = _safe_json_loads(info_json) info["state"] = state info["state_since"] = event_time info["pause_reason"] = reason if task_key is not None: info["current_task"] = task_key self.redis.hset("workers:info", worker_id, _safe_json_dumps(info)) def set_worker_state( self, worker_id: str, state: str, reason: Optional[str] = None, task_key: Optional[str] = None, event_time: Optional[float] = None, ): with self._lock: event_time = event_time or time.time() current = self._get_current_state(worker_id) if ( current.get("state") == state and current.get("reason") == reason and current.get("task_key") == task_key ): return payload = {"state": state, "reason": reason, "task_key": task_key, "since": event_time} self._set_current_state(worker_id, payload) self._update_worker_info_state(worker_id, state, task_key, reason, event_time) self.repo.insert_worker_state_event( worker_id=worker_id, session_id=self.current_session_id, event_time=event_time, state=state, reason=reason, task_key=task_key, ) def register_worker(self, worker_id: str, task: Optional[Dict[str, Any]] = None, event_time: Optional[float] = None): with self._lock: event_time = event_time or time.time() if task and task.get("task_key"): self._update_worker_info_state(worker_id, "idle_waiting_task", task.get("task_key"), None, event_time) else: self.set_worker_state(worker_id, "idle_waiting_task", reason="waiting_task", event_time=event_time) def handle_task_assigned(self, worker_id: str, task_key: str, event_time: Optional[float] = None): with self._lock: event_time = event_time or time.time() info_json = self.redis.hget("workers:info", worker_id) if not info_json: return info = _safe_json_loads(info_json) info["current_task"] = task_key info["last_update"] = event_time self.redis.hset("workers:info", worker_id, _safe_json_dumps(info)) # 任务分发后将 worker state 标记为 running（等 stage_started 事件进一步细化为 download/collect） self.set_worker_state(worker_id, "running_download", task_key=task_key, event_time=event_time) if not self.redis.get(self._execution_key(task_key)): status_payload = _safe_json_loads(self.redis.hget("task:status", task_key)) retry_count = int(status_payload.get("retry_count") or 0) task_metadata = self._resolve_task_metadata(task_key) execution_id = f"{task_key}:{int(event_time * 1000)}" self.repo.upsert_task_execution( execution_id=execution_id, task_key=task_key, worker_id=worker_id, session_id=self.current_session_id, stat_date=_date_str_from_ts(event_time), task_started_at=event_time, app_name=task_metadata.get("app_name"), package_name=task_metadata.get("package_name"), last_updated=task_metadata.get("last_updated"), collection_task_type=task_metadata.get("collection_task_type"), retry_count=retry_count, ) self.redis.set(self._execution_key(task_key), execution_id) def _resolve_task_details(self, task_key: str) -> Tuple[Optional[str], Optional[str]]: details = self._resolve_task_metadata(task_key) return details.get("app_name"), details.get("package_name") def _resolve_task_metadata(self, task_key: str) -> Dict[str, Any]: raw = self.redis.hget("task:details", task_key) if not raw: return {} details = _safe_json_loads(raw) return details if isinstance(details, dict) else {} def _ensure_execution_id( self, worker_id: str, task_key: str, event_time: float, retry_count: int = 0, ) -> str: with self._lock: execution_id = self.redis.get(self._execution_key(task_key)) if execution_id: return execution_id execution_id = f"{task_key}:{int(event_time * 1000)}" task_metadata = self._resolve_task_metadata(task_key) self.repo.upsert_task_execution( execution_id=execution_id, task_key=task_key, worker_id=worker_id, session_id=self.current_session_id, stat_date=_date_str_from_ts(event_time), task_started_at=event_time, app_name=task_metadata.get("app_name"), package_name=task_metadata.get("package_name"), last_updated=task_metadata.get("last_updated"), collection_task_type=task_metadata.get("collection_task_type"), retry_count=retry_count, ) self.redis.set(self._execution_key(task_key), execution_id) return execution_id def _resolve_report_execution_id( self, worker_id: str, task_key: str, event_time: float, retry_count: int = 0, ) -> str: execution_id = self.redis.get(self._execution_key(task_key)) if execution_id: return execution_id execution_id = self.repo.get_latest_execution_id(task_key) if execution_id: return execution_id return self._ensure_execution_id(worker_id, task_key, event_time, retry_count=retry_count) def _resolve_failure_fields( self, error_type: Optional[str], error_message: Optional[str], download_errors: Any = None, trace_payload: Optional[Dict[str, Any]] = None, ) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]: trace_payload = dict(trace_payload) if isinstance(trace_payload, dict) else {} normalized_download_errors = _normalize_download_errors(download_errors) if normalized_download_errors: trace_payload["download_errors"] = normalized_download_errors failure_domain = trace_payload.get("failure_domain") failure_subtype = trace_payload.get("failure_subtype") has_failure = bool(error_type or error_message or normalized_download_errors) if has_failure and (not failure_domain or not failure_subtype): failure_domain, failure_subtype = _classify_failure( error_type, error_message, download_errors=normalized_download_errors, ) elif not has_failure: failure_domain = None failure_subtype = None if not trace_payload: trace_payload = None elif has_failure and not trace_payload.get("failure_domain"): trace_payload["failure_domain"] = failure_domain trace_payload["failure_subtype"] = failure_subtype return failure_domain, failure_subtype, trace_payload def handle_worker_event(self, payload: Dict[str, Any]): with self._lock: worker_id = normalize_worker_id(payload.get("worker_id")) task_key = payload.get("task_key") event_type = payload.get("event_type") if not worker_id or not event_type: return event_time = float(payload.get("event_time") or time.time()) stage = payload.get("stage") state = payload.get("state") status = payload.get("status") reason = payload.get("reason") metrics = payload.get("metrics") or {} if event_type == "worker_state_changed" and state: self.set_worker_state(worker_id, state, reason=reason, task_key=task_key, event_time=event_time) return if not task_key: return retry_count = int(metrics.get("retry_count") or 0) execution_id = self._ensure_execution_id(worker_id, task_key, event_time, retry_count=retry_count) self.repo.insert_task_stage_event( execution_id=execution_id, task_key=task_key, worker_id=worker_id, session_id=self.current_session_id, event_time=event_time, event_type=event_type, stage=stage, status=status, reason=reason, metrics=metrics, ) if event_type == "task_started": task_metadata = self._resolve_task_metadata(task_key) self.repo.upsert_task_execution( execution_id=execution_id, task_key=task_key, worker_id=worker_id, session_id=self.current_session_id, stat_date=_date_str_from_ts(event_time), task_started_at=event_time, app_name=task_metadata.get("app_name"), package_name=task_metadata.get("package_name"), last_updated=task_metadata.get("last_updated"), collection_task_type=task_metadata.get("collection_task_type"), retry_count=retry_count, ) self.handle_task_assigned(worker_id, task_key, event_time) return if event_type == "stage_started": if stage == "download": self.set_worker_state(worker_id, "running_download", task_key=task_key, event_time=event_time) elif stage == "collect": self.set_worker_state(worker_id, "running_collect", task_key=task_key, event_time=event_time) return if event_type == "stage_finished": self.repo.update_task_execution_stage( execution_id=execution_id, stage=stage or "", duration_seconds=metrics.get("duration_seconds"), download_source=metrics.get("download_source"), ) return if event_type == "task_finished": error_type = metrics.get("error_type") error_message = metrics.get("error_message") or reason total_duration = metrics.get("total_duration_seconds") failed_total_duration = metrics.get("failed_total_duration_seconds") failed_stage = metrics.get("failed_stage") failure_domain, failure_subtype, trace_payload = self._resolve_failure_fields( error_type, error_message, download_errors=metrics.get("download_errors"), trace_payload=metrics.get("trace"), ) self.repo.finish_task_execution( execution_id=execution_id, task_ended_at=event_time, status=status or "unknown", failed_stage=failed_stage, failure_bucket="failed" if (status or "unknown") == "failed" else status, total_duration_seconds=total_duration, failed_total_duration_seconds=failed_total_duration, error_type=error_type, error_message=error_message, result_detail=error_message, droidbot_steps=None, guiagent_steps=None, total_steps=None, num_nodes=None, num_reached_activities=None, app_num_total_activities=None, failure_domain=failure_domain, failure_subtype=failure_subtype, trace_payload=trace_payload, retry_count=retry_count, ) self.redis.delete(self._execution_key(task_key)) self.set_worker_state(worker_id, "idle_waiting_task", reason="waiting_task", event_time=event_time) def handle_worker_report( self, worker_id: str, previous_task_key: str, report_data: Dict[str, Any], current_time: Optional[float] = None, retry_count: int = 0, resolved_message: Optional[str] = None, ): worker_id = normalize_worker_id(worker_id) with self._lock: if not previous_task_key: return current_time = current_time or time.time() execution_id = self._resolve_report_execution_id( worker_id, previous_task_key, current_time, retry_count=retry_count, ) statistics = report_data.get("statistics") or {} error_data = report_data.get("error") or {} trace_payload = report_data.get("trace") status = report_data.get("status", "unknown") current_state = self._get_current_state(worker_id).get("state") failed_stage = "download" if not statistics.get("download_source") else "collect" if status == "success": failed_stage = None elif current_state == "running_download": failed_stage = "download" elif current_state == "running_collect": failed_stage = "collect" error_type = ( f"{error_data.get('category')}/{error_data.get('code')}" if error_data.get("category") is not None and error_data.get("code") is not None else None ) error_message = resolved_message or error_data.get("reason") or statistics.get("error_reason") failure_domain, failure_subtype, trace_payload = self._resolve_failure_fields( error_type, error_message, download_errors=report_data.get("download_errors"), trace_payload=trace_payload, ) self.repo.finish_task_execution( execution_id=execution_id, task_ended_at=current_time, status=status, failed_stage=failed_stage, failure_bucket=status if status != "failed" else "failed", total_duration_seconds=statistics.get("duration_seconds"), failed_total_duration_seconds=statistics.get("duration_seconds") if status != "success" else None, error_type=error_type, error_message=error_message, result_detail=resolved_message or error_message, droidbot_steps=statistics.get("droidbot_steps"), guiagent_steps=statistics.get("guiagent_steps"), total_steps=statistics.get("total_steps"), num_nodes=statistics.get("num_nodes"), num_reached_activities=statistics.get("num_reached_activities"), app_num_total_activities=statistics.get("app_num_total_activities"), failure_domain=failure_domain, failure_subtype=failure_subtype, trace_payload=trace_payload, retry_count=retry_count, ) self.redis.delete(self._execution_key(previous_task_key)) def mark_worker_disabled(self, worker_id: str, reason: str): self.set_worker_state(worker_id, "disabled", reason=reason, event_time=time.time()) def mark_worker_offline(self, worker_id: str, reason: str): self.set_worker_state(worker_id, "offline", reason=reason, event_time=time.time()) def set_failure_bucket(self, task_key: str, failure_bucket: str): with self._lock: execution_id = self.redis.get(self._execution_key(task_key)) if execution_id: self.repo.update_failure_bucket(execution_id, failure_bucket) return self.repo.update_failure_bucket_by_task_key(task_key, failure_bucket) def _controller_up_seconds(self, date_str: str, now_ts: Optional[float] = None) -> float: day_start, day_end, _ = _day_bounds(date_str, now_ts) total = 0.0 for row in self.repo.list_sessions_overlapping(day_start, day_end): start = max(float(row["started_at"]), day_start) end = min(float(row["ended_at"] or row["last_heartbeat_at"] or day_end), day_end) total += max(0.0, end - start) return round(total, 2) def _resolve_worker_ids( self, date_str: str, dispatcher=None, worker_id: Optional[str] = None, ) -> List[str]: if worker_id: return [worker_id] _, day_end, is_today = _day_bounds(date_str) # 以 managed_worker_ids（inventory.json）为准 worker_ids = set(self.managed_worker_ids) if not is_today: # 仅在查看历史日期时合并 SQLite 中的 worker（兼容查看旧数据） worker_ids.update(self.repo.list_worker_ids_up_to(day_end)) if dispatcher is not None and is_today: for worker in dispatcher.get_dashboard_workers(): worker_ids.add(worker.get("worker_id")) worker_ids.discard(None) worker_ids.discard("") return sorted(worker_ids) def _build_worker_options(self, date_str: str, dispatcher=None) -> List[Dict[str, Any]]: worker_ids = self._resolve_worker_ids(date_str, dispatcher=dispatcher) current_workers: Dict[str, Dict[str, Any]] = {} if dispatcher is not None: current_workers = {worker.get("worker_id"): worker for worker in dispatcher.get_dashboard_workers()} options: List[Dict[str, Any]] = [] for worker_id in worker_ids: current = current_workers.get(worker_id, {}) ip_address = current.get("ip_address") or worker_id.split("_", 1)[0] hostname = current.get("hostname") or "" label = f"{ip_address} · {hostname}" if hostname else ip_address options.append( { "worker_id": worker_id, "ip_address": ip_address, "hostname": hostname, "label": label, } ) return options def _build_worker_intervals( self, date_str: str, dispatcher=None, worker_id: Optional[str] = None, ) -> Dict[str, List[Tuple[float, float, str]]]: day_start, day_end, _ = _day_bounds(date_str) worker_ids = self._resolve_worker_ids(date_str, dispatcher=dispatcher, worker_id=worker_id) rows = self.repo.list_state_events_before(day_end, worker_ids=worker_ids) grouped: Dict[str, List[sqlite3.Row]] = defaultdict(list) for row in rows: grouped[row["worker_id"]].append(row) intervals: Dict[str, List[Tuple[float, float, str]]] = {} for worker_id in worker_ids: events = grouped.get(worker_id, []) prev_state = "offline" prev_time = day_start worker_intervals: List[Tuple[float, float, str]] = [] for row in events: event_time = float(row["event_time"]) if event_time < day_start: prev_state = row["state"] continue if event_time > day_end: break if event_time > prev_time: worker_intervals.append((prev_time, event_time, prev_state)) prev_state = row["state"] prev_time = event_time if day_end > prev_time: worker_intervals.append((prev_time, day_end, prev_state)) intervals[worker_id] = worker_intervals return intervals def _summarize_worker_seconds( self, date_str: str, dispatcher=None, worker_id: Optional[str] = None, ) -> Dict[str, Counter]: intervals = self._build_worker_intervals(date_str, dispatcher=dispatcher, worker_id=worker_id) summary: Dict[str, Counter] = {} for worker_id, worker_intervals in intervals.items(): counts = Counter() for start, end, state in worker_intervals: counts[state] += max(0.0, end - start) summary[worker_id] = counts return summary def _runtime_state_breakdown( self, date_str: str, dispatcher=None, worker_id: Optional[str] = None, ) -> List[Dict[str, Any]]: totals = Counter() for counts in self._summarize_worker_seconds(date_str, dispatcher=dispatcher, worker_id=worker_id).values(): for state in RUNTIME_STATE_ORDER: totals[state] += float(counts.get(state, 0.0)) total_seconds = sum(totals.values()) return [ { "state": state, "label": RUNTIME_STATE_LABELS[state], "duration_seconds": round(float(totals[state]), 2), "share_percent": round((float(totals[state]) / total_seconds) * 100, 2) if total_seconds > 0 else 0.0, } for state in RUNTIME_STATE_ORDER ] def get_overview( self, date_str: Optional[str] = None, dispatcher=None, worker_id: Optional[str] = None, ) -> Dict[str, Any]: now_ts = time.time() date_str = date_str or _date_str_from_ts(now_ts) day_start, day_end, is_today = _day_bounds(date_str, now_ts=now_ts) up_seconds = self._controller_up_seconds(date_str, now_ts=now_ts) down_seconds = round(max(0.0, day_end - day_start - up_seconds), 2) metrics = self.repo.get_task_overview_metrics(date_str, worker_id=worker_id) overall_totals = self.repo.get_task_execution_totals(worker_id=worker_id) overall_success_total = overall_totals["task_success"] # The global dashboard should reflect the dispatcher's in-memory success set, # including tasks preloaded from the current active catalog queue. if dispatcher is not None and not worker_id: overall_success_total = int(dispatcher.get_statistics().get("completed") or 0) state_counts = { "online_workers": 0, "running_workers": 0, "running_download_workers": 0, "running_collect_workers": 0, "idle_workers": 0, "offline_workers": 0, "disabled_workers": 0, "pending_tasks": 0, } if is_today and dispatcher is not None: dashboard_workers = dispatcher.get_dashboard_workers() if worker_id: dashboard_workers = [worker for worker in dashboard_workers if worker.get("worker_id") == worker_id] state_counts["pending_tasks"] = 0 if worker_id else dispatcher.get_statistics().get("pending", 0) for worker in dashboard_workers: current_worker_id = worker.get("worker_id", "") state = worker.get("state") or self._get_current_state(current_worker_id).get("state") or "offline" if not worker.get("online"): state = "offline" if state in ONLINE_STATES: state_counts["online_workers"] += 1 if state in RUNNING_STATES: state_counts["running_workers"] += 1 if state == "running_download": state_counts["running_download_workers"] += 1 elif state == "running_collect": state_counts["running_collect_workers"] += 1 elif state == "idle_waiting_task": state_counts["idle_workers"] += 1 elif state == "disabled": state_counts["disabled_workers"] += 1 else: state_counts["offline_workers"] += 1 else: timeline = self.get_timeline(date_str=date_str, dispatcher=dispatcher, worker_id=worker_id) last = timeline["points"][-1] if timeline["points"] else {} state_counts = { "online_workers": last.get("online_workers", 0), "running_workers": last.get("running_workers", 0), "running_download_workers": last.get("running_download_workers", 0), "running_collect_workers": last.get("running_collect_workers", 0), "idle_workers": last.get("idle_workers", 0), "offline_workers": last.get("offline_workers", 0), "disabled_workers": last.get("disabled_workers", 0), } return { "date": date_str, "is_today": is_today, "controller_status": "running" if (is_today and self.current_session_id) else "stopped", "controller_up_seconds": up_seconds, "controller_down_seconds": down_seconds, "task_total": metrics["task_total"], "task_success": metrics["task_success"], "task_non_success": metrics["task_non_success"], "task_failed": metrics["task_failed"], "task_total_all": overall_totals["task_total"], "task_success_all": overall_success_total, "task_non_success_all": overall_totals["task_non_success"], "task_failed_all": overall_totals["task_failed"], "avg_download_duration_seconds": metrics["avg_download_duration_seconds"], "avg_collect_duration_seconds": metrics["avg_collect_duration_seconds"], "p50_download_duration_seconds": metrics["p50_download_duration_seconds"], "p90_download_duration_seconds": metrics["p90_download_duration_seconds"], "p50_collect_duration_seconds": metrics["p50_collect_duration_seconds"], "p90_collect_duration_seconds": metrics["p90_collect_duration_seconds"], **state_counts, "selected_worker_id": worker_id or "", "worker_options": self._build_worker_options(date_str, dispatcher=dispatcher), "generated_at": datetime.fromtimestamp(now_ts, SHANGHAI_TZ).strftime("%Y-%m-%d %H:%M:%S"), } def get_distributions( self, date_str: Optional[str] = None, worker_id: Optional[str] = None, dispatcher=None, ) -> Dict[str, Any]: date_str = date_str or _date_str_from_ts(time.time()) task_rows = self.repo.list_task_executions(date_str, worker_id=worker_id, limit=100000) _, severe_items = self._split_rows_by_quality(task_rows) download_values = [float(row["download_duration_seconds"]) for row in task_rows if row["download_duration_seconds"] is not None] collect_values = [float(row["collect_duration_seconds"]) for row in task_rows if row["collect_duration_seconds"] is not None] total_values = [float(row["total_duration_seconds"]) for row in task_rows if row["total_duration_seconds"] is not None] severe_values = [ float(row["failed_total_duration_seconds"] or row["total_duration_seconds"] or 0.0) for row in severe_items if (row["failed_total_duration_seconds"] is not None or row["total_duration_seconds"] is not None) ] bucket_edges = [60, 180, 300, 600, 1200, 1800, 3600] return { "date": date_str, "download_duration": _build_histogram(download_values, bucket_edges), "collect_duration": _build_histogram(collect_values, bucket_edges), "total_duration": _build_histogram(total_values, bucket_edges), "non_success_total_duration": _build_histogram(severe_values, bucket_edges), **self.get_stage_analysis(date_str, worker_id=worker_id, dispatcher=dispatcher, rows=task_rows), } def get_stage_analysis( self, date_str: Optional[str] = None, worker_id: Optional[str] = None, dispatcher=None, rows: Optional[List[sqlite3.Row]] = None, ) -> Dict[str, Any]: date_str = date_str or _date_str_from_ts(time.time()) rows = rows if rows is not None else self.repo.list_task_executions(date_str, worker_id=worker_id, limit=100000) worker_options = self._build_worker_options(date_str, dispatcher=dispatcher) worker_label_map = { item["worker_id"]: item.get("label") or item["worker_id"] for item in worker_options } qualified_items, severe_items, collection_quality = self._classify_rows_by_collection_quality(rows) def row_total(row: sqlite3.Row) -> float: return _task_row_total_duration(row) def summarize(items: List[sqlite3.Row]) -> List[Dict[str, Any]]: if not items: return [] total_sum = sum(max(row_total(item), 0.0) for item in items) result = [] for stage in STAGE_ORDER: key = f"{stage}_duration_seconds" stage_sum = sum(float(item[key] or 0.0) for item in items) result.append( { "stage": stage, "label": STAGE_LABELS.get(stage, stage), "avg_duration_seconds": round(stage_sum / len(items), 2), "share_percent": round((stage_sum / total_sum) * 100, 2) if total_sum > 0 else 0.0, "total_duration_seconds": round(stage_sum, 2), } ) return result qualified_total = round(sum(row_total(item) for item in qualified_items), 2) severe_total = round(sum(row_total(item) for item in severe_items), 2) total_runtime = qualified_total + severe_total time_composition = [ { "bucket": "qualified", "label": "合格", "duration_seconds": qualified_total, "share_percent": round((qualified_total / total_runtime) * 100, 2) if total_runtime > 0 else 0.0, }, { "bucket": "severe_restricted", "label": "严重受限", "duration_seconds": severe_total, "share_percent": round((severe_total / total_runtime) * 100, 2) if total_runtime > 0 else 0.0, }, ] domain_buckets: Dict[str, Dict[str, Any]] = {} for row in severe_items: failure_domain = row["failure_domain"] failure_subtype = row["failure_subtype"] if not failure_domain or not failure_subtype: failure_domain, failure_subtype = _classify_failure(row["error_type"], row["error_message"]) duration_seconds = row_total(row) reason_label = _humanize_failure_reason(row["error_type"], row["error_message"]) message_label = row["error_message"] or reason_label sample = _build_failure_task_sample(row) domain_entry = domain_buckets.setdefault( failure_domain, { "domain": failure_domain, "label": _humanize_failure_domain(failure_domain), "duration_seconds": 0.0, "task_count": 0, "_subtypes": {}, "_reasons": {}, "_messages": {}, "_tasks": [], }, ) domain_entry["duration_seconds"] += duration_seconds domain_entry["task_count"] += 1 domain_entry["_tasks"].append(sample) domain_reason = domain_entry["_reasons"].setdefault(reason_label, {"label": reason_label, "duration_seconds": 0.0, "task_count": 0}) domain_reason["duration_seconds"] += duration_seconds domain_reason["task_count"] += 1 domain_message = domain_entry["_messages"].setdefault(message_label, {"label": message_label, "duration_seconds": 0.0, "task_count": 0}) domain_message["duration_seconds"] += duration_seconds domain_message["task_count"] += 1 subtype_entry = domain_entry["_subtypes"].setdefault( failure_subtype, { "subtype": failure_subtype, "label": _humanize_failure_subtype(failure_domain, failure_subtype, row["error_type"], row["error_message"]), "duration_seconds": 0.0, "task_count": 0, "_reasons": {}, "_messages": {}, "_tasks": [], }, ) subtype_entry["duration_seconds"] += duration_seconds subtype_entry["task_count"] += 1 subtype_entry["_tasks"].append(sample) subtype_reason = subtype_entry["_reasons"].setdefault(reason_label, {"label": reason_label, "duration_seconds": 0.0, "task_count": 0}) subtype_reason["duration_seconds"] += duration_seconds subtype_reason["task_count"] += 1 subtype_message = subtype_entry["_messages"].setdefault(message_label, {"label": message_label, "duration_seconds": 0.0, "task_count": 0}) subtype_message["duration_seconds"] += duration_seconds subtype_message["task_count"] += 1 failure_domain_breakdown = [] failure_subtype_breakdown = [] for domain_entry in sorted(domain_buckets.values(), key=lambda item: item["duration_seconds"], reverse=True): domain_duration = float(domain_entry["duration_seconds"]) subtypes = [] for subtype_entry in sorted(domain_entry["_subtypes"].values(), key=lambda item: item["duration_seconds"], reverse=True): subtype_duration = float(subtype_entry["duration_seconds"]) subtype_reasons = sorted(subtype_entry["_reasons"].values(), key=lambda item: item["duration_seconds"], reverse=True) subtype_messages = sorted(subtype_entry["_messages"].values(), key=lambda item: item["duration_seconds"], reverse=True) subtype_tasks = sorted( subtype_entry["_tasks"], key=lambda item: ( float(item.get("failed_total_duration_seconds") or item.get("total_duration_seconds") or 0.0), item.get("task_ended_at", ""), ), reverse=True, ) reasons = [ { "label": item["label"], "duration_seconds": round(float(item["duration_seconds"]), 2), "task_count": int(item["task_count"]), "overall_share_percent": round((float(item["duration_seconds"]) / total_runtime) * 100, 2) if total_runtime > 0 else 0.0, "error_share_percent": round((float(item["duration_seconds"]) / severe_total) * 100, 2) if severe_total > 0 else 0.0, "domain_share_percent": round((float(item["duration_seconds"]) / domain_duration) * 100, 2) if domain_duration > 0 else 0.0, "subtype_share_percent": round((float(item["duration_seconds"]) / subtype_duration) * 100, 2) if subtype_duration > 0 else 0.0, } for item in subtype_reasons ] messages = [ { "label": item["label"], "message": item["label"], "duration_seconds": round(float(item["duration_seconds"]), 2), "task_count": int(item["task_count"]), "overall_share_percent": round((float(item["duration_seconds"]) / total_runtime) * 100, 2) if total_runtime > 0 else 0.0, "error_share_percent": round((float(item["duration_seconds"]) / severe_total) * 100, 2) if severe_total > 0 else 0.0, "domain_share_percent": round((float(item["duration_seconds"]) / domain_duration) * 100, 2) if domain_duration > 0 else 0.0, "subtype_share_percent": round((float(item["duration_seconds"]) / subtype_duration) * 100, 2) if subtype_duration > 0 else 0.0, } for item in subtype_messages ] subtype_item = { "domain": domain_entry["domain"], "subtype": subtype_entry["subtype"], "label": subtype_entry["label"], "duration_seconds": round(subtype_duration, 2), "task_count": int(subtype_entry["task_count"]), "avg_duration_seconds": round(subtype_duration / subtype_entry["task_count"], 2) if subtype_entry["task_count"] else 0.0, "overall_share_percent": round((subtype_duration / total_runtime) * 100, 2) if total_runtime > 0 else 0.0, "error_share_percent": round((subtype_duration / severe_total) * 100, 2) if severe_total > 0 else 0.0, "domain_share_percent": round((subtype_duration / domain_duration) * 100, 2) if domain_duration > 0 else 0.0, "reasons": reasons, "messages": messages, "tasks": subtype_tasks, } subtypes.append(subtype_item) failure_subtype_breakdown.append(subtype_item) domain_reasons = sorted(domain_entry["_reasons"].values(), key=lambda item: item["duration_seconds"], reverse=True) domain_messages = sorted(domain_entry["_messages"].values(), key=lambda item: item["duration_seconds"], reverse=True) domain_tasks = sorted( domain_entry["_tasks"], key=lambda item: ( float(item.get("failed_total_duration_seconds") or item.get("total_duration_seconds") or 0.0), item.get("task_ended_at", ""), ), reverse=True, ) failure_domain_breakdown.append( { "domain": domain_entry["domain"], "label": domain_entry["label"], "duration_seconds": round(domain_duration, 2), "task_count": int(domain_entry["task_count"]), "overall_share_percent": round((domain_duration / total_runtime) * 100, 2) if total_runtime > 0 else 0.0, "error_share_percent": round((domain_duration / severe_total) * 100, 2) if severe_total > 0 else 0.0, "share_percent": round((domain_duration / severe_total) * 100, 2) if severe_total > 0 else 0.0, "subtypes": subtypes, "reasons": [ { "label": item["label"], "duration_seconds": round(float(item["duration_seconds"]), 2), "task_count": int(item["task_count"]), "overall_share_percent": round((float(item["duration_seconds"]) / total_runtime) * 100, 2) if total_runtime > 0 else 0.0, "error_share_percent": round((float(item["duration_seconds"]) / severe_total) * 100, 2) if severe_total > 0 else 0.0, "domain_share_percent": round((float(item["duration_seconds"]) / domain_duration) * 100, 2) if domain_duration > 0 else 0.0, } for item in domain_reasons ], "messages": [ { "label": item["label"], "message": item["label"], "duration_seconds": round(float(item["duration_seconds"]), 2), "task_count": int(item["task_count"]), "overall_share_percent": round((float(item["duration_seconds"]) / total_runtime) * 100, 2) if total_runtime > 0 else 0.0, "error_share_percent": round((float(item["duration_seconds"]) / severe_total) * 100, 2) if severe_total > 0 else 0.0, "domain_share_percent": round((float(item["duration_seconds"]) / domain_duration) * 100, 2) if domain_duration > 0 else 0.0, } for item in domain_messages ], "tasks": domain_tasks, } ) failure_reasons = Counter() for row in severe_items: label = _humanize_failure_reason(row["error_type"], row["error_message"]) failure_reasons[label] += 1 if len(time_composition) > 1: time_composition[1]["details"] = [ { "domain": item["domain"], "label": item["label"], "duration_seconds": item["duration_seconds"], "overall_share_percent": item["overall_share_percent"], "error_share_percent": item["error_share_percent"], "task_count": item["task_count"], } for item in failure_domain_breakdown ] worker_stage_buckets: Dict[str, Dict[str, Any]] = {} total_collect_seconds = 0.0 for row in rows: current_worker_id = row["worker_id"] if not current_worker_id: continue download_seconds = float(row["download_duration_seconds"] or 0.0) collect_seconds = float(row["collect_duration_seconds"] or 0.0) if download_seconds <= 0 and collect_seconds <= 0: continue stage_total_seconds = download_seconds + collect_seconds bucket = worker_stage_buckets.setdefault( current_worker_id, { "worker_id": current_worker_id, "label": worker_label_map.get(current_worker_id, current_worker_id), "download_duration_seconds": 0.0, "collect_duration_seconds": 0.0, "stage_total_seconds": 0.0, "task_count": 0, "success_count": 0, "non_success_count": 0, }, ) bucket["download_duration_seconds"] += download_seconds bucket["collect_duration_seconds"] += collect_seconds bucket["stage_total_seconds"] += stage_total_seconds bucket["task_count"] += 1 if row["status"] == "success": bucket["success_count"] += 1 else: bucket["non_success_count"] += 1 total_collect_seconds += collect_seconds worker_collect_share_breakdown = [ { "worker_id": item["worker_id"], "label": item["label"], "download_duration_seconds": round(float(item["download_duration_seconds"]), 2), "collect_duration_seconds": round(float(item["collect_duration_seconds"]), 2), "stage_total_seconds": round(float(item["stage_total_seconds"]), 2), "collect_share_percent": round((float(item["collect_duration_seconds"]) / float(item["stage_total_seconds"])) * 100, 2) if item["stage_total_seconds"] > 0 else 0.0, "overall_collect_share_percent": round((float(item["collect_duration_seconds"]) / total_collect_seconds) * 100, 2) if total_collect_seconds > 0 else 0.0, "task_count": int(item["task_count"]), "success_count": int(item["success_count"]), "non_success_count": int(item["non_success_count"]), } for item in sorted( worker_stage_buckets.values(), key=lambda current: ( round((float(current["collect_duration_seconds"]) / float(current["stage_total_seconds"])) * 100, 6) if current["stage_total_seconds"] > 0 else 0.0, float(current["collect_duration_seconds"]), ), reverse=True, ) ] return { "success_stage_breakdown": summarize(qualified_items), "non_success_stage_breakdown": summarize(severe_items), "time_composition": time_composition, "daily_collection_overview": { "qualified_task_count": int(collection_quality["qualified_task_count"]), "qualified_success_task_count": int(collection_quality["qualified_success_task_count"]), "qualified_light_task_count": int(collection_quality["qualified_light_task_count"]), "non_success_task_count": int(collection_quality["non_success_task_count"]), "non_retryable_task_count": int(collection_quality["non_retryable_task_count"]), }, "runtime_state_breakdown": self._runtime_state_breakdown( date_str, dispatcher=dispatcher, worker_id=worker_id, ), "failure_domain_breakdown": failure_domain_breakdown, "failure_subtype_breakdown": failure_subtype_breakdown, "failure_reasons": [ {"label": label, "count": count} for label, count in failure_reasons.most_common() ], "success_task_count": len(qualified_items), "non_success_task_count": len(severe_items), "worker_collect_share_breakdown": worker_collect_share_breakdown, "total_collect_duration_seconds": round(total_collect_seconds, 2), "selected_worker_id": worker_id or "", "worker_options": worker_options, } def get_timeline( self, date_str: Optional[str] = None, dispatcher=None, bucket_minutes: int = MONITORING_TIMELINE_BUCKET_MINUTES, worker_id: Optional[str] = None, ) -> Dict[str, Any]: date_str = date_str or _date_str_from_ts(time.time()) day_start, day_end, _ = _day_bounds(date_str) intervals = self._build_worker_intervals(date_str, dispatcher=dispatcher, worker_id=worker_id) bucket_seconds = max(1, bucket_minutes) * 60 buckets: List[Dict[str, Any]] = [] bucket_start = day_start while bucket_start < day_end: buckets.append( { "timestamp": bucket_start, "label": datetime.fromtimestamp(bucket_start, SHANGHAI_TZ).strftime("%H:%M"), "online_workers": 0, "running_workers": 0, "running_download_workers": 0, "running_collect_workers": 0, "idle_workers": 0, "offline_workers": 0, "disabled_workers": 0, } ) bucket_start += bucket_seconds for worker_intervals in intervals.values(): for interval_start, interval_end, state in worker_intervals: for bucket in buckets: bucket_start = bucket["timestamp"] bucket_end = bucket_start + bucket_seconds overlap_start = max(interval_start, bucket_start) overlap_end = min(interval_end, bucket_end) if overlap_start >= overlap_end: continue weight = (overlap_end - overlap_start) / bucket_seconds if state in ONLINE_STATES: bucket["online_workers"] += weight if state in RUNNING_STATES: bucket["running_workers"] += weight if state == "running_download": bucket["running_download_workers"] += weight elif state == "running_collect": bucket["running_collect_workers"] += weight elif state == "idle_waiting_task": bucket["idle_workers"] += weight elif state == "disabled": bucket["disabled_workers"] += weight else: bucket["offline_workers"] += weight for bucket in buckets: for key in ( "online_workers", "running_workers", "running_download_workers", "running_collect_workers", "idle_workers", "offline_workers", "disabled_workers", ): bucket[key] = round(float(bucket[key]), 2) return {"date": date_str, "bucket_minutes": bucket_minutes, "points": buckets} def get_worker_stats( self, date_str: Optional[str] = None, dispatcher=None, ) -> List[Dict[str, Any]]: now_ts = time.time() date_str = date_str or _date_str_from_ts(now_ts) is_today = _date_str_from_ts(now_ts) == date_str seconds_by_worker = self._summarize_worker_seconds(date_str, dispatcher=dispatcher) task_rows = self.repo.list_task_executions(date_str, limit=100000) task_stats: Dict[str, Dict[str, Any]] = defaultdict( lambda: { "task_count": 0, "success_count": 0, "failed_count": 0, "download_values": [], "collect_values": [], "last_task_at": None, } ) for row in task_rows: entry = task_stats[row["worker_id"]] entry["task_count"] += 1 if row["status"] == "success": entry["success_count"] += 1 elif row["status"] == "failed": entry["failed_count"] += 1 if row["download_duration_seconds"] is not None: entry["download_values"].append(float(row["download_duration_seconds"])) if row["collect_duration_seconds"] is not None: entry["collect_values"].append(float(row["collect_duration_seconds"])) last_task_at = row["task_ended_at"] or row["task_started_at"] if last_task_at is not None: if entry["last_task_at"] is None or float(last_task_at) > entry["last_task_at"]: entry["last_task_at"] = float(last_task_at) current_workers = {} if dispatcher is not None and is_today: current_workers = {worker.get("worker_id"): worker for worker in dispatcher.get_dashboard_workers()} results: List[Dict[str, Any]] = [] for worker_id in sorted(set(self._resolve_worker_ids(date_str, dispatcher=dispatcher))): seconds = seconds_by_worker.get(worker_id, Counter()) metrics = task_stats.get(worker_id, {}) current_worker = current_workers.get(worker_id, {}) current_state = ( current_worker.get("state") or self._get_current_state(worker_id).get("state") or ("offline" if not current_worker.get("online") else "idle_waiting_task") ) if is_today and current_worker and not current_worker.get("online"): current_state = "offline" last_task_at = metrics.get("last_task_at") results.append( { "worker_id": worker_id, "ip_address": current_worker.get("ip_address") or worker_id.split("_", 1)[0], "status": DISPLAY_STATE_LABELS.get(current_state, current_state), "running_seconds": round(seconds.get("running_download", 0.0) + seconds.get("running_collect", 0.0), 2), "idle_waiting_seconds": round(seconds.get("idle_waiting_task", 0.0), 2), "offline_seconds": round(seconds.get("offline", 0.0), 2), "disabled_seconds": round(seconds.get("disabled", 0.0), 2), "task_count": metrics.get("task_count", 0), "success_count": metrics.get("success_count", 0), "failed_count": metrics.get("failed_count", 0), "avg_download_duration_seconds": round(sum(metrics.get("download_values", [])) / len(metrics.get("download_values", [])), 2) if metrics.get("download_values") else 0.0, "avg_collect_duration_seconds": round(sum(metrics.get("collect_values", [])) / len(metrics.get("collect_values", [])), 2) if metrics.get("collect_values") else 0.0, "last_task_at": datetime.fromtimestamp(last_task_at, SHANGHAI_TZ).strftime("%Y-%m-%d %H:%M:%S") if last_task_at else "--", } ) return results def get_task_rows( self, date_str: Optional[str] = None, worker_id: Optional[str] = None, limit: int = 200, ) -> List[Dict[str, Any]]: date_str = date_str or _date_str_from_ts(time.time()) rows = self.repo.list_task_executions(date_str, worker_id=worker_id, limit=limit) return [_build_failure_task_sample(row) for row in rows]