diff --git a/backend/app/config.py b/backend/app/config.py index 49aefcb..4d3c5bc 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -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( @@ -490,10 +506,49 @@ def get_rcon_historical_capture_retry_delay_seconds() -> int: if retry_delay_seconds < 0: raise ValueError( "HLL_RCON_HISTORICAL_CAPTURE_RETRY_DELAY_SECONDS must be zero or positive." - ) + ) 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( diff --git a/backend/app/rcon_historical_worker.py b/backend/app/rcon_historical_worker.py index d308e7a..6a76993 100644 --- a/backend/app/rcon_historical_worker.py +++ b/backend/app/rcon_historical_worker.py @@ -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,13 +166,16 @@ def run_rcon_historical_capture_unlocked( result=admin_log_result, ) - materialization_result = materialize_rcon_admin_log() - stats.materialized_matches_inserted = int( - materialization_result.get("matches_materialized") or 0 - ) - stats.materialized_matches_updated = int( - materialization_result.get("matches_updated") or 0 + 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 + ) + stats.materialized_matches_updated = int( + materialization_result.get("matches_updated") or 0 + ) status = "success" if not errors else ("partial" if items else "failed") finalize_rcon_historical_capture_run( @@ -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 diff --git a/backend/tests/test_rcon_historical_worker.py b/backend/tests/test_rcon_historical_worker.py new file mode 100644 index 0000000..028a0d8 --- /dev/null +++ b/backend/tests/test_rcon_historical_worker.py @@ -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 diff --git a/deploy/jta/.env.example b/deploy/jta/.env.example index 42d7fab..c00551a 100644 --- a/deploy/jta/.env.example +++ b/deploy/jta/.env.example @@ -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 diff --git a/deploy/jta/docker-compose.yml b/deploy/jta/docker-compose.yml index a610c56..b5e0ee0 100644 --- a/deploy/jta/docker-compose.yml +++ b/deploy/jta/docker-compose.yml @@ -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: