diff --git a/ai/tasks/pending/TASK-120-integrate-adminlog-ingestion-into-rcon-historical-worker.md b/ai/tasks/done/TASK-120-integrate-adminlog-ingestion-into-rcon-historical-worker.md similarity index 75% rename from ai/tasks/pending/TASK-120-integrate-adminlog-ingestion-into-rcon-historical-worker.md rename to ai/tasks/done/TASK-120-integrate-adminlog-ingestion-into-rcon-historical-worker.md index d6b7176..b1131ff 100644 --- a/ai/tasks/pending/TASK-120-integrate-adminlog-ingestion-into-rcon-historical-worker.md +++ b/ai/tasks/done/TASK-120-integrate-adminlog-ingestion-into-rcon-historical-worker.md @@ -1,7 +1,7 @@ --- id: TASK-120 title: Integrate AdminLog ingestion into RCON historical worker -status: pending +status: done type: backend team: Backend Senior supporting_teams: @@ -81,3 +81,19 @@ HLL Vietnam is moving to an RCON-first historical/live data platform. Manual Adm - Stage only intended files. - Commit the completed implementation. - Push the branch to origin. + +## Outcome + +- Integrated AdminLog ingestion into the RCON historical capture worker using the existing ingestion/storage path. +- Added `HLL_BACKEND_RCON_ADMIN_LOG_LOOKBACK_MINUTES`, defaulting to 60 minutes. +- Added worker result totals for AdminLog events seen, inserted, duplicated and failed targets. +- Verified a temporary bad RCON target reports one session failure and one AdminLog failure while a valid target still captures successfully. + +## Validation Result + +- `python -m compileall backend/app` passed. +- `docker compose up -d --build backend rcon-historical-worker` passed. +- `docker compose exec backend python -m app.rcon_historical_worker capture` passed and returned AdminLog metrics. +- `docker compose exec backend python -m app.rcon_admin_log_ingestion --minutes 1440` run twice; second run reported 143 duplicate events and 0 inserted events. +- `/health` returned `status: ok`. +- `git diff --name-only` matched the expected task scope. diff --git a/backend/app/rcon_historical_worker.py b/backend/app/rcon_historical_worker.py index 1516930..940c169 100644 --- a/backend/app/rcon_historical_worker.py +++ b/backend/app/rcon_historical_worker.py @@ -4,6 +4,7 @@ from __future__ import annotations import argparse import json +import os import time from dataclasses import dataclass from typing import Iterable @@ -14,6 +15,7 @@ from .config import ( get_rcon_historical_capture_retry_delay_seconds, get_rcon_request_timeout_seconds, ) +from .rcon_admin_log_ingestion import ingest_rcon_admin_logs from .rcon_client import ( RconQueryError, build_rcon_target_key, @@ -38,6 +40,10 @@ class RconHistoricalCaptureStats: samples_inserted: int = 0 duplicate_samples: int = 0 failed_targets: int = 0 + admin_log_events_seen: int = 0 + admin_log_events_inserted: int = 0 + admin_log_duplicate_events: int = 0 + admin_log_failed_targets: int = 0 def run_rcon_historical_capture( @@ -60,12 +66,14 @@ def run_rcon_historical_capture_unlocked( """Capture one prospective RCON sample assuming the shared writer lock is already held.""" initialize_rcon_historical_storage() selected_targets = _select_targets(target_key) + admin_log_lookback_minutes = get_rcon_admin_log_lookback_minutes() captured_at = utc_now().isoformat().replace("+00:00", "Z") target_scope = target_key or "all-configured-rcon-targets" run_id = start_rcon_historical_capture_run(mode="capture", target_scope=target_scope) stats = RconHistoricalCaptureStats() items: list[dict[str, object]] = [] errors: list[dict[str, object]] = [] + admin_log_errors: list[dict[str, object]] = [] timeout_seconds = get_rcon_request_timeout_seconds() try: @@ -108,6 +116,17 @@ def run_rcon_historical_capture_unlocked( ) errors.append(_serialize_capture_error(target, exc, timeout_seconds=timeout_seconds)) + admin_log_result = _ingest_target_admin_log( + target_key=str(target_metadata["target_key"]), + minutes=admin_log_lookback_minutes, + ) + _merge_admin_log_result( + stats=stats, + admin_log_errors=admin_log_errors, + target=target_metadata, + result=admin_log_result, + ) + status = "success" if not errors else ("partial" if items else "failed") finalize_rcon_historical_capture_run( run_id, @@ -135,14 +154,20 @@ def run_rcon_historical_capture_unlocked( "run_status": status, "captured_at": captured_at, "target_scope": target_scope, + "admin_log_lookback_minutes": admin_log_lookback_minutes, "targets": items, "errors": errors, + "admin_log_errors": admin_log_errors, "storage_status": list_rcon_historical_target_statuses(), "totals": { "targets_seen": stats.targets_seen, "samples_inserted": stats.samples_inserted, "duplicate_samples": stats.duplicate_samples, "failed_targets": stats.failed_targets, + "admin_log_events_seen": stats.admin_log_events_seen, + "admin_log_events_inserted": stats.admin_log_events_inserted, + "admin_log_duplicate_events": stats.admin_log_duplicate_events, + "admin_log_failed_targets": stats.admin_log_failed_targets, }, } @@ -231,6 +256,72 @@ def _select_targets(target_key: str | None) -> list[object]: return selected +def get_rcon_admin_log_lookback_minutes() -> int: + """Return the AdminLog lookback window used by periodic RCON capture.""" + configured_value = os.getenv("HLL_BACKEND_RCON_ADMIN_LOG_LOOKBACK_MINUTES", "60") + lookback_minutes = int(configured_value) + if lookback_minutes <= 0: + raise ValueError("HLL_BACKEND_RCON_ADMIN_LOG_LOOKBACK_MINUTES must be positive.") + return lookback_minutes + + +def _ingest_target_admin_log( + *, + target_key: str, + minutes: int, +) -> dict[str, object]: + try: + return ingest_rcon_admin_logs(minutes=minutes, target_key=target_key) + except Exception as exc: # noqa: BLE001 - worker reports per-target AdminLog failures + return { + "status": "error", + "errors": [ + { + "target_key": target_key, + "status": "error", + "error_type": type(exc).__name__, + "message": str(exc), + } + ], + "totals": { + "events_seen": 0, + "events_inserted": 0, + "duplicate_events": 0, + "failed_targets": 1, + }, + } + + +def _merge_admin_log_result( + *, + stats: RconHistoricalCaptureStats, + admin_log_errors: list[dict[str, object]], + target: dict[str, object], + result: dict[str, object], +) -> None: + totals = result.get("totals") + if isinstance(totals, dict): + stats.admin_log_events_seen += int(totals.get("events_seen") or 0) + stats.admin_log_events_inserted += int(totals.get("events_inserted") or 0) + stats.admin_log_duplicate_events += int(totals.get("duplicate_events") or 0) + stats.admin_log_failed_targets += int(totals.get("failed_targets") or 0) + + errors = result.get("errors") + if isinstance(errors, list): + for error in errors: + if isinstance(error, dict): + admin_log_errors.append( + { + "target_key": target["target_key"], + "external_server_id": target.get("external_server_id"), + "name": target.get("name"), + "status": "error", + "error_type": error.get("error_type"), + "message": error.get("message"), + } + ) + + def _serialize_target(target: object) -> dict[str, object]: return { "target_key": build_rcon_target_key(target),