fix(iwooos): isolate control plane source reads
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m34s
CD Pipeline / build-and-deploy (push) Failing after 8m21s
AWOOOI Harbor 110 Local Repair / workflow-shape (push) Successful in 1s
AWOOOI Harbor 110 Local Repair / harbor-110-local-repair (push) Successful in 15s
CD Pipeline / post-deploy-checks (push) Has been skipped

This commit is contained in:
ogt
2026-07-11 19:59:18 +08:00
parent 5e286f25be
commit ea85739951
2 changed files with 191 additions and 37 deletions

View File

@@ -1069,6 +1069,9 @@ def build_unavailable_iwooos_security_asset_control_plane(
"source_status": "live_database_unavailable", "source_status": "live_database_unavailable",
"reason_code": reason_code, "reason_code": reason_code,
"summary": { "summary": {
"source_count": 9,
"ready_source_count": 0,
"unavailable_source_count": 9,
"managed_asset_count": 0, "managed_asset_count": 0,
"expected_asset_type_count": len(_ASSET_TYPES), "expected_asset_type_count": len(_ASSET_TYPES),
"present_asset_type_count": 0, "present_asset_type_count": 0,
@@ -1212,6 +1215,24 @@ def build_unavailable_iwooos_security_asset_control_plane(
"accepted_ai_trace_count_24h": 0, "accepted_ai_trace_count_24h": 0,
"immutable_external_audit_store_evidenced": False, "immutable_external_audit_store_evidenced": False,
}, },
"source_health": [
{
"source_id": source_id,
"status": "unavailable",
"reason_code": reason_code,
}
for source_id in (
"asset_discovery",
"asset_inventory",
"automation_coverage",
"asset_relationship",
"security_compliance",
"automation_operations",
"ai_decision_trace",
"siem_runtime",
"audit_runtime",
)
],
"work_items": [ "work_items": [
_work_item( _work_item(
"AIA-P0-006-01", "AIA-P0-006-01",
@@ -1250,6 +1271,44 @@ def build_unavailable_iwooos_security_asset_control_plane(
} }
async def _read_public_source(
db: Any,
*,
source_id: str,
statement: Any,
result_mode: str,
default: Any,
) -> tuple[Any, dict[str, Any]]:
"""Isolate one aggregate query so an optional source cannot abort the readback."""
try:
async with db.begin_nested():
result = await db.execute(statement)
if result_mode == "all":
value = result.fetchall()
elif result_mode == "one":
value = result.one()
elif result_mode == "one_or_none":
value = result.one_or_none()
else:
raise ValueError("unsupported_result_mode")
return value, {
"source_id": source_id,
"status": "ready",
"reason_code": None,
}
except Exception as exc: # noqa: BLE001 - isolate and redact source failures
logger.warning(
"iwooos_security_asset_control_plane_source_unavailable",
source_id=source_id,
error_type=type(exc).__name__,
)
return default, {
"source_id": source_id,
"status": "unavailable",
"reason_code": f"{source_id}_query_failed",
}
class IwoooSSecurityAssetControlPlaneService: class IwoooSSecurityAssetControlPlaneService:
"""Read aggregate production security state with a bounded DB timeout.""" """Read aggregate production security state with a bounded DB timeout."""
@@ -1274,8 +1333,12 @@ class IwoooSSecurityAssetControlPlaneService:
async def _load_snapshot(self) -> dict[str, Any]: async def _load_snapshot(self) -> dict[str, Any]:
async with get_db_context("awoooi") as db: async with get_db_context("awoooi") as db:
discovery_result = await db.execute( source_health: list[dict[str, Any]] = []
_sql(
discovery_row, health = await _read_public_source(
db,
source_id="asset_discovery",
statement=_sql(
""" """
SELECT SELECT
latest.status AS latest_status, latest.status AS latest_status,
@@ -1303,12 +1366,16 @@ class IwoooSSecurityAssetControlPlaneService:
LIMIT 1 LIMIT 1
) successful ON true ) successful ON true
""" """
) ),
result_mode="one_or_none",
default=None,
) )
discovery_row = discovery_result.one_or_none() source_health.append(health)
inventory_result = await db.execute( inventory_rows, health = await _read_public_source(
_sql( db,
source_id="asset_inventory",
statement=_sql(
""" """
SELECT SELECT
asset_type, asset_type,
@@ -1324,12 +1391,16 @@ class IwoooSSecurityAssetControlPlaneService:
GROUP BY asset_type GROUP BY asset_type
ORDER BY asset_type ORDER BY asset_type
""" """
) ),
result_mode="all",
default=[],
) )
inventory_rows = inventory_result.fetchall() source_health.append(health)
coverage_result = await db.execute( coverage_rows, health = await _read_public_source(
_sql( db,
source_id="automation_coverage",
statement=_sql(
""" """
SELECT dimension, coverage_status AS status, count(*) AS cnt SELECT dimension, coverage_status AS status, count(*) AS cnt
FROM asset_coverage_snapshot FROM asset_coverage_snapshot
@@ -1343,12 +1414,16 @@ class IwoooSSecurityAssetControlPlaneService:
GROUP BY dimension, coverage_status GROUP BY dimension, coverage_status
ORDER BY dimension, coverage_status ORDER BY dimension, coverage_status
""" """
) ),
result_mode="all",
default=[],
) )
coverage_rows = coverage_result.fetchall() source_health.append(health)
relationship_result = await db.execute( relationship_row, health = await _read_public_source(
_sql( db,
source_id="asset_relationship",
statement=_sql(
""" """
SELECT SELECT
(SELECT count(*) FROM asset_relationship WHERE is_active) AS relationship_count, (SELECT count(*) FROM asset_relationship WHERE is_active) AS relationship_count,
@@ -1363,12 +1438,16 @@ class IwoooSSecurityAssetControlPlaneService:
FROM asset_inventory asset FROM asset_inventory asset
WHERE asset.lifecycle_state = 'active' WHERE asset.lifecycle_state = 'active'
""" """
) ),
result_mode="one",
default={},
) )
relationship_row = relationship_result.one() source_health.append(health)
compliance_result = await db.execute( compliance_rows, health = await _read_public_source(
_sql( db,
source_id="security_compliance",
statement=_sql(
""" """
WITH latest AS ( WITH latest AS (
SELECT DISTINCT ON (asset_id, dimension) SELECT DISTINCT ON (asset_id, dimension)
@@ -1381,12 +1460,16 @@ class IwoooSSecurityAssetControlPlaneService:
GROUP BY dimension, status GROUP BY dimension, status
ORDER BY dimension, status ORDER BY dimension, status
""" """
) ),
result_mode="all",
default=[],
) )
compliance_rows = compliance_result.fetchall() source_health.append(health)
automation_result = await db.execute( automation_rows, health = await _read_public_source(
_sql( db,
source_id="automation_operations",
statement=_sql(
""" """
SELECT operation_type, status, count(*) AS cnt SELECT operation_type, status, count(*) AS cnt
FROM automation_operation_log FROM automation_operation_log
@@ -1394,12 +1477,16 @@ class IwoooSSecurityAssetControlPlaneService:
GROUP BY operation_type, status GROUP BY operation_type, status
ORDER BY operation_type, status ORDER BY operation_type, status
""" """
) ),
result_mode="all",
default=[],
) )
automation_rows = automation_result.fetchall() source_health.append(health)
ai_trace_result = await db.execute( ai_trace_row, health = await _read_public_source(
_sql( db,
source_id="ai_decision_trace",
statement=_sql(
""" """
SELECT SELECT
count(*) AS trace_count, count(*) AS trace_count,
@@ -1408,12 +1495,16 @@ class IwoooSSecurityAssetControlPlaneService:
JOIN automation_operation_log operation ON operation.op_id = trace.op_id JOIN automation_operation_log operation ON operation.op_id = trace.op_id
WHERE operation.created_at >= NOW() - INTERVAL '24 hours' WHERE operation.created_at >= NOW() - INTERVAL '24 hours'
""" """
) ),
result_mode="one",
default={},
) )
ai_trace_row = ai_trace_result.one() source_health.append(health)
siem_result = await db.execute( siem_row, health = await _read_public_source(
_sql( db,
source_id="siem_runtime",
statement=_sql(
""" """
SELECT SELECT
(SELECT count(*) FROM alert_rule_catalog) AS rule_total, (SELECT count(*) FROM alert_rule_catalog) AS rule_total,
@@ -1454,12 +1545,16 @@ class IwoooSSecurityAssetControlPlaneService:
AND resolved_at >= NOW() - INTERVAL '24 hours') AND resolved_at >= NOW() - INTERVAL '24 hours')
AS mttr_minutes_24h AS mttr_minutes_24h
""" """
) ),
result_mode="one",
default={},
) )
siem_row = siem_result.one() source_health.append(health)
audit_result = await db.execute( audit_row, health = await _read_public_source(
_sql( db,
source_id="audit_runtime",
statement=_sql(
""" """
SELECT SELECT
(SELECT count(*) FROM audit_logs (SELECT count(*) FROM audit_logs
@@ -1486,9 +1581,11 @@ class IwoooSSecurityAssetControlPlaneService:
AND ai_analysis IS NOT NULL) AND ai_analysis IS NOT NULL)
AS asset_change_ai_analysis_count_24h AS asset_change_ai_analysis_count_24h
""" """
) ),
result_mode="one",
default={},
) )
audit_row = audit_result.one() source_health.append(health)
return build_iwooos_security_asset_control_plane( return build_iwooos_security_asset_control_plane(
discovery_row=discovery_row, discovery_row=discovery_row,
@@ -1500,6 +1597,7 @@ class IwoooSSecurityAssetControlPlaneService:
ai_trace_row=ai_trace_row, ai_trace_row=ai_trace_row,
siem_row=siem_row, siem_row=siem_row,
audit_row=audit_row, audit_row=audit_row,
source_health=source_health,
) )

View File

@@ -16,6 +16,7 @@ from fastapi.testclient import TestClient
from src.api.v1.iwooos import router from src.api.v1.iwooos import router
from src.services.iwooos_security_asset_control_plane import ( from src.services.iwooos_security_asset_control_plane import (
IwoooSSecurityAssetControlPlaneService, IwoooSSecurityAssetControlPlaneService,
_read_public_source,
build_iwooos_security_asset_control_plane, build_iwooos_security_asset_control_plane,
build_unavailable_iwooos_security_asset_control_plane, build_unavailable_iwooos_security_asset_control_plane,
) )
@@ -25,7 +26,7 @@ def _row(**values):
return SimpleNamespace(**values) return SimpleNamespace(**values)
def _ready_payload() -> dict: def _ready_payload(source_health: list[dict] | None = None) -> dict:
now = datetime(2026, 7, 11, 4, 0, tzinfo=UTC) now = datetime(2026, 7, 11, 4, 0, tzinfo=UTC)
discovery = _row( discovery = _row(
latest_status="success", latest_status="success",
@@ -111,6 +112,7 @@ def _ready_payload() -> dict:
asset_change_count_24h=4, asset_change_count_24h=4,
asset_change_ai_analysis_count_24h=3, asset_change_ai_analysis_count_24h=3,
), ),
source_health=source_health,
generated_at=now, generated_at=now,
) )
@@ -201,6 +203,60 @@ def test_unknown_coverage_and_compliance_create_p0_work_items() -> None:
} <= work_item_ids } <= work_item_ids
def test_partial_source_failure_is_visible_without_erasing_core_inventory() -> None:
source_health = [
{"source_id": "asset_discovery", "status": "ready", "reason_code": None},
{"source_id": "asset_inventory", "status": "ready", "reason_code": None},
{
"source_id": "audit_runtime",
"status": "unavailable",
"reason_code": "audit_runtime_query_failed",
},
]
payload = _ready_payload(source_health)
assert payload["status"] == "degraded"
assert payload["source_status"] == "live_database_partial"
assert payload["summary"]["managed_asset_count"] == 24
assert payload["summary"]["ready_source_count"] == 2
assert payload["summary"]["unavailable_source_count"] == 1
assert payload["source_health"] == source_health
assert payload["work_items"][0]["work_item_id"] == "AIA-P0-006-00"
def test_read_public_source_returns_redacted_reason_code_on_query_failure() -> None:
class NestedContext:
async def __aenter__(self):
return None
async def __aexit__(self, exc_type, exc, traceback):
return False
class FailingDb:
def begin_nested(self):
return NestedContext()
async def execute(self, statement):
raise RuntimeError("private table detail")
value, health = asyncio.run(
_read_public_source(
FailingDb(),
source_id="audit_runtime",
statement="SELECT 1",
result_mode="one",
default={},
)
)
assert value == {}
assert health == {
"source_id": "audit_runtime",
"status": "unavailable",
"reason_code": "audit_runtime_query_failed",
}
def test_unavailable_payload_is_fixed_degraded_contract_without_raw_error() -> None: def test_unavailable_payload_is_fixed_degraded_contract_without_raw_error() -> None:
payload = build_unavailable_iwooos_security_asset_control_plane( payload = build_unavailable_iwooos_security_asset_control_plane(
"database_query_failed" "database_query_failed"