123
This commit is contained in:
parent
b230025ab4
commit
48b1b737f3
2 changed files with 57 additions and 51 deletions
|
|
@ -18,6 +18,7 @@ logging.basicConfig(
|
||||||
level=logging.INFO,
|
level=logging.INFO,
|
||||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||||||
)
|
)
|
||||||
|
logging.getLogger('app.services.dashboard_monitor').setLevel(logging.DEBUG)
|
||||||
|
|
||||||
|
|
||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
|
|
|
||||||
|
|
@ -362,24 +362,24 @@ async def _run_check_on_device(
|
||||||
check: DashboardCheckDefinition,
|
check: DashboardCheckDefinition,
|
||||||
device: Device,
|
device: Device,
|
||||||
obj: Object,
|
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:
|
async with sem:
|
||||||
action = action_registry.get(check.plugin_action)
|
action = action_registry.get(check.plugin_action)
|
||||||
now = datetime.now(timezone.utc)
|
now = datetime.now(timezone.utc)
|
||||||
if action is None:
|
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()
|
started = time.monotonic()
|
||||||
try:
|
try:
|
||||||
result = await action.execute(device, obj, {}, _noop_emit)
|
result = await action.execute(device, obj, {}, _noop_emit)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
duration_ms = int((time.monotonic() - started) * 1000)
|
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)
|
duration_ms = int((time.monotonic() - started) * 1000)
|
||||||
status, value_json = _derive_status(check.check_key, result.success, result.data or {})
|
status, value_json = _derive_status(check.check_key, result.success, result.data or {})
|
||||||
error = result.error or None
|
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(
|
def _effective_check_settings(
|
||||||
|
|
@ -537,11 +537,57 @@ async def run_dashboard_monitor_cycle(
|
||||||
if not due_devices:
|
if not due_devices:
|
||||||
continue
|
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))
|
sem = asyncio.Semaphore(max(1, check.concurrency))
|
||||||
results = await asyncio.gather(
|
tasks = [
|
||||||
*[_run_check_on_device(sem, check, device, device.object) for device in due_devices],
|
asyncio.ensure_future(_run_check_on_device(sem, check, device, device.object))
|
||||||
return_exceptions=True,
|
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 AsyncSessionLocal() as dbw:
|
||||||
async with dbw.begin():
|
async with dbw.begin():
|
||||||
|
|
@ -562,34 +608,8 @@ async def run_dashboard_monitor_cycle(
|
||||||
dbw, [d.id for d in due_devices]
|
dbw, [d.id for d in due_devices]
|
||||||
)
|
)
|
||||||
|
|
||||||
for device, result in zip(due_devices, results):
|
for result in results:
|
||||||
if isinstance(result, Exception):
|
device, success, status, value_json, error, checked_at, duration_ms = result
|
||||||
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["checks_executed"] += 1
|
||||||
counts["devices_checked"] += 1
|
counts["devices_checked"] += 1
|
||||||
if not success:
|
if not success:
|
||||||
|
|
@ -639,21 +659,6 @@ async def run_dashboard_monitor_cycle(
|
||||||
now=checked_at,
|
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 AsyncSessionLocal() as db:
|
||||||
async with db.begin():
|
async with db.begin():
|
||||||
run_state = await _get_run_state(db)
|
run_state = await _get_run_state(db)
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue