From 48b1b737f33cd02b6bc205cf9942c9bb2cb190dc Mon Sep 17 00:00:00 2001 From: dv Date: Wed, 6 May 2026 11:46:53 +0300 Subject: [PATCH] 123 --- backend/app/main.py | 1 + backend/app/services/dashboard_monitor.py | 107 +++++++++++----------- 2 files changed, 57 insertions(+), 51 deletions(-) diff --git a/backend/app/main.py b/backend/app/main.py index 560ed18..b324c96 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -18,6 +18,7 @@ logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', ) +logging.getLogger('app.services.dashboard_monitor').setLevel(logging.DEBUG) @asynccontextmanager diff --git a/backend/app/services/dashboard_monitor.py b/backend/app/services/dashboard_monitor.py index ddedee4..c0a1d2d 100644 --- a/backend/app/services/dashboard_monitor.py +++ b/backend/app/services/dashboard_monitor.py @@ -362,24 +362,24 @@ async def _run_check_on_device( check: DashboardCheckDefinition, device: Device, obj: Object, -) -> tuple[bool, str, dict[str, Any] | None, str | None, datetime, int]: +) -> tuple[Device, 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 + return device, 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 + return device, 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 + return device, result.success, status, value_json, error, datetime.now(timezone.utc), duration_ms def _effective_check_settings( @@ -537,11 +537,57 @@ async def run_dashboard_monitor_cycle( if not due_devices: continue + # Группируем для логирования + by_object: dict[int, list[str]] = {} + for d in due_devices: + by_object.setdefault(d.object_id, []).append(d.ip) + for obj_id, ips in by_object.items(): + obj_name = next((o.name for o in objects if o.id == obj_id), str(obj_id)) + logger.info( + "Dashboard monitor [%s] %s: %d устройств → %s", + check.check_key, obj_name, len(ips), + ", ".join(ips[:10]) + ("…" if len(ips) > 10 else ""), + ) + 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, - ) + tasks = [ + asyncio.ensure_future(_run_check_on_device(sem, check, device, device.object)) + for device in due_devices + ] + + completed_count = 0 + failed_count = 0 + results = [] + + for coro in asyncio.as_completed(tasks): + result = await coro + results.append(result) + completed_count += 1 + device_done = result[0] + success_done = result[1] + if not success_done: + failed_count += 1 + logger.debug( + "Dashboard monitor [%s] %s (%s): %d/%d done", + check.check_key, device_done.ip, + "ok" if success_done else "fail", + completed_count, len(due_devices), + ) + # Шлём промежуточный прогресс каждые 5 устройств или на последнем + if completed_count % 5 == 0 or completed_count == len(due_devices): + await sse_manager.publish( + SSEEvent( + type="dashboard_monitor_update", + task_id=DASHBOARD_MONITOR_CHANNEL, + data={ + "check_key": check.check_key, + "processed": completed_count, + "due": len(due_devices), + "failed": failed_count, + "ts": datetime.now(timezone.utc).isoformat(), + }, + ) + ) async with AsyncSessionLocal() as dbw: async with dbw.begin(): @@ -562,34 +608,8 @@ async def run_dashboard_monitor_cycle( dbw, [d.id for d in due_devices] ) - 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 + for result in results: + device, success, status, value_json, error, checked_at, duration_ms = result counts["checks_executed"] += 1 counts["devices_checked"] += 1 if not success: @@ -639,21 +659,6 @@ async def run_dashboard_monitor_cycle( now=checked_at, ) - check_failed = sum(1 for r in results if isinstance(r, Exception) or (not isinstance(r, Exception) and not r[0])) - await sse_manager.publish( - SSEEvent( - type="dashboard_monitor_update", - task_id=DASHBOARD_MONITOR_CHANNEL, - data={ - "check_key": check.check_key, - "processed": len(due_devices), - "due": len(due_devices), - "failed": check_failed, - "ts": datetime.now(timezone.utc).isoformat(), - }, - ) - ) - async with AsyncSessionLocal() as db: async with db.begin(): run_state = await _get_run_state(db)