mirror of
https://github.com/blackboxprogramming/BlackRoad-Operating-System.git
synced 2026-03-17 06:57:17 -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>
57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
"""Offline canary analysis."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Dict
|
|
from datetime import datetime
|
|
|
|
import yaml
|
|
|
|
from . import ARTIFACTS, ROOT, _inc
|
|
|
|
CONFIG = ROOT / "configs" / "aiops" / "canary.yaml"
|
|
|
|
|
|
def _load_thresholds() -> Dict[str, float]:
|
|
with open(CONFIG, "r", encoding="utf-8") as fh:
|
|
data = yaml.safe_load(fh) or {}
|
|
return data.get("thresholds", {})
|
|
|
|
|
|
def analyze(
|
|
base_path: Path,
|
|
canary_path: Path,
|
|
artifacts_dir: Path = ARTIFACTS,
|
|
) -> dict:
|
|
"""Compare two metric snapshots and output diff."""
|
|
with open(base_path, "r", encoding="utf-8") as fh:
|
|
base = json.load(fh)
|
|
with open(canary_path, "r", encoding="utf-8") as fh:
|
|
canary = json.load(fh)
|
|
|
|
thr = _load_thresholds()
|
|
deltas = {}
|
|
failed = False
|
|
for key in ["latency_p50", "latency_p95", "error_rate"]:
|
|
b = base.get(key, 0)
|
|
c = canary.get(key, 0)
|
|
delta = c - b
|
|
deltas[key] = delta
|
|
if abs(delta) > thr.get(key, float("inf")):
|
|
failed = True
|
|
result = "FAIL" if failed else "PASS"
|
|
|
|
ts = datetime.utcnow().strftime("%Y%m%d%H%M%S")
|
|
out_dir = artifacts_dir / "aiops" / f"canary_{ts}"
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
with open(out_dir / "diff.json", "w", encoding="utf-8") as fh:
|
|
json.dump({"deltas": deltas, "result": result}, fh, indent=2)
|
|
with open(out_dir / "report.md", "w", encoding="utf-8") as fh:
|
|
fh.write(f"Result: {result}\n")
|
|
for k, v in deltas.items():
|
|
fh.write(f"{k}: {v}\n")
|
|
|
|
_inc("aiops_correlations") # reuse metric for simplicity
|
|
return {"deltas": deltas, "result": result}
|