diff --git a/.env.example b/.env.example index d96ef7a..5605b64 100644 --- a/.env.example +++ b/.env.example @@ -5,6 +5,8 @@ DATABASE_MAX_OVERFLOW=20 DATABASE_POOL_TIMEOUT=30 TASK_DEVICE_CONCURRENCY=10 MONITOR_PING_CONCURRENCY=20 +DASHBOARD_MONITOR_TICK_SECONDS=60 +DASHBOARD_MONITOR_SYNC_DEVICE_STATUS=false # ── Security ────────────────────────────────────────────────────────────────── # Generate with: python -c "import secrets; print(secrets.token_hex(32))" diff --git a/backend/alembic/versions/0016_dashboard_monitor.py b/backend/alembic/versions/0016_dashboard_monitor.py new file mode 100644 index 0000000..3619f95 --- /dev/null +++ b/backend/alembic/versions/0016_dashboard_monitor.py @@ -0,0 +1,106 @@ +"""0016 dashboard monitor tables + +Revision ID: 0016 +Revises: 0015 +Create Date: 2026-05-05 +""" + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +revision = "0016" +down_revision = "0015" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "dashboard_check_definitions", + sa.Column("check_key", sa.String(length=64), primary_key=True), + sa.Column("enabled", sa.Boolean(), nullable=False, server_default="true"), + sa.Column("interval_sec_default", sa.Integer(), nullable=False), + sa.Column("plugin_action", sa.String(length=128), nullable=False), + sa.Column("categories", postgresql.ARRAY(sa.String()), nullable=False), + sa.Column("timeout_sec", sa.Integer(), nullable=False), + sa.Column("concurrency", sa.Integer(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + ) + + op.create_table( + "dashboard_object_monitor_policy", + sa.Column("object_id", sa.Integer(), sa.ForeignKey("objects.id", ondelete="CASCADE"), primary_key=True), + sa.Column("mode", sa.String(length=16), nullable=False, server_default="inherit"), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + ) + + op.create_table( + "dashboard_tag_monitor_policy", + sa.Column("tag_id", sa.Integer(), sa.ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True), + sa.Column("enabled", sa.Boolean(), nullable=False, server_default="true"), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + ) + + op.create_table( + "dashboard_object_check_overrides", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("object_id", sa.Integer(), sa.ForeignKey("objects.id", ondelete="CASCADE"), nullable=False), + sa.Column("check_key", sa.String(length=64), sa.ForeignKey("dashboard_check_definitions.check_key", ondelete="CASCADE"), nullable=False), + sa.Column("enabled_override", sa.Boolean(), nullable=True), + sa.Column("interval_sec_override", sa.Integer(), nullable=True), + sa.Column("categories_override", postgresql.ARRAY(sa.String()), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.UniqueConstraint("object_id", "check_key", name="uq_dashboard_obj_check"), + ) + op.create_index("ix_dashboard_object_check_overrides_object_id", "dashboard_object_check_overrides", ["object_id"]) + + op.create_table( + "dashboard_device_live_state", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("device_id", sa.Integer(), sa.ForeignKey("devices.id", ondelete="CASCADE"), nullable=False), + sa.Column("check_key", sa.String(length=64), sa.ForeignKey("dashboard_check_definitions.check_key", ondelete="CASCADE"), nullable=False), + sa.Column("status", sa.String(length=20), nullable=False), + sa.Column("value_json", postgresql.JSONB(), nullable=True), + sa.Column("error", sa.Text(), nullable=True), + sa.Column("checked_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("duration_ms", sa.Integer(), nullable=True), + sa.Column("source", sa.String(length=32), nullable=False, server_default="dashboard_monitor"), + sa.UniqueConstraint("device_id", "check_key", name="uq_dashboard_device_check"), + ) + op.create_index("ix_dashboard_device_live_state_device_id", "dashboard_device_live_state", ["device_id"]) + op.create_index("ix_dashboard_device_live_state_check_key", "dashboard_device_live_state", ["check_key"]) + op.create_index("ix_dashboard_device_live_state_checked_at", "dashboard_device_live_state", ["checked_at"]) + + op.create_table( + "dashboard_monitor_run_state", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("status", sa.String(length=20), nullable=False, server_default="idle"), + sa.Column("run_type", sa.String(length=20), nullable=True), + sa.Column("scope", postgresql.JSONB(), nullable=True), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_success_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_error", sa.Text(), nullable=True), + sa.Column("last_counts", postgresql.JSONB(), nullable=True), + ) + + +def downgrade() -> None: + op.drop_table("dashboard_monitor_run_state") + + op.drop_index("ix_dashboard_device_live_state_checked_at", table_name="dashboard_device_live_state") + op.drop_index("ix_dashboard_device_live_state_check_key", table_name="dashboard_device_live_state") + op.drop_index("ix_dashboard_device_live_state_device_id", table_name="dashboard_device_live_state") + op.drop_table("dashboard_device_live_state") + + op.drop_index("ix_dashboard_object_check_overrides_object_id", table_name="dashboard_object_check_overrides") + op.drop_table("dashboard_object_check_overrides") + + op.drop_table("dashboard_tag_monitor_policy") + op.drop_table("dashboard_object_monitor_policy") + op.drop_table("dashboard_check_definitions") diff --git a/backend/app/config.py b/backend/app/config.py index 9f0629e..ebbdae7 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -23,6 +23,8 @@ class Settings(BaseSettings): TASK_DEVICE_CONCURRENCY: int = 10 MONITOR_PING_CONCURRENCY: int = 20 + DASHBOARD_MONITOR_TICK_SECONDS: int = 60 + DASHBOARD_MONITOR_SYNC_DEVICE_STATUS: bool = False settings = Settings() diff --git a/backend/app/main.py b/backend/app/main.py index 3239193..560ed18 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -7,7 +7,9 @@ from fastapi.openapi.utils import get_openapi from app.middleware.auth import AuthMiddleware from app.plugins.loader import load_plugins -from app.routers import admin, alerts, auth, devices, events, global_devices, objects, racks, snapshots, ssh_keys, tags, tasks, zones +from app.routers import admin, alerts, auth, dashboard, devices, events, global_devices, objects, racks, snapshots, ssh_keys, tags, tasks, zones +from app.services.dashboard_monitor import ensure_dashboard_bootstrap +from app.workers.dashboard_monitor import start_dashboard_monitor, stop_dashboard_monitor from app.workers.monitor import start_monitor, stop_monitor from app.workers.startup import recover_stale_tasks @@ -22,8 +24,11 @@ logging.basicConfig( async def lifespan(app: FastAPI): await load_plugins() await recover_stale_tasks() + await ensure_dashboard_bootstrap() start_monitor() + start_dashboard_monitor() yield + await stop_dashboard_monitor() await stop_monitor() @@ -59,6 +64,7 @@ def create_app() -> FastAPI: app.include_router(snapshots.router) app.include_router(tasks.router) app.include_router(events.router) + app.include_router(dashboard.router) def custom_openapi(): if app.openapi_schema: diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index bfd6782..ceb6517 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -10,6 +10,14 @@ from .snapshot import ConfigSnapshot, SnapshotContent from .alert import Alert from .plugin import Plugin from .ssh_key import SSHKey, ObjectSSHCredential, ObjectCategoryCredential, GlobalSSHCredential +from .dashboard import ( + DashboardCheckDefinition, + DashboardObjectCheckOverride, + DashboardDeviceLiveState, + DashboardObjectMonitorPolicy, + DashboardTagMonitorPolicy, + DashboardMonitorRunState, +) __all__ = [ "Base", @@ -33,4 +41,10 @@ __all__ = [ "ObjectSSHCredential", "ObjectCategoryCredential", "GlobalSSHCredential", + "DashboardCheckDefinition", + "DashboardObjectCheckOverride", + "DashboardDeviceLiveState", + "DashboardObjectMonitorPolicy", + "DashboardTagMonitorPolicy", + "DashboardMonitorRunState", ] diff --git a/backend/app/models/dashboard.py b/backend/app/models/dashboard.py new file mode 100644 index 0000000..822a15b --- /dev/null +++ b/backend/app/models/dashboard.py @@ -0,0 +1,98 @@ +from datetime import datetime +from typing import Optional + +from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint +from sqlalchemy.dialects.postgresql import ARRAY, JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from .base import Base, TimestampMixin + +CHECK_STATUS_VALUES = ( + "online", + "offline", + "unknown", + "ok", + "warn", + "failed", +) + +MONITOR_POLICY_MODES = ("inherit", "include", "exclude") + + +class DashboardCheckDefinition(Base, TimestampMixin): + __tablename__ = "dashboard_check_definitions" + + check_key: Mapped[str] = mapped_column(String(64), primary_key=True) + enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + interval_sec_default: Mapped[int] = mapped_column(Integer, nullable=False, default=300) + plugin_action: Mapped[str] = mapped_column(String(128), nullable=False) + categories: Mapped[list[str]] = mapped_column(ARRAY(String), nullable=False, default=list) + timeout_sec: Mapped[int] = mapped_column(Integer, nullable=False, default=30) + concurrency: Mapped[int] = mapped_column(Integer, nullable=False, default=20) + + +class DashboardObjectCheckOverride(Base, TimestampMixin): + __tablename__ = "dashboard_object_check_overrides" + __table_args__ = (UniqueConstraint("object_id", "check_key", name="uq_dashboard_obj_check"),) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + object_id: Mapped[int] = mapped_column( + Integer, ForeignKey("objects.id", ondelete="CASCADE"), nullable=False, index=True + ) + check_key: Mapped[str] = mapped_column( + String(64), ForeignKey("dashboard_check_definitions.check_key", ondelete="CASCADE"), nullable=False + ) + enabled_override: Mapped[Optional[bool]] = mapped_column(Boolean, nullable=True) + interval_sec_override: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) + categories_override: Mapped[Optional[list[str]]] = mapped_column(ARRAY(String), nullable=True) + + +class DashboardDeviceLiveState(Base): + __tablename__ = "dashboard_device_live_state" + __table_args__ = (UniqueConstraint("device_id", "check_key", name="uq_dashboard_device_check"),) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + device_id: Mapped[int] = mapped_column( + Integer, ForeignKey("devices.id", ondelete="CASCADE"), nullable=False, index=True + ) + check_key: Mapped[str] = mapped_column( + String(64), ForeignKey("dashboard_check_definitions.check_key", ondelete="CASCADE"), nullable=False, index=True + ) + status: Mapped[str] = mapped_column(String(20), nullable=False, default="unknown") + value_json: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) + error: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + checked_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True) + duration_ms: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) + source: Mapped[str] = mapped_column(String(32), nullable=False, default="dashboard_monitor") + + +class DashboardObjectMonitorPolicy(Base, TimestampMixin): + __tablename__ = "dashboard_object_monitor_policy" + + object_id: Mapped[int] = mapped_column( + Integer, ForeignKey("objects.id", ondelete="CASCADE"), primary_key=True + ) + mode: Mapped[str] = mapped_column(String(16), nullable=False, default="inherit") + + +class DashboardTagMonitorPolicy(Base, TimestampMixin): + __tablename__ = "dashboard_tag_monitor_policy" + + tag_id: Mapped[int] = mapped_column( + Integer, ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True + ) + enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + + +class DashboardMonitorRunState(Base): + __tablename__ = "dashboard_monitor_run_state" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + status: Mapped[str] = mapped_column(String(20), nullable=False, default="idle") + run_type: Mapped[Optional[str]] = mapped_column(String(20), nullable=True) + scope: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) + started_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) + finished_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) + last_success_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) + last_error: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + last_counts: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) diff --git a/backend/app/plugins/builtin/ssh/disk_usage.py b/backend/app/plugins/builtin/ssh/disk_usage.py new file mode 100644 index 0000000..f572760 --- /dev/null +++ b/backend/app/plugins/builtin/ssh/disk_usage.py @@ -0,0 +1,106 @@ +""" +Collects disk usage stats over SSH using df. +Designed for dashboard healthcheck loop. +""" + +from app.models.device import Device +from app.models.object import Object +from app.plugins.base import ActionResult, BaseAction, Emitter +from app.services import ssh as ssh_service + +_CMD = "df -P -k -x tmpfs -x devtmpfs" + + +class DiskUsageAction(BaseAction): + """Collects disk usage (df) and returns normalized stats.""" + + name = "disk_usage" + display_name = "Disk Usage" + compatible_categories = ["main_server", "vm"] + params_schema = [] + + async def execute(self, device: Device, obj: Object, params: dict, emit: Emitter, db=None) -> ActionResult: + await emit(f"[{device.ip}] Checking disk usage...", "info") + + result = await ssh_service.run_command(device, obj, _CMD, timeout=30, db=db) + if result.error: + await emit(f"[{device.ip}] disk usage error: {result.error}", "error") + return ActionResult(success=False, error=result.error, stdout=result.stdout, stderr=result.stderr) + + if result.exit_code != 0: + return ActionResult( + success=False, + error=f"df returned non-zero exit code: {result.exit_code}", + stdout=result.stdout, + stderr=result.stderr, + exit_code=result.exit_code, + ) + + disks = _parse_df(result.stdout) + if not disks: + return ActionResult( + success=False, + error="No disk rows parsed from df output", + stdout=result.stdout, + stderr=result.stderr, + exit_code=result.exit_code, + ) + + top = max(disks, key=lambda d: d["pct"]) + max_pct = top["pct"] + max_mount = top["mount"] + + await emit(f"[{device.ip}] max disk usage: {max_pct}% ({max_mount})", "info") + return ActionResult( + success=True, + stdout=result.stdout, + stderr=result.stderr, + exit_code=result.exit_code, + data={ + "disks": disks, + "max_pct": max_pct, + "max_mount": max_mount, + }, + ) + + +def _parse_df(output: str) -> list[dict]: + rows: list[dict] = [] + for line in output.splitlines(): + line = line.strip() + if not line or line.startswith("Filesystem"): + continue + + parts = line.split() + if len(parts) < 6: + continue + + filesystem = parts[0] + total_kb = _to_int(parts[1]) + used_kb = _to_int(parts[2]) + avail_kb = _to_int(parts[3]) + pct_raw = parts[4].rstrip("%") + mount = parts[5] + pct = _to_int(pct_raw) + + if pct is None: + continue + + rows.append( + { + "filesystem": filesystem, + "total_kb": total_kb, + "used_kb": used_kb, + "avail_kb": avail_kb, + "pct": pct, + "mount": mount, + } + ) + return rows + + +def _to_int(value: str): + try: + return int(value) + except Exception: + return None diff --git a/backend/app/plugins/loader.py b/backend/app/plugins/loader.py index 72af1e2..d0e4b0c 100644 --- a/backend/app/plugins/loader.py +++ b/backend/app/plugins/loader.py @@ -26,6 +26,7 @@ from app.plugins.builtin.mikrotik.get_config import MikrotikGetConfigAction from app.plugins.builtin.mikrotik.get_ntp import MikrotikGetNTPAction from app.plugins.builtin.mikrotik.set_ntp import MikrotikSetNTPAction from app.plugins.builtin.ssh.debian_system_info import DebianSystemInfoAction +from app.plugins.builtin.ssh.disk_usage import DiskUsageAction from app.plugins.builtin.ssh.get_file import GetFileAction from app.plugins.builtin.ssh.get_timedatectl import GetTimedatectlAction from app.plugins.builtin.ssh.ssh_command import SSHCommandAction @@ -60,6 +61,7 @@ BUILTIN_ACTIONS: dict[str, BaseAction] = { PingAction(), SSHCommandAction(), DebianSystemInfoAction(), + DiskUsageAction(), _timedatectl, GetFileAction(), MikrotikGetConfigAction(), diff --git a/backend/app/routers/dashboard.py b/backend/app/routers/dashboard.py new file mode 100644 index 0000000..b5a6994 --- /dev/null +++ b/backend/app/routers/dashboard.py @@ -0,0 +1,92 @@ +from fastapi import APIRouter, Depends + +from app.dependencies import get_current_user +from app.models.user import User +from app.schemas.dashboard import ( + DashboardCheckDefinitionPatchItem, + DashboardChecksResponse, + DashboardHomeSummaryResponse, + DashboardMonitorStatus, + DashboardObjectOverridePatchItem, + DashboardObjectPolicyPatchItem, + DashboardPolicyResponse, + DashboardRunRequest, + DashboardRunResponse, + DashboardTagPolicyPatchItem, +) +from app.services import dashboard_monitor as dashboard_service + +router = APIRouter(prefix="/api/v1/dashboard", tags=["dashboard"]) + + +@router.get("/home-summary", response_model=DashboardHomeSummaryResponse) +async def home_summary( + city: str | None = None, + client: str | None = None, + object_ids: list[int] | None = None, + tag_ids: list[int] | None = None, + _: User = Depends(get_current_user), +): + scope = { + "city": city, + "client": client, + "object_ids": object_ids or [], + "tag_ids": tag_ids or [], + } + data = await dashboard_service.get_home_summary(scope) + return DashboardHomeSummaryResponse.model_validate(data) + + +@router.get("/monitor/status", response_model=DashboardMonitorStatus) +async def monitor_status(_: User = Depends(get_current_user)): + row = await dashboard_service.get_monitor_status() + return DashboardMonitorStatus.model_validate(row) + + +@router.post("/monitor/run", response_model=DashboardRunResponse) +async def monitor_run(body: DashboardRunRequest, _: User = Depends(get_current_user)): + accepted, running, detail = await dashboard_service.request_manual_run( + scope=body.model_dump(exclude={"force"}), + force=body.force, + ) + return DashboardRunResponse(accepted=accepted, running=running, detail=detail) + + +@router.get("/monitor/policy", response_model=DashboardPolicyResponse) +async def monitor_policy(_: User = Depends(get_current_user)): + object_policies, tag_policies = await dashboard_service.get_policy() + return DashboardPolicyResponse(object_policies=object_policies, tag_policies=tag_policies) + + +@router.patch("/monitor/policy/objects", response_model=DashboardPolicyResponse) +async def patch_object_policy(body: list[DashboardObjectPolicyPatchItem], _: User = Depends(get_current_user)): + await dashboard_service.patch_object_policy([i.model_dump() for i in body]) + object_policies, tag_policies = await dashboard_service.get_policy() + return DashboardPolicyResponse(object_policies=object_policies, tag_policies=tag_policies) + + +@router.patch("/monitor/policy/tags", response_model=DashboardPolicyResponse) +async def patch_tag_policy(body: list[DashboardTagPolicyPatchItem], _: User = Depends(get_current_user)): + await dashboard_service.patch_tag_policy([i.model_dump() for i in body]) + object_policies, tag_policies = await dashboard_service.get_policy() + return DashboardPolicyResponse(object_policies=object_policies, tag_policies=tag_policies) + + +@router.get("/checks", response_model=DashboardChecksResponse) +async def dashboard_checks(_: User = Depends(get_current_user)): + checks, overrides = await dashboard_service.list_checks_with_overrides() + return DashboardChecksResponse(checks=checks, object_overrides=overrides) + + +@router.patch("/checks", response_model=DashboardChecksResponse) +async def patch_checks(body: list[DashboardCheckDefinitionPatchItem], _: User = Depends(get_current_user)): + await dashboard_service.patch_checks([item.model_dump(exclude_none=True) for item in body]) + checks, overrides = await dashboard_service.list_checks_with_overrides() + return DashboardChecksResponse(checks=checks, object_overrides=overrides) + + +@router.patch("/checks/object-overrides", response_model=DashboardChecksResponse) +async def patch_object_overrides(body: list[DashboardObjectOverridePatchItem], _: User = Depends(get_current_user)): + await dashboard_service.patch_object_overrides([item.model_dump(exclude_none=False) for item in body]) + checks, overrides = await dashboard_service.list_checks_with_overrides() + return DashboardChecksResponse(checks=checks, object_overrides=overrides) diff --git a/backend/app/schemas/dashboard.py b/backend/app/schemas/dashboard.py new file mode 100644 index 0000000..bc01256 --- /dev/null +++ b/backend/app/schemas/dashboard.py @@ -0,0 +1,139 @@ +from datetime import datetime +from typing import Any, Literal, Optional + +from pydantic import BaseModel, Field + + +MonitorPolicyMode = Literal["inherit", "include", "exclude"] + + +class DashboardScope(BaseModel): + city: Optional[str] = None + client: Optional[str] = None + object_ids: list[int] = Field(default_factory=list) + tag_ids: list[int] = Field(default_factory=list) + + +class DashboardCheckDefinitionRead(BaseModel): + check_key: str + enabled: bool + interval_sec_default: int + plugin_action: str + categories: list[str] + timeout_sec: int + concurrency: int + updated_at: datetime + + model_config = {"from_attributes": True} + + +class DashboardCheckDefinitionPatchItem(BaseModel): + check_key: str + enabled: Optional[bool] = None + interval_sec_default: Optional[int] = None + categories: Optional[list[str]] = None + timeout_sec: Optional[int] = None + concurrency: Optional[int] = None + + +class DashboardObjectOverrideRead(BaseModel): + object_id: int + check_key: str + enabled_override: Optional[bool] = None + interval_sec_override: Optional[int] = None + categories_override: Optional[list[str]] = None + updated_at: datetime + + model_config = {"from_attributes": True} + + +class DashboardObjectOverridePatchItem(BaseModel): + object_id: int + check_key: str + enabled_override: Optional[bool] = None + interval_sec_override: Optional[int] = None + categories_override: Optional[list[str]] = None + + +class DashboardObjectPolicyRead(BaseModel): + object_id: int + mode: MonitorPolicyMode + updated_at: datetime + + model_config = {"from_attributes": True} + + +class DashboardObjectPolicyPatchItem(BaseModel): + object_id: int + mode: MonitorPolicyMode + + +class DashboardTagPolicyRead(BaseModel): + tag_id: int + enabled: bool + updated_at: datetime + + model_config = {"from_attributes": True} + + +class DashboardTagPolicyPatchItem(BaseModel): + tag_id: int + enabled: bool + + +class DashboardPolicyResponse(BaseModel): + object_policies: list[DashboardObjectPolicyRead] + tag_policies: list[DashboardTagPolicyRead] + + +class DashboardChecksResponse(BaseModel): + checks: list[DashboardCheckDefinitionRead] + object_overrides: list[DashboardObjectOverrideRead] + + +class DashboardRunRequest(DashboardScope): + force: bool = True + + +class DashboardRunResponse(BaseModel): + accepted: bool + running: bool + detail: str + + +class DashboardMonitorStatus(BaseModel): + status: str + run_type: Optional[str] = None + scope: Optional[dict[str, Any]] = None + started_at: Optional[datetime] = None + finished_at: Optional[datetime] = None + last_success_at: Optional[datetime] = None + last_error: Optional[str] = None + last_counts: Optional[dict[str, Any]] = None + + +class DashboardObjectCheckSummary(BaseModel): + status_counts: dict[str, int] + last_checked_at: Optional[datetime] = None + max_age_sec: Optional[int] = None + + +class DashboardObjectSummary(BaseModel): + object_id: int + object_name: str + city: Optional[str] = None + client: Optional[str] = None + allowed: bool + monitor_mode: MonitorPolicyMode + device_count: int + checks: dict[str, DashboardObjectCheckSummary] + health_score: Optional[float] = None + + +class DashboardHomeSummaryResponse(BaseModel): + generated_at: datetime + scope: DashboardScope + totals: dict[str, int] + checks_meta: list[DashboardCheckDefinitionRead] + objects: list[DashboardObjectSummary] + problem_objects: list[DashboardObjectSummary] diff --git a/backend/app/services/dashboard_monitor.py b/backend/app/services/dashboard_monitor.py new file mode 100644 index 0000000..a7eff0d --- /dev/null +++ b/backend/app/services/dashboard_monitor.py @@ -0,0 +1,804 @@ +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, + } diff --git a/backend/app/workers/dashboard_monitor.py b/backend/app/workers/dashboard_monitor.py new file mode 100644 index 0000000..8e6751e --- /dev/null +++ b/backend/app/workers/dashboard_monitor.py @@ -0,0 +1,41 @@ +import asyncio + +from app.config import settings +from app.services.dashboard_monitor import run_dashboard_monitor_cycle + +_stop_event: asyncio.Event | None = None +_worker_task: asyncio.Task | None = None + + +async def _loop() -> None: + assert _stop_event is not None + while not _stop_event.is_set(): + try: + await run_dashboard_monitor_cycle(run_type="auto", scope=None, force=False) + except Exception as exc: + print(f"[dashboard_monitor] Cycle error: {exc}") + + try: + await asyncio.wait_for(_stop_event.wait(), timeout=settings.DASHBOARD_MONITOR_TICK_SECONDS) + except asyncio.TimeoutError: + pass + + +def start_dashboard_monitor() -> None: + global _stop_event, _worker_task + _stop_event = asyncio.Event() + _worker_task = asyncio.create_task(_loop()) + print("[dashboard_monitor] Worker started") + + +async def stop_dashboard_monitor() -> None: + global _stop_event, _worker_task + if _stop_event: + _stop_event.set() + if _worker_task: + _worker_task.cancel() + try: + await _worker_task + except asyncio.CancelledError: + pass + print("[dashboard_monitor] Worker stopped") diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index e488693..81a8318 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -3,6 +3,7 @@ import { AppLayout } from './components/AppLayout' import { AdminPage } from './pages/Admin' import { AlertsPage } from './pages/Alerts' import { DeviceDetailPage } from './pages/DeviceDetail' +import { HomePage } from './pages/Home' import { LoginPage } from './pages/Login' import { ObjectsPage } from './pages/Objects' import { SnapshotsPage } from './pages/Snapshots' @@ -29,7 +30,8 @@ export default function App() { } > - } /> + } /> + } /> {/* Инвентарь */} } /> diff --git a/frontend/src/api/dashboard.ts b/frontend/src/api/dashboard.ts new file mode 100644 index 0000000..989fab0 --- /dev/null +++ b/frontend/src/api/dashboard.ts @@ -0,0 +1,202 @@ +import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query' +import { api } from './client' + +export interface DashboardScope { + city?: string + client?: string + object_ids?: number[] + tag_ids?: number[] +} + +export interface DashboardCheckDefinition { + check_key: string + enabled: boolean + interval_sec_default: number + plugin_action: string + categories: string[] + timeout_sec: number + concurrency: number + updated_at: string +} + +export interface DashboardObjectOverride { + object_id: number + check_key: string + enabled_override: boolean | null + interval_sec_override: number | null + categories_override: string[] | null + updated_at: string +} + +export interface DashboardMonitorStatus { + status: string + run_type: string | null + scope: Record | null + started_at: string | null + finished_at: string | null + last_success_at: string | null + last_error: string | null + last_counts: Record | null +} + +export interface DashboardObjectPolicy { + object_id: number + mode: 'inherit' | 'include' | 'exclude' + updated_at: string +} + +export interface DashboardTagPolicy { + tag_id: number + enabled: boolean + updated_at: string +} + +export interface DashboardPolicyResponse { + object_policies: DashboardObjectPolicy[] + tag_policies: DashboardTagPolicy[] +} + +export interface DashboardObjectCheckSummary { + status_counts: Record + last_checked_at: string | null + max_age_sec: number | null +} + +export interface DashboardObjectSummary { + object_id: number + object_name: string + city: string | null + client: string | null + allowed: boolean + monitor_mode: 'inherit' | 'include' | 'exclude' + device_count: number + checks: Record + health_score: number | null +} + +export interface DashboardHomeSummary { + generated_at: string + scope: DashboardScope + totals: Record + checks_meta: DashboardCheckDefinition[] + objects: DashboardObjectSummary[] + problem_objects: DashboardObjectSummary[] +} + +export interface DashboardChecksResponse { + checks: DashboardCheckDefinition[] + object_overrides: DashboardObjectOverride[] +} + +export const useDashboardHomeSummary = (scope: DashboardScope) => + useQuery({ + queryKey: ['dashboard-home-summary', scope], + queryFn: () => + api + .get('/api/v1/dashboard/home-summary', { params: scope }) + .then((r) => r.data), + refetchInterval: 30_000, + }) + +export const useDashboardMonitorStatus = () => + useQuery({ + queryKey: ['dashboard-monitor-status'], + queryFn: () => + api + .get('/api/v1/dashboard/monitor/status') + .then((r) => r.data), + refetchInterval: 10_000, + }) + +export const useDashboardRunNow = () => { + const qc = useQueryClient() + return useMutation({ + mutationFn: (body: DashboardScope & { force?: boolean }) => + api.post('/api/v1/dashboard/monitor/run', body).then((r) => r.data), + onSuccess: () => { + qc.invalidateQueries({ queryKey: ['dashboard-monitor-status'] }) + qc.invalidateQueries({ queryKey: ['dashboard-home-summary'] }) + }, + }) +} + +export const useDashboardPolicy = () => + useQuery({ + queryKey: ['dashboard-policy'], + queryFn: () => + api + .get('/api/v1/dashboard/monitor/policy') + .then((r) => r.data), + }) + +export const usePatchDashboardObjectPolicy = () => { + const qc = useQueryClient() + return useMutation({ + mutationFn: (body: Array<{ object_id: number; mode: 'inherit' | 'include' | 'exclude' }>) => + api.patch('/api/v1/dashboard/monitor/policy/objects', body).then((r) => r.data), + onSuccess: () => { + qc.invalidateQueries({ queryKey: ['dashboard-policy'] }) + qc.invalidateQueries({ queryKey: ['dashboard-home-summary'] }) + }, + }) +} + +export const usePatchDashboardTagPolicy = () => { + const qc = useQueryClient() + return useMutation({ + mutationFn: (body: Array<{ tag_id: number; enabled: boolean }>) => + api.patch('/api/v1/dashboard/monitor/policy/tags', body).then((r) => r.data), + onSuccess: () => { + qc.invalidateQueries({ queryKey: ['dashboard-policy'] }) + qc.invalidateQueries({ queryKey: ['dashboard-home-summary'] }) + }, + }) +} + +export const useDashboardChecks = () => + useQuery({ + queryKey: ['dashboard-checks'], + queryFn: () => + api + .get('/api/v1/dashboard/checks') + .then((r) => r.data), + }) + +export const usePatchDashboardChecks = () => { + const qc = useQueryClient() + return useMutation({ + mutationFn: ( + body: Array<{ + check_key: string + enabled?: boolean + interval_sec_default?: number + categories?: string[] + timeout_sec?: number + concurrency?: number + }> + ) => api.patch('/api/v1/dashboard/checks', body).then((r) => r.data), + onSuccess: () => { + qc.invalidateQueries({ queryKey: ['dashboard-checks'] }) + qc.invalidateQueries({ queryKey: ['dashboard-home-summary'] }) + }, + }) +} + +export const usePatchDashboardObjectOverrides = () => { + const qc = useQueryClient() + return useMutation({ + mutationFn: ( + body: Array<{ + object_id: number + check_key: string + enabled_override?: boolean | null + interval_sec_override?: number | null + categories_override?: string[] | null + }> + ) => api.patch('/api/v1/dashboard/checks/object-overrides', body).then((r) => r.data), + onSuccess: () => { + qc.invalidateQueries({ queryKey: ['dashboard-checks'] }) + qc.invalidateQueries({ queryKey: ['dashboard-home-summary'] }) + }, + }) +} diff --git a/frontend/src/api/tags.ts b/frontend/src/api/tags.ts new file mode 100644 index 0000000..c16f371 --- /dev/null +++ b/frontend/src/api/tags.ts @@ -0,0 +1,16 @@ +import { useQuery } from '@tanstack/react-query' +import { api } from './client' + +export interface TagItem { + id: number + name: string + color: string + tag_type: string + created_at: string +} + +export const useTags = () => + useQuery({ + queryKey: ['tags'], + queryFn: () => api.get('/api/v1/tags').then((r) => r.data), + }) diff --git a/frontend/src/components/AppLayout.tsx b/frontend/src/components/AppLayout.tsx index 2aeb344..02791b7 100644 --- a/frontend/src/components/AppLayout.tsx +++ b/frontend/src/components/AppLayout.tsx @@ -1,4 +1,5 @@ import { + HomeOutlined, AppstoreOutlined, BellOutlined, ControlOutlined, @@ -38,7 +39,9 @@ export function AppLayout() { }, }) - const selectedKey = location.pathname.startsWith('/inventory/objects') + const selectedKey = location.pathname.startsWith('/home') + ? 'home' + : location.pathname.startsWith('/inventory/objects') ? 'inventory-objects' : location.pathname.startsWith('/work/devices') ? 'work-devices' @@ -74,6 +77,13 @@ export function AppLayout() { mode="inline" selectedKeys={[selectedKey]} items={[ + { + key: 'home', + icon: , + label: 'Главная', + onClick: () => navigate('/home'), + }, + { type: 'divider' }, { key: 'inventory-group', type: 'group', diff --git a/frontend/src/hooks/useDashboardMonitorSSE.ts b/frontend/src/hooks/useDashboardMonitorSSE.ts new file mode 100644 index 0000000..48f9e9f --- /dev/null +++ b/frontend/src/hooks/useDashboardMonitorSSE.ts @@ -0,0 +1,45 @@ +import { fetchEventSource } from '@microsoft/fetch-event-source' +import { useEffect, useRef } from 'react' +import { useAuthStore } from '../store/auth' + +interface DashboardMonitorHandlers { + onUpdate?: (eventType: string, data: Record) => void +} + +const DASHBOARD_MONITOR_CHANNEL = '__dashboard_monitor__' + +export function useDashboardMonitorSSE(handlers: DashboardMonitorHandlers) { + const accessToken = useAuthStore((s) => s.accessToken) + const handlersRef = useRef(handlers) + handlersRef.current = handlers + + useEffect(() => { + if (!accessToken) return + const ctrl = new AbortController() + + fetchEventSource(`/api/v1/events?task_id=${DASHBOARD_MONITOR_CHANNEL}`, { + headers: { Authorization: `Bearer ${accessToken}` }, + signal: ctrl.signal, + onmessage(ev) { + try { + const data = JSON.parse(ev.data) + if ( + ev.event === 'dashboard_monitor_update' || + ev.event === 'dashboard_monitor_cycle_done' || + ev.event === 'dashboard_monitor_cycle_failed' + ) { + handlersRef.current.onUpdate?.(ev.event, data) + } + } catch { + // ignore parse errors + } + }, + onerror(err) { + ctrl.abort() + throw err + }, + }) + + return () => ctrl.abort() + }, [accessToken]) +} diff --git a/frontend/src/pages/Home.tsx b/frontend/src/pages/Home.tsx new file mode 100644 index 0000000..6664ffd --- /dev/null +++ b/frontend/src/pages/Home.tsx @@ -0,0 +1,569 @@ +import { + Alert, + Button, + Card, + Col, + Divider, + InputNumber, + Row, + Select, + Space, + Statistic, + Switch, + Table, + Tag, + Typography, + message, +} from 'antd' +import dayjs from 'dayjs' +import { useEffect, useMemo, useState } from 'react' +import { useQueryClient } from '@tanstack/react-query' +import { useObjects, useObjectsMeta } from '../api/objects' +import { + useDashboardChecks, + useDashboardHomeSummary, + useDashboardMonitorStatus, + useDashboardPolicy, + useDashboardRunNow, + usePatchDashboardChecks, + usePatchDashboardObjectOverrides, + usePatchDashboardObjectPolicy, + usePatchDashboardTagPolicy, + type DashboardObjectOverride, +} from '../api/dashboard' +import { useTags } from '../api/tags' +import { useDashboardMonitorSSE } from '../hooks/useDashboardMonitorSSE' + +const CATEGORY_OPTIONS = [ + 'main_server', + 'vm', + 'router', + 'embedded', + 'camera', + 'io_board', + 'bank_terminal', + 'cash_register', + 'other', +] + +type OverrideEnabledMode = 'inherit' | 'enabled' | 'disabled' +type OverrideCategoriesMode = 'inherit' | 'custom' + +function formatAge(seconds: number | null | undefined): string { + if (seconds == null) return '-' + if (seconds < 60) return `${seconds}с` + if (seconds < 3600) return `${Math.floor(seconds / 60)}м` + return `${Math.floor(seconds / 3600)}ч ${Math.floor((seconds % 3600) / 60)}м` +} + +export function HomePage() { + const qc = useQueryClient() + + const { data: meta } = useObjectsMeta() + const { data: objects } = useObjects() + const { data: tags } = useTags() + + const [city, setCity] = useState() + const [client, setClient] = useState() + const [objectIds, setObjectIds] = useState([]) + const [tagIds, setTagIds] = useState([]) + + const scope = useMemo( + () => ({ city, client, object_ids: objectIds, tag_ids: tagIds }), + [city, client, objectIds, tagIds] + ) + + const { data: summary, isLoading } = useDashboardHomeSummary(scope) + const { data: monitorStatus } = useDashboardMonitorStatus() + const { data: policy } = useDashboardPolicy() + const { data: checksData } = useDashboardChecks() + + const runNow = useDashboardRunNow() + const patchTagPolicy = usePatchDashboardTagPolicy() + const patchObjectPolicy = usePatchDashboardObjectPolicy() + const patchChecks = usePatchDashboardChecks() + const patchOverrides = usePatchDashboardObjectOverrides() + + const [overrideObjectId, setOverrideObjectId] = useState() + const [overrideCheckKey, setOverrideCheckKey] = useState() + const [overrideInterval, setOverrideInterval] = useState() + const [overrideEnabledMode, setOverrideEnabledMode] = useState('inherit') + const [overrideCategoriesMode, setOverrideCategoriesMode] = useState('inherit') + const [overrideCategories, setOverrideCategories] = useState([]) + + const tagPolicyMap = useMemo(() => { + const map = new Map() + for (const row of policy?.tag_policies ?? []) map.set(row.tag_id, row.enabled) + return map + }, [policy]) + + const objectPolicyMap = useMemo(() => { + const map = new Map() + for (const row of policy?.object_policies ?? []) map.set(row.object_id, row.mode) + return map + }, [policy]) + + const overrideMap = useMemo(() => { + const map = new Map() + for (const row of checksData?.object_overrides ?? []) { + map.set(`${row.object_id}:${row.check_key}`, row) + } + return map + }, [checksData]) + + useEffect(() => { + if (!overrideObjectId || !overrideCheckKey) { + setOverrideEnabledMode('inherit') + setOverrideInterval(undefined) + setOverrideCategoriesMode('inherit') + setOverrideCategories([]) + return + } + + const current = overrideMap.get(`${overrideObjectId}:${overrideCheckKey}`) + if (!current) { + setOverrideEnabledMode('inherit') + setOverrideInterval(undefined) + setOverrideCategoriesMode('inherit') + setOverrideCategories([]) + return + } + + if (current.enabled_override === true) setOverrideEnabledMode('enabled') + else if (current.enabled_override === false) setOverrideEnabledMode('disabled') + else setOverrideEnabledMode('inherit') + + setOverrideInterval(current.interval_sec_override ?? undefined) + + if (current.categories_override && current.categories_override.length > 0) { + setOverrideCategoriesMode('custom') + setOverrideCategories(current.categories_override) + } else { + setOverrideCategoriesMode('inherit') + setOverrideCategories([]) + } + }, [overrideObjectId, overrideCheckKey, overrideMap]) + + useDashboardMonitorSSE({ + onUpdate: () => { + qc.invalidateQueries({ queryKey: ['dashboard-home-summary'] }) + qc.invalidateQueries({ queryKey: ['dashboard-monitor-status'] }) + }, + }) + + const handleRunNow = async () => { + try { + await runNow.mutateAsync({ ...scope, force: true }) + message.success('Мониторинг запущен') + } catch { + message.error('Не удалось запустить мониторинг') + } + } + + const totals = summary?.totals ?? { objects: 0, devices: 0, allowed_objects: 0 } + const problemObjects = summary?.problem_objects ?? [] + + const summaryColumns = [ + { + title: 'Объект', + key: 'object', + render: (_: unknown, row: any) => ( + + {row.object_name} + {row.city && {row.city}} + {row.client && {row.client}} + + ), + }, + { + title: 'Политика', + key: 'policy', + render: (_: unknown, row: any) => {row.monitor_mode}, + }, + { + title: 'Устройств', + dataIndex: 'device_count', + key: 'device_count', + }, + { + title: 'Ping', + key: 'ping', + render: (_: unknown, row: any) => { + const ping = row.checks?.ping + const cnt = ping?.status_counts ?? {} + return ( + + online {cnt.online ?? 0} / offline {cnt.offline ?? 0} / unknown {cnt.unknown ?? 0} + + age: {formatAge(ping?.max_age_sec)} + + + ) + }, + }, + { + title: 'Диски', + key: 'disk', + render: (_: unknown, row: any) => { + const disk = row.checks?.disk_usage + const cnt = disk?.status_counts ?? {} + return ( + + ok {cnt.ok ?? 0} / warn {cnt.warn ?? 0} / failed {cnt.failed ?? 0} + + age: {formatAge(disk?.max_age_sec)} + + + ) + }, + }, + { + title: 'Health', + key: 'health', + render: (_: unknown, row: any) => { + if (row.health_score == null) return '-' + const color = row.health_score >= 80 ? 'green' : row.health_score >= 60 ? 'orange' : 'red' + return {row.health_score} + }, + }, + ] + + return ( +
+ + Главная + + + + + ({ value: c, label: c }))} + value={client} + onChange={setClient} + /> + + + ({ value: t.id, label: t.name }))} + /> + + + + + + + + + + Статус: {monitorStatus?.status ?? '-'} + {monitorStatus?.last_success_at + ? ` · успешный запуск ${dayjs(monitorStatus.last_success_at).format('DD.MM HH:mm:ss')}` + : ''} + + + + + {monitorStatus?.last_error && ( + + )} + + + + + + + + + + + + + + + + {(checksData?.checks ?? []).map((check) => ( + + {check.check_key} + + { + await patchChecks.mutateAsync([{ check_key: check.check_key, enabled }]) + }} + /> + + + { + const val = Number((e.target as HTMLInputElement).value) + if (!Number.isFinite(val) || val <= 0) return + await patchChecks.mutateAsync([{ check_key: check.check_key, interval_sec_default: val }]) + }} + /> + + + { + await patchObjectPolicy.mutateAsync([{ object_id: row.id, mode }]) + }} + /> + ), + }, + ]} + /> + + + + + ({ value: c.check_key, label: c.check_key }))} + /> + setOverrideCategoriesMode(v as OverrideCategoriesMode)} + options={[ + { value: 'inherit', label: 'inherit' }, + { value: 'custom', label: 'custom' }, + ]} + /> +