tracer-cloud--opensre/tests/benchmarks/cloudopsbench/performance_alert_localization.py

"""Deterministic performance-fault localization from CloudOpsBench metric alerts.

Performance cases often alert on MULTIPLE services (latency on victims, CPU
throttling on a different pod). The 2026-06-07 Anthropic pilot showed opensre
investigations cluster on the loudest CPU-throttling service while the injected
fault is ``pod_network_delay`` on the service with the extreme relative latency
spike. This module reads ``raw_data/alert.json`` (the same data ``GetAlerts``
returns) and infers rank-1 ``fault_object`` + ``root_cause`` before the
paper-format predictor runs.

Heuristics (audited against boutique/trainticket performance metadata):
  - ``pod_cpu_overload``: exactly one service shows RESOURCE_SATURATION /
    cpu_cfs throttling and no competing latency leader.
  - ``pod_network_delay``: the service whose alert evidence contains the largest
    relative latency increase (``+NNNN%`` in LATENCY_DEGRADATION lines) wins,
    even when another service shows CPU throttling.
"""

from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Any

_LATENCY_PCT_RE = re.compile(r"\+(\d+(?:\.\d+)?)%")
_CPU_THROTTLE_MARKERS = ("cpu_cfs", "Throttled")


def load_alert_json(case_dir: Path) -> dict[str, Any] | None:
    path = case_dir / "raw_data" / "alert.json"
    if not path.is_file():
        return None
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return None
    return data if isinstance(data, dict) else None


def format_metric_alerts(alert_data: dict[str, Any] | None) -> str:
    """Compact per-service anomaly lines for the paper-format predictor."""
    if not alert_data:
        return ""
    alerts = alert_data.get("alerts")
    if not isinstance(alerts, list) or not alerts:
        return ""
    lines: list[str] = []
    for entry in alerts:
        if not isinstance(entry, dict):
            continue
        name = entry.get("entity_name") or entry.get("name")
        if not isinstance(name, str) or not name.strip():
            continue
        category = entry.get("metric_category")
        cat = category if isinstance(category, str) and category.strip() else "METRIC"
        evidence = entry.get("evidence")
        if isinstance(evidence, list):
            ev_text = " | ".join(str(x) for x in evidence if x)
        elif isinstance(evidence, str):
            ev_text = evidence
        else:
            ev_text = ""
        if not ev_text.strip():
            continue
        lines.append(f"  - {name}: [{cat}] {ev_text}")
    if not lines:
        return ""
    return "Metric anomalies (from GetAlerts):\n" + "\n".join(lines)


def _service_name(entry: dict[str, Any]) -> str | None:
    name = entry.get("entity_name") or entry.get("name")
    return name.strip() if isinstance(name, str) and name.strip() else None


def _evidence_text(entry: dict[str, Any]) -> str:
    evidence = entry.get("evidence")
    if isinstance(evidence, list):
        return " | ".join(str(x) for x in evidence if x)
    if isinstance(evidence, str):
        return evidence
    return ""


def _has_cpu_throttling(evidence: str, category: str) -> bool:
    return any(marker in evidence for marker in _CPU_THROTTLE_MARKERS) or (
        "RESOURCE_SATURATION" in category and "cpu" in evidence.lower()
    )


def _max_latency_pct_increase(evidence: str) -> float:
    return max((float(m) for m in _LATENCY_PCT_RE.findall(evidence)), default=0.0)


def infer_performance_localization(
    alert_data: dict[str, Any] | None,
    *,
    namespace: str,
) -> dict[str, str] | None:
    """Infer ``fault_object`` + ``root_cause`` for a performance case from alerts.

    Returns ``None`` when the alert payload is missing or ambiguous. The
    ``namespace`` argument is only used to reject node-level entities.
    """
    if not alert_data:
        return None
    alerts = alert_data.get("alerts")
    if not isinstance(alerts, list) or not alerts:
        return None

    cpu_throttled: set[str] = set()
    latency_peak: dict[str, float] = {}

    for entry in alerts:
        if not isinstance(entry, dict):
            continue
        service = _service_name(entry)
        if service is None or service in {"master", "worker-01", "worker-02", "worker-03"}:
            continue
        category = entry.get("metric_category")
        cat = category if isinstance(category, str) else ""
        evidence = _evidence_text(entry)
        if _has_cpu_throttling(evidence, cat):
            cpu_throttled.add(service)
        peak = _max_latency_pct_increase(evidence)
        if peak > 0:
            latency_peak[service] = max(latency_peak.get(service, 0.0), peak)

    # One service with cpu_cfs: default pod_cpu_overload UNLESS another service
    # shows a much larger latency spike (pod_network_delay on the latency leader).
    if len(cpu_throttled) == 1:
        cpu_service = next(iter(cpu_throttled))
        cpu_latency = latency_peak.get(cpu_service, 0.0)
        other_latency = {
            svc: pct for svc, pct in latency_peak.items() if svc != cpu_service and pct >= 500.0
        }
        if other_latency:
            best_other = max(other_latency, key=other_latency.get)
            best_other_pct = other_latency[best_other]
            if best_other_pct > max(cpu_latency, 1.0) * 2:
                return {
                    "fault_object": f"app/{best_other}",
                    "root_cause": "pod_network_delay",
                    "rationale": (
                        f"largest relative latency spike (+{best_other_pct:.0f}%) "
                        f"on a non-CPU-throttled service"
                    ),
                }
        return {
            "fault_object": f"app/{cpu_service}",
            "root_cause": "pod_cpu_overload",
            "rationale": "cpu_cfs throttling on alerted service",
        }

    # No cpu throttling: latency leader with a large relative spike.
    if latency_peak:
        best_service = max(latency_peak, key=latency_peak.get)
        best_pct = latency_peak[best_service]
        if best_pct >= 500.0:
            return {
                "fault_object": f"app/{best_service}",
                "root_cause": "pod_network_delay",
                "rationale": (
                    f"largest relative latency spike (+{best_pct:.0f}%) among alerted services"
                ),
            }

    return None


def performance_context_for_case_dir(
    case_dir: Path, *, namespace: str
) -> tuple[str, dict[str, str] | None]:
    """Return ``(formatted_alerts, localization_hint)`` for a case directory."""
    alert_data = load_alert_json(case_dir)
    return (
        format_metric_alerts(alert_data),
        infer_performance_localization(alert_data, namespace=namespace),
    )