фикс воркеров

backend/app/services/dashboard_monitor.py

@@ -11,6 +11,7 @@ from sqlalchemy.orm import selectinload
 from app.config import settings
 
 logger = logging.getLogger(__name__)
+from app.models.alert import Alert
 from app.models.dashboard import (
     DashboardCheckDefinition,
     DashboardDeviceCheckHistory,
@@ -26,7 +27,7 @@ from app.models.object import Object
 from app.models.rack import Rack
 from app.models.tag import Tag
 from app.plugins.registry import action_registry
-from app.services.database import AsyncSessionLocal
+from app.services.database import MonitorSessionLocal as AsyncSessionLocal
 from app.services.sse import SSEEvent, sse_manager
 
 DASHBOARD_MONITOR_CHANNEL = "__dashboard_monitor__"
@@ -530,6 +531,23 @@ async def run_dashboard_monitor_cycle(
 
                     async with AsyncSessionLocal() as dbw:
                         async with dbw.begin():
+                            # Загружаем предыдущие статусы и baseline только для ping
+                            prev_status_map: dict[int, str | None] = {}
+                            baseline_map_for_check: dict[int, str] = {}
+                            if check.check_key == "ping":
+                                prev_state_rows = (
+                                    await dbw.execute(
+                                        select(DashboardDeviceLiveState).where(
+                                            DashboardDeviceLiveState.check_key == "ping",
+                                            DashboardDeviceLiveState.device_id.in_([d.id for d in due_devices]),
+                                        )
+                                    )
+                                ).scalars().all()
+                                prev_status_map = {row.device_id: row.status for row in prev_state_rows}
+                                baseline_map_for_check = await _load_baseline_map_for_devices(
+                                    dbw, [d.id for d in due_devices]
+                                )
+
                             for device, result in zip(due_devices, results):
                                 if isinstance(result, Exception):
                                     counts["failed"] += 1
@@ -595,6 +613,18 @@ async def run_dashboard_monitor_cycle(
                                             if ping_ms is not None:
                                                 device_row.last_ping_ms = int(ping_ms)
 
+                                if check.check_key == "ping":
+                                    bl = baseline_map_for_check.get(device.id)
+                                    baseline_suppressed = bl in ("offline", "unknown") and status != "online"
+                                    await _process_ping_alerts(
+                                        dbw,
+                                        device_id=device.id,
+                                        prev_status=prev_status_map.get(device.id),
+                                        new_status=status,
+                                        baseline_suppressed=baseline_suppressed,
+                                        now=checked_at,
+                                    )
+
                     await sse_manager.publish(
                         SSEEvent(
                             type="dashboard_monitor_update",
@@ -998,7 +1028,67 @@ def _device_noc_item(
     }
 
 
-async def save_object_baseline(object_id: int) -> int:
+async def _process_ping_alerts(
+    db,
+    *,
+    device_id: int,
+    prev_status: str | None,
+    new_status: str,
+    baseline_suppressed: bool,
+    now: datetime,
+) -> None:
+    """Создаёт алерт при переходе в offline, резолвит при возврате в online."""
+    if new_status == "offline" and prev_status != "offline" and not baseline_suppressed:
+        # Проверяем нет ли уже открытого алерта
+        existing = (
+            await db.execute(
+                select(Alert).where(
+                    Alert.device_id == device_id,
+                    Alert.status.in_(["open", "acknowledged"]),
+                )
+            )
+        ).scalar_one_or_none()
+        if existing is None:
+            db.add(Alert(
+                device_id=device_id,
+                status="open",
+                message="Устройство недоступно (ping offline)",
+                created_at=now,
+            ))
+            logger.info("Alert created for device_id=%d (offline)", device_id)
+
+    elif new_status == "online" and prev_status == "offline":
+        # Резолвим все открытые алерты устройства
+        open_alerts = (
+            await db.execute(
+                select(Alert).where(
+                    Alert.device_id == device_id,
+                    Alert.status.in_(["open", "acknowledged"]),
+                )
+            )
+        ).scalars().all()
+        for alert in open_alerts:
+            alert.status = "resolved"
+            alert.resolved_at = now
+        if open_alerts:
+            logger.info("Alert resolved for device_id=%d (back online)", device_id)
+
+
+async def _load_baseline_map_for_devices(db, device_ids: list[int]) -> dict[int, str]:
+    """Загружает baseline статусы для списка устройств."""
+    if not device_ids:
+        return {}
+    rows = (
+        await db.execute(
+            select(DashboardObjectStatusBaseline).where(
+                DashboardObjectStatusBaseline.device_id.in_(device_ids)
+            )
+        )
+    ).scalars().all()
+    return {row.device_id: row.baseline_status for row in rows}
+
+
+
     """Сохраняет текущие ping-статусы всех устройств объекта как baseline. Возвращает кол-во записей."""
     now = datetime.now(timezone.utc)
     async with AsyncSessionLocal() as db:

backend/app/services/database.py

@@ -14,6 +14,16 @@ engine = create_async_engine(
     pool_timeout=settings.DATABASE_POOL_TIMEOUT,
 )
 
+# Отдельный пул для фонового монитора — не конкурирует с HTTP-запросами
+monitor_engine = create_async_engine(
+    settings.DATABASE_URL,
+    echo=False,
+    pool_pre_ping=True,
+    pool_size=3,
+    max_overflow=2,
+    pool_timeout=60,
+)
+
 
 #if not settings.SQLALCHEMY_LOG:
 #    for _log_name in ("sqlalchemy.engine", "sqlalchemy.pool", "sqlalchemy.dialects", "sqlalchemy.orm"):
@@ -36,3 +46,10 @@ AsyncSessionLocal = async_sessionmaker(
     expire_on_commit=False,
     autoflush=False,
 )
+
+MonitorSessionLocal = async_sessionmaker(
+    bind=monitor_engine,
+    class_=AsyncSession,
+    expire_on_commit=False,
+    autoflush=False,
+)