diff --git a/deploy/backup-equipment-window.sh b/deploy/backup-equipment-window.sh index 18c626a..f99cc1e 100644 --- a/deploy/backup-equipment-window.sh +++ b/deploy/backup-equipment-window.sh @@ -21,6 +21,16 @@ container=$(docker compose ps -q inference) echo 'equipment inference is not running' >&2 exit 1 } +old_health=$(docker inspect -f '{{.State.Health.Status}}' "$container") +schema_recovery=0 +old_logs=$(docker logs --tail 100 "$container" 2>&1) +if grep -Fq 'column "journal_is_alarm" does not exist' <<<"$old_logs"; then + schema_recovery=1 + echo 'old equipment is blocked by missing journal_is_alarm; preserving its window before repair' +elif [[ $old_health != healthy ]]; then + echo "old equipment is not healthy: $old_health" >&2 + exit 1 +fi image=$(docker image inspect -f '{{.Id}}' inference:current) [[ -n $image ]] || { echo 'inference:current image is missing' >&2; exit 1; } volume=$(docker inspect -f '{{range .Mounts}}{{if eq .Destination "/app/runtime"}}{{.Name}}{{end}}{{end}}' "$container") @@ -43,7 +53,7 @@ cleanup() { local result=$? trap - EXIT if (( stopped )); then - if ! docker compose start inference || ! wait_healthy; then + if ! docker compose start inference || ! verify_old_running; then echo 'CRITICAL: old equipment inference did not recover after backup failure' >&2 result=1 fi @@ -68,6 +78,14 @@ wait_healthy() { done } +verify_old_running() { + if (( schema_recovery )); then + [[ $(docker inspect -f '{{.State.Running}}' "$container") == true ]] + else + wait_healthy + fi +} + # Stop before copying the cursor so it matches a completed batch. stopped=1 docker compose stop inference @@ -104,16 +122,16 @@ copy_query "SELECT id, event_id, channel_id, ts, raw_value, is_alarm FROM events copy_query "SELECT id, event_id, channel_id, ts, raw_value, is_alarm FROM events WHERE ts > '${snapshot_at}'::timestamptz" future_events.csv.gz copy_query 'SELECT channel_id, eng_system_type, sensor_type, system_tag, sensor_name, object_id FROM dim_channels_current' channels.csv.gz copy_query 'SELECT object_id, hierarchy_level, parent_id, object_kind, dispatcher_name FROM dim_objects_current' objects.csv.gz -wait_healthy +verify_old_running -python3 - "$tmp" "$snapshot_at" "$image" "$sha" "$volume" <<'PY' +python3 - "$tmp" "$snapshot_at" "$image" "$sha" "$volume" "$old_health" "$schema_recovery" <<'PY' import csv import gzip import json import sys from pathlib import Path -directory, snapshot_at, image, sha, volume = sys.argv[1:] +directory, snapshot_at, image, sha, volume, old_health, schema_recovery = sys.argv[1:] root = Path(directory) counts = {} for name in ('events', 'future_events', 'channels', 'objects'): @@ -143,6 +161,8 @@ manifest = { 'window_hours': 72, 'image_id': image, 'runtime_volume': volume, + 'old_health': old_health, + 'schema_recovery': schema_recovery == '1', 'deploy_sha': sha, 'cursor_low': data['low'], 'cursor_seen': len(data['seen']), diff --git a/inference/runner.py b/inference/runner.py index a7c8f67..bf6ba50 100644 --- a/inference/runner.py +++ b/inference/runner.py @@ -54,7 +54,7 @@ # Фильтр по ts отсекает старые партиции и импорт истории вне окон. SELECT_NEW = """ - SELECT id, channel_id, ts, raw_value, is_alarm, journal_is_alarm FROM events + SELECT id, channel_id, ts, raw_value, is_alarm, {journal_alarm} FROM events WHERE ts >= %(since)s AND id > %(low)s AND id <> ALL(%(seen)s::bigint[]) ORDER BY id @@ -326,8 +326,11 @@ def tick(self, conn: psycopg.Connection) -> bool: def _fetch(self, conn: psycopg.Connection) -> list[EventRow]: assert self.cursor is not None + journal_alarm = ( + "journal_is_alarm" if self.targets is not None else "NULL::boolean" + ) rows = conn.execute( - SELECT_NEW, + SELECT_NEW.format(journal_alarm=journal_alarm), { "since": self.clock() - self.window, "low": self.cursor.low, diff --git a/tests/test_equipment_window_backup.py b/tests/test_equipment_window_backup.py index 4e53ae4..54f3dc5 100644 --- a/tests/test_equipment_window_backup.py +++ b/tests/test_equipment_window_backup.py @@ -29,7 +29,12 @@ def _mock_tools(tmp_path: Path) -> Path: 'compose ps -q inference') echo old-container ;; 'inspect -f {{.State.Running}} old-container') echo true ;; 'inspect -f {{range .Mounts}}{{if eq .Destination "/app/runtime"}}{{.Name}}{{end}}{{end}} old-container') echo backend_inference-runtime ;; - 'inspect -f {{.State.Health.Status}} old-container') echo healthy ;; + 'inspect -f {{.State.Health.Status}} old-container') + if [ "${MOCK_SCHEMA_ERROR:-0}" = 1 ] && [ "${MOCK_HEALTHY_ERROR:-0}" != 1 ]; then echo starting; + else echo healthy; fi ;; + 'logs --tail 100 old-container') + if [ "${MOCK_SCHEMA_ERROR:-0}" = 1 ]; then + echo 'column "journal_is_alarm" does not exist'; fi ;; 'image inspect -f {{.Id}} inference:current') echo sha256:old-image ;; 'compose stop inference') : ;; 'compose start inference') [ "${MOCK_FAIL_START:-0}" != 1 ] ;; @@ -54,7 +59,9 @@ def _mock_tools(tmp_path: Path) -> Path: def _run(tmp_path: Path, fail_copy: bool = False, fail_start: bool = False, - no_inference: bool = False) -> tuple[subprocess.CompletedProcess[str], Path]: + no_inference: bool = False, schema_error: bool = False, + healthy_error: bool = False + ) -> tuple[subprocess.CompletedProcess[str], Path]: bin_dir = _mock_tools(tmp_path) runtime = tmp_path / "runtime" runtime.mkdir() @@ -77,6 +84,8 @@ def _run(tmp_path: Path, fail_copy: bool = False, fail_start: bool = False, "MOCK_FAIL_CP": "1" if fail_copy else "0", "MOCK_FAIL_START": "1" if fail_start else "0", "MOCK_NO_INFERENCE": "1" if no_inference else "0", + "MOCK_SCHEMA_ERROR": "1" if schema_error else "0", + "MOCK_HEALTHY_ERROR": "1" if healthy_error else "0", } result = subprocess.run(["bash", str(SCRIPT)], env=env, text=True, capture_output=True, check=False) @@ -125,3 +134,22 @@ def test_missing_equipment_service_blocks_image_switch(tmp_path: Path) -> None: assert result.returncode != 0 assert "window cannot be preserved" in result.stderr assert not list((backup / "inference-windows").glob("equipment-*")) + + +def test_known_schema_failure_still_preserves_window(tmp_path: Path) -> None: + result, backup = _run(tmp_path, schema_error=True) + assert result.returncode == 0, result.stderr + snapshot = next((backup / "inference-windows").glob("equipment-*")) + manifest = json.loads((snapshot / "manifest.json").read_text()) + assert manifest["old_health"] == "starting" + assert manifest["schema_recovery"] is True + assert "compose start inference" in (tmp_path / "calls.log").read_text() + + +def test_schema_error_overrides_stale_healthy_status(tmp_path: Path) -> None: + result, backup = _run(tmp_path, schema_error=True, healthy_error=True) + assert result.returncode == 0, result.stderr + snapshot = next((backup / "inference-windows").glob("equipment-*")) + manifest = json.loads((snapshot / "manifest.json").read_text()) + assert manifest["old_health"] == "healthy" + assert manifest["schema_recovery"] is True diff --git a/tests/test_runner.py b/tests/test_runner.py index 1cb53b2..5d74432 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -139,6 +139,16 @@ def test_tick_writes_prediction_rows(db, make_runner) -> None: assert "null" not in [item["feature_value"] for item in shap["top"]] +def test_equipment_reads_events_before_incident_migration(db, make_runner) -> None: + with db.admin() as admin: + admin.execute("ALTER TABLE events DROP COLUMN journal_is_alarm") + runner, conn = started(make_runner, db) + with db.admin() as admin: + add_event(admin, CHANNEL, datetime.now(timezone.utc), "Норма") + runner.tick(conn) + assert len(predictions(db)) == 1 + + def test_shap_null_below_threshold(db, make_runner) -> None: runner, conn = started(make_runner, db, threshold=1.0)