Tribh/devops-copilot

1from prometheus_client import Counter, Histogram, Gauge, start_http_server2import time3from functools import wraps4from typing import Callable, Any5from devops_copilot.utils.logger import logger6import os7 8# === General Agent Metrics ===9TOOL_CALL_SUCCESS = Counter("tool_call_success_total", "Total successful tool calls", ["tool_name"])10TOOL_CALL_FAILURE = Counter("tool_call_failure_total", "Total failed tool calls", ["tool_name"])11TOOL_CALL_LATENCY = Histogram("tool_call_latency_seconds", "Latency of tool calls in seconds", ["tool_name"])12AGENT_FAILURE = Counter("agent_failure_total", "Total agent failures", ["agent_name"])13 14# === DevOps Copilot Metrics ===15ANOMALY_DETECTION_TIME = Histogram(16    "devops_mttd_seconds",17    "Mean time to detection - time from anomaly start to agent detection",18    buckets=[1, 5, 10, 30, 60, 120, 300]19)20FALSE_POSITIVE_TOTAL = Counter("devops_false_positive_total", "Total false positive anomaly alerts")21ACTIVE_INCIDENTS = Gauge("devops_active_incidents", "Number of currently open incidents")22REMEDIATION_SUCCESS = Counter("devops_remediation_success_total", "Successful auto-remediations", ["service"])23REMEDIATION_FAILURE = Counter("devops_remediation_failure_total", "Failed auto-remediations", ["service"])24 25def start_metrics_server():26    port = int(os.getenv("PROMETHEUS_PORT", 8000))27    start_http_server(port)28    logger.info(f"Prometheus metrics server started on port {port}")29 30def track_tool_metrics(tool_name: str):31    """Decorator to track tool execution metrics."""32    def decorator(func: Callable):33        @wraps(func)34        def wrapper(*args, **kwargs):35            start_time = time.time()36            try:37                result = func(*args, **kwargs)38                TOOL_CALL_SUCCESS.labels(tool_name=tool_name).inc()39                return result40            except Exception as e:41                TOOL_CALL_FAILURE.labels(tool_name=tool_name).inc()42                raise e43            finally:44                latency = time.time() - start_time45                TOOL_CALL_LATENCY.labels(tool_name=tool_name).observe(latency)46        return wrapper47    return decorator48 49class RateLimiter:50    """Sliding window rate limiter for precision."""51    def __init__(self, requests_per_minute: int = 60):52        self.rpm = requests_per_minute53        self.requests = []54 55    def acquire(self):56        now = time.time()57        # Remove requests older than 60s58        self.requests = [r for r in self.requests if r > now - 60]59        60        if len(self.requests) >= self.rpm:61            wait_time = 60 - (now - self.requests[0])62            logger.warning(f"Sliding window full. Waiting {wait_time:.2f}s")63            time.sleep(wait_time)64            now = time.time() # Update now after sleep65            self.requests = [r for r in self.requests if r > now - 60]66        67        self.requests.append(now)68