import asyncio import time from collections import defaultdict from datetime import datetime, timedelta, timezone from typing import Any from sqlalchemy import select from sqlalchemy.orm import selectinload from app.config import settings from app.models.dashboard import ( DashboardCheckDefinition, DashboardDeviceCheckHistory, DashboardDeviceLiveState, DashboardMonitorRunState, DashboardObjectCheckOverride, DashboardObjectMonitorPolicy, DashboardTagMonitorPolicy, ) from app.models.device import Device from app.models.object import Object from app.models.rack import Rack from app.models.tag import Tag from app.plugins.registry import action_registry from app.services.database import AsyncSessionLocal from app.services.sse import SSEEvent, sse_manager DASHBOARD_MONITOR_CHANNEL = "__dashboard_monitor__" DEFAULT_CHECKS = [ { "check_key": "ping", "enabled": True, "interval_sec_default": 300, "plugin_action": "ping", "categories": [], "timeout_sec": 15, "concurrency": 40, }, { "check_key": "disk_usage", "enabled": True, "interval_sec_default": 3600, "plugin_action": "disk_usage", "categories": ["main_server", "vm"], "timeout_sec": 45, "concurrency": 10, }, ] WINDOW_SECONDS = { "24h": 24 * 3600, "3d": 3 * 24 * 3600, "7d": 7 * 24 * 3600, } WINDOW_BUCKET_SECONDS = { "24h": 3600, "3d": 7200, "7d": 21600, } _RUN_LOCK = asyncio.Lock() _MANUAL_TASK: asyncio.Task | None = None async def _noop_emit(_: str, __: str = "info") -> None: return def _scope_to_dict(scope: dict[str, Any] | None) -> dict[str, Any]: if not scope: return {} return { "city": scope.get("city"), "client": scope.get("client"), "object_ids": list(scope.get("object_ids") or []), "tag_ids": list(scope.get("tag_ids") or []), } def _window_seconds(window: str) -> int: return WINDOW_SECONDS.get(window, WINDOW_SECONDS["24h"]) def _bucket_seconds(window: str) -> int: return WINDOW_BUCKET_SECONDS.get(window, WINDOW_BUCKET_SECONDS["24h"]) def _is_object_in_scope(obj: Object, scope: dict[str, Any]) -> bool: city = scope.get("city") if city and obj.city != city: return False client = scope.get("client") if client and obj.client != client: return False object_ids = list(scope.get("object_ids") or []) if object_ids and obj.id not in object_ids: return False tag_ids = set(scope.get("tag_ids") or []) if tag_ids: obj_tag_ids = {t.id for t in obj.tags} if obj_tag_ids.isdisjoint(tag_ids): return False return True async def ensure_dashboard_bootstrap() -> None: async with AsyncSessionLocal() as db: async with db.begin(): existing = { c.check_key: c for c in ( await db.execute(select(DashboardCheckDefinition)) ).scalars().all() } for payload in DEFAULT_CHECKS: row = existing.get(payload["check_key"]) if row is None: db.add(DashboardCheckDefinition(**payload)) continue if row.check_key == "disk_usage" and row.plugin_action == "debian_system_info": row.plugin_action = "disk_usage" if not row.plugin_action: row.plugin_action = payload["plugin_action"] if not row.categories: row.categories = payload["categories"] if row.timeout_sec <= 0: row.timeout_sec = payload["timeout_sec"] if row.concurrency <= 0: row.concurrency = payload["concurrency"] run_state = await db.get(DashboardMonitorRunState, 1) if run_state is None: db.add(DashboardMonitorRunState(id=1, status="idle")) async def _get_run_state(db) -> DashboardMonitorRunState: row = await db.get(DashboardMonitorRunState, 1) if row is None: row = DashboardMonitorRunState(id=1, status="idle") db.add(row) await db.flush() return row async def get_monitor_status() -> DashboardMonitorRunState: async with AsyncSessionLocal() as db: return await _get_run_state(db) async def _load_objects_with_scope(db, scope: dict[str, Any] | None) -> list[Object]: scope = scope or {} query = ( select(Object) .options(selectinload(Object.tags)) .where(Object.is_active == True) .order_by(Object.city, Object.name) ) if scope.get("city"): query = query.where(Object.city == scope["city"]) if scope.get("client"): query = query.where(Object.client == scope["client"]) if scope.get("object_ids"): query = query.where(Object.id.in_(scope["object_ids"])) if scope.get("tag_ids"): query = query.where(Object.tags.any(Tag.id.in_(scope["tag_ids"]))) result = await db.execute(query) return list(result.scalars().all()) async def _load_object_with_tags(db, obj_id: int) -> Object | None: row = ( await db.execute( select(Object) .options(selectinload(Object.tags)) .where(Object.id == obj_id, Object.is_active == True) ) ).scalar_one_or_none() return row def _is_object_allowed( obj: Object, object_modes: dict[int, str], tag_modes: dict[int, bool], ) -> tuple[bool, str, str]: mode = object_modes.get(obj.id, "inherit") if mode == "include": return True, mode, "object_include_policy" if mode == "exclude": return False, mode, "object_exclude_policy" tag_values = [tag_modes[t.id] for t in obj.tags if t.id in tag_modes] if not tag_values: return True, mode, "inherit_default" if any(tag_values): return True, mode, "tag_policy_enabled" return False, mode, "tag_policy_disabled" async def _build_object_policy_map(db, objects: list[Object]) -> dict[int, tuple[bool, str, str]]: object_ids = [o.id for o in objects] if not object_ids: return {} obj_policy_rows = ( await db.execute( select(DashboardObjectMonitorPolicy).where(DashboardObjectMonitorPolicy.object_id.in_(object_ids)) ) ).scalars().all() object_modes = {row.object_id: row.mode for row in obj_policy_rows} tag_ids = {t.id for obj in objects for t in obj.tags} tag_modes: dict[int, bool] = {} if tag_ids: tag_rows = ( await db.execute( select(DashboardTagMonitorPolicy).where(DashboardTagMonitorPolicy.tag_id.in_(tag_ids)) ) ).scalars().all() tag_modes = {row.tag_id: row.enabled for row in tag_rows} return {obj.id: _is_object_allowed(obj, object_modes, tag_modes) for obj in objects} def _pct_from_value(value: Any) -> float | None: if value is None: return None if isinstance(value, (int, float)): return float(value) text = str(value).strip() if not text: return None cleaned = text.replace("%", "") try: return float(cleaned) except ValueError: return None def _derive_status(check_key: str, success: bool, data: dict[str, Any]) -> tuple[str, dict[str, Any] | None]: if check_key == "ping": ping_ms = data.get("ping_ms") return ("online" if success else "offline", {"ping_ms": ping_ms}) if check_key == "disk_usage": if not success: return "failed", None if isinstance(data.get("max_pct"), (int, float)): max_pct = float(data.get("max_pct")) status = "warn" if max_pct >= 90 else "ok" return status, data disks = data.get("disks") or [] max_pct = 0.0 top_mount = None for disk in disks: pct = _pct_from_value(disk.get("pct")) if pct is None: continue if pct >= max_pct: max_pct = pct top_mount = disk.get("mount") status = "warn" if max_pct >= 90 else "ok" return status, { "max_pct": max_pct, "max_mount": top_mount, "disks": disks, } return ("ok" if success else "failed", data or None) async def _upsert_live_state( db, *, device_id: int, check_key: str, status: str, value_json: dict[str, Any] | None, error: str | None, checked_at: datetime, duration_ms: int, ) -> None: row = ( await db.execute( select(DashboardDeviceLiveState).where( DashboardDeviceLiveState.device_id == device_id, DashboardDeviceLiveState.check_key == check_key, ) ) ).scalar_one_or_none() if row is None: db.add( DashboardDeviceLiveState( device_id=device_id, check_key=check_key, status=status, value_json=value_json, error=error, checked_at=checked_at, duration_ms=duration_ms, source="dashboard_monitor", ) ) return row.status = status row.value_json = value_json row.error = error row.checked_at = checked_at row.duration_ms = duration_ms row.source = "dashboard_monitor" async def _append_check_history( db, *, device_id: int, check_key: str, status: str, value_json: dict[str, Any] | None, error: str | None, checked_at: datetime, duration_ms: int, ) -> None: ping_ms = None if check_key == "ping" and value_json is not None: raw = value_json.get("ping_ms") if raw is not None: try: ping_ms = int(raw) except (TypeError, ValueError): ping_ms = None db.add( DashboardDeviceCheckHistory( device_id=device_id, check_key=check_key, status=status, ping_ms=ping_ms, checked_at=checked_at, duration_ms=duration_ms, error=error, source="dashboard_monitor", ) ) async def _run_check_on_device( sem: asyncio.Semaphore, check: DashboardCheckDefinition, device: Device, obj: Object, ) -> tuple[bool, str, dict[str, Any] | None, str | None, datetime, int]: async with sem: action = action_registry.get(check.plugin_action) now = datetime.now(timezone.utc) if action is None: return False, "failed", None, f"Unknown action: {check.plugin_action}", now, 0 started = time.monotonic() try: result = await action.execute(device, obj, {}, _noop_emit) except Exception as exc: duration_ms = int((time.monotonic() - started) * 1000) return False, "failed", None, str(exc), now, duration_ms duration_ms = int((time.monotonic() - started) * 1000) status, value_json = _derive_status(check.check_key, result.success, result.data or {}) error = result.error or None return result.success, status, value_json, error, datetime.now(timezone.utc), duration_ms def _effective_check_settings( check: DashboardCheckDefinition, override: DashboardObjectCheckOverride | None, ) -> tuple[bool, int, list[str]]: enabled = check.enabled interval = check.interval_sec_default categories = list(check.categories or []) if override is not None: if override.enabled_override is not None: enabled = override.enabled_override if override.interval_sec_override is not None: interval = override.interval_sec_override if override.categories_override is not None: categories = list(override.categories_override) return enabled, interval, categories async def run_dashboard_monitor_cycle( *, run_type: str, scope: dict[str, Any] | None = None, force: bool = False, ) -> dict[str, Any] | None: if _RUN_LOCK.locked(): return None async with _RUN_LOCK: await ensure_dashboard_bootstrap() started_at = datetime.now(timezone.utc) async with AsyncSessionLocal() as db: async with db.begin(): run_state = await _get_run_state(db) run_state.status = "running" run_state.run_type = run_type run_state.scope = _scope_to_dict(scope) run_state.started_at = started_at run_state.finished_at = None run_state.last_error = None counts = { "objects_in_scope": 0, "objects_allowed": 0, "devices_checked": 0, "checks_executed": 0, "failed": 0, "skipped_not_due": 0, "skipped_policy": 0, } try: async with AsyncSessionLocal() as db: objects = await _load_objects_with_scope(db, scope) counts["objects_in_scope"] = len(objects) object_policy_map = await _build_object_policy_map(db, objects) allowed_object_ids = [oid for oid, (allowed, _mode, _reason) in object_policy_map.items() if allowed] counts["objects_allowed"] = len(allowed_object_ids) counts["skipped_policy"] = max(0, len(objects) - len(allowed_object_ids)) if not allowed_object_ids: async with AsyncSessionLocal() as dbw: async with dbw.begin(): run_state = await _get_run_state(dbw) run_state.status = "idle" run_state.finished_at = datetime.now(timezone.utc) run_state.last_success_at = datetime.now(timezone.utc) run_state.last_counts = counts return counts device_rows = ( await db.execute( select(Device) .options( selectinload(Device.object).selectinload(Object.ssh_credentials), selectinload(Device.object).selectinload(Object.category_credentials), ) .where(Device.is_active == True, Device.object_id.in_(allowed_object_ids)) ) ).scalars().all() checks = (await db.execute(select(DashboardCheckDefinition))).scalars().all() overrides = ( await db.execute( select(DashboardObjectCheckOverride).where( DashboardObjectCheckOverride.object_id.in_(allowed_object_ids) ) ) ).scalars().all() override_map = {(o.object_id, o.check_key): o for o in overrides} for check in checks: check_devices = [] for device in device_rows: override = override_map.get((device.object_id, check.check_key)) enabled, interval_sec, categories = _effective_check_settings(check, override) if not enabled: continue if categories and device.category not in categories: continue check_devices.append((device, interval_sec)) if not check_devices: continue state_rows = ( await db.execute( select(DashboardDeviceLiveState).where( DashboardDeviceLiveState.check_key == check.check_key, DashboardDeviceLiveState.device_id.in_([d.id for d, _ in check_devices]), ) ) ).scalars().all() state_map = {row.device_id: row for row in state_rows} due_devices: list[Device] = [] for device, interval_sec in check_devices: if force: due_devices.append(device) continue state = state_map.get(device.id) if state is None: due_devices.append(device) continue age = (started_at - state.checked_at).total_seconds() if age >= interval_sec: due_devices.append(device) else: counts["skipped_not_due"] += 1 if not due_devices: continue sem = asyncio.Semaphore(max(1, check.concurrency)) results = await asyncio.gather( *[_run_check_on_device(sem, check, device, device.object) for device in due_devices], return_exceptions=True, ) async with AsyncSessionLocal() as dbw: async with dbw.begin(): for device, result in zip(due_devices, results): if isinstance(result, Exception): counts["failed"] += 1 counts["checks_executed"] += 1 checked_at = datetime.now(timezone.utc) await _upsert_live_state( dbw, device_id=device.id, check_key=check.check_key, status="failed", value_json=None, error=str(result), checked_at=checked_at, duration_ms=0, ) await _append_check_history( dbw, device_id=device.id, check_key=check.check_key, status="failed", value_json=None, error=str(result), checked_at=checked_at, duration_ms=0, ) continue success, status, value_json, error, checked_at, duration_ms = result counts["checks_executed"] += 1 counts["devices_checked"] += 1 if not success: counts["failed"] += 1 await _upsert_live_state( dbw, device_id=device.id, check_key=check.check_key, status=status, value_json=value_json, error=error, checked_at=checked_at, duration_ms=duration_ms, ) await _append_check_history( dbw, device_id=device.id, check_key=check.check_key, status=status, value_json=value_json, error=error, checked_at=checked_at, duration_ms=duration_ms, ) if settings.DASHBOARD_MONITOR_SYNC_DEVICE_STATUS and check.check_key == "ping": device_row = await dbw.get(Device, device.id) if device_row is not None: device_row.status = "online" if status == "online" else "offline" device_row.last_ping_at = checked_at if status == "online": device_row.last_seen = checked_at ping_ms = (value_json or {}).get("ping_ms") if ping_ms is not None: device_row.last_ping_ms = int(ping_ms) await sse_manager.publish( SSEEvent( type="dashboard_monitor_update", task_id=DASHBOARD_MONITOR_CHANNEL, data={ "check_key": check.check_key, "processed": len(due_devices), "ts": datetime.now(timezone.utc).isoformat(), }, ) ) async with AsyncSessionLocal() as db: async with db.begin(): run_state = await _get_run_state(db) run_state.status = "idle" run_state.finished_at = datetime.now(timezone.utc) run_state.last_success_at = datetime.now(timezone.utc) run_state.last_counts = counts await sse_manager.publish( SSEEvent( type="dashboard_monitor_cycle_done", task_id=DASHBOARD_MONITOR_CHANNEL, data={ "run_type": run_type, "scope": _scope_to_dict(scope), "counts": counts, "ts": datetime.now(timezone.utc).isoformat(), }, ) ) return counts except Exception as exc: async with AsyncSessionLocal() as db: async with db.begin(): run_state = await _get_run_state(db) run_state.status = "idle" run_state.finished_at = datetime.now(timezone.utc) run_state.last_error = str(exc) run_state.last_counts = counts await sse_manager.publish( SSEEvent( type="dashboard_monitor_cycle_failed", task_id=DASHBOARD_MONITOR_CHANNEL, data={ "run_type": run_type, "scope": _scope_to_dict(scope), "error": str(exc), "ts": datetime.now(timezone.utc).isoformat(), }, ) ) raise async def request_manual_run(scope: dict[str, Any] | None, force: bool) -> tuple[bool, bool, str]: global _MANUAL_TASK if _RUN_LOCK.locked(): return False, True, "Dashboard monitor is already running" async def _runner() -> None: try: await run_dashboard_monitor_cycle(run_type="manual", scope=scope, force=force) finally: global _MANUAL_TASK _MANUAL_TASK = None _MANUAL_TASK = asyncio.create_task(_runner()) return True, True, "Manual run started" async def list_checks_with_overrides() -> tuple[list[DashboardCheckDefinition], list[DashboardObjectCheckOverride]]: await ensure_dashboard_bootstrap() async with AsyncSessionLocal() as db: checks = (await db.execute(select(DashboardCheckDefinition).order_by(DashboardCheckDefinition.check_key))).scalars().all() overrides = ( await db.execute( select(DashboardObjectCheckOverride).order_by( DashboardObjectCheckOverride.object_id, DashboardObjectCheckOverride.check_key, ) ) ).scalars().all() return list(checks), list(overrides) async def patch_checks(items: list[dict[str, Any]]) -> None: if not items: return await ensure_dashboard_bootstrap() async with AsyncSessionLocal() as db: async with db.begin(): rows = ( await db.execute( select(DashboardCheckDefinition).where( DashboardCheckDefinition.check_key.in_([i["check_key"] for i in items]) ) ) ).scalars().all() row_map = {r.check_key: r for r in rows} for item in items: row = row_map.get(item["check_key"]) if row is None: continue for field in ("enabled", "interval_sec_default", "categories", "timeout_sec", "concurrency"): if field in item and item[field] is not None: setattr(row, field, item[field]) async def patch_object_overrides(items: list[dict[str, Any]]) -> None: async with AsyncSessionLocal() as db: async with db.begin(): for item in items: row = ( await db.execute( select(DashboardObjectCheckOverride).where( DashboardObjectCheckOverride.object_id == item["object_id"], DashboardObjectCheckOverride.check_key == item["check_key"], ) ) ).scalar_one_or_none() has_payload = any( item.get(k) is not None for k in ("enabled_override", "interval_sec_override", "categories_override") ) if not has_payload: if row is not None: await db.delete(row) continue if row is None: row = DashboardObjectCheckOverride( object_id=item["object_id"], check_key=item["check_key"], ) db.add(row) row.enabled_override = item.get("enabled_override") row.interval_sec_override = item.get("interval_sec_override") row.categories_override = item.get("categories_override") async def get_policy() -> tuple[list[DashboardObjectMonitorPolicy], list[DashboardTagMonitorPolicy]]: async with AsyncSessionLocal() as db: obj = (await db.execute(select(DashboardObjectMonitorPolicy).order_by(DashboardObjectMonitorPolicy.object_id))).scalars().all() tags = (await db.execute(select(DashboardTagMonitorPolicy).order_by(DashboardTagMonitorPolicy.tag_id))).scalars().all() return list(obj), list(tags) async def patch_object_policy(items: list[dict[str, Any]]) -> None: async with AsyncSessionLocal() as db: async with db.begin(): for item in items: row = await db.get(DashboardObjectMonitorPolicy, item["object_id"]) if row is None: row = DashboardObjectMonitorPolicy(object_id=item["object_id"], mode=item["mode"]) db.add(row) else: row.mode = item["mode"] async def patch_tag_policy(items: list[dict[str, Any]]) -> None: async with AsyncSessionLocal() as db: async with db.begin(): for item in items: row = await db.get(DashboardTagMonitorPolicy, item["tag_id"]) if row is None: row = DashboardTagMonitorPolicy(tag_id=item["tag_id"], enabled=item["enabled"]) db.add(row) else: row.enabled = item["enabled"] def _calc_health( checks: dict[str, dict[str, int]], *, device_count: int, ping_interval_sec: int | None, ping_max_age_sec: int | None, ) -> float | None: ping = checks.get("ping") if not ping: return None total = sum(ping.values()) or device_count if total == 0: return None online = ping.get("online", 0) unknown = ping.get("unknown", 0) offline = ping.get("offline", 0) known = max(0, total - unknown) availability = online / total coverage = known / total freshness = 0.4 if ping_interval_sec and ping_interval_sec > 0 and ping_max_age_sec is not None: if ping_max_age_sec <= ping_interval_sec: freshness = 1.0 elif ping_max_age_sec <= ping_interval_sec * 3: freshness = 0.6 else: freshness = 0.2 warn = checks.get("disk_usage", {}).get("warn", 0) failed = checks.get("disk_usage", {}).get("failed", 0) disk_penalty = min(25.0, warn * 1.5 + failed * 4.0) unknown_penalty = min(15.0, (unknown / total) * 20.0) offline_penalty = min(20.0, (offline / total) * 25.0) base = 100.0 * (0.65 * availability + 0.20 * freshness + 0.15 * coverage) score = base - disk_penalty - unknown_penalty - offline_penalty return round(max(0.0, score), 1) def _build_check_summary( *, now: datetime, obj_devices: list[Device], checks: list[DashboardCheckDefinition], state_map: dict[tuple[int, str], DashboardDeviceLiveState], ) -> dict[str, dict[str, Any]]: check_summary: dict[str, dict[str, Any]] = {} for check in checks: counts: dict[str, int] = {} last_checked = None max_age = None for dev in obj_devices: state = state_map.get((dev.id, check.check_key)) if state is None: counts["unknown"] = counts.get("unknown", 0) + 1 continue counts[state.status] = counts.get(state.status, 0) + 1 if last_checked is None or state.checked_at > last_checked: last_checked = state.checked_at age_sec = int((now - state.checked_at).total_seconds()) if max_age is None or age_sec > max_age: max_age = age_sec check_summary[check.check_key] = { "status_counts": counts, "last_checked_at": last_checked, "max_age_sec": max_age, } return check_summary def _to_object_summary( *, now: datetime, obj: Object, allowed: bool, monitor_mode: str, inclusion_reason: str, checks: list[DashboardCheckDefinition], obj_devices: list[Device], state_map: dict[tuple[int, str], DashboardDeviceLiveState], ) -> dict[str, Any]: check_summary = _build_check_summary(now=now, obj_devices=obj_devices, checks=checks, state_map=state_map) check_interval_map = {c.check_key: c.interval_sec_default for c in checks} health = None if allowed: ping_meta = check_summary.get("ping", {}) health = _calc_health( {k: v["status_counts"] for k, v in check_summary.items()}, device_count=len(obj_devices), ping_interval_sec=check_interval_map.get("ping"), ping_max_age_sec=ping_meta.get("max_age_sec"), ) return { "object_id": obj.id, "object_name": obj.name, "city": obj.city, "client": obj.client, "allowed": allowed, "monitor_mode": monitor_mode, "inclusion_reason": inclusion_reason, "device_count": len(obj_devices), "checks": check_summary, "health_score": health, } def _device_noc_item( *, now: datetime, device: Device, rack_name: str | None, ping_state: DashboardDeviceLiveState | None, disk_state: DashboardDeviceLiveState | None, ping_interval_sec: int, ) -> dict[str, Any]: ping_age_sec = None ping_checked_at = None ping_status = "unknown" ping_fresh = False if ping_state is not None: ping_status = ping_state.status ping_checked_at = ping_state.checked_at ping_age_sec = int((now - ping_state.checked_at).total_seconds()) ping_fresh = ping_age_sec <= ping_interval_sec * 3 disk_status = disk_state.status if disk_state is not None else None disk_max_pct = None if disk_state and isinstance(disk_state.value_json, dict): val = disk_state.value_json.get("max_pct") if isinstance(val, (int, float)): disk_max_pct = float(val) has_problem = ( ping_status == "offline" or (ping_status == "unknown" and not ping_fresh) or disk_status in ("warn", "failed") ) return { "device_id": device.id, "object_id": device.object_id, "rack_id": device.rack_id, "rack_name": rack_name, "ip": device.ip, "hostname": device.hostname, "category": device.category, "role": device.role, "status": device.status, "monitor_enabled": device.monitor_enabled, "last_seen": device.last_seen, "last_ping_ms": device.last_ping_ms, "ping_status": ping_status, "ping_checked_at": ping_checked_at, "ping_age_sec": ping_age_sec, "ping_fresh": ping_fresh, "disk_status": disk_status, "disk_max_pct": disk_max_pct, "has_problem": has_problem, } async def get_home_summary(scope: dict[str, Any] | None) -> dict[str, Any]: await ensure_dashboard_bootstrap() now = datetime.now(timezone.utc) scope = _scope_to_dict(scope) async with AsyncSessionLocal() as db: objects = await _load_objects_with_scope(db, scope) policy_map = await _build_object_policy_map(db, objects) object_ids = [obj.id for obj in objects] if object_ids: device_rows = ( await db.execute( select(Device).where(Device.is_active == True, Device.object_id.in_(object_ids)) ) ).scalars().all() else: device_rows = [] devices_by_object: dict[int, list[Device]] = {} for d in device_rows: devices_by_object.setdefault(d.object_id, []).append(d) state_rows = [] if device_rows: state_rows = ( await db.execute( select(DashboardDeviceLiveState).where( DashboardDeviceLiveState.device_id.in_([d.id for d in device_rows]) ) ) ).scalars().all() state_map: dict[tuple[int, str], DashboardDeviceLiveState] = { (row.device_id, row.check_key): row for row in state_rows } checks = (await db.execute(select(DashboardCheckDefinition).order_by(DashboardCheckDefinition.check_key))).scalars().all() summaries = [] for obj in objects: allowed, mode, reason = policy_map.get(obj.id, (True, "inherit", "inherit_default")) obj_devices = devices_by_object.get(obj.id, []) summaries.append( _to_object_summary( now=now, obj=obj, allowed=allowed, monitor_mode=mode, inclusion_reason=reason, checks=checks, obj_devices=obj_devices, state_map=state_map, ) ) included_objects = [s for s in summaries if s["allowed"]] excluded_objects = [s for s in summaries if not s["allowed"]] problems = sorted( included_objects, key=lambda x: x["health_score"] if x["health_score"] is not None else 999, )[:10] totals = { "objects": len(objects), "allowed_objects": len(included_objects), "excluded_objects": len(excluded_objects), "devices": len(device_rows), } return { "generated_at": now, "scope": scope, "totals": totals, "checks_meta": checks, "objects": summaries, "problem_objects": problems, "included_count": len(included_objects), "excluded_count": len(excluded_objects), "included_objects": included_objects, "excluded_objects": excluded_objects, } async def get_object_devices(object_id: int, scope: dict[str, Any] | None) -> dict[str, Any] | None: await ensure_dashboard_bootstrap() now = datetime.now(timezone.utc) scope = _scope_to_dict(scope) async with AsyncSessionLocal() as db: obj = await _load_object_with_tags(db, object_id) if obj is None: return None device_rows = ( await db.execute( select(Device) .where(Device.object_id == object_id, Device.is_active == True) .order_by(Device.category, Device.ip) ) ).scalars().all() rack_name_by_id: dict[int, str] = {} rack_ids = {d.rack_id for d in device_rows if d.rack_id is not None} if rack_ids: rack_rows = ( await db.execute( select(Rack.id, Rack.name).where(Rack.id.in_(rack_ids)) ) ).all() rack_name_by_id = {rid: name for rid, name in rack_rows} state_rows = [] if device_rows: state_rows = ( await db.execute( select(DashboardDeviceLiveState).where( DashboardDeviceLiveState.device_id.in_([d.id for d in device_rows]), DashboardDeviceLiveState.check_key.in_(["ping", "disk_usage"]), ) ) ).scalars().all() check_rows = ( await db.execute( select(DashboardCheckDefinition).where(DashboardCheckDefinition.check_key == "ping") ) ).scalars().all() ping_interval = check_rows[0].interval_sec_default if check_rows else 300 state_map = {(row.device_id, row.check_key): row for row in state_rows} rows = [] for device in device_rows: rows.append( _device_noc_item( now=now, device=device, rack_name=rack_name_by_id.get(device.rack_id) if device.rack_id else None, ping_state=state_map.get((device.id, "ping")), disk_state=state_map.get((device.id, "disk_usage")), ping_interval_sec=ping_interval, ) ) rows.sort( key=lambda r: ( not r["has_problem"], 0 if r["ping_status"] == "offline" else 1, r["ip"], ) ) return { "generated_at": now, "scope": scope, "object_id": object_id, "devices": rows, } async def get_object_summary(object_id: int, scope: dict[str, Any] | None) -> dict[str, Any] | None: await ensure_dashboard_bootstrap() now = datetime.now(timezone.utc) scope = _scope_to_dict(scope) async with AsyncSessionLocal() as db: obj = await _load_object_with_tags(db, object_id) if obj is None: return None policy_map = await _build_object_policy_map(db, [obj]) allowed, mode, reason = policy_map.get(obj.id, (True, "inherit", "inherit_default")) if not _is_object_in_scope(obj, scope): allowed = False reason = "scope_filtered" checks = (await db.execute(select(DashboardCheckDefinition).order_by(DashboardCheckDefinition.check_key))).scalars().all() device_rows = ( await db.execute( select(Device).where(Device.object_id == object_id, Device.is_active == True) ) ).scalars().all() rack_name_by_id: dict[int, str] = {} rack_ids = {d.rack_id for d in device_rows if d.rack_id is not None} if rack_ids: rack_rows = ( await db.execute( select(Rack.id, Rack.name).where(Rack.id.in_(rack_ids)) ) ).all() rack_name_by_id = {rid: name for rid, name in rack_rows} state_rows = [] if device_rows: state_rows = ( await db.execute( select(DashboardDeviceLiveState).where( DashboardDeviceLiveState.device_id.in_([d.id for d in device_rows]) ) ) ).scalars().all() state_map: dict[tuple[int, str], DashboardDeviceLiveState] = { (row.device_id, row.check_key): row for row in state_rows } summary = _to_object_summary( now=now, obj=obj, allowed=allowed, monitor_mode=mode, inclusion_reason=reason, checks=checks, obj_devices=device_rows, state_map=state_map, ) ping_interval = next((c.interval_sec_default for c in checks if c.check_key == "ping"), 300) top_problem_devices = [ _device_noc_item( now=now, device=device, rack_name=rack_name_by_id.get(device.rack_id) if device.rack_id else None, ping_state=state_map.get((device.id, "ping")), disk_state=state_map.get((device.id, "disk_usage")), ping_interval_sec=ping_interval, ) for device in device_rows ] top_problem_devices.sort( key=lambda r: ( not r["has_problem"], 0 if r["ping_status"] == "offline" else 1, r["ip"], ) ) totals = { "devices": len(device_rows), "problem_devices": sum(1 for d in top_problem_devices if d["has_problem"]), "offline": sum(1 for d in top_problem_devices if d["ping_status"] == "offline"), "warn": sum(1 for d in top_problem_devices if d["disk_status"] in ("warn", "failed")), } return { "generated_at": now, "scope": scope, "object": summary, "totals": totals, "checks_meta": checks, "top_problem_devices": top_problem_devices[:12], } def _build_bucket_points(start_at: datetime, end_at: datetime, bucket_sec: int) -> list[tuple[datetime, datetime]]: buckets = [] cur = start_at while cur < end_at: nxt = min(end_at, cur + timedelta(seconds=bucket_sec)) buckets.append((cur, nxt)) cur = nxt return buckets def _series_from_rows( rows: list[DashboardDeviceCheckHistory], *, start_at: datetime, end_at: datetime, bucket_sec: int, ) -> list[dict[str, Any]]: buckets = _build_bucket_points(start_at, end_at, bucket_sec) points = [] for b_start, b_end in buckets: bucket_rows = [r for r in rows if b_start <= r.checked_at < b_end and r.status in ("online", "offline")] online = sum(1 for r in bucket_rows if r.status == "online") offline = sum(1 for r in bucket_rows if r.status == "offline") samples = online + offline availability = round((online / samples) * 100, 1) if samples > 0 else None points.append( { "bucket_start": b_start, "bucket_end": b_end, "availability_pct": availability, "samples": samples, "online": online, "offline": offline, } ) return points def _intervals_from_rows(rows: list[DashboardDeviceCheckHistory], *, end_at: datetime) -> list[dict[str, Any]]: if not rows: return [] intervals = [] cur_status = rows[0].status cur_start = rows[0].checked_at for row in rows[1:]: if row.status == cur_status: continue duration = int((row.checked_at - cur_start).total_seconds()) intervals.append( { "status": cur_status, "started_at": cur_start, "ended_at": row.checked_at, "duration_sec": max(0, duration), } ) cur_status = row.status cur_start = row.checked_at intervals.append( { "status": cur_status, "started_at": cur_start, "ended_at": end_at, "duration_sec": max(0, int((end_at - cur_start).total_seconds())), } ) return intervals async def get_object_availability( object_id: int, *, window: str, device_ids: list[int] | None = None, compact: bool = False, ) -> dict[str, Any] | None: await ensure_dashboard_bootstrap() now = datetime.now(timezone.utc) window_key = window if window in WINDOW_SECONDS else "24h" start_at = now - timedelta(seconds=_window_seconds(window_key)) bucket_sec = _bucket_seconds(window_key) async with AsyncSessionLocal() as db: obj = await _load_object_with_tags(db, object_id) if obj is None: return None q_devices = select(Device).where(Device.object_id == object_id, Device.is_active == True) if device_ids: q_devices = q_devices.where(Device.id.in_(device_ids)) devices = (await db.execute(q_devices.order_by(Device.ip))).scalars().all() if not devices: return { "generated_at": now, "object_id": object_id, "window": window_key, "start_at": start_at, "end_at": now, "bucket_sec": bucket_sec, "overall_availability_pct": None, "overall_series": _series_from_rows([], start_at=start_at, end_at=now, bucket_sec=bucket_sec), "devices": [], "insufficient_data": True, } rows = ( await db.execute( select(DashboardDeviceCheckHistory).where( DashboardDeviceCheckHistory.check_key == "ping", DashboardDeviceCheckHistory.device_id.in_([d.id for d in devices]), DashboardDeviceCheckHistory.checked_at >= start_at, ) ) ).scalars().all() rows_by_device: dict[int, list[DashboardDeviceCheckHistory]] = defaultdict(list) for row in rows: rows_by_device[row.device_id].append(row) for lst in rows_by_device.values(): lst.sort(key=lambda r: r.checked_at) device_payload = [] all_rows: list[DashboardDeviceCheckHistory] = [] for device in devices: d_rows = rows_by_device.get(device.id, []) all_rows.extend(d_rows) online = sum(1 for r in d_rows if r.status == "online") offline = sum(1 for r in d_rows if r.status == "offline") samples = online + offline availability = round((online / samples) * 100, 1) if samples > 0 else None last_row = d_rows[-1] if d_rows else None device_payload.append( { "device_id": device.id, "ip": device.ip, "hostname": device.hostname, "availability_pct": availability, "samples": samples, "online": online, "offline": offline, "last_checked_at": last_row.checked_at if last_row else None, "last_status": last_row.status if last_row else None, "last_ping_ms": last_row.ping_ms if last_row else None, "series": _series_from_rows(d_rows, start_at=start_at, end_at=now, bucket_sec=bucket_sec), "intervals": [] if compact else _intervals_from_rows(d_rows, end_at=now), } ) all_rows.sort(key=lambda r: r.checked_at) overall_online = sum(1 for r in all_rows if r.status == "online") overall_offline = sum(1 for r in all_rows if r.status == "offline") overall_samples = overall_online + overall_offline overall_availability = round((overall_online / overall_samples) * 100, 1) if overall_samples > 0 else None insufficient_data = overall_samples == 0 return { "generated_at": now, "object_id": object_id, "window": window_key, "start_at": start_at, "end_at": now, "bucket_sec": bucket_sec, "overall_availability_pct": overall_availability, "overall_series": _series_from_rows(all_rows, start_at=start_at, end_at=now, bucket_sec=bucket_sec), "devices": device_payload, "insufficient_data": insufficient_data, }