feat: persist rcon admin log events
This commit is contained in:
devRaGonSa
147
backend/app/rcon_admin_log_ingestion.py
Normal file
147
backend/app/rcon_admin_log_ingestion.py
Normal file
@@ -0,0 +1,147 @@
|
||||
"""Manual ingestion of Hell Let Loose RCON AdminLog events."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
|
||||
from .config import get_rcon_request_timeout_seconds
|
||||
from .rcon_admin_log_storage import (
|
||||
list_rcon_admin_log_event_counts,
|
||||
persist_rcon_admin_log_entries,
|
||||
)
|
||||
from .rcon_client import HllRconConnection, build_rcon_target_key, load_rcon_targets
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class AdminLogIngestionStats:
|
||||
targets_seen: int = 0
|
||||
events_seen: int = 0
|
||||
events_inserted: int = 0
|
||||
duplicate_events: int = 0
|
||||
failed_targets: int = 0
|
||||
|
||||
|
||||
def ingest_rcon_admin_logs(
|
||||
*,
|
||||
minutes: int,
|
||||
target_key: str | None = None,
|
||||
) -> dict[str, object]:
|
||||
"""Fetch and persist recent AdminLog entries from configured RCON targets."""
|
||||
selected_targets = _select_targets(target_key)
|
||||
stats = AdminLogIngestionStats()
|
||||
targets: list[dict[str, object]] = []
|
||||
errors: list[dict[str, object]] = []
|
||||
timeout_seconds = get_rcon_request_timeout_seconds()
|
||||
|
||||
for target in selected_targets:
|
||||
stats.targets_seen += 1
|
||||
target_metadata = _serialize_target(target)
|
||||
|
||||
try:
|
||||
with HllRconConnection(timeout_seconds=timeout_seconds) as connection:
|
||||
connection.connect(host=target.host, port=target.port, password=target.password)
|
||||
payload = connection.execute_json(
|
||||
"GetAdminLog",
|
||||
{
|
||||
"LogBackTrackTime": minutes * 60,
|
||||
"Filters": [],
|
||||
},
|
||||
)
|
||||
|
||||
entries = payload.get("entries")
|
||||
if not isinstance(entries, list):
|
||||
entries = []
|
||||
|
||||
normalized_entries = [entry for entry in entries if isinstance(entry, dict)]
|
||||
delta = persist_rcon_admin_log_entries(
|
||||
target=target_metadata,
|
||||
entries=normalized_entries,
|
||||
)
|
||||
|
||||
stats.events_seen += int(delta["events_seen"])
|
||||
stats.events_inserted += int(delta["events_inserted"])
|
||||
stats.duplicate_events += int(delta["duplicate_events"])
|
||||
targets.append(
|
||||
{
|
||||
**target_metadata,
|
||||
"status": "ok",
|
||||
"minutes": minutes,
|
||||
**delta,
|
||||
}
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001 - manual diagnostic command reports per-target failures
|
||||
stats.failed_targets += 1
|
||||
errors.append(
|
||||
{
|
||||
**target_metadata,
|
||||
"status": "error",
|
||||
"error_type": type(exc).__name__,
|
||||
"message": str(exc),
|
||||
}
|
||||
)
|
||||
|
||||
return {
|
||||
"status": "ok" if not errors else ("partial" if targets else "error"),
|
||||
"target_scope": target_key or "all-configured-rcon-targets",
|
||||
"minutes": minutes,
|
||||
"targets": targets,
|
||||
"errors": errors,
|
||||
"totals": {
|
||||
"targets_seen": stats.targets_seen,
|
||||
"events_seen": stats.events_seen,
|
||||
"events_inserted": stats.events_inserted,
|
||||
"duplicate_events": stats.duplicate_events,
|
||||
"failed_targets": stats.failed_targets,
|
||||
},
|
||||
"event_counts": list_rcon_admin_log_event_counts(),
|
||||
}
|
||||
|
||||
|
||||
def _select_targets(target_key: str | None) -> list[object]:
|
||||
configured_targets = list(load_rcon_targets())
|
||||
if not configured_targets:
|
||||
raise RuntimeError("No RCON targets configured in HLL_BACKEND_RCON_TARGETS.")
|
||||
if target_key is None:
|
||||
return configured_targets
|
||||
|
||||
normalized = target_key.strip()
|
||||
selected = [
|
||||
target
|
||||
for target in configured_targets
|
||||
if build_rcon_target_key(target) == normalized
|
||||
]
|
||||
if not selected:
|
||||
raise ValueError(f"Unknown RCON target key: {target_key}")
|
||||
return selected
|
||||
|
||||
|
||||
def _serialize_target(target: object) -> dict[str, object]:
|
||||
return {
|
||||
"target_key": build_rcon_target_key(target),
|
||||
"external_server_id": target.external_server_id,
|
||||
"name": target.name,
|
||||
"host": target.host,
|
||||
"port": target.port,
|
||||
"source_name": target.source_name,
|
||||
}
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--minutes", type=int, default=60)
|
||||
parser.add_argument("--target", default=None)
|
||||
args = parser.parse_args()
|
||||
|
||||
print(
|
||||
json.dumps(
|
||||
ingest_rcon_admin_logs(minutes=args.minutes, target_key=args.target),
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
127
backend/app/rcon_admin_log_storage.py
Normal file
127
backend/app/rcon_admin_log_storage.py
Normal file
@@ -0,0 +1,127 @@
|
||||
"""Storage helpers for parsed RCON AdminLog events."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
from collections.abc import Mapping
|
||||
from pathlib import Path
|
||||
|
||||
from .config import get_storage_path
|
||||
from .rcon_admin_log_parser import parse_rcon_admin_log_entry
|
||||
from .rcon_historical_storage import initialize_rcon_historical_storage
|
||||
from .sqlite_utils import connect_sqlite_writer
|
||||
|
||||
|
||||
def initialize_rcon_admin_log_storage(*, db_path: Path | None = None) -> Path:
|
||||
"""Create SQLite structures for parsed RCON AdminLog events."""
|
||||
resolved_path = initialize_rcon_historical_storage(db_path=db_path)
|
||||
|
||||
with connect_sqlite_writer(resolved_path) as connection:
|
||||
connection.executescript(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS rcon_admin_log_events (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
target_key TEXT NOT NULL,
|
||||
external_server_id TEXT,
|
||||
event_timestamp TEXT,
|
||||
server_time INTEGER,
|
||||
relative_time TEXT,
|
||||
event_type TEXT NOT NULL,
|
||||
raw_message TEXT NOT NULL,
|
||||
parsed_payload_json TEXT NOT NULL,
|
||||
raw_entry_json TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
UNIQUE(target_key, server_time, raw_message)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_rcon_admin_log_events_target_time
|
||||
ON rcon_admin_log_events(target_key, server_time DESC);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_rcon_admin_log_events_type
|
||||
ON rcon_admin_log_events(event_type);
|
||||
"""
|
||||
)
|
||||
|
||||
return resolved_path
|
||||
|
||||
|
||||
def persist_rcon_admin_log_entries(
|
||||
*,
|
||||
target: Mapping[str, object],
|
||||
entries: list[dict[str, object]],
|
||||
db_path: Path | None = None,
|
||||
) -> dict[str, int]:
|
||||
"""Persist raw and parsed AdminLog entries idempotently."""
|
||||
resolved_path = initialize_rcon_admin_log_storage(db_path=db_path)
|
||||
target_key = str(target.get("target_key") or target.get("external_server_id") or "")
|
||||
if not target_key:
|
||||
raise ValueError("target must include target_key or external_server_id")
|
||||
|
||||
external_server_id = target.get("external_server_id")
|
||||
inserted = 0
|
||||
duplicates = 0
|
||||
|
||||
with connect_sqlite_writer(resolved_path) as connection:
|
||||
for entry in entries:
|
||||
parsed = parse_rcon_admin_log_entry(entry)
|
||||
cursor = connection.execute(
|
||||
"""
|
||||
INSERT OR IGNORE INTO rcon_admin_log_events (
|
||||
target_key,
|
||||
external_server_id,
|
||||
event_timestamp,
|
||||
server_time,
|
||||
relative_time,
|
||||
event_type,
|
||||
raw_message,
|
||||
parsed_payload_json,
|
||||
raw_entry_json
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(
|
||||
target_key,
|
||||
external_server_id,
|
||||
parsed.get("timestamp"),
|
||||
parsed.get("server_time"),
|
||||
parsed.get("relative_time"),
|
||||
parsed.get("event_type") or "unknown",
|
||||
parsed.get("raw_message") or "",
|
||||
json.dumps(parsed, ensure_ascii=False, separators=(",", ":")),
|
||||
json.dumps(entry, ensure_ascii=False, separators=(",", ":")),
|
||||
),
|
||||
)
|
||||
if int(cursor.rowcount or 0):
|
||||
inserted += 1
|
||||
else:
|
||||
duplicates += 1
|
||||
|
||||
return {
|
||||
"events_seen": len(entries),
|
||||
"events_inserted": inserted,
|
||||
"duplicate_events": duplicates,
|
||||
}
|
||||
|
||||
|
||||
def list_rcon_admin_log_event_counts(*, db_path: Path | None = None) -> list[dict[str, object]]:
|
||||
"""Return event counts grouped by target and event type."""
|
||||
resolved_path = db_path or get_storage_path()
|
||||
initialize_rcon_admin_log_storage(db_path=resolved_path)
|
||||
|
||||
with sqlite3.connect(resolved_path) as connection:
|
||||
connection.row_factory = sqlite3.Row
|
||||
rows = connection.execute(
|
||||
"""
|
||||
SELECT
|
||||
target_key,
|
||||
event_type,
|
||||
COUNT(*) AS event_count,
|
||||
MIN(server_time) AS first_server_time,
|
||||
MAX(server_time) AS last_server_time
|
||||
FROM rcon_admin_log_events
|
||||
GROUP BY target_key, event_type
|
||||
ORDER BY target_key ASC, event_count DESC
|
||||
"""
|
||||
).fetchall()
|
||||
|
||||
return [dict(row) for row in rows]
|
||||
Reference in New Issue
Block a user