From 8862da39297e91bd9b6cffedea4c1dd91422ebb8 Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 00:42:23 +0800 Subject: [PATCH] fix(agent): route disk pressure to bounded repair --- .../services/awooop_ansible_audit_service.py | 107 ++++++- .../awooop_ansible_check_mode_service.py | 290 +++++++++++++++++- .../services/awooop_ansible_post_verifier.py | 30 ++ apps/api/src/services/telegram_gateway.py | 18 +- .../tests/test_ansible_verified_closure.py | 57 ++++ ...test_host_disk_pressure_ansible_catalog.py | 265 ++++++++++++++++ .../host-disk-pressure-bounded-cleanup.yml | 124 ++++++++ .../docker-disk-pressure-retention-cleanup.py | 197 +++++++++++- ..._docker_disk_pressure_retention_cleanup.py | 88 +++++- 9 files changed, 1145 insertions(+), 31 deletions(-) create mode 100644 apps/api/tests/test_host_disk_pressure_ansible_catalog.py create mode 100644 infra/ansible/playbooks/host-disk-pressure-bounded-cleanup.yml diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py index bbe4a6c3a..7155ba6e1 100644 --- a/apps/api/src/services/awooop_ansible_audit_service.py +++ b/apps/api/src/services/awooop_ansible_audit_service.py @@ -417,6 +417,11 @@ _CATALOG: tuple[dict[str, Any], ...] = ( "gitea_service", "systemd_control_plane", ], + "activation_keywords": [ + "host110sustainedmoderatepressure", + "host-pressure-controller", + "sustained pressure", + ], "supports_check_mode": True, "auto_apply_enabled": False, "approval_required": False, @@ -424,6 +429,74 @@ _CATALOG: tuple[dict[str, Any], ...] = ( "no_write_only": True, "catalog_revision": "2026-07-14-host110-inputs-v2", }, + { + "catalog_id": "ansible:110-disk-pressure", + "playbook_path": ( + "infra/ansible/playbooks/host-disk-pressure-bounded-cleanup.yml" + ), + "check_mode_playbook_path": ( + "infra/ansible/playbooks/host-disk-pressure-bounded-cleanup.yml" + ), + "inventory_hosts": ["host_110"], + "domains": ["disk_pressure", "docker_artifacts", "host_110"], + "keywords": [ + "hostoutofdiskspace", + "hostdiskusagehigh", + "hostdiskusagecritical", + "hostdiskwillfillin24hours", + "disk_full", + "node-exporter-110", + "192.168.0.110", + ], + "activation_keyword_groups": [ + [ + "hostoutofdiskspace", + "hostdiskusagehigh", + "hostdiskusagecritical", + "hostdiskwillfillin24hours", + ], + ["node-exporter-110", "192.168.0.110", "host_110", "host110"], + ], + "supports_check_mode": True, + "auto_apply_enabled": True, + "approval_required": False, + "risk_level": "medium", + "catalog_revision": "2026-07-14-disk-pressure-bounded-v1", + }, + { + "catalog_id": "ansible:188-disk-pressure", + "playbook_path": ( + "infra/ansible/playbooks/host-disk-pressure-bounded-cleanup.yml" + ), + "check_mode_playbook_path": ( + "infra/ansible/playbooks/host-disk-pressure-bounded-cleanup.yml" + ), + "inventory_hosts": ["host_188"], + "domains": ["disk_pressure", "docker_artifacts", "host_188"], + "keywords": [ + "hostoutofdiskspace", + "hostdiskusagehigh", + "hostdiskusagecritical", + "hostdiskwillfillin24hours", + "disk_full", + "node-exporter-188", + "192.168.0.188", + ], + "activation_keyword_groups": [ + [ + "hostoutofdiskspace", + "hostdiskusagehigh", + "hostdiskusagecritical", + "hostdiskwillfillin24hours", + ], + ["node-exporter-188", "192.168.0.188", "host_188", "host188"], + ], + "supports_check_mode": True, + "auto_apply_enabled": True, + "approval_required": False, + "risk_level": "medium", + "catalog_revision": "2026-07-14-disk-pressure-bounded-v1", + }, { "catalog_id": "ansible:110-devops", "playbook_path": "infra/ansible/playbooks/110-node-exporter-repair.yml", @@ -437,9 +510,13 @@ _CATALOG: tuple[dict[str, Any], ...] = ( "node_exporter", "node-exporter", ], - "activation_keywords": [ - "node-exporter-110", - "node exporter 110", + "activation_keyword_groups": [ + [ + "nodeexporterdown", + "nodeexporterscrapedown", + "node_exporter_down", + ], + ["node-exporter-110", "192.168.0.110", "host_110", "host110"], ], "supports_check_mode": True, "auto_apply_enabled": True, @@ -914,7 +991,15 @@ def _catalog_hints(incident: dict[str, Any] | None, drift: dict[str, Any] | None matched_activation_keywords = [ keyword for keyword in activation_keywords if keyword in haystack ] - semantic_match = not activation_keywords or bool(matched_activation_keywords) + activation_keyword_groups = item.get("activation_keyword_groups") or [] + matched_activation_keyword_groups = [ + [keyword for keyword in group if keyword in haystack] + for group in activation_keyword_groups + ] + semantic_match = ( + (not activation_keywords or bool(matched_activation_keywords)) + and all(matched_activation_keyword_groups) + ) public_item = { key: value for key, value in item.items() @@ -939,6 +1024,9 @@ def _catalog_hints(incident: dict[str, Any] | None, drift: dict[str, Any] | None "matched_keywords": matched, "semantic_match": True, "matched_activation_keywords": matched_activation_keywords, + "matched_activation_keyword_groups": ( + matched_activation_keyword_groups + ), }) else: unmatched.append(item["catalog_id"]) @@ -952,6 +1040,17 @@ def _catalog_hints(incident: dict[str, Any] | None, drift: dict[str, Any] | None } +def get_ansible_catalog_candidates( + incident: dict[str, Any] | None, + drift: dict[str, Any] | None = None, +) -> list[dict[str, Any]]: + """Return current semantically matched catalog rows for replay routing.""" + + hints = _catalog_hints(incident, drift) + candidates = hints.get("candidates") + return [dict(row) for row in candidates or [] if isinstance(row, dict)] + + def build_ansible_truth( automation_ops: list[dict[str, Any]], *, diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py index 4363333e5..903899a83 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -19,7 +19,7 @@ from dataclasses import dataclass, replace from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any -from uuid import UUID, uuid4 +from uuid import NAMESPACE_URL, UUID, uuid4, uuid5 import structlog from sqlalchemy import select, text @@ -31,7 +31,10 @@ from src.models.approval import ApprovalStatus from src.services.ai_automation_runtime_contract import ( AI_AUTOMATION_STAGE_RECEIPT_SCHEMA_VERSION, ) -from src.services.awooop_ansible_audit_service import get_ansible_catalog_item +from src.services.awooop_ansible_audit_service import ( + get_ansible_catalog_candidates, + get_ansible_catalog_item, +) from src.services.awooop_ansible_learning_writeback import ( canonical_ansible_playbook_id, ensure_ansible_rag_writeback, @@ -52,6 +55,7 @@ _CONTROLLED_RETRY_MAX_ATTEMPTS = 1 _CONTROLLED_RETRY_ERROR_BACKOFF_MIN_SECONDS = 15 _CONTROLLED_RETRY_ERROR_BACKOFF_MAX_SECONDS = 30 _STDIN_BOUNDARY_REPLAY_WINDOW_HOURS = 7 * 24 +_CATALOG_REPAIR_REPLAY_WINDOW_HOURS = 72 _STDIN_BOUNDARY_TERMINAL_PROJECTION_BATCH_LIMIT = 5 _EXECUTION_CAPABILITY_MIN_TTL_SECONDS = 300 _EXECUTION_CAPABILITY_MAX_TTL_SECONDS = 1_800 @@ -1062,9 +1066,28 @@ def _claim_from_stale_check_mode_row( or input_payload.get("incident_id") or "" ) - catalog_id = str(input_payload.get("catalog_id") or "") + previous_catalog_id = str(input_payload.get("catalog_id") or "") + catalog_id = previous_catalog_id + incident_routing_evaluated = bool( + row.get("incident_alertname") + or row.get("incident_alert_category") + or row.get("incident_affected_services") + ) + if incident_routing_evaluated: + candidates = get_ansible_catalog_candidates( + { + "alertname": row.get("incident_alertname"), + "alert_category": row.get("incident_alert_category"), + "affected_services": row.get("incident_affected_services") or [], + } + ) + if not candidates: + return None + catalog_id = str(candidates[0].get("catalog_id") or "") catalog_item = get_ansible_catalog_item(catalog_id) or {} - inventory_hosts = input_payload.get("inventory_hosts") + inventory_hosts = catalog_item.get("inventory_hosts") or input_payload.get( + "inventory_hosts" + ) source_playbook_path = str( input_payload.get("source_candidate_playbook_path") or input_payload.get("catalog_playbook_path") @@ -1106,6 +1129,13 @@ def _claim_from_stale_check_mode_row( ) except ValueError: return None + claim_input.update( + { + "previous_catalog_id": previous_catalog_id or None, + "catalog_route_changed": catalog_id != previous_catalog_id, + "incident_semantic_route_rechecked": incident_routing_evaluated, + } + ) return AnsibleCheckModeClaim( op_id=str(row.get("op_id") or ""), source_candidate_op_id=source_candidate_op_id, @@ -5663,6 +5693,10 @@ async def _send_controlled_apply_telegram_receipt( try: from src.services.telegram_gateway import get_telegram_gateway + effective_provider_delivery = str( + claim.input_payload.get("telegram_provider_delivery") + or provider_delivery + ) response = await get_telegram_gateway().send_controlled_apply_result_receipt( automation_run_id=str( claim.input_payload.get("automation_run_id") @@ -5680,7 +5714,7 @@ async def _send_controlled_apply_telegram_receipt( verifier_written=bool(writeback.get("verification")), learning_written=bool(writeback.get("learning")), execution_kind=execution_kind, - provider_delivery=provider_delivery, + provider_delivery=effective_provider_delivery, project_id=project_id, ) return bool( @@ -7654,9 +7688,10 @@ async def claim_catalog_drift_failed_check_modes( """Replay a failed check once when its canonical check playbook changed.""" claims: list[AnsibleCheckModeClaim] = [] - max_age_hours = ( + max_age_hours = max( candidate_max_age_hours - or settings.AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS + or settings.AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS, + _CATALOG_REPAIR_REPLAY_WINDOW_HOURS, ) async with get_db_context(project_id) as db: result = await db.execute( @@ -7668,10 +7703,20 @@ async def claim_catalog_drift_failed_check_modes( check_mode.input ->> 'incident_id', check_mode.incident_id::text ) AS incident_id, - check_mode.input + check_mode.input, + incident.alertname AS incident_alertname, + incident.alert_category AS incident_alert_category, + incident.affected_services AS incident_affected_services FROM automation_operation_log check_mode + JOIN incidents incident + ON incident.incident_id = coalesce( + check_mode.input ->> 'incident_id', + check_mode.incident_id::text + ) + AND incident.project_id = :project_id WHERE check_mode.operation_type = 'ansible_check_mode_executed' AND check_mode.status = 'failed' + AND upper(incident.status::text) NOT IN ('RESOLVED', 'CLOSED') AND check_mode.created_at >= NOW() - ( :candidate_max_age_hours * INTERVAL '1 hour' ) @@ -7686,6 +7731,7 @@ async def claim_catalog_drift_failed_check_modes( FOR UPDATE SKIP LOCKED """), { + "project_id": project_id, "candidate_max_age_hours": max(1, max_age_hours), "scan_limit": min(100, max(10, limit * 10)), }, @@ -7733,6 +7779,8 @@ async def claim_catalog_drift_failed_check_modes( "replay_of_check_mode_op_id": str(row.get("op_id") or ""), "previous_check_mode_playbook_path": previous_check_path, "previous_catalog_revision": previous_catalog_revision or None, + "historical_projection_backfill": True, + "telegram_provider_delivery": "shadow_only", } inserted = await db.execute( text(""" @@ -7761,6 +7809,9 @@ async def claim_catalog_drift_failed_check_modes( = :catalog_replay_revision AND existing_replay.input ->> 'catalog_drift_replay' = 'true' + AND existing_replay.input + ->> 'replay_of_check_mode_op_id' + = :failed_check_mode_op_id ) RETURNING op_id """), @@ -7786,6 +7837,7 @@ async def claim_catalog_drift_failed_check_modes( "parent_op_id": canonical_claim.source_candidate_op_id, "catalog_id": canonical_claim.catalog_id, "catalog_replay_revision": catalog_replay_revision, + "failed_check_mode_op_id": str(row.get("op_id") or ""), "tags": [ "ansible", "check_mode", @@ -7939,6 +7991,197 @@ async def claim_catalog_drift_failed_applies( return claims +async def claim_semantically_misrouted_verified_applies( + *, + project_id: str = "awoooi", + limit: int = 1, + window_hours: int = _CATALOG_REPAIR_REPLAY_WINDOW_HOURS, +) -> list[AnsibleCheckModeClaim]: + """Re-verify terminal applies whose catalog no longer matches the alert. + + Resolved incidents remain eligible because this path corrects false terminal + attribution; it does not reopen incidents from an unresolved alert signal. + """ + + claims: list[AnsibleCheckModeClaim] = [] + async with get_db_context(project_id) as db: + result = await db.execute( + text(""" + SELECT + source_check.op_id, + source_check.parent_op_id, + incident.incident_id, + source_check.input, + incident.alertname AS incident_alertname, + incident.alert_category AS incident_alert_category, + incident.affected_services AS incident_affected_services, + apply.op_id::text AS previous_apply_op_id + FROM automation_operation_log apply + JOIN automation_operation_log source_check + ON source_check.op_id = apply.parent_op_id + AND source_check.operation_type + = 'ansible_check_mode_executed' + JOIN incidents incident + ON incident.incident_id = coalesce( + apply.input ->> 'incident_id', + apply.incident_id::text + ) + AND incident.project_id = :project_id + WHERE apply.operation_type = 'ansible_apply_executed' + AND apply.status = 'success' + AND apply.output ->> 'independent_post_verifier_passed' + = 'true' + AND apply.created_at >= NOW() - ( + :window_hours * INTERVAL '1 hour' + ) + ORDER BY apply.created_at DESC + LIMIT :scan_limit + FOR UPDATE OF apply SKIP LOCKED + """), + { + "project_id": project_id, + "window_hours": max(1, min(window_hours, 7 * 24)), + "scan_limit": min(100, max(10, limit * 10)), + }, + ) + seen_incident_ids: set[str] = set() + for row_value in result.mappings().all(): + row = dict(row_value) + incident_id = str(row.get("incident_id") or "") + if not incident_id or incident_id in seen_incident_ids: + continue + seen_incident_ids.add(incident_id) + previous_input = _json_loads(row.get("input")) + previous_catalog_id = str( + previous_input.get("catalog_id") or "" + ) + canonical_claim = _claim_from_stale_check_mode_row(row) + if ( + canonical_claim is None + or not previous_catalog_id + or canonical_claim.catalog_id == previous_catalog_id + ): + continue + current_revision = str( + canonical_claim.input_payload.get("catalog_revision") or "" + ) + previous_apply_op_id = str( + row.get("previous_apply_op_id") or "" + ) + if not current_revision or not previous_apply_op_id: + continue + semantic_repair_run_id = str( + uuid5( + NAMESPACE_URL, + ( + "awoooi:ansible-semantic-repair:" + f"{project_id}:{previous_apply_op_id}:" + f"{current_revision}" + ), + ) + ) + replay_input = { + **canonical_claim.input_payload, + "automation_run_id": semantic_repair_run_id, + "semantic_route_reconciliation": True, + "semantic_repair_of_apply_op_id": previous_apply_op_id, + "catalog_replay_revision": current_revision, + "historical_projection_backfill": True, + "telegram_provider_delivery": "shadow_only", + "apply_idempotency_key": ( + "ansible-semantic-repair:" + f"{previous_apply_op_id}:{current_revision}" + ), + "bounded_execution_scope": ( + "semantic_catalog_reconciliation_single_apply" + ), + } + inserted = await db.execute( + text(""" + INSERT INTO automation_operation_log ( + operation_type, actor, status, incident_id, + input, output, dry_run_result, + parent_op_id, tags + ) SELECT + 'ansible_check_mode_executed', + 'ansible_check_mode_worker', + 'pending', + :incident_db_id, + CAST(:input AS jsonb), + '{}'::jsonb, + CAST(:dry_run_result AS jsonb), + CAST(:parent_op_id AS uuid), + :tags + WHERE NOT EXISTS ( + SELECT 1 + FROM automation_operation_log existing_replay + WHERE existing_replay.operation_type + = 'ansible_check_mode_executed' + AND existing_replay.input + ->> 'semantic_repair_of_apply_op_id' + = :previous_apply_op_id + AND existing_replay.input + ->> 'catalog_replay_revision' + = :catalog_replay_revision + ) + RETURNING op_id + """), + { + "incident_db_id": _automation_operation_log_incident_id( + canonical_claim.incident_id + ), + "input": json.dumps(replay_input, ensure_ascii=False), + "dry_run_result": json.dumps( + { + "check_mode_executed": False, + "apply_executed": False, + "claim_state": ( + "semantic_catalog_reconciliation" + ), + "controlled_apply_allowed": bool( + replay_input.get("controlled_apply_allowed") + ), + }, + ensure_ascii=False, + ), + "parent_op_id": ( + canonical_claim.source_candidate_op_id + ), + "previous_apply_op_id": previous_apply_op_id, + "catalog_replay_revision": current_revision, + "tags": [ + "ansible", + "check_mode", + "pending", + "semantic_catalog_reconciliation", + ], + }, + ) + op_id = inserted.scalar_one_or_none() + if op_id is None: + continue + claims.append( + AnsibleCheckModeClaim( + op_id=str(op_id), + source_candidate_op_id=( + canonical_claim.source_candidate_op_id + ), + incident_id=canonical_claim.incident_id, + catalog_id=canonical_claim.catalog_id, + playbook_path=canonical_claim.playbook_path, + apply_playbook_path=( + canonical_claim.apply_playbook_path + ), + inventory_hosts=canonical_claim.inventory_hosts, + risk_level=canonical_claim.risk_level, + input_payload=replay_input, + ) + ) + if len(claims) >= max(1, limit): + break + return claims + + async def recent_ansible_transport_blockers( *, project_id: str = "awoooi", @@ -8716,6 +8959,30 @@ async def run_pending_check_modes_once( 0, remaining_limit - len(stdin_boundary_replay_claims), ) + semantic_catalog_reconciliation_claims: list[ + AnsibleCheckModeClaim + ] = [] + semantic_catalog_reconciliation_error: str | None = None + if remaining_limit: + try: + semantic_catalog_reconciliation_claims = ( + await claim_semantically_misrouted_verified_applies( + project_id=project_id, + limit=remaining_limit, + ) + ) + except Exception as exc: + semantic_catalog_reconciliation_error = type(exc).__name__ + logger.warning( + "ansible_semantic_catalog_reconciliation_claim_failed", + project_id=project_id, + error_type=semantic_catalog_reconciliation_error, + fresh_candidate_claim_continues=True, + ) + remaining_limit = max( + 0, + remaining_limit - len(semantic_catalog_reconciliation_claims), + ) failed_apply_catalog_replay_claims: list[AnsibleCheckModeClaim] = [] failed_apply_catalog_replay_error: str | None = None if remaining_limit: @@ -8775,6 +9042,7 @@ async def run_pending_check_modes_once( claims = [ *reclaimed_claims, *stdin_boundary_replay_claims, + *semantic_catalog_reconciliation_claims, *failed_apply_catalog_replay_claims, *catalog_replay_claims, *fresh_claims, @@ -8924,6 +9192,12 @@ async def run_pending_check_modes_once( stdin_boundary_replay_terminal_written ), "stdin_boundary_replay_error": stdin_boundary_replay_error, + "semantic_catalog_reconciled": len( + semantic_catalog_reconciliation_claims + ), + "semantic_catalog_reconciliation_error": ( + semantic_catalog_reconciliation_error + ), "failed_apply_catalog_replayed": len( failed_apply_catalog_replay_claims ), diff --git a/apps/api/src/services/awooop_ansible_post_verifier.py b/apps/api/src/services/awooop_ansible_post_verifier.py index be5806d5a..c822f6e16 100644 --- a/apps/api/src/services/awooop_ansible_post_verifier.py +++ b/apps/api/src/services/awooop_ansible_post_verifier.py @@ -98,6 +98,36 @@ _POSTCONDITION_REGISTRY: dict[str, tuple[AssetPostcondition, ...]] = { "test -s /home/wooo/node_exporter_textfiles/docker_stats.prom", ), ), + "ansible:110-disk-pressure": ( + AssetPostcondition( + "host_110_root_disk_below_alert_threshold", + "metric", + "host_110", + "usage=$(df -P / | awk 'END {gsub(/%/, \"\", $5); print $5}'); " + "test -n \"$usage\" && test \"$usage\" -lt 85", + ), + AssetPostcondition( + "host_110_docker_runtime_after_cleanup", + "service", + "host_110", + "docker info >/dev/null 2>&1 && test \"$(docker ps -q | wc -l)\" -gt 0", + ), + ), + "ansible:188-disk-pressure": ( + AssetPostcondition( + "host_188_root_disk_below_alert_threshold", + "metric", + "host_188", + "usage=$(df -P / | awk 'END {gsub(/%/, \"\", $5); print $5}'); " + "test -n \"$usage\" && test \"$usage\" -lt 85", + ), + AssetPostcondition( + "host_188_docker_runtime_after_cleanup", + "service", + "host_188", + "docker info >/dev/null 2>&1 && test \"$(docker ps -q | wc -l)\" -gt 0", + ), + ), "ansible:110-devops": ( AssetPostcondition( "host_110_node_exporter_container", diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index a13e22dfb..9b3dd07f8 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -5312,9 +5312,9 @@ class TelegramGateway: row = existing.mappings().one_or_none() if row is not None: existing_row = dict(row) - elif failure_fingerprint: + elif force_suppress or failure_fingerprint: should_suppress = force_suppress - if not should_suppress: + if not should_suppress and failure_fingerprint: group_reservation = await db.execute( text(""" SELECT message_id::text AS message_id @@ -10370,6 +10370,20 @@ class TelegramGateway: "km_rag_playbook", ], } + elif provider_delivery == "shadow_only": + source_extra["notification_policy"] = { + "policy_version": "telegram_historical_repair_shadow_v1", + "failure_fingerprint": "", + "group_scope": "historical_projection_backfill", + "provider_delivery": "shadow_only", + "disposition": "suppressed", + "provider_send_performed": False, + "details_retained_in": [ + "automation_operation_log", + "awooop_runs", + "km_rag_playbook", + ], + } if no_write_replay: lines = [ f"{html.escape(title)}", diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py index fd7bcd5e9..de64fff4a 100644 --- a/apps/api/tests/test_ansible_verified_closure.py +++ b/apps/api/tests/test_ansible_verified_closure.py @@ -992,6 +992,63 @@ async def test_historical_no_write_backfill_is_provider_silent( assert policy["suppression_reason"] == "historical_projection_backfill" +@pytest.mark.asyncio +async def test_historical_verified_repair_is_provider_silent( + monkeypatch: pytest.MonkeyPatch, +) -> None: + db = _SequenceDB( + _MappingResult(), + _MappingResult(row=None), + ) + + @asynccontextmanager + async def fake_get_db_context(_project_id): + yield db + + record_outbound = AsyncMock( + return_value="00000000-0000-0000-0000-000000000498" + ) + monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context) + monkeypatch.setattr( + "src.services.channel_hub.record_outbound_message", + record_outbound, + ) + gateway = TelegramGateway() + identity = { + "project_id": "awoooi", + "automation_run_id": "00000000-0000-0000-0000-000000000402", + "incident_id": "INC-20260711-HISTORICAL-SUCCESS", + "apply_op_id": "00000000-0000-0000-0000-000000000404", + } + source_extra = { + "callback_reply": { + "action": "controlled_apply_result", + "execution_kind": "controlled_apply", + **identity, + }, + "notification_policy": { + "failure_fingerprint": "", + "disposition": "suppressed", + "provider_delivery": "shadow_only", + }, + } + + reservation = await gateway._reserve_controlled_apply_result_outbound( + identity=identity, + payload={"chat_id": "sre-chat", "text": "historical repair closed"}, + source_envelope_extra=source_extra, + ) + + assert reservation["status"] == "suppressed" + assert len(db.statements) == 2 + record_kwargs = record_outbound.await_args.kwargs + assert record_kwargs["send_status"] == "shadow" + assert record_kwargs["provider_message_id"] is None + policy = record_kwargs["source_envelope"]["notification_policy"] + assert policy["suppression_reason"] == "historical_projection_backfill" + assert policy["provider_send_performed"] is False + + def test_controlled_result_uses_durable_reservation_and_single_reconcile_entry() -> None: reserve_source = inspect.getsource( TelegramGateway._reserve_controlled_apply_result_outbound diff --git a/apps/api/tests/test_host_disk_pressure_ansible_catalog.py b/apps/api/tests/test_host_disk_pressure_ansible_catalog.py new file mode 100644 index 000000000..d373e032c --- /dev/null +++ b/apps/api/tests/test_host_disk_pressure_ansible_catalog.py @@ -0,0 +1,265 @@ +from __future__ import annotations + +import inspect +from contextlib import asynccontextmanager +from pathlib import Path +from uuid import UUID + +import pytest +import yaml + +from src.services.awooop_ansible_audit_service import ( + _catalog_hints, + get_ansible_catalog_item, +) +from src.services.awooop_ansible_check_mode_service import ( + _claim_from_stale_check_mode_row, + _send_controlled_apply_telegram_receipt, + claim_catalog_drift_failed_check_modes, + claim_semantically_misrouted_verified_applies, + run_pending_check_modes_once, +) +from src.services.awooop_ansible_post_verifier import postconditions_for_catalog + +ROOT = Path(__file__).resolve().parents[3] +PLAYBOOK = ( + ROOT + / "infra" + / "ansible" + / "playbooks" + / "host-disk-pressure-bounded-cleanup.yml" +) + + +def _disk_incident(host: str) -> dict: + return { + "alertname": "HostOutOfDiskSpace", + "alert_category": "host_resource", + "affected_services": [f"node-exporter-{host}"], + } + + +def test_disk_alert_routes_to_the_matching_host_catalog() -> None: + host110 = _catalog_hints(_disk_incident("110"), None) + host188 = _catalog_hints(_disk_incident("188"), None) + + assert [row["catalog_id"] for row in host110["candidates"]] == [ + "ansible:110-disk-pressure" + ] + assert [row["catalog_id"] for row in host188["candidates"]] == [ + "ansible:188-disk-pressure" + ] + assert all( + row["catalog_id"] != "ansible:110-devops" + for row in [*host110["candidates"], *host188["candidates"]] + ) + + +def test_node_exporter_repair_requires_a_down_signal_not_a_disk_alert() -> None: + disk_hints = _catalog_hints(_disk_incident("110"), None) + down_hints = _catalog_hints( + { + "alertname": "NodeExporterDown", + "affected_services": ["node-exporter-110"], + }, + None, + ) + + assert "ansible:110-devops" not in { + row["catalog_id"] for row in disk_hints["candidates"] + } + assert [row["catalog_id"] for row in down_hints["candidates"]] == [ + "ansible:110-devops" + ] + + +def test_disk_cleanup_playbook_excludes_destructive_asset_classes() -> None: + payload = yaml.safe_load(PLAYBOOK.read_text(encoding="utf-8")) + source = PLAYBOOK.read_text(encoding="utf-8").lower() + + assert payload[0]["hosts"] == "devops:ai_web" + assert payload[0]["become"] is False + assert "docker-disk-pressure-retention-cleanup.py" in source + assert "--apply" in source + assert "--max-image-removals" in source + assert "single_apply_image_removal_is_count_bounded" in source + assert "--include-builder-cache" not in source + for forbidden in ( + "docker system prune", + "docker volume prune", + "truncate -s", + "rm -rf", + "reboot", + "systemctl restart", + ): + assert forbidden not in source + + +def test_stale_generic_host188_row_is_semantically_remapped() -> None: + claim = _claim_from_stale_check_mode_row( + { + "op_id": "00000000-0000-0000-0000-000000000188", + "parent_op_id": "00000000-0000-0000-0000-000000000100", + "incident_id": "INC-DISK-188", + "incident_alertname": "HostOutOfDiskSpace", + "incident_alert_category": "host_resource", + "incident_affected_services": ["node-exporter-188"], + "input": { + "source_candidate_op_id": ( + "00000000-0000-0000-0000-000000000100" + ), + "incident_id": "INC-DISK-188", + "catalog_id": "ansible:110-devops", + "catalog_playbook_path": "infra/ansible/playbooks/110-devops.yml", + "playbook_path": "infra/ansible/playbooks/110-devops.yml", + "inventory_hosts": ["host_110"], + "risk_level": "medium", + }, + } + ) + + assert claim is not None + assert claim.catalog_id == "ansible:188-disk-pressure" + assert claim.inventory_hosts == ("host_188",) + assert claim.input_payload["catalog_route_changed"] is True + assert claim.input_payload["incident_semantic_route_rechecked"] is True + + +def test_disk_catalogs_have_independent_threshold_verifiers() -> None: + for host in ("110", "188"): + catalog_id = f"ansible:{host}-disk-pressure" + catalog = get_ansible_catalog_item(catalog_id) + conditions = postconditions_for_catalog(catalog_id) + + assert catalog is not None + assert catalog["auto_apply_enabled"] is True + assert catalog["risk_level"] == "medium" + assert len(conditions) == 2 + assert any("df -P /" in condition.probe for condition in conditions) + assert any("docker info" in condition.probe for condition in conditions) + + +def test_historical_catalog_replay_is_per_failed_row_and_provider_silent() -> None: + claim_source = inspect.getsource(claim_catalog_drift_failed_check_modes) + send_source = inspect.getsource(_send_controlled_apply_telegram_receipt) + + assert "replay_of_check_mode_op_id" in claim_source + assert "failed_check_mode_op_id" in claim_source + assert '"historical_projection_backfill": True' in claim_source + assert '"telegram_provider_delivery": "shadow_only"' in claim_source + assert "telegram_provider_delivery" in send_source + + +def test_verified_wrong_catalog_is_reconciled_before_failure_replays() -> None: + claim_source = inspect.getsource( + claim_semantically_misrouted_verified_applies + ) + worker_source = inspect.getsource(run_pending_check_modes_once) + + assert "semantic_route_reconciliation" in claim_source + assert "semantic_repair_of_apply_op_id" in claim_source + assert '"telegram_provider_delivery": "shadow_only"' in claim_source + assert "independent_post_verifier_passed" in claim_source + assert "seen_incident_ids" in claim_source + assert "RESOLVED" not in claim_source + assert worker_source.index( + "claim_semantically_misrouted_verified_applies" + ) < worker_source.index("claim_catalog_drift_failed_applies") + assert "semantic_catalog_reconciled" in worker_source + + +@pytest.mark.asyncio +async def test_verified_wrong_catalog_enqueues_one_shadow_disk_repair( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.services import awooop_ansible_check_mode_service as service + + previous_apply_op_id = "00000000-0000-0000-0000-000000000202" + + class Result: + def __init__(self, *, rows=None, scalar=None): + self._rows = rows or [] + self._scalar = scalar + + def mappings(self): + return self + + def all(self): + return self._rows + + def scalar_one_or_none(self): + return self._scalar + + class DB: + def __init__(self): + self.statements: list[str] = [] + + async def execute(self, statement, _params): + self.statements.append(str(statement)) + if len(self.statements) == 1: + return Result( + rows=[ + { + "op_id": ( + "00000000-0000-0000-0000-000000000201" + ), + "parent_op_id": ( + "00000000-0000-0000-0000-000000000200" + ), + "incident_id": "INC-DISK-188-VERIFIED-WRONG", + "incident_alertname": "HostOutOfDiskSpace", + "incident_alert_category": "host_resource", + "incident_affected_services": [ + "node-exporter-188" + ], + "previous_apply_op_id": previous_apply_op_id, + "input": { + "source_candidate_op_id": ( + "00000000-0000-0000-0000-000000000200" + ), + "incident_id": ( + "INC-DISK-188-VERIFIED-WRONG" + ), + "catalog_id": "ansible:110-devops", + "catalog_playbook_path": ( + "infra/ansible/playbooks/" + "110-node-exporter-repair.yml" + ), + "playbook_path": ( + "infra/ansible/playbooks/" + "110-node-exporter-repair.yml" + ), + "inventory_hosts": ["host_110"], + "risk_level": "low", + }, + } + ] + ) + return Result( + scalar="00000000-0000-0000-0000-000000000203" + ) + + db = DB() + + @asynccontextmanager + async def fake_get_db_context(_project_id): + yield db + + monkeypatch.setattr(service, "get_db_context", fake_get_db_context) + + claims = await claim_semantically_misrouted_verified_applies(limit=1) + + assert len(claims) == 1 + claim = claims[0] + assert claim.catalog_id == "ansible:188-disk-pressure" + assert claim.inventory_hosts == ("host_188",) + assert claim.input_payload["semantic_repair_of_apply_op_id"] == ( + previous_apply_op_id + ) + assert UUID(claim.input_payload["automation_run_id"]) + assert claim.input_payload["automation_run_id"] != ( + "00000000-0000-0000-0000-000000000200" + ) + assert claim.input_payload["telegram_provider_delivery"] == "shadow_only" + assert "DISTINCT ON" not in db.statements[0] + assert "FOR UPDATE OF apply SKIP LOCKED" in db.statements[0] diff --git a/infra/ansible/playbooks/host-disk-pressure-bounded-cleanup.yml b/infra/ansible/playbooks/host-disk-pressure-bounded-cleanup.yml new file mode 100644 index 000000000..18bee4b32 --- /dev/null +++ b/infra/ansible/playbooks/host-disk-pressure-bounded-cleanup.yml @@ -0,0 +1,124 @@ +--- +- name: Host disk pressure bounded Docker artifact cleanup + hosts: devops:ai_web + gather_facts: false + become: false + serial: 1 + any_errors_fatal: true + + vars: + disk_cleanup_script: >- + {{ playbook_dir }}/../../../scripts/ops/docker-disk-pressure-retention-cleanup.py + disk_cleanup_gate_percent: 85 + disk_cleanup_min_age_hours: 24 + disk_cleanup_keep_dangling_newest: >- + {{ 3 if inventory_hostname == 'host_188' else 20 }} + disk_cleanup_include_tagged_images: >- + {{ inventory_hostname == 'host_188' }} + disk_cleanup_tagged_min_age_hours: 168 + disk_cleanup_keep_tagged_newest_per_repository: 3 + disk_cleanup_max_image_removals: >- + {{ 50 if inventory_hostname == 'host_188' else 100 }} + + tasks: + - name: Read root filesystem usage before cleanup + ansible.builtin.command: + argv: + - df + - -P + - / + register: disk_before + check_mode: false + changed_when: false + + - name: Normalize the disk pressure gate + ansible.builtin.set_fact: + disk_before_used_percent: >- + {{ disk_before.stdout_lines[-1].split()[4] | replace('%', '') | int }} + disk_cleanup_should_apply: >- + {{ + (not ansible_check_mode) + and ( + disk_before.stdout_lines[-1].split()[4] + | replace('%', '') + | int + ) >= disk_cleanup_gate_percent + }} + changed_when: false + + - name: Plan or apply bounded unreferenced image cleanup + ansible.builtin.script: + cmd: >- + {{ disk_cleanup_script }} + --host-label {{ inventory_hostname }} + --min-age-hours {{ disk_cleanup_min_age_hours }} + --keep-dangling-newest {{ disk_cleanup_keep_dangling_newest }} + --tagged-min-age-hours {{ disk_cleanup_tagged_min_age_hours }} + --keep-tagged-newest-per-repository + {{ disk_cleanup_keep_tagged_newest_per_repository }} + --max-image-removals {{ disk_cleanup_max_image_removals }} + {{ '--include-tagged-images' if disk_cleanup_include_tagged_images else '' }} + {{ '--apply' if disk_cleanup_should_apply else '' }} + register: disk_cleanup_command + check_mode: false + changed_when: >- + disk_cleanup_should_apply + and ((disk_cleanup_command.stdout | from_json).removed_image_ids | length > 0) + + - name: Parse the bounded cleanup receipt + ansible.builtin.set_fact: + disk_cleanup_receipt: "{{ disk_cleanup_command.stdout | from_json }}" + changed_when: false + + - name: Enforce the cleanup safety boundary + ansible.builtin.assert: + that: + - not disk_cleanup_receipt.boundaries.touches_containers + - not disk_cleanup_receipt.boundaries.touches_volumes + - not disk_cleanup_receipt.boundaries.touches_databases + - not disk_cleanup_receipt.boundaries.touches_backups + - not disk_cleanup_receipt.boundaries.uses_docker_system_prune + - disk_cleanup_receipt.boundaries.removes_only_container_unreferenced_images + - disk_cleanup_receipt.boundaries.tagged_image_cleanup_preserves_repository_rollback_window + - disk_cleanup_receipt.boundaries.single_apply_image_removal_is_count_bounded + - >- + (disk_cleanup_receipt.removed_image_ids | length) + <= (disk_cleanup_max_image_removals | int) + fail_msg: bounded_disk_cleanup_boundary_failed + success_msg: bounded_disk_cleanup_boundary_verified + changed_when: false + + - name: Read root filesystem usage after cleanup + ansible.builtin.command: + argv: + - df + - -P + - / + register: disk_after + check_mode: false + changed_when: false + + - name: Emit public-safe disk cleanup receipt + ansible.builtin.debug: + msg: + schema_version: host_disk_pressure_bounded_cleanup_v1 + host: "{{ inventory_hostname }}" + requested_mode: "{{ 'apply' if not ansible_check_mode else 'check' }}" + runtime_apply_executed: "{{ disk_cleanup_should_apply }}" + disk_before_used_percent: "{{ disk_before_used_percent }}" + disk_after_used_percent: >- + {{ disk_after.stdout_lines[-1].split()[4] | replace('%', '') | int }} + dangling_candidate_count: >- + {{ disk_cleanup_receipt.dangling_image_removal_plan.count }} + tagged_candidate_count: >- + {{ disk_cleanup_receipt.tagged_image_removal_plan.count }} + eligible_candidate_count: >- + {{ disk_cleanup_receipt.eligible_image_removal_plan.count }} + max_image_removals: "{{ disk_cleanup_max_image_removals }}" + removed_image_count: "{{ disk_cleanup_receipt.removed_image_ids | length }}" + touches_containers: false + touches_volumes: false + touches_databases: false + touches_backups: false + uses_docker_system_prune: false + secret_value_exposed: false diff --git a/scripts/ops/docker-disk-pressure-retention-cleanup.py b/scripts/ops/docker-disk-pressure-retention-cleanup.py index b8722df98..106422c87 100755 --- a/scripts/ops/docker-disk-pressure-retention-cleanup.py +++ b/scripts/ops/docker-disk-pressure-retention-cleanup.py @@ -2,9 +2,10 @@ """Bounded Docker disk-pressure cleanup for host reboot recovery. This controller intentionally avoids `docker system prune`, volumes, containers, -running images, databases, backups, and logs. It only removes dangling images -that are not referenced by any container, and can optionally run a bounded -BuildKit cache cleanup with an explicit keep-storage floor. +running images, databases, backups, and logs. It removes only images that are +not referenced by any container. Tagged-image cleanup is explicit, age-bounded, +and preserves a per-repository rollback window. BuildKit cleanup is separately +opt-in and preserves an explicit storage floor. """ from __future__ import annotations @@ -13,14 +14,18 @@ import argparse import json import subprocess import sys +from collections.abc import Iterable from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path -from typing import Any, Iterable - +from typing import Any +UTC = timezone.utc # noqa: UP017 - controller still runs on Python 3.9 hosts. DEFAULT_MIN_AGE_HOURS = 24 DEFAULT_KEEP_DANGLING_NEWEST = 20 +DEFAULT_TAGGED_MIN_AGE_HOURS = 168 +DEFAULT_KEEP_TAGGED_NEWEST_PER_REPOSITORY = 3 +DEFAULT_MAX_IMAGE_REMOVALS = 100 DEFAULT_BUILDER_KEEP_STORAGE = "30GB" @@ -42,6 +47,30 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--host-label", default="") parser.add_argument("--min-age-hours", type=int, default=DEFAULT_MIN_AGE_HOURS) parser.add_argument("--keep-dangling-newest", type=int, default=DEFAULT_KEEP_DANGLING_NEWEST) + parser.add_argument( + "--include-tagged-images", + action="store_true", + help=( + "Also remove old, container-unreferenced tagged images while preserving " + "the newest images for every repository." + ), + ) + parser.add_argument( + "--tagged-min-age-hours", + type=int, + default=DEFAULT_TAGGED_MIN_AGE_HOURS, + ) + parser.add_argument( + "--keep-tagged-newest-per-repository", + type=int, + default=DEFAULT_KEEP_TAGGED_NEWEST_PER_REPOSITORY, + ) + parser.add_argument( + "--max-image-removals", + type=int, + default=DEFAULT_MAX_IMAGE_REMOVALS, + help="Maximum image IDs removed by one controlled apply.", + ) parser.add_argument( "--skip-dangling-images", action="store_true", @@ -102,8 +131,8 @@ def parse_docker_datetime(value: str) -> datetime: text = f"{head}.{frac_text}{suffix}" parsed = datetime.fromisoformat(text) if parsed.tzinfo is None: - parsed = parsed.replace(tzinfo=timezone.utc) - return parsed.astimezone(timezone.utc) + parsed = parsed.replace(tzinfo=UTC) + return parsed.astimezone(UTC) def chunked(values: list[str], size: int) -> Iterable[list[str]]: @@ -170,6 +199,45 @@ def get_dangling_images(docker_bin: str) -> list[ImageInfo]: return images +def get_all_images(docker_bin: str) -> list[ImageInfo]: + image_ids = { + normalize_image_id(value) + for value in docker( + ["image", "ls", "--all", "--quiet", "--no-trunc"], + docker_bin, + ).stdout.split() + if normalize_image_id(value) + } + images: list[ImageInfo] = [] + for group in chunked(sorted(image_ids), 100): + result = docker(["image", "inspect", *group], docker_bin) + payload = json.loads(result.stdout or "[]") + for item in payload: + image_id = normalize_image_id(str(item.get("Id") or "")) + if not image_id: + continue + tags = item.get("RepoTags") or [] + images.append( + ImageInfo( + image_id=image_id, + created_at=parse_docker_datetime(str(item.get("Created") or "")), + size_bytes=int(item.get("Size") or 0), + repo_tags=tuple(str(tag) for tag in tags if tag), + ) + ) + return images + + +def repository_from_tag(tag: str) -> str: + value = str(tag or "").strip() + if "@" in value: + return value.split("@", 1)[0] + tail = value.rsplit("/", 1)[-1] + if ":" not in tail: + return value + return value.rsplit(":", 1)[0] + + def select_dangling_image_removals( images: list[ImageInfo], protected_ids: set[str], @@ -192,6 +260,45 @@ def select_dangling_image_removals( return sorted(dangling, key=lambda image: image.created_at) +def select_tagged_image_removals( + images: list[ImageInfo], + protected_ids: set[str], + *, + now: datetime, + min_age_hours: int, + keep_newest_per_repository: int, +) -> list[ImageInfo]: + unreferenced = [ + image + for image in images + if normalize_image_id(image.image_id) not in protected_ids + and image.repo_tags + ] + repository_images: dict[str, list[ImageInfo]] = {} + for image in unreferenced: + for repository in { + repository_from_tag(tag) for tag in image.repo_tags if tag + }: + repository_images.setdefault(repository, []).append(image) + + retained_ids: set[str] = set() + for repository_rows in repository_images.values(): + repository_rows.sort(key=lambda image: image.created_at, reverse=True) + retained_ids.update( + normalize_image_id(image.image_id) + for image in repository_rows[:keep_newest_per_repository] + ) + + cutoff_seconds = min_age_hours * 3600 + selected = [ + image + for image in unreferenced + if normalize_image_id(image.image_id) not in retained_ids + and (now - image.created_at).total_seconds() >= cutoff_seconds + ] + return sorted(selected, key=lambda image: image.created_at) + + def summarize_images(images: list[ImageInfo]) -> dict[str, Any]: return { "count": len(images), @@ -202,13 +309,29 @@ def summarize_images(images: list[ImageInfo]) -> dict[str, Any]: } +def limit_image_removals( + images: list[ImageInfo], + *, + max_removals: int, +) -> list[ImageInfo]: + return sorted(images, key=lambda image: image.created_at)[:max_removals] + + def remove_images(images: list[ImageInfo], docker_bin: str) -> list[str]: removed: list[str] = [] - for group in chunked([image.image_id for image in images], 25): - if not group: - continue + dangling_ids = [image.image_id for image in images if not image.repo_tags] + for group in chunked(dangling_ids, 25): docker(["image", "rm", *group], docker_bin) removed.extend(group) + tagged_images = [image for image in images if image.repo_tags] + tagged_references = [ + reference + for image in tagged_images + for reference in image.repo_tags + ] + for group in chunked(tagged_references, 25): + docker(["image", "rm", *group], docker_bin) + removed.extend(image.image_id for image in tagged_images) return removed @@ -227,11 +350,11 @@ def builder_prune_command(args: argparse.Namespace) -> list[str]: def build_receipt(args: argparse.Namespace) -> dict[str, Any]: - now = datetime.now(timezone.utc) + now = datetime.now(UTC) before = current_disk_bytes(args.disk_path) protected_ids = get_container_image_ids(args.docker_bin) dangling_images = get_dangling_images(args.docker_bin) - removal_candidates = ( + dangling_removal_candidates = ( [] if args.skip_dangling_images else select_dangling_image_removals( @@ -242,6 +365,29 @@ def build_receipt(args: argparse.Namespace) -> dict[str, Any]: keep_newest=args.keep_dangling_newest, ) ) + all_images = get_all_images(args.docker_bin) if args.include_tagged_images else [] + tagged_removal_candidates = ( + select_tagged_image_removals( + all_images, + protected_ids, + now=now, + min_age_hours=args.tagged_min_age_hours, + keep_newest_per_repository=args.keep_tagged_newest_per_repository, + ) + if args.include_tagged_images + else [] + ) + candidates_by_id = { + normalize_image_id(image.image_id): image + for image in [*dangling_removal_candidates, *tagged_removal_candidates] + } + eligible_removal_candidates = sorted( + candidates_by_id.values(), key=lambda image: image.created_at + ) + removal_candidates = limit_image_removals( + eligible_removal_candidates, + max_removals=args.max_image_removals, + ) receipt: dict[str, Any] = { "schema_version": "awoooi_docker_disk_pressure_retention_cleanup_v1", "generated_at": now.isoformat(), @@ -254,12 +400,21 @@ def build_receipt(args: argparse.Namespace) -> dict[str, Any]: "touches_databases": False, "touches_backups": False, "uses_docker_system_prune": False, - "removes_only_unreferenced_dangling_images": True, + "removes_only_container_unreferenced_images": True, + "tagged_image_cleanup_requires_explicit_flag": True, + "tagged_image_cleanup_preserves_repository_rollback_window": True, + "single_apply_image_removal_is_count_bounded": True, "builder_cache_cleanup_requires_explicit_flag": True, }, "parameters": { "min_age_hours": args.min_age_hours, "keep_dangling_newest": args.keep_dangling_newest, + "include_tagged_images": args.include_tagged_images, + "tagged_min_age_hours": args.tagged_min_age_hours, + "keep_tagged_newest_per_repository": ( + args.keep_tagged_newest_per_repository + ), + "max_image_removals": args.max_image_removals, "include_builder_cache": args.include_builder_cache, "builder_keep_storage": args.builder_keep_storage, "skip_dangling_images": args.skip_dangling_images, @@ -267,7 +422,12 @@ def build_receipt(args: argparse.Namespace) -> dict[str, Any]: "disk_before": before, "protected_container_image_count": len(protected_ids), "dangling_image_total_count": len(dangling_images), - "dangling_image_removal_plan": summarize_images(removal_candidates), + "dangling_image_removal_plan": summarize_images(dangling_removal_candidates), + "tagged_image_removal_plan": summarize_images(tagged_removal_candidates), + "eligible_image_removal_plan": summarize_images( + eligible_removal_candidates + ), + "total_image_removal_plan": summarize_images(removal_candidates), "builder_cache_command": builder_prune_command(args)[1:] if args.include_builder_cache else None, "removed_image_ids": [], "builder_cache_cleanup_executed": False, @@ -292,6 +452,15 @@ def main() -> int: if args.keep_dangling_newest < 0: print("keep-dangling-newest must be >= 0", file=sys.stderr) return 2 + if args.tagged_min_age_hours <= 0: + print("tagged-min-age-hours must be > 0", file=sys.stderr) + return 2 + if args.keep_tagged_newest_per_repository < 1: + print("keep-tagged-newest-per-repository must be >= 1", file=sys.stderr) + return 2 + if args.max_image_removals < 1: + print("max-image-removals must be >= 1", file=sys.stderr) + return 2 receipt = build_receipt(args) text = json.dumps(receipt, ensure_ascii=False, indent=2, sort_keys=True) + "\n" if args.output: diff --git a/scripts/ops/tests/test_docker_disk_pressure_retention_cleanup.py b/scripts/ops/tests/test_docker_disk_pressure_retention_cleanup.py index e1e67d655..9027b0a77 100644 --- a/scripts/ops/tests/test_docker_disk_pressure_retention_cleanup.py +++ b/scripts/ops/tests/test_docker_disk_pressure_retention_cleanup.py @@ -6,7 +6,6 @@ from datetime import datetime, timedelta, timezone from pathlib import Path from types import SimpleNamespace - ROOT = Path(__file__).resolve().parents[3] SCRIPT = ROOT / "scripts" / "ops" / "docker-disk-pressure-retention-cleanup.py" spec = importlib.util.spec_from_file_location("docker_disk_pressure_retention_cleanup", SCRIPT) @@ -14,6 +13,7 @@ module = importlib.util.module_from_spec(spec) assert spec and spec.loader sys.modules[spec.name] = module spec.loader.exec_module(module) +UTC = timezone.utc # noqa: UP017 - mirrors supported Python 3.9 runtime. def image(image_id: str, created_at: datetime, tags: tuple[str, ...] = ()): @@ -26,7 +26,7 @@ def image(image_id: str, created_at: datetime, tags: tuple[str, ...] = ()): def test_select_dangling_images_keeps_newest_and_protects_running_images() -> None: - now = datetime(2026, 7, 1, 12, tzinfo=timezone.utc) + now = datetime(2026, 7, 1, 12, tzinfo=UTC) images = [ image("sha256:running", now - timedelta(hours=72)), image("oldest", now - timedelta(hours=72)), @@ -73,7 +73,7 @@ def test_parse_docker_datetime_accepts_nanosecond_fraction() -> None: def test_summary_never_reports_volumes_or_container_cleanup_boundary() -> None: - now = datetime(2026, 7, 1, 12, tzinfo=timezone.utc) + now = datetime(2026, 7, 1, 12, tzinfo=UTC) selected = module.select_dangling_image_removals( [image("old", now - timedelta(days=3))], set(), @@ -87,6 +87,88 @@ def test_summary_never_reports_volumes_or_container_cleanup_boundary() -> None: assert summary["sample_image_ids"] == ["old"] +def test_tagged_cleanup_preserves_running_and_repository_rollback_window() -> None: + now = datetime(2026, 7, 14, 12, tzinfo=UTC) + images = [ + image("running", now - timedelta(days=30), ("registry:5000/app:running",)), + image("latest", now - timedelta(days=1), ("registry:5000/app:latest",)), + image("rollback-1", now - timedelta(days=2), ("registry:5000/app:sha-1",)), + image("rollback-2", now - timedelta(days=3), ("registry:5000/app:sha-2",)), + image("expired", now - timedelta(days=14), ("registry:5000/app:sha-old",)), + image("other-repo", now - timedelta(days=14), ("registry:5000/other:only",)), + ] + + selected = module.select_tagged_image_removals( + images, + {"running"}, + now=now, + min_age_hours=7 * 24, + keep_newest_per_repository=3, + ) + + assert [item.image_id for item in selected] == ["expired"] + + +def test_multi_repository_image_is_retained_when_any_repository_needs_it() -> None: + now = datetime(2026, 7, 14, 12, tzinfo=UTC) + shared = image( + "shared", + now - timedelta(days=30), + ("registry:5000/app:old", "registry:5000/critical:latest"), + ) + selected = module.select_tagged_image_removals( + [shared, image("app-new", now - timedelta(days=1), ("registry:5000/app:new",))], + set(), + now=now, + min_age_hours=7 * 24, + keep_newest_per_repository=1, + ) + + assert selected == [] + + +def test_repository_parser_preserves_registry_port() -> None: + assert module.repository_from_tag("registry:5000/team/app:sha") == "registry:5000/team/app" + + +def test_single_apply_image_removal_is_count_bounded() -> None: + now = datetime(2026, 7, 14, 12, tzinfo=UTC) + candidates = [ + image(f"image-{index}", now - timedelta(days=index + 1)) + for index in range(5) + ] + + selected = module.limit_image_removals(candidates, max_removals=2) + + assert [item.image_id for item in selected] == ["image-4", "image-3"] + + +def test_remove_images_batches_tagged_references(monkeypatch) -> None: + calls: list[list[str]] = [] + + def fake_docker(args: list[str], _docker_bin: str): + calls.append(args) + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(module, "docker", fake_docker) + now = datetime(2026, 7, 14, 12, tzinfo=UTC) + + removed = module.remove_images( + [ + image("dangling", now), + image("tagged-a", now, ("repo-a:old", "repo-b:old")), + image("tagged-b", now, ("repo-c:old",)), + ], + "docker", + ) + + assert removed == ["dangling", "tagged-a", "tagged-b"] + assert calls == [ + ["image", "rm", "dangling"], + ["image", "rm", "repo-a:old", "repo-b:old", "repo-c:old"], + ] + + def test_cli_exposes_builder_cache_only_flag() -> None: help_text = module.run_command(["python3", str(SCRIPT), "--help"]).stdout