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

fix: decouple live RCON capture from historical materialization

Browse Source
This commit is contained in:
devRaGonSa
2026-06-05 16:55:14 +02:00
parent af4ee2f767
commit ede22fd62e
5 changed files with 372 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
backend/app/config.py
View File

@@ -35,6 +35,8 @@ DEFAULT_PLAYER_EVENT_REFRESH_RETRY_DELAY_SECONDS = 30
DEFAULT_RCON_HISTORICAL_CAPTURE_INTERVAL_SECONDS = 600
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_RCON_BACKFILL_CHUNK_HOURS = 6
DEFAULT_RCON_BACKFILL_SLEEP_SECONDS = 1.0
DEFAULT_RCON_BACKFILL_MAX_DAYS_BACK = 45
@@ -279,6 +281,20 @@ def _read_float_env(name: str, default_value: str, *, minimum: float) -> float:
return value
def _read_bool_env(name: str, *, default: bool) -> bool:
"""Read one boolean env var using a small set of explicit truthy/falsey values."""
configured_value = os.getenv(name)
if configured_value is None:
return default
normalized = configured_value.strip().lower()
if normalized in {"1", "true", "yes", "on"}:
return True
if normalized in {"0", "false", "no", "off"}:
return False
raise ValueError(f"{name} must be a boolean value.")
def get_historical_refresh_overlap_hours() -> int:
"""Return the overlap window used by incremental historical refreshes."""
configured_value = os.getenv(
@@ -494,6 +510,45 @@ def get_rcon_historical_capture_retry_delay_seconds() -> int:
return retry_delay_seconds
def get_rcon_capture_mode() -> str:
"""Return whether the worker runs the normal historical pipeline or live capture only."""
configured_mode = os.getenv("HLL_RCON_CAPTURE_MODE")
if configured_mode is not None:
normalized_mode = configured_mode.strip().lower()
if normalized_mode in {"historical", "current-live"}:
return normalized_mode
raise ValueError("HLL_RCON_CAPTURE_MODE must be 'historical' or 'current-live'.")
if _read_bool_env("HLL_RCON_CURRENT_MATCH_MODE", default=False):
return "current-live"
if _read_bool_env("HLL_RCON_SKIP_HISTORICAL_MATERIALIZATION", default=False):
return "current-live"
return "historical"
def get_rcon_skip_historical_materialization() -> bool:
"""Return whether the worker must skip heavy historical AdminLog materialization."""
return get_rcon_capture_mode() == "current-live"
def get_rcon_current_match_capture_interval_seconds() -> int:
"""Return the preferred live current-match capture interval."""
return _read_int_env(
"HLL_RCON_CURRENT_MATCH_CAPTURE_INTERVAL_SECONDS",
str(DEFAULT_RCON_CURRENT_MATCH_CAPTURE_INTERVAL_SECONDS),
minimum=1,
)
def get_rcon_current_match_writer_lock_timeout_seconds() -> float:
"""Return the lock wait budget for the lightweight current-match worker."""
return _read_float_env(
"HLL_RCON_CURRENT_MATCH_WRITER_LOCK_TIMEOUT_SECONDS",
str(DEFAULT_RCON_CURRENT_MATCH_WRITER_LOCK_TIMEOUT_SECONDS),
minimum=0,
)
def get_rcon_backfill_chunk_hours() -> int:
"""Return the AdminLog backfill chunk size in hours."""
return _read_int_env(

127
backend/app/rcon_historical_worker.py
View File

@@ -11,10 +11,14 @@ from dataclasses import dataclass
from typing import Iterable
from .config import (
get_rcon_capture_mode,
get_rcon_current_match_capture_interval_seconds,
get_rcon_current_match_writer_lock_timeout_seconds,
get_rcon_historical_capture_interval_seconds,
get_rcon_historical_capture_max_retries,
get_rcon_historical_capture_retry_delay_seconds,
get_rcon_request_timeout_seconds,
get_rcon_skip_historical_materialization,
)
from .rcon_admin_log_ingestion import ingest_rcon_admin_logs
from .rcon_admin_log_materialization import materialize_rcon_admin_log
@@ -33,7 +37,13 @@ from .rcon_historical_storage import (
start_rcon_historical_capture_run,
)
from .snapshots import utc_now
from .writer_lock import backend_writer_lock, build_writer_lock_holder
from .writer_lock import (
backend_writer_lock,
build_writer_lock_holder,
)
CAPTURE_MODE_HISTORICAL = "historical"
CAPTURE_MODE_CURRENT_LIVE = "current-live"
@dataclass(slots=True)
@@ -53,28 +63,52 @@ class RconHistoricalCaptureStats:
def run_rcon_historical_capture(
*,
target_key: str | None = None,
capture_mode: str | None = None,
skip_materialization: bool | None = None,
writer_lock_timeout_seconds: float | None = None,
) -> dict[str, object]:
"""Capture one prospective RCON sample for one or all configured targets."""
resolved_capture_mode, resolved_skip_materialization = _resolve_capture_options(
capture_mode=capture_mode,
skip_materialization=skip_materialization,
)
resolved_lock_timeout = writer_lock_timeout_seconds
if resolved_lock_timeout is None and resolved_capture_mode == CAPTURE_MODE_CURRENT_LIVE:
resolved_lock_timeout = get_rcon_current_match_writer_lock_timeout_seconds()
with backend_writer_lock(
holder=build_writer_lock_holder(
f"app.rcon_historical_worker capture:{target_key or 'all-targets'}"
)
),
timeout_seconds=resolved_lock_timeout,
):
return run_rcon_historical_capture_unlocked(target_key=target_key)
return run_rcon_historical_capture_unlocked(
target_key=target_key,
capture_mode=resolved_capture_mode,
skip_materialization=resolved_skip_materialization,
)
def run_rcon_historical_capture_unlocked(
*,
target_key: str | None = None,
capture_mode: str | None = None,
skip_materialization: bool | None = None,
) -> dict[str, object]:
"""Capture one prospective RCON sample assuming the shared writer lock is already held."""
resolved_capture_mode, resolved_skip_materialization = _resolve_capture_options(
capture_mode=capture_mode,
skip_materialization=skip_materialization,
)
initialize_rcon_historical_storage()
selected_targets = _select_targets(target_key)
selected_target_keys = {build_rcon_target_key(target) for target in selected_targets}
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)
run_id = start_rcon_historical_capture_run(
mode=resolved_capture_mode,
target_scope=target_scope,
)
stats = RconHistoricalCaptureStats()
items: list[dict[str, object]] = []
errors: list[dict[str, object]] = []
@@ -132,7 +166,10 @@ def run_rcon_historical_capture_unlocked(
result=admin_log_result,
)
materialization_result = materialize_rcon_admin_log()
materialization_result = _run_materialization_if_enabled(
skip_materialization=resolved_skip_materialization
)
if not resolved_skip_materialization:
stats.materialized_matches_inserted = int(
materialization_result.get("matches_materialized") or 0
)
@@ -167,7 +204,13 @@ def run_rcon_historical_capture_unlocked(
"run_status": status,
"captured_at": captured_at,
"target_scope": target_scope,
"capture_mode": resolved_capture_mode,
"materialization_skipped": resolved_skip_materialization,
"admin_log_lookback_minutes": admin_log_lookback_minutes,
"admin_log_events_seen": stats.admin_log_events_seen,
"admin_log_events_inserted": stats.admin_log_events_inserted,
"duplicate_events": stats.admin_log_duplicate_events,
"samples_inserted": stats.samples_inserted,
"targets": items,
"errors": errors,
"admin_log_errors": admin_log_errors,
@@ -198,9 +241,15 @@ def run_periodic_rcon_historical_capture(
max_retries: int,
retry_delay_seconds: int,
target_key: str | None = None,
capture_mode: str | None = None,
skip_materialization: bool | None = None,
max_runs: int | None = None,
) -> None:
"""Run prospective RCON capture in a local loop."""
resolved_capture_mode, resolved_skip_materialization = _resolve_capture_options(
capture_mode=capture_mode,
skip_materialization=skip_materialization,
)
completed_runs = 0
startup_targets = _describe_loop_targets(target_key)
_emit_worker_event(
@@ -208,6 +257,8 @@ def run_periodic_rcon_historical_capture(
interval_seconds=interval_seconds,
max_retries=max_retries,
retry_delay_seconds=retry_delay_seconds,
capture_mode=resolved_capture_mode,
materialization_skipped=resolved_skip_materialization,
target_scope=target_key or "all-configured-rcon-targets",
target_count=len(startup_targets),
targets=startup_targets,
@@ -225,6 +276,8 @@ def run_periodic_rcon_historical_capture(
max_retries=max_retries,
retry_delay_seconds=retry_delay_seconds,
target_key=target_key,
capture_mode=resolved_capture_mode,
skip_materialization=resolved_skip_materialization,
)
_emit_worker_event(
"rcon-historical-capture-cycle-finished",
@@ -255,6 +308,8 @@ def _run_capture_with_retries(
max_retries: int,
retry_delay_seconds: int,
target_key: str | None,
capture_mode: str,
skip_materialization: bool,
) -> dict[str, object]:
attempt = 0
while True:
@@ -263,7 +318,11 @@ def _run_capture_with_retries(
return {
"status": "ok",
"attempts_used": attempt,
"capture_result": run_rcon_historical_capture(target_key=target_key),
"capture_result": run_rcon_historical_capture(
target_key=target_key,
capture_mode=capture_mode,
skip_materialization=skip_materialization,
),
}
except Exception as exc:
if attempt > max_retries:
@@ -480,6 +539,34 @@ def _format_error_message(error: Exception) -> str:
return f"[{error_type}] {error}"
def _resolve_capture_options(
*,
capture_mode: str | None,
skip_materialization: bool | None,
) -> tuple[str, bool]:
resolved_capture_mode = capture_mode or get_rcon_capture_mode()
if resolved_capture_mode not in {CAPTURE_MODE_HISTORICAL, CAPTURE_MODE_CURRENT_LIVE}:
raise ValueError("capture_mode must be 'historical' or 'current-live'.")
if skip_materialization is None:
resolved_skip_materialization = get_rcon_skip_historical_materialization()
else:
resolved_skip_materialization = skip_materialization
if resolved_capture_mode == CAPTURE_MODE_CURRENT_LIVE:
resolved_skip_materialization = True
return resolved_capture_mode, resolved_skip_materialization
def _run_materialization_if_enabled(*, skip_materialization: bool) -> dict[str, object]:
if skip_materialization:
return {
"status": "skipped",
"reason": "skip-materialization-enabled",
}
return materialize_rcon_admin_log()
def build_arg_parser() -> argparse.ArgumentParser:
"""Create the CLI parser for manual or periodic prospective RCON capture."""
parser = argparse.ArgumentParser(
@@ -518,6 +605,18 @@ def build_arg_parser() -> argparse.ArgumentParser:
type=int,
help="optional safety cap for loop mode",
)
parser.add_argument(
"--capture-mode",
choices=(CAPTURE_MODE_HISTORICAL, CAPTURE_MODE_CURRENT_LIVE),
default=get_rcon_capture_mode(),
help="historical keeps materialization; current-live only captures lightweight live data",
)
parser.add_argument(
"--skip-materialization",
action="store_true",
default=None,
help="capture AdminLog and live snapshots without running heavy historical materialization",
)
return parser
@@ -527,10 +626,22 @@ def main(argv: Iterable[str] | None = None) -> int:
args = parser.parse_args(list(argv) if argv is not None else None)
if args.mode == "capture":
result = run_rcon_historical_capture(target_key=args.target_key)
result = run_rcon_historical_capture(
target_key=args.target_key,
capture_mode=args.capture_mode,
skip_materialization=args.skip_materialization,
)
print(json.dumps(result, indent=2, default=_json_default))
return 0
default_interval = (
get_rcon_current_match_capture_interval_seconds()
if args.capture_mode == CAPTURE_MODE_CURRENT_LIVE
and "--interval" not in (argv or [])
else args.interval
)
args.interval = default_interval
if args.interval <= 0:
raise ValueError("--interval must be a positive integer.")
if args.retries < 0:
@@ -545,6 +656,8 @@ def main(argv: Iterable[str] | None = None) -> int:
max_retries=args.retries,
retry_delay_seconds=args.retry_delay,
target_key=args.target_key,
capture_mode=args.capture_mode,
skip_materialization=args.skip_materialization,
max_runs=args.max_runs,
)
return 0

171
backend/tests/test_rcon_historical_worker.py Normal file
View File

@@ -0,0 +1,171 @@
from __future__ import annotations
import os
import unittest
from contextlib import contextmanager
from types import SimpleNamespace
from unittest.mock import patch
from app.rcon_historical_worker import (
CAPTURE_MODE_CURRENT_LIVE,
CAPTURE_MODE_HISTORICAL,
build_arg_parser,
main,
run_rcon_historical_capture,
run_rcon_historical_capture_unlocked,
)
TARGET = SimpleNamespace(
external_server_id="comunidad-hispana-01",
name="Comunidad Hispana #01",
host="203.0.113.10",
port=7779,
region="ES",
game_port=7777,
query_port=7778,
source_name="community-hispana-rcon",
)
class RconHistoricalWorkerTests(unittest.TestCase):
def test_current_live_capture_skips_materialization(self) -> None:
with (
patch("app.rcon_historical_worker.initialize_rcon_historical_storage"),
patch("app.rcon_historical_worker._select_targets", return_value=[TARGET]),
patch(
"app.rcon_historical_worker.query_live_server_sample",
return_value={"normalized": {"players": 10}, "raw_session": {"raw": True}},
),
patch(
"app.rcon_historical_worker.persist_rcon_historical_sample",
return_value={"samples_inserted": 1, "duplicate_samples": 0},
),
patch(
"app.rcon_historical_worker._ingest_target_admin_log",
return_value={
"status": "ok",
"errors": [],
"totals": {
"events_seen": 4,
"events_inserted": 2,
"duplicate_events": 2,
"failed_targets": 0,
},
},
),
patch("app.rcon_historical_worker.start_rcon_historical_capture_run", return_value=101),
patch("app.rcon_historical_worker.finalize_rcon_historical_capture_run"),
patch(
"app.rcon_historical_worker.list_rcon_historical_target_statuses",
return_value=[{"target_key": "comunidad-hispana-01"}],
),
patch("app.rcon_historical_worker.materialize_rcon_admin_log") as materialize,
):
payload = run_rcon_historical_capture_unlocked(capture_mode=CAPTURE_MODE_CURRENT_LIVE)
materialize.assert_not_called()
self.assertEqual(payload["capture_mode"], CAPTURE_MODE_CURRENT_LIVE)
self.assertIs(payload["materialization_skipped"], True)
self.assertEqual(payload["admin_log_events_seen"], 4)
self.assertEqual(payload["admin_log_events_inserted"], 2)
self.assertEqual(payload["duplicate_events"], 2)
self.assertEqual(payload["samples_inserted"], 1)
self.assertEqual(payload["materialization_result"]["status"], "skipped")
def test_historical_capture_keeps_materialization(self) -> None:
with (
patch("app.rcon_historical_worker.initialize_rcon_historical_storage"),
patch("app.rcon_historical_worker._select_targets", return_value=[TARGET]),
patch(
"app.rcon_historical_worker.query_live_server_sample",
return_value={"normalized": {"players": 10}, "raw_session": {"raw": True}},
),
patch(
"app.rcon_historical_worker.persist_rcon_historical_sample",
return_value={"samples_inserted": 1, "duplicate_samples": 0},
),
patch(
"app.rcon_historical_worker._ingest_target_admin_log",
return_value={
"status": "ok",
"errors": [],
"totals": {
"events_seen": 1,
"events_inserted": 1,
"duplicate_events": 0,
"failed_targets": 0,
},
},
),
patch("app.rcon_historical_worker.start_rcon_historical_capture_run", return_value=102),
patch("app.rcon_historical_worker.finalize_rcon_historical_capture_run"),
patch(
"app.rcon_historical_worker.list_rcon_historical_target_statuses",
return_value=[{"target_key": "comunidad-hispana-01"}],
),
patch(
"app.rcon_historical_worker.materialize_rcon_admin_log",
return_value={"matches_materialized": 3, "matches_updated": 2},
) as materialize,
):
payload = run_rcon_historical_capture_unlocked(capture_mode=CAPTURE_MODE_HISTORICAL)
materialize.assert_called_once_with()
self.assertEqual(payload["capture_mode"], CAPTURE_MODE_HISTORICAL)
self.assertIs(payload["materialization_skipped"], False)
self.assertEqual(payload["totals"]["materialized_matches_inserted"], 3)
self.assertEqual(payload["totals"]["materialized_matches_updated"], 2)
def test_cli_and_env_can_activate_current_live_mode(self) -> None:
with _temporary_env(
HLL_RCON_CURRENT_MATCH_MODE="true",
HLL_RCON_CURRENT_MATCH_CAPTURE_INTERVAL_SECONDS="5",
):
args = build_arg_parser().parse_args(["loop"])
with patch("app.rcon_historical_worker.run_periodic_rcon_historical_capture") as runner:
exit_code = main(["loop"])
self.assertEqual(args.capture_mode, CAPTURE_MODE_CURRENT_LIVE)
self.assertEqual(exit_code, 0)
runner.assert_called_once()
self.assertEqual(runner.call_args.kwargs["capture_mode"], CAPTURE_MODE_CURRENT_LIVE)
self.assertEqual(runner.call_args.kwargs["interval_seconds"], 5)
explicit = build_arg_parser().parse_args(["capture", "--skip-materialization"])
self.assertIs(explicit.skip_materialization, True)
def test_current_live_capture_uses_short_lock_timeout(self) -> None:
seen: dict[str, object] = {}
@contextmanager
def fake_lock(**kwargs):
seen.update(kwargs)
yield {"holder": kwargs["holder"]}
with (
_temporary_env(HLL_RCON_CURRENT_MATCH_WRITER_LOCK_TIMEOUT_SECONDS="3.5"),
patch("app.rcon_historical_worker.backend_writer_lock", side_effect=fake_lock),
patch(
"app.rcon_historical_worker.run_rcon_historical_capture_unlocked",
return_value={"status": "ok"},
),
):
run_rcon_historical_capture(capture_mode=CAPTURE_MODE_CURRENT_LIVE)
self.assertEqual(seen["timeout_seconds"], 3.5)
@contextmanager
def _temporary_env(**values: str):
previous = {name: os.environ.get(name) for name in values}
try:
for name, value in values.items():
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

6
deploy/jta/.env.example
View File

@@ -21,5 +21,9 @@ HLL_ADMIN_LOG_CRITICAL_RETENTION_DAYS=90
HLL_SERVER_SNAPSHOT_RETENTION_DAYS=14
HLL_DB_MAINTENANCE_BATCH_SIZE=5000
HLL_BACKEND_RCON_HISTORICAL_INTERVAL_SECONDS=600
HLL_BACKEND_RCON_HISTORICAL_INTERVAL_SECONDS=5
HLL_RCON_CURRENT_MATCH_CAPTURE_INTERVAL_SECONDS=5
HLL_RCON_HISTORICAL_CAPTURE_MAX_RETRIES=0
HLL_RCON_HISTORICAL_CAPTURE_RETRY_DELAY_SECONDS=0
HLL_RCON_CURRENT_MATCH_WRITER_LOCK_TIMEOUT_SECONDS=4
HLL_BACKEND_RCON_ADMIN_LOG_LOOKBACK_MINUTES=10

16
deploy/jta/docker-compose.yml
View File

@@ -117,9 +117,23 @@ services:
- -m
- app.rcon_historical_worker
- loop
- --capture-mode
- current-live
- --interval
- "5"
- --retries
- "0"
- --retry-delay
- "0"
environment:
<<: *backend_environment
HLL_RCON_HISTORICAL_CAPTURE_INTERVAL_SECONDS: ${HLL_BACKEND_RCON_HISTORICAL_INTERVAL_SECONDS:-600}
HLL_RCON_HISTORICAL_CAPTURE_INTERVAL_SECONDS: ${HLL_BACKEND_RCON_HISTORICAL_INTERVAL_SECONDS:-5}
HLL_RCON_CURRENT_MATCH_CAPTURE_INTERVAL_SECONDS: ${HLL_RCON_CURRENT_MATCH_CAPTURE_INTERVAL_SECONDS:-5}
HLL_RCON_CURRENT_MATCH_MODE: "true"
HLL_RCON_SKIP_HISTORICAL_MATERIALIZATION: "true"
HLL_RCON_HISTORICAL_CAPTURE_MAX_RETRIES: ${HLL_RCON_HISTORICAL_CAPTURE_MAX_RETRIES:-0}
HLL_RCON_HISTORICAL_CAPTURE_RETRY_DELAY_SECONDS: ${HLL_RCON_HISTORICAL_CAPTURE_RETRY_DELAY_SECONDS:-0}
HLL_RCON_CURRENT_MATCH_WRITER_LOCK_TIMEOUT_SECONDS: ${HLL_RCON_CURRENT_MATCH_WRITER_LOCK_TIMEOUT_SECONDS:-4}
HLL_BACKEND_RCON_ADMIN_LOG_LOOKBACK_MINUTES: ${HLL_BACKEND_RCON_ADMIN_LOG_LOOKBACK_MINUTES:-10}
depends_on:
postgres: