rgonsal/comunidadhll
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add current match AdminLog freshness worker

Browse Source
This commit is contained in:
devRaGonSa
2026-06-18 09:23:21 +02:00
parent 7fcd307ec0
commit 7f37d198e4
6 changed files with 788 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
backend/app/config.py
View File

@@ -47,6 +47,9 @@ DEFAULT_RCON_HISTORICAL_CAPTURE_MAX_RETRIES = 2
DEFAULT_RCON_HISTORICAL_CAPTURE_RETRY_DELAY_SECONDS = 15
DEFAULT_RCON_CURRENT_MATCH_CAPTURE_INTERVAL_SECONDS = 5
DEFAULT_RCON_CURRENT_MATCH_WRITER_LOCK_TIMEOUT_SECONDS = 4.0
DEFAULT_CURRENT_MATCH_ADMINLOG_INTERVAL_SECONDS = 10
DEFAULT_CURRENT_MATCH_ADMINLOG_LOOKBACK_SECONDS = 180
DEFAULT_CURRENT_MATCH_ADMINLOG_ENABLED = False
DEFAULT_KPM_MIN_ACTIVE_SECONDS = 60
DEFAULT_RCON_BACKFILL_CHUNK_HOURS = 6
DEFAULT_RCON_BACKFILL_SLEEP_SECONDS = 1.0
@@ -678,6 +681,32 @@ def get_rcon_current_match_writer_lock_timeout_seconds() -> float:
)
def get_current_match_adminlog_interval_seconds() -> int:
"""Return the dedicated current-match AdminLog worker interval."""
return _read_int_env(
"CURRENT_MATCH_ADMINLOG_INTERVAL_SECONDS",
str(DEFAULT_CURRENT_MATCH_ADMINLOG_INTERVAL_SECONDS),
minimum=1,
)
def get_current_match_adminlog_lookback_seconds() -> int:
"""Return the overlap-safe AdminLog lookback for the dedicated current-match worker."""
return _read_int_env(
"CURRENT_MATCH_ADMINLOG_LOOKBACK_SECONDS",
str(DEFAULT_CURRENT_MATCH_ADMINLOG_LOOKBACK_SECONDS),
minimum=1,
)
def get_current_match_adminlog_enabled() -> bool:
"""Return whether a deployment should enable the dedicated current-match worker."""
return _read_bool_env(
"CURRENT_MATCH_ADMINLOG_ENABLED",
default=DEFAULT_CURRENT_MATCH_ADMINLOG_ENABLED,
)
def get_kpm_min_active_seconds() -> int:
"""Return the minimum observed active seconds required before KPM is considered valid."""
return _read_int_env(

53
backend/app/rcon_admin_log_ingestion.py
View File

@@ -11,7 +11,7 @@ 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
from .rcon_client import HllRconConnection, RconServerTarget, build_rcon_target_key, load_rcon_targets
@dataclass(slots=True)
@@ -37,24 +37,14 @@ def ingest_rcon_admin_logs(
for target in selected_targets:
stats.targets_seen += 1
target_metadata = _serialize_target(target)
target_metadata = serialize_rcon_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)]
normalized_entries = fetch_recent_admin_log_entries(
target,
lookback_seconds=minutes * 60,
timeout_seconds=timeout_seconds,
)
delta = persist_rcon_admin_log_entries(
target=target_metadata,
entries=normalized_entries,
@@ -117,7 +107,34 @@ def _select_targets(target_key: str | None) -> list[object]:
return selected
def _serialize_target(target: object) -> dict[str, object]:
def fetch_recent_admin_log_entries(
target: RconServerTarget,
*,
lookback_seconds: int,
timeout_seconds: float | None = None,
) -> list[dict[str, object]]:
"""Fetch recent raw AdminLog entries for one configured target."""
if lookback_seconds <= 0:
raise ValueError("lookback_seconds must be positive.")
resolved_timeout = (
get_rcon_request_timeout_seconds() if timeout_seconds is None else timeout_seconds
)
with HllRconConnection(timeout_seconds=resolved_timeout) as connection:
connection.connect(host=target.host, port=target.port, password=target.password)
payload = connection.execute_json(
"GetAdminLog",
{
"LogBackTrackTime": lookback_seconds,
"Filters": [],
},
)
entries = payload.get("entries")
if not isinstance(entries, list):
return []
return [entry for entry in entries if isinstance(entry, dict)]
def serialize_rcon_target(target: object) -> dict[str, object]:
return {
"target_key": build_rcon_target_key(target),
"external_server_id": target.external_server_id,

256
backend/app/rcon_current_match_worker.py Normal file
View File

@@ -0,0 +1,256 @@
"""Dedicated lightweight AdminLog freshness worker for current-match pages."""
from __future__ import annotations
import argparse
import json
import time
from collections.abc import Callable, Iterable, Sequence
from datetime import datetime, timezone
from pathlib import Path
from .config import (
get_current_match_adminlog_enabled,
get_current_match_adminlog_interval_seconds,
get_current_match_adminlog_lookback_seconds,
get_rcon_current_match_writer_lock_timeout_seconds,
)
from .rcon_admin_log_ingestion import fetch_recent_admin_log_entries, serialize_rcon_target
from .rcon_admin_log_storage import persist_rcon_admin_log_entries
from .rcon_client import RconServerTarget, build_rcon_target_key, load_rcon_targets
from .scoreboard_origins import list_trusted_public_scoreboard_origins
from .writer_lock import backend_writer_lock, build_writer_lock_holder
def list_current_match_trusted_targets() -> list[RconServerTarget]:
"""Return only the configured RCON targets trusted for public current-match pages."""
trusted_keys = {
origin.slug for origin in list_trusted_public_scoreboard_origins()
}
return [
target
for target in load_rcon_targets()
if build_rcon_target_key(target) in trusted_keys
]
def run_current_match_adminlog_refresh_once(
*,
lookback_seconds: int | None = None,
targets: Sequence[RconServerTarget] | None = None,
fetch_entries_fn: Callable[..., list[dict[str, object]]] = fetch_recent_admin_log_entries,
persist_entries_fn: Callable[..., dict[str, int]] = persist_rcon_admin_log_entries,
db_path: object = None,
) -> dict[str, object]:
"""Refresh recent AdminLog rows once for trusted current-match targets."""
with backend_writer_lock(
holder=build_writer_lock_holder("app.rcon_current_match_worker once"),
timeout_seconds=get_rcon_current_match_writer_lock_timeout_seconds(),
):
return run_current_match_adminlog_refresh_once_unlocked(
lookback_seconds=lookback_seconds,
targets=targets,
fetch_entries_fn=fetch_entries_fn,
persist_entries_fn=persist_entries_fn,
db_path=db_path,
)
def run_current_match_adminlog_refresh_once_unlocked(
*,
lookback_seconds: int | None = None,
targets: Sequence[RconServerTarget] | None = None,
fetch_entries_fn: Callable[..., list[dict[str, object]]] = fetch_recent_admin_log_entries,
persist_entries_fn: Callable[..., dict[str, int]] = persist_rcon_admin_log_entries,
db_path: object = None,
) -> dict[str, object]:
"""Refresh recent AdminLog rows once assuming the shared writer lock is already held."""
resolved_lookback_seconds = (
get_current_match_adminlog_lookback_seconds()
if lookback_seconds is None
else int(lookback_seconds)
)
if resolved_lookback_seconds <= 0:
raise ValueError("lookback_seconds must be positive.")
resolved_db_path = Path(db_path) if isinstance(db_path, str) else db_path
selected_targets = list(targets) if targets is not None else list_current_match_trusted_targets()
if not selected_targets:
raise RuntimeError("No trusted current-match RCON targets are configured.")
timeout_seconds = None
refreshed_at = datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
items: list[dict[str, object]] = []
errors: list[dict[str, object]] = []
totals = {
"targets_seen": 0,
"entries_seen": 0,
"events_inserted": 0,
"duplicate_events": 0,
"failed_targets": 0,
}
for target in selected_targets:
totals["targets_seen"] += 1
target_metadata = serialize_rcon_target(target)
started = time.perf_counter()
try:
entries = fetch_entries_fn(
target,
lookback_seconds=resolved_lookback_seconds,
timeout_seconds=timeout_seconds,
)
delta = persist_entries_fn(
target=target_metadata,
entries=entries,
db_path=resolved_db_path,
)
duration_ms = round((time.perf_counter() - started) * 1000, 2)
totals["entries_seen"] += int(delta.get("events_seen") or 0)
totals["events_inserted"] += int(delta.get("events_inserted") or 0)
totals["duplicate_events"] += int(delta.get("duplicate_events") or 0)
items.append(
{
"target_key": target_metadata["target_key"],
"external_server_id": target_metadata["external_server_id"],
"name": target_metadata["name"],
"entries_seen": int(delta.get("events_seen") or 0),
"events_inserted": int(delta.get("events_inserted") or 0),
"duplicate_events": int(delta.get("duplicate_events") or 0),
"duration_ms": duration_ms,
}
)
except Exception as exc: # noqa: BLE001 - per-target failure must not kill the loop
totals["failed_targets"] += 1
errors.append(
{
"target_key": target_metadata["target_key"],
"external_server_id": target_metadata["external_server_id"],
"name": target_metadata["name"],
"error_type": type(exc).__name__,
"message": str(exc),
}
)
return {
"status": "ok" if not errors else ("partial" if items else "error"),
"worker_enabled": get_current_match_adminlog_enabled(),
"refreshed_at": refreshed_at,
"lookback_seconds": resolved_lookback_seconds,
"targets": items,
"errors": errors,
"totals": totals,
}
def run_current_match_adminlog_refresh_loop(
*,
interval_seconds: int | None = None,
lookback_seconds: int | None = None,
max_runs: int | None = None,
) -> None:
"""Run the lightweight current-match AdminLog refresher in a loop."""
resolved_interval_seconds = (
get_current_match_adminlog_interval_seconds()
if interval_seconds is None
else int(interval_seconds)
)
if resolved_interval_seconds <= 0:
raise ValueError("interval_seconds must be positive.")
if max_runs is not None and max_runs <= 0:
raise ValueError("max_runs must be positive when provided.")
run_count = 0
_emit_worker_event(
"current-match-adminlog-worker-started",
enabled=get_current_match_adminlog_enabled(),
interval_seconds=resolved_interval_seconds,
lookback_seconds=(
get_current_match_adminlog_lookback_seconds()
if lookback_seconds is None
else int(lookback_seconds)
),
targets=[
{
"target_key": build_rcon_target_key(target),
"external_server_id": target.external_server_id,
"name": target.name,
}
for target in list_current_match_trusted_targets()
],
)
try:
while max_runs is None or run_count < max_runs:
run_count += 1
result = run_current_match_adminlog_refresh_once(
lookback_seconds=lookback_seconds,
)
_emit_worker_event(
"current-match-adminlog-cycle-finished",
run=run_count,
result=result,
)
if max_runs is not None and run_count >= max_runs:
break
time.sleep(resolved_interval_seconds)
except KeyboardInterrupt:
_emit_worker_event("current-match-adminlog-worker-stopped", reason="keyboard-interrupt")
def build_arg_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Lightweight current-match AdminLog freshness worker.",
)
parser.add_argument(
"mode",
choices=("once", "loop"),
help="run once or keep polling trusted current-match targets",
)
parser.add_argument(
"--interval",
type=int,
default=get_current_match_adminlog_interval_seconds(),
help="seconds between loop iterations",
)
parser.add_argument(
"--lookback-seconds",
type=int,
default=get_current_match_adminlog_lookback_seconds(),
help="overlap-safe AdminLog lookback window in seconds",
)
parser.add_argument(
"--max-runs",
type=int,
help="optional safety cap for loop mode",
)
return parser
def main(argv: Iterable[str] | None = None) -> int:
parser = build_arg_parser()
args = parser.parse_args(list(argv) if argv is not None else None)
if args.mode == "once":
print(
json.dumps(
run_current_match_adminlog_refresh_once(
lookback_seconds=args.lookback_seconds,
),
indent=2,
)
)
return 0
run_current_match_adminlog_refresh_loop(
interval_seconds=args.interval,
lookback_seconds=args.lookback_seconds,
max_runs=args.max_runs,
)
return 0
def _emit_worker_event(event: str, **fields: object) -> None:
print(json.dumps({"event": event, **fields}, indent=2, default=str), flush=True)
if __name__ == "__main__":
raise SystemExit(main())

193
backend/tests/test_rcon_current_match_worker.py Normal file
View File

@@ -0,0 +1,193 @@
from __future__ import annotations
import os
import tempfile
import unittest
from contextlib import contextmanager
from types import SimpleNamespace
from unittest.mock import patch
from app.config import (
get_current_match_adminlog_enabled,
get_current_match_adminlog_interval_seconds,
get_current_match_adminlog_lookback_seconds,
)
from app.rcon_current_match_worker import (
list_current_match_trusted_targets,
run_current_match_adminlog_refresh_once_unlocked,
)
TARGET_01 = SimpleNamespace(
external_server_id="comunidad-hispana-01",
name="Comunidad Hispana #01",
host="203.0.113.10",
port=7779,
password="secret-01",
region="ES",
game_port=None,
query_port=None,
source_name="community-hispana-rcon",
)
TARGET_02 = SimpleNamespace(
external_server_id="comunidad-hispana-02",
name="Comunidad Hispana #02",
host="203.0.113.11",
port=7879,
password="secret-02",
region="ES",
game_port=None,
query_port=None,
source_name="community-hispana-rcon",
)
TARGET_03 = SimpleNamespace(
external_server_id="comunidad-hispana-03",
name="Comunidad Hispana #03",
host="203.0.113.12",
port=7979,
password="secret-03",
region="ES",
game_port=None,
query_port=None,
source_name="community-hispana-rcon",
)
class RconCurrentMatchWorkerTests(unittest.TestCase):
def test_list_current_match_trusted_targets_filters_only_01_and_02(self) -> None:
with patch(
"app.rcon_current_match_worker.load_rcon_targets",
return_value=(TARGET_01, TARGET_02, TARGET_03),
):
selected = list_current_match_trusted_targets()
self.assertEqual(
[target.external_server_id for target in selected],
["comunidad-hispana-01", "comunidad-hispana-02"],
)
def test_once_unlocked_calls_existing_persistence_path(self) -> None:
fetch_calls: list[object] = []
persist_calls: list[dict[str, object]] = []
def fake_fetch(target, *, lookback_seconds, timeout_seconds):
fetch_calls.append((target.external_server_id, lookback_seconds, timeout_seconds))
return [{"timestamp": "2026-06-18T18:00:00Z", "message": "[1 (1)] Killer(Allies) -> Victim(Axis) with Rifle"}]
def fake_persist(*, target, entries, db_path=None):
persist_calls.append({"target": target, "entries": entries, "db_path": db_path})
return {
"events_seen": len(entries),
"events_inserted": len(entries),
"duplicate_events": 0,
}
result = run_current_match_adminlog_refresh_once_unlocked(
lookback_seconds=180,
targets=[TARGET_01],
fetch_entries_fn=fake_fetch,
persist_entries_fn=fake_persist,
)
self.assertEqual(fetch_calls, [("comunidad-hispana-01", 180, None)])
self.assertEqual(len(persist_calls), 1)
self.assertEqual(persist_calls[0]["target"]["target_key"], "comunidad-hispana-01")
self.assertEqual(result["totals"]["events_inserted"], 1)
self.assertEqual(result["status"], "ok")
def test_failing_target_does_not_block_other_target(self) -> None:
def fake_fetch(target, *, lookback_seconds, timeout_seconds):
if target.external_server_id == "comunidad-hispana-01":
raise RuntimeError("boom")
return [{"timestamp": "2026-06-18T18:00:00Z", "message": "[1 (1)] Killer(Allies) -> Victim(Axis) with Rifle"}]
def fake_persist(*, target, entries, db_path=None):
return {
"events_seen": len(entries),
"events_inserted": len(entries),
"duplicate_events": 0,
}
result = run_current_match_adminlog_refresh_once_unlocked(
lookback_seconds=180,
targets=[TARGET_01, TARGET_02],
fetch_entries_fn=fake_fetch,
persist_entries_fn=fake_persist,
)
self.assertEqual(result["status"], "partial")
self.assertEqual(result["totals"]["failed_targets"], 1)
self.assertEqual(result["totals"]["events_inserted"], 1)
self.assertEqual(len(result["targets"]), 1)
self.assertEqual(result["targets"][0]["target_key"], "comunidad-hispana-02")
def test_overlapping_windows_remain_idempotent_via_existing_persistence(self) -> None:
entry = {
"timestamp": "2026-06-18T18:00:00Z",
"message": (
"[5:00 min (321)] KILL: Alpha(Allies/76561198000000001) -> "
"Bravo(Axis/76561198000000002) with M1 GARAND"
),
}
with tempfile.TemporaryDirectory() as temp_dir:
db_path = os.path.join(temp_dir, "current_match.sqlite3")
def fake_fetch(target, *, lookback_seconds, timeout_seconds):
return [entry]
first = run_current_match_adminlog_refresh_once_unlocked(
lookback_seconds=180,
targets=[TARGET_01],
fetch_entries_fn=fake_fetch,
db_path=db_path,
)
second = run_current_match_adminlog_refresh_once_unlocked(
lookback_seconds=180,
targets=[TARGET_01],
fetch_entries_fn=fake_fetch,
db_path=db_path,
)
self.assertEqual(first["totals"]["events_inserted"], 1)
self.assertEqual(first["totals"]["duplicate_events"], 0)
self.assertEqual(second["totals"]["events_inserted"], 0)
self.assertEqual(second["totals"]["duplicate_events"], 1)
def test_current_match_adminlog_config_defaults_and_overrides(self) -> None:
with _temporary_env(
CURRENT_MATCH_ADMINLOG_INTERVAL_SECONDS=None,
CURRENT_MATCH_ADMINLOG_LOOKBACK_SECONDS=None,
CURRENT_MATCH_ADMINLOG_ENABLED=None,
):
self.assertEqual(get_current_match_adminlog_interval_seconds(), 10)
self.assertEqual(get_current_match_adminlog_lookback_seconds(), 180)
self.assertIs(get_current_match_adminlog_enabled(), False)
with _temporary_env(
CURRENT_MATCH_ADMINLOG_INTERVAL_SECONDS="15",
CURRENT_MATCH_ADMINLOG_LOOKBACK_SECONDS="120",
CURRENT_MATCH_ADMINLOG_ENABLED="true",
):
self.assertEqual(get_current_match_adminlog_interval_seconds(), 15)
self.assertEqual(get_current_match_adminlog_lookback_seconds(), 120)
self.assertIs(get_current_match_adminlog_enabled(), True)
@contextmanager
def _temporary_env(**values: str | None):
previous = {name: os.environ.get(name) for name in values}
try:
for name, value in values.items():
if value is None:
os.environ.pop(name, None)
else:
os.environ[name] = value
yield
finally:
for name, value in previous.items():
if value is None:
os.environ.pop(name, None)
else:
os.environ[name] = value