1329 lines
46 KiB
Python
1329 lines
46 KiB
Python
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.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]]:
|
|
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)
|
|
|
|
|
|
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,
|
|
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,
|
|
"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]:
|
|
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:
|
|
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()
|
|
|
|
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,
|
|
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:
|
|
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()
|
|
|
|
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,
|
|
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,
|
|
) -> dict[str, Any] | None:
|
|
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": _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,
|
|
}
|