mirror of
https://github.com/blackboxprogramming/BlackRoad-Operating-System.git
synced 2026-03-17 03:57:13 -05:00
9644737ba7
## Domain Architecture - Complete domain-to-service mapping for 16 verified domains - Subdomain architecture for blackroad.systems and blackroad.io - GitHub organization mapping (BlackRoad-OS repos) - Railway service-to-domain configuration - DNS configuration templates for Cloudflare ## Extracted Services ### AIops Service (services/aiops/) - Canary analysis for deployment validation - Config drift detection - Event correlation engine - Auto-remediation with runbook mapping - SLO budget management ### Analytics Service (services/analytics/) - Rule-based anomaly detection with safe expression evaluation - Cohort analysis with multi-metric aggregation - Decision engine with credit budget constraints - Narrative report generation ### Codex Governance (services/codex/) - 82+ governance principles (entries) - Codex Pantheon with 48+ agent archetypes - Manifesto defining ethical framework ## Integration Points - AIops → infra.blackroad.systems (blackroad-os-infra) - Analytics → core.blackroad.systems (blackroad-os-core) - Codex → operator.blackroad.systems (blackroad-os-operator) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
61 lines
2.1 KiB
Python
61 lines
2.1 KiB
Python
"""Config drift detection."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Dict, Tuple
|
|
|
|
from . import ARTIFACTS, _inc
|
|
|
|
BASELINE = ARTIFACTS / "aiops" / "config_baseline.json"
|
|
|
|
SEVERITY_RULES = {"version": "critical"}
|
|
|
|
|
|
def record_baseline(snapshot: Dict, artifacts_dir: Path = ARTIFACTS) -> None:
|
|
out_dir = artifacts_dir / "aiops"
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
path = out_dir / "config_baseline.json"
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
json.dump(snapshot, fh, indent=2)
|
|
|
|
|
|
def _diff(base: Dict, curr: Dict, prefix: str = "") -> Dict[str, Tuple[object, object]]:
|
|
changes: Dict[str, Tuple[object, object]] = {}
|
|
keys = set(base) | set(curr)
|
|
for k in keys:
|
|
p = f"{prefix}{k}"
|
|
if k not in base:
|
|
changes[p] = (None, curr[k])
|
|
elif k not in curr:
|
|
changes[p] = (base[k], None)
|
|
elif isinstance(base[k], dict) and isinstance(curr[k], dict):
|
|
changes.update(_diff(base[k], curr[k], p + "."))
|
|
elif base[k] != curr[k]:
|
|
changes[p] = (base[k], curr[k])
|
|
return changes
|
|
|
|
|
|
def compare(snapshot: Dict = None, artifacts_dir: Path = ARTIFACTS) -> dict:
|
|
path = artifacts_dir / "aiops" / "config_baseline.json"
|
|
if not path.exists():
|
|
raise FileNotFoundError("baseline not recorded")
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
base = json.load(fh)
|
|
curr = snapshot if snapshot is not None else base
|
|
diffs = _diff(base, curr)
|
|
items = []
|
|
for k, (b, c) in diffs.items():
|
|
items.append({"path": k, "baseline": b, "current": c, "severity": SEVERITY_RULES.get(k, "warning")})
|
|
ts = datetime.utcnow().strftime("%Y%m%d%H%M%S")
|
|
out_dir = artifacts_dir / "aiops"
|
|
with open(out_dir / f"drift_{ts}.json", "w", encoding="utf-8") as fh:
|
|
json.dump({"diff": items}, fh, indent=2)
|
|
with open(out_dir / "drift.md", "w", encoding="utf-8") as fh:
|
|
for i in items:
|
|
fh.write(f"- {i['path']}: {i['baseline']} -> {i['current']} ({i['severity']})\n")
|
|
if items:
|
|
_inc("aiops_drift_detected")
|
|
return {"diff": items}
|