314 lines
11 KiB
Python
314 lines
11 KiB
Python
"""Run bounded M14 alert lifecycle rehearsals against the configured database.
|
|
|
|
The harness creates only isolated, recognisably named LAB fixtures. It never changes an
|
|
existing node, deployment, migration, lifecycle operation, artifact, alias, or external
|
|
process. Alert and incident history remains; transient queue/migration/rollback triggers
|
|
are removed so their alerts can prove recovery.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any, TypeVar
|
|
|
|
from modelforge_api.db import engine
|
|
from modelforge_api.domain.observability import AlertAction
|
|
from modelforge_api.persistence.models import (
|
|
Accelerator,
|
|
AlertHistoryEvent,
|
|
CapabilityDeployment,
|
|
ComputeNode,
|
|
LifecycleOperation,
|
|
MigrationPlan,
|
|
OperationalAlert,
|
|
SchedulerAcceleratorState,
|
|
ServingJob,
|
|
StorageRoot,
|
|
)
|
|
from modelforge_api.services.observability import ObservabilityService
|
|
from sqlalchemy import delete, inspect, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
ModelT = TypeVar("ModelT")
|
|
CONFIRMATION = "--confirm-live-isolated-fixtures"
|
|
|
|
|
|
def clone_row(model: type[ModelT], source: ModelT, **overrides: Any) -> ModelT:
|
|
"""Clone mapped scalar columns while leaving identity/timestamps to the database."""
|
|
omitted = {"id", "created_at", "updated_at"}
|
|
values = {
|
|
column.key: getattr(source, column.key)
|
|
for column in inspect(model).columns
|
|
if column.key not in omitted
|
|
}
|
|
values.update(overrides)
|
|
return model(**values)
|
|
|
|
|
|
def history(session: Session, alert_id: uuid.UUID) -> list[dict[str, Any]]:
|
|
rows = session.scalars(
|
|
select(AlertHistoryEvent)
|
|
.where(AlertHistoryEvent.alert_id == alert_id)
|
|
.order_by(AlertHistoryEvent.occurred_at)
|
|
)
|
|
return [
|
|
{
|
|
"state": row.to_state,
|
|
"at": row.occurred_at.isoformat(),
|
|
"actor": row.actor,
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument(CONFIRMATION, action="store_true")
|
|
args = parser.parse_args()
|
|
if not args.confirm_live_isolated_fixtures:
|
|
parser.error(f"explicit {CONFIRMATION} is required")
|
|
|
|
run_id = uuid.uuid4().hex[:12]
|
|
now = datetime.now(UTC)
|
|
first_observation = now - timedelta(seconds=90)
|
|
firing_observation = now - timedelta(seconds=20)
|
|
|
|
with Session(engine) as session:
|
|
service = ObservabilityService(session, actor="m14-live-rehearsal")
|
|
service.ensure_defaults()
|
|
|
|
source_deployment = session.scalar(
|
|
select(CapabilityDeployment)
|
|
.where(CapabilityDeployment.production.is_(False))
|
|
.order_by(CapabilityDeployment.created_at)
|
|
.limit(1)
|
|
)
|
|
source_migration = session.scalar(
|
|
select(MigrationPlan).order_by(MigrationPlan.created_at).limit(1)
|
|
)
|
|
source_lifecycle = session.scalar(
|
|
select(LifecycleOperation).order_by(LifecycleOperation.started_at).limit(1)
|
|
)
|
|
if not source_deployment or not source_migration or not source_lifecycle:
|
|
raise RuntimeError(
|
|
"required pre-existing LAB rehearsal sources are unavailable"
|
|
)
|
|
|
|
node = ComputeNode(
|
|
key=f"m14-rehearsal-node-{run_id}",
|
|
hostname=f"m14-rehearsal-{run_id}.invalid",
|
|
display_name=f"M14 isolated rehearsal {run_id}",
|
|
identity_source="m14_rehearsal",
|
|
status="active",
|
|
inventory={"fixture": "M14"},
|
|
enabled=True,
|
|
production_eligible=True,
|
|
lab_eligible=True,
|
|
benchmark_eligible=False,
|
|
observation_source="local_control_plane",
|
|
liveness_state="offline",
|
|
last_heartbeat_at=first_observation,
|
|
total_ram_bytes=8 * 1024**3,
|
|
labels={"purpose": "m14_live_rehearsal", "run_id": run_id},
|
|
)
|
|
session.add(node)
|
|
session.flush()
|
|
|
|
accelerator = Accelerator(
|
|
compute_node_id=node.id,
|
|
device_index=0,
|
|
device_uuid=f"M14-REHEARSAL-{run_id}",
|
|
name="M14 isolated virtual GPU fixture",
|
|
vendor="NVIDIA",
|
|
total_vram_bytes=4 * 1024**3,
|
|
memory_total_mb=4096,
|
|
status="active",
|
|
inventory_source="m14_rehearsal",
|
|
capabilities={"fixture": True},
|
|
)
|
|
session.add(accelerator)
|
|
session.flush()
|
|
gpu_state = SchedulerAcceleratorState(
|
|
accelerator_id=accelerator.id,
|
|
pressure_state="HIGH",
|
|
pressure_changed_at=first_observation,
|
|
last_observed_at=first_observation,
|
|
)
|
|
session.add(gpu_state)
|
|
|
|
storage = StorageRoot(
|
|
compute_node_id=node.id,
|
|
name=f"m14-rehearsal-storage-{run_id}",
|
|
purpose="m14_rehearsal",
|
|
path=f"fixture://m14/{run_id}",
|
|
status="ready",
|
|
writable=False,
|
|
capacity_bytes=100 * 1024**3,
|
|
free_bytes=12 * 1024**3,
|
|
reserve_bytes=10 * 1024**3,
|
|
reserve_percent=10,
|
|
capacity_observed_at=first_observation,
|
|
validation_details={"fixture": True, "no_real_storage": True},
|
|
)
|
|
session.add(storage)
|
|
|
|
lab_deployment = clone_row(
|
|
CapabilityDeployment,
|
|
source_deployment,
|
|
compute_node_id=node.id,
|
|
accelerator_id=accelerator.id,
|
|
channel="m14_rehearsal",
|
|
status="stable",
|
|
production=False,
|
|
health_status="unavailable",
|
|
routing_weight=0,
|
|
config_fingerprint=(f"m14{run_id}" * 6)[:64],
|
|
provenance={"fixture": "M14", "run_id": run_id, "routable": False},
|
|
promoted_at=None,
|
|
draining_at=None,
|
|
deprecated_at=None,
|
|
)
|
|
session.add(lab_deployment)
|
|
session.flush()
|
|
lab_deployment_id = lab_deployment.id
|
|
|
|
queue_jobs = [
|
|
ServingJob(
|
|
capability_deployment_id=lab_deployment.id,
|
|
compute_node_id=node.id,
|
|
operation="invoke",
|
|
status="queued",
|
|
priority="background",
|
|
idempotency_key=f"m14-{run_id}-queue-{index}",
|
|
payload_reference=None,
|
|
result_summary={"fixture": True},
|
|
)
|
|
for index in range(9)
|
|
]
|
|
session.add_all(queue_jobs)
|
|
|
|
migration = clone_row(
|
|
MigrationPlan,
|
|
source_migration,
|
|
environment="LAB",
|
|
target_shadow_target=f"m14-rehearsal-shadow-{run_id}",
|
|
idempotency_key=f"m14-rehearsal-{run_id}",
|
|
state="FAILED",
|
|
plan_fingerprint=(f"a{run_id}" * 6)[:64],
|
|
approval_fingerprint=(f"b{run_id}" * 6)[:64],
|
|
failure_code="VALIDATION_FAILED",
|
|
failure_details={
|
|
"fixture": True,
|
|
"reason": "controlled M14 validation failure",
|
|
},
|
|
started_at=first_observation,
|
|
completed_at=first_observation,
|
|
)
|
|
session.add(migration)
|
|
|
|
rollback = clone_row(
|
|
LifecycleOperation,
|
|
source_lifecycle,
|
|
stage="ROLLED_BACK",
|
|
idempotency_key=f"m14-rehearsal-rollback-{run_id}",
|
|
failure_code="ROLLBACK_FAILED",
|
|
failure_details={"fixture": True, "reason": "controlled adapter refusal"},
|
|
started_at=first_observation,
|
|
finished_at=first_observation,
|
|
)
|
|
session.add(rollback)
|
|
session.commit()
|
|
|
|
service.evaluate_alerts(first_observation)
|
|
service.evaluate_alerts(firing_observation)
|
|
|
|
expected_subjects = {
|
|
"NODE_OFFLINE": str(node.id),
|
|
"GPU_PRESSURE": str(accelerator.id),
|
|
"STORAGE_LOW": str(storage.id),
|
|
"CAPABILITY_UNAVAILABLE": None,
|
|
"QUEUE_SATURATION": "scheduler-queue",
|
|
"MIGRATION_FAILURE": str(migration.id),
|
|
"ROLLBACK_FAILURE": str(rollback.id),
|
|
}
|
|
rehearsed: dict[str, OperationalAlert] = {}
|
|
for alert_type, subject_ref in expected_subjects.items():
|
|
statement = select(OperationalAlert).where(
|
|
OperationalAlert.alert_type == alert_type
|
|
)
|
|
if subject_ref is not None:
|
|
statement = statement.where(OperationalAlert.subject_ref == subject_ref)
|
|
else:
|
|
statement = statement.where(
|
|
OperationalAlert.source == "capability_deployment",
|
|
OperationalAlert.details["node_id"].as_string() == str(node.id),
|
|
)
|
|
alert = session.scalar(
|
|
statement.order_by(OperationalAlert.first_seen_at.desc())
|
|
)
|
|
if not alert or alert.state != "FIRING":
|
|
raise RuntimeError(f"{alert_type} did not reach FIRING")
|
|
rehearsed[alert_type] = alert
|
|
|
|
node_ack = service.acknowledge(
|
|
rehearsed["NODE_OFFLINE"].id,
|
|
AlertAction(
|
|
actor="m14-rehearsal-operator",
|
|
reason="Controlled isolated node rehearsal acknowledged",
|
|
),
|
|
)
|
|
if node_ack.state.value != "ACKNOWLEDGED":
|
|
raise RuntimeError("node acknowledgement was not persisted")
|
|
|
|
node.liveness_state = "online"
|
|
node.last_heartbeat_at = now
|
|
node.enabled = False
|
|
node.status = "unavailable"
|
|
accelerator.status = "missing"
|
|
gpu_state.pressure_state = "NORMAL"
|
|
gpu_state.last_observed_at = now
|
|
storage.free_bytes = 90 * 1024**3
|
|
storage.deprecated_at = now
|
|
lab_deployment.health_status = "ready_on_demand"
|
|
session.execute(
|
|
delete(ServingJob).where(ServingJob.id.in_([row.id for row in queue_jobs]))
|
|
)
|
|
session.delete(lab_deployment)
|
|
session.delete(migration)
|
|
session.delete(rollback)
|
|
session.commit()
|
|
|
|
service.evaluate_alerts(now + timedelta(seconds=1))
|
|
session.expire_all()
|
|
|
|
result: dict[str, Any] = {
|
|
"run_id": run_id,
|
|
"fixture_node_id": str(node.id),
|
|
"fixture_deployment_id": str(lab_deployment_id),
|
|
"queue_depth_trigger": len(queue_jobs),
|
|
"first_observation": first_observation.isoformat(),
|
|
"firing_observation": firing_observation.isoformat(),
|
|
"recovery_observation": (now + timedelta(seconds=1)).isoformat(),
|
|
"alerts": {},
|
|
}
|
|
for alert_type, prior in rehearsed.items():
|
|
current = session.get(OperationalAlert, prior.id)
|
|
if not current or current.state != "RESOLVED":
|
|
raise RuntimeError(f"{alert_type} did not reach RESOLVED")
|
|
result["alerts"][alert_type] = {
|
|
"id": str(current.id),
|
|
"severity": current.severity,
|
|
"state": current.state,
|
|
"occurrences": current.occurrence_count,
|
|
"history": history(session, current.id),
|
|
}
|
|
|
|
print(json.dumps(result, indent=2, sort_keys=True))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|