Tribh/devops-copilot

1"""2config.py — Centralized threshold and tuning configuration.3All values can be overridden via environment variables or per-service overrides.4"""5import os6from typing import Dict, Optional7 8 9class ThresholdConfig:10    """11    Anomaly detection thresholds, loaded from environment with sensible defaults.12 13    Per-service overrides via env:14        THRESHOLD_<SERVICE_UPPER>_ERROR_RATE=0.2515        WINDOW_<SERVICE_UPPER>_SECONDS=12016 17    Example:18        THRESHOLD_PAYMENT_GATEWAY_ERROR_RATE=0.2019        WINDOW_PAYMENT_GATEWAY_SECONDS=12020    """21 22    # === Global defaults ======================================================23    # Error rate above which an anomaly is flagged (0.0 – 1.0)24    DEFAULT_ERROR_RATE_THRESHOLD: float = float(25        os.getenv("ANOMALY_ERROR_RATE_THRESHOLD", "0.10")26    )27    # Rolling window for error rate calculation (seconds)28    DEFAULT_WINDOW_SECONDS: int = int(29        os.getenv("ANOMALY_WINDOW_SECONDS", "300")30    )31    # Minimum total log volume before anomaly detection fires (avoids cold-start noise)32    MIN_LOG_VOLUME: int = int(33        os.getenv("ANOMALY_MIN_LOG_VOLUME", "5")34    )35 36    # === MTTD ================================================================37    # Maximum MTTD to record in Prometheus (seconds); spikes > this are clamped38    MTTD_CEILING_SECONDS: float = float(39        os.getenv("MTTD_CEILING_SECONDS", "3600")40    )41 42    # === Remediation =========================================================43    # Number of consecutive anomaly checks before auto-escalation is considered44    ESCALATION_THRESHOLD: int = int(45        os.getenv("ESCALATION_THRESHOLD_COUNT", "3")46    )47 48    @classmethod49    def error_rate_threshold(cls, service: str) -> float:50        """Return error rate threshold for a specific service, with env override."""51        key = f"THRESHOLD_{_env_key(service)}_ERROR_RATE"52        return float(os.getenv(key, str(cls.DEFAULT_ERROR_RATE_THRESHOLD)))53 54    @classmethod55    def window_seconds(cls, service: str) -> int:56        """Return the lookback window for a specific service."""57        key = f"WINDOW_{_env_key(service)}_SECONDS"58        return int(os.getenv(key, str(cls.DEFAULT_WINDOW_SECONDS)))59 60    @classmethod61    def summary(cls) -> Dict:62        """Return current effective config as a dict (for /health response)."""63        return {64            "default_error_rate_threshold": cls.DEFAULT_ERROR_RATE_THRESHOLD,65            "default_window_seconds": cls.DEFAULT_WINDOW_SECONDS,66            "min_log_volume": cls.MIN_LOG_VOLUME,67            "mttd_ceiling_seconds": cls.MTTD_CEILING_SECONDS,68            "escalation_threshold": cls.ESCALATION_THRESHOLD,69        }70 71    # === Service Mappings ====================================================72    SERVICE_URLS: Dict[str, str] = {73        "docugenie": "https://huggingface.co/spaces/Tribh/docugenie/metrics",74        "portfolio": "https://tribhangind.vercel.app/metrics",75        "incident-command": "https://incident-command-center.vercel.app/metrics",76        "ml-pipeline": "https://tribh-real-time-ml-pipeline.hf.space/metrics",77        "order-service": "https://tribh-order-processing-system.hf.space/metrics"78    }79 80    @classmethod81    def get_service_url(cls, service: str) -> Optional[str]:82        return cls.SERVICE_URLS.get(service.lower())83 84 85def _env_key(service: str) -> str:86    """Convert 'payment-gateway' → 'PAYMENT_GATEWAY' for env lookups."""87    return service.upper().replace("-", "_").replace(".", "_")88 89 90# Singleton instance used by tools91thresholds = ThresholdConfig()92