add regression report generator

ghstack-source-id: b022472
Pull-Request: #7094

Author: yangw-dev

aws/lambda/benchmark_regression_summary_report/common/regression_utils.py

@@ -0,0 +1,268 @@
+import logging
+import math
+from typing import Any, Dict, List, Literal, Optional, Tuple, TypedDict
+import statistics
+from dateutil.parser import isoparse
+from common.config_model import BenchmarkConfig, RegressionPolicy
+from common.benchmark_time_series_api_model import (
+    BenchmarkTimeSeriesApiData,
+    BenchmarkTimeSeriesItem,
+)
+import pprint
+
+logger = logging.getLogger()
+
+RegressionClassifyLabel = Literal[
+    "regression", "suspicious", "no_regression", "insufficient_data"
+]
+
+
+class BaselineItem(TypedDict):
+    group_info: Dict[str, Any]
+    value: float
+
+
+class BenchmarkValueItem(TypedDict):
+    group_info: Dict[str, Any]
+    values: List[Dict[str, Any]]
+
+
+class PerGroupResult(TypedDict, total=True):
+    group_info: Dict[str, Any]
+    baseline: Optional[float]
+    points: List[Any]
+    label: RegressionClassifyLabel
+    policy: Optional["RegressionPolicy"]
+
+
+def percentile(values: list[float], q: float):
+    v = sorted(values)
+    k = (len(v) - 1) * q
+    f = math.floor(k)
+    c = math.ceil(k)
+    if f == c:
+        return v[int(k)]
+    return v[f] + (v[c] - v[f]) * (k - f)
+
+
+class BenchmarkRegressionReportGenerator:
+    def __init__(
+        self,
+        config: BenchmarkConfig,
+        latest_ts: BenchmarkTimeSeriesApiData,
+        baseline_ts: BenchmarkTimeSeriesApiData,
+    ) -> None:
+        self.metric_policies = config.policy.metrics
+        self.latest_ts = self._to_data_map(latest_ts)
+        self.baseline_raw = self._to_data_map(baseline_ts)
+
+    def generate(self) -> Tuple[List[PerGroupResult], bool]:
+        return self.detect_regressions_with_policies(
+            self.baseline_raw,
+            self.latest_ts,
+            metric_policies=self.metric_policies,
+        )
+
+    def detect_regressions_with_policies(
+        self,
+        baseline_map: Dict[tuple, BenchmarkValueItem],
+        dp_map: Dict[tuple, BenchmarkValueItem],
+        *,
+        metric_policies: Dict[str, RegressionPolicy],
+        min_points: int = 2,
+    ) -> Tuple[List[PerGroupResult], bool]:
+        """
+        For each group:
+        - choose policy by group_info['metric']
+        - compute flags via policy.is_violation(value, baseline)
+        - classify with classify_flags
+        Returns a list of {group_info, baseline, values, flags, label, policy}
+        """
+        results: List[PerGroupResult] = []
+
+        is_any_regression = False
+
+        for key in sorted(dp_map.keys()):
+            cur_item = dp_map.get(key)
+            gi = cur_item["group_info"] if cur_item else {}
+            points: List[Any] = cur_item["values"] if cur_item else []
+
+            base_item = baseline_map.get(key)
+            if not base_item:
+                logger.warning("Skip. No baseline item found for %s", gi)
+                results.append(
+                    PerGroupResult(
+                        group_info=gi,
+                        baseline=None,
+                        points=[],
+                        label="insufficient_data",
+                        policy=None,
+                    )
+                )
+                continue
+            policy = self._resolve_policy(metric_policies, gi.get("metric", ""))
+            if not policy:
+                logger.warning("No policy for %s", gi)
+                results.append(
+                    PerGroupResult(
+                        group_info=gi,
+                        baseline=None,
+                        points=[],
+                        label="insufficient_data",
+                        policy=None,
+                    )
+                )
+                continue
+
+            baseline_aggre_mode = policy.baseline_aggregation
+            baseline_value = self._get_baseline(base_item,baseline_aggre_mode)
+            if baseline_value is None or len(points) == 0:
+                logger.warning("baseline_value is %s, len(points) == %s", baseline_value,len(points))
+                results.append(
+                    PerGroupResult(
+                        group_info=gi,
+                        baseline=None,
+                        points=[],
+                        label="insufficient_data",
+                        policy=policy,
+                    )
+                )
+                continue
+
+            # Per-point violations (True = regression)
+            flags: List[bool] = [
+                policy.is_violation(p["value"], baseline_value["value"]) for p in points
+            ]
+            label = self.classify_flags(flags, min_points=min_points)
+
+            enriched_points = [{**p, "flag": f} for p, f in zip(points, flags)]
+            results.append(
+                PerGroupResult(
+                    group_info=gi,
+                    baseline= baseline_value["value"],
+                    points=enriched_points,
+                    label=label,
+                    policy=policy,
+                )
+            )
+            if label == "regression":
+                is_any_regression = True
+        return results, is_any_regression
+
+    def _to_data_map(
+        self, data: "BenchmarkTimeSeriesApiData", field: str = "value"
+    ) -> Dict[tuple, BenchmarkValueItem]:
+        result: Dict[tuple, BenchmarkValueItem] = {}
+        for ts_group in data.time_series:
+            group_keys = tuple(sorted(ts_group.group_info.items()))
+            points: List[Dict[str, Any]] = []
+            for d in sorted(
+                ts_group.data, key=lambda d: isoparse(d["granularity_bucket"])
+            ):
+                if field not in d:
+                    continue
+                points.append(
+                    {
+                        "value": float(d[field]),
+                        "commit": d.get("commit"),
+                        "branch": d.get("branch"),
+                        "timestamp": isoparse(d["granularity_bucket"]),
+                    }
+                )
+            result[group_keys] = {
+                "group_info": ts_group.group_info,
+                "values": points,
+            }
+        return result
+
+    def _get_baseline(
+        self,
+        data: BenchmarkValueItem,
+        mode: str = "mean",
+        field: str = "value",
+    ) -> Optional[BaselineItem]:
+        values = [float(d[field]) for d in data["values"] if field in d]
+        if not values:
+            return None
+
+        if mode == "mean":
+            val = statistics.fmean(values)
+        elif mode == "p90":
+            val = percentile(values, 0.9)
+        elif mode == "max":
+            val = max(values)
+        elif mode == "min":
+            val = min(values)
+        elif mode == "latest":
+            val = values[-1]
+        elif mode == "earliest":
+            val = values[0]
+        elif mode == "p50":
+            val = percentile(values, 0.5)
+        elif mode == "p95":
+            val = percentile(values, 0.95)
+        else:
+            logger.warning("Unknown mode: %s", mode)
+            return None
+        result:BaselineItem =  {
+            "group_info": data["group_info"],
+            "value": val,
+        }
+        return result
+
+    def classify_flags(
+        self, flags: list[bool], min_points: int = 3
+    ) -> RegressionClassifyLabel:
+        """
+        Classify a sequence of boolean flags to detect regression.
+
+        - regression: last run has >= 2 consecutive True values
+        - suspicious: there is a run of >= 3 consecutive True values, but not at the end
+        - no_regression: all other cases
+        - insufficient_data: not enough data points (< min_points)
+
+        Special case:
+        - If min_points == 1, then just look at the last flag:
+            True  -> regression
+            False -> no_regression
+        """
+        n = len(flags)
+        if n == 0:
+            return "insufficient_data"
+
+        if min_points == 1:
+            return "regression" if flags[-1] else "no_regression"
+
+        if n < min_points:
+            return "insufficient_data"
+
+        # trailing run length
+        t = 0
+        for v in reversed(flags):
+            if v:
+                t += 1
+            else:
+                break
+        if t >= 2:
+            return "regression"
+
+        # longest run anywhere
+        longest = cur = 0
+        for v in flags:
+            cur = cur + 1 if v else 0
+            longest = max(longest, cur)
+
+        if longest >= 3:
+            return "suspicious"
+
+        return "no_regression"
+
+    def _resolve_policy(
+        self,
+        metric_policies: Dict[str, RegressionPolicy],
+        metric: str,
+    ) -> Optional[RegressionPolicy]:
+        if not metric:
+            return None
+        m = metric.lower()
+        return metric_policies.get(m)

aws/lambda/benchmark_regression_summary_report/lambda_function.py

@@ -1,12 +1,14 @@
 #!/usr/bin/env python
 import argparse
 import datetime as dt
+import json
 import logging
 import os
 import threading
 from concurrent.futures import as_completed, ThreadPoolExecutor
 from typing import Any, Optional
 
+from common.regression_utils import BenchmarkRegressionReportGenerator
 import clickhouse_connect
 import requests
 from common.benchmark_time_series_api_model import BenchmarkTimeSeriesApiResponse
@@ -138,6 +140,15 @@ def log_error(msg: str):
             )
             return
 
+        generator = BenchmarkRegressionReportGenerator(
+            config=config, latest_ts=latest, baseline_ts=baseline
+        )
+        result, regression_detected = generator.generate()
+        if self.is_dry_run:
+            print("regression_detected: ", regression_detected)
+            print(json.dumps(result, indent=2, default=str))
+        return
+
     def get_latest(self, config: BenchmarkConfig, end_time: dt.datetime):
         data_range = config.policy.range
         latest_s = end_time - data_range.comparison_timedelta()

aws/lambda/benchmark_regression_summary_report/requirements.txt

@@ -0,0 +1,5 @@
+clickhouse_connect==0.8.5
+boto3==1.35.33
+PyGithub==1.59.0
+python-dateutil==2.8.2
+PyYAML==6.0.1