804 lines
30 KiB
Python
804 lines
30 KiB
Python
import asyncio
|
|
import time
|
|
from datetime import datetime, 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,
|
|
DashboardDeviceLiveState,
|
|
DashboardMonitorRunState,
|
|
DashboardObjectCheckOverride,
|
|
DashboardObjectMonitorPolicy,
|
|
DashboardTagMonitorPolicy,
|
|
)
|
|
from app.models.device import Device
|
|
from app.models.object import Object
|
|
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,
|
|
},
|
|
]
|
|
|
|
_RUN_LOCK = asyncio.Lock()
|
|
_MANUAL_TASK: asyncio.Task | None = None
|
|
|
|
|
|
async def _noop_emit(_: str, __: str = "info") -> None:
|
|
return
|
|
|
|
|
|
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
|
|
|
|
# keep user-tuned runtime settings, but backfill key defaults and
|
|
# transparently switch legacy disk_usage action source
|
|
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
|
|
|
|
|
|
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 []),
|
|
}
|
|
|
|
|
|
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())
|
|
|
|
|
|
def _is_object_allowed(
|
|
obj: Object,
|
|
object_modes: dict[int, str],
|
|
tag_modes: dict[int, bool],
|
|
) -> tuple[bool, str]:
|
|
mode = object_modes.get(obj.id, "inherit")
|
|
if mode == "include":
|
|
return True, mode
|
|
if mode == "exclude":
|
|
return False, mode
|
|
|
|
tag_values = [tag_modes[t.id] for t in obj.tags if t.id in tag_modes]
|
|
if not tag_values:
|
|
return True, mode
|
|
return any(tag_values), mode
|
|
|
|
|
|
async def _build_object_policy_map(db, objects: list[Object]) -> dict[int, tuple[bool, 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 _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) 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
|
|
await _upsert_live_state(
|
|
dbw,
|
|
device_id=device.id,
|
|
check_key=check.check_key,
|
|
status="failed",
|
|
value_json=None,
|
|
error=str(result),
|
|
checked_at=datetime.now(timezone.utc),
|
|
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,
|
|
)
|
|
|
|
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]]:
|
|
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
|
|
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)
|
|
|
|
|
|
async def get_home_summary(scope: dict[str, Any] | None) -> dict[str, Any]:
|
|
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()
|
|
|
|
check_interval_map = {c.check_key: c.interval_sec_default for c in checks}
|
|
summaries = []
|
|
for obj in objects:
|
|
allowed, mode = policy_map.get(obj.id, (True, "inherit"))
|
|
obj_devices = devices_by_object.get(obj.id, [])
|
|
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,
|
|
}
|
|
|
|
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"),
|
|
)
|
|
summaries.append(
|
|
{
|
|
"object_id": obj.id,
|
|
"object_name": obj.name,
|
|
"city": obj.city,
|
|
"client": obj.client,
|
|
"allowed": allowed,
|
|
"monitor_mode": mode,
|
|
"device_count": len(obj_devices),
|
|
"checks": check_summary,
|
|
"health_score": health,
|
|
}
|
|
)
|
|
|
|
problems = sorted(
|
|
summaries,
|
|
key=lambda x: (
|
|
x["allowed"] is False,
|
|
x["health_score"] if x["health_score"] is not None else 999,
|
|
),
|
|
)[:10]
|
|
|
|
totals = {
|
|
"objects": len(objects),
|
|
"allowed_objects": sum(1 for s in summaries if s["allowed"]),
|
|
"devices": len(device_rows),
|
|
}
|
|
|
|
return {
|
|
"generated_at": now,
|
|
"scope": scope,
|
|
"totals": totals,
|
|
"checks_meta": checks,
|
|
"objects": summaries,
|
|
"problem_objects": problems,
|
|
}
|