Update

[ghstack-poisoned]

Author: yangw-dev

aws/lambda/benchmark_regression_summary_report/common/config_model.py

@@ -36,6 +36,9 @@ def to_timedelta(self) -> timedelta:
         else:
             raise ValueError(f"Unsupported unit: {self.unit}")
 
+    def to_timedelta_s(self) -> int:
+        return int(self.to_timedelta().total_seconds())
+
     def get_text(self):
         return f"{self.value}_{self.unit}"
 
@@ -97,12 +100,23 @@ class RangeConfig:
     def total_timedelta(self) -> timedelta:
         return timedelta(days=self.baseline.value + self.comparison.value)
 
+    def total_timedelta_s(self) -> int:
+        return int(
+            timedelta(days=self.baseline.value + self.comparison.value).total_seconds()
+        )
+
     def comparison_timedelta(self) -> timedelta:
         return timedelta(days=self.comparison.value)
 
+    def comparison_timedelta_s(self) -> int:
+        return int(self.comparison_timedelta().total_seconds())
+
     def baseline_timedelta(self) -> timedelta:
         return timedelta(days=self.baseline.value)
 
+    def baseline_timedelta_s(self) -> int:
+        return int(self.baseline_timedelta().total_seconds())
+
 
 # -------- Policy: metrics --------
 @dataclass
@@ -122,9 +136,7 @@ class RegressionPolicy:
         "greater_than", "less_than", "equal_to", "greater_equal", "less_equal"
     ]
     threshold: float
-    baseline_aggregation: Literal[
-        "avg", "max", "min", "p50", "p90", "p95", "latest", "earliest"
-    ] = "max"
+    baseline_aggregation: Literal["max", "min", "latest", "earliest"] = "max"
     rel_tol: float = 1e-3  # used only for "equal_to"
 
     def is_violation(self, value: float, baseline: float) -> bool:

aws/lambda/benchmark_regression_summary_report/common/regression_utils.py

@@ -1,6 +1,5 @@
 import logging
 import math
-import statistics
 from typing import Any, Counter, Dict, List, Literal, Optional, Tuple, TypedDict
 
 from common.benchmark_time_series_api_model import BenchmarkTimeSeriesApiData
@@ -15,8 +14,9 @@
 ]
 
 
-class BaselineItem(TypedDict):
+class BaselineResult(TypedDict):
     group_info: Dict[str, Any]
+    orignal_item: Dict[str, Any]
     value: float
 
 
@@ -27,7 +27,7 @@ class BenchmarkValueItem(TypedDict):
 
 class PerGroupResult(TypedDict, total=True):
     group_info: Dict[str, Any]
-    baseline: Optional[float]
+    baseline_item: Optional[Dict[str, Any]]
     points: List[Any]
     label: RegressionClassifyLabel
     policy: Optional["RegressionPolicy"]
@@ -87,11 +87,11 @@ def detect_regressions_with_policies(
 
             base_item = baseline_map.get(key)
             if not base_item:
-                logger.warning("Skip. No baseline item found for %s", gi)
+                logger.warning("Skip. No baseline item found for %s", key)
                 results.append(
                     PerGroupResult(
                         group_info=gi,
-                        baseline=None,
+                        baseline_item=None,
                         points=[],
                         label="insufficient_data",
                         policy=None,
@@ -104,44 +104,50 @@ def detect_regressions_with_policies(
                 results.append(
                     PerGroupResult(
                         group_info=gi,
-                        baseline=None,
+                        baseline_item=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:
+            baseline_result = self._get_baseline(base_item, baseline_aggre_mode)
+            if (
+                not baseline_result
+                or not baseline_result["orignal_item"]
+                or len(points) == 0
+            ):
                 logger.warning(
-                    "baseline_value is %s, len(points) == %s",
-                    baseline_value,
+                    "No valid baseline result found, baseline_item is %s, len(points) == %s",
+                    baseline_result,
                     len(points),
                 )
                 results.append(
                     PerGroupResult(
                         group_info=gi,
-                        baseline=None,
+                        baseline_item=None,
                         points=[],
                         label="insufficient_data",
                         policy=policy,
                     )
                 )
                 continue
 
+            orignal_baseline_obj = baseline_result["orignal_item"]
+
             # Per-point violations (True = regression)
             flags: List[bool] = [
-                policy.is_violation(p["value"], baseline_value["value"]) for p in points
+                policy.is_violation(p["value"], baseline_result["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"],
+                    baseline_item=orignal_baseline_obj,
                     points=enriched_points,
                     label=label,
                     policy=policy,
@@ -202,39 +208,33 @@ def _to_data_map(
     def _get_baseline(
         self,
         data: BenchmarkValueItem,
-        mode: str = "mean",
+        mode: str = "max",
         field: str = "value",
-    ) -> Optional[BaselineItem]:
+    ) -> Optional[BaselineResult]:
         """
         calculate the baseline value based on the mode
         mode: mean, p90, max, min, latest, p50, p95
         """
-        values = [float(d[field]) for d in data["values"] if field in d]
-        if not values:
+        items = [d for d in data["values"] if field in d]
+        if not items:
             return None
 
-        if mode == "mean":
-            val = statistics.fmean(values)
-        elif mode == "p90":
-            val = percentile(values, 0.9)
-        elif mode == "max":
-            val = max(values)
+        if mode == "max":
+            baseline_obj = max(items, key=lambda d: float(d[field]))
         elif mode == "min":
-            val = min(values)
+            baseline_obj = min(items, key=lambda d: float(d[field]))
         elif mode == "latest":
-            val = values[-1]
+            baseline_obj = items[-1]
         elif mode == "earliest":
-            val = values[0]
-        elif mode == "p50":
-            val = percentile(values, 0.5)
-        elif mode == "p95":
-            val = percentile(values, 0.95)
+            baseline_obj = items[0]
         else:
             logger.warning("Unknown mode: %s", mode)
             return None
-        result: BaselineItem = {
+
+        result: BaselineResult = {
             "group_info": data["group_info"],
-            "value": val,
+            "value": float(baseline_obj[field]),
+            "orignal_item": baseline_obj,
         }
         return result
 

aws/lambda/benchmark_regression_summary_report/common/report_manager.py

@@ -27,11 +27,6 @@
 - **Start:** `{{ time_range.start }}`
 - **End:** `{{ time_range.end }}`
 
-## Latest Benchmark Run We Used for Report
-- **Timestamp:** `{{ latest.timestamp | default('') }}`
-- **Commit:** `{{ latest.commit | default('') }}`
-- **Branch:** `{{ latest.branch | default('') }}`
-- **Workflow ID:** `{{ latest.workflow_id | default('') }}`
 
 ## Summary
 | Metric | Value |
@@ -42,14 +37,25 @@
 | No Regression | {{ summary.no_regression_count | default(0) }} |
 | Insufficient Data | {{ summary.insufficient_data_count | default(0) }} |
 
+## Latest commit
+- **Timestamp:** `{{ latest.timestamp | default('') }}`
+- **Commit:** `{{ latest.commit | default('') }}`
+- **Branch:** `{{ latest.branch | default('') }}`
+- **Workflow ID:** `{{ latest.workflow_id | default('') }}`
+
 {% if regression_items and regression_items|length > 0 %}
 ## Regression Glance
 
 {% set items = regression_items if regression_items|length <= 10 else regression_items[:10] %}
 {% for item in items %}
 - **{% for k, v in item.group_info.items() %}{{ k }}={{ v }}{% if not loop.last %}, {% endif %}{% endfor %}**
-{% endfor %}
-
+{% if item.baseline_item %}
+    baseline commit: {{ item.baseline_item.commit }}),
+    workflow_id: {{ item.baseline_item.workflow_id }}),
+    timestamp:{{item.baseline_item.granularity_bucket}}
+{% else %}
+    baseline item is missing
+{% endif %}
 {% if regression_items|length > 10 %}
 ... (showing first 10 only, total {{ regression_items|length }} regressions)
 {% endif %}
@@ -67,7 +73,6 @@ class ReportManager:
     def __init__(
         self,
         db_table_name: str,
-        config_id: str,
         config: BenchmarkConfig,
         regression_summary: Dict[str, Any],
         latest_meta_info: Dict[str, Any],
@@ -79,12 +84,12 @@ def __init__(
     ):
         self.regression_summary = regression_summary
         self.regression_result = result
-        self.config_id = config_id
+        self.config_id = config.id
         self.config = config
         self.status = self._resolve_status(regression_summary)
         self.latest_meta_info = self._validate_latest_meta_info(latest_meta_info)
         self.report_data = self._to_report_data(
-            config_id=config_id,
+            config_id=config.id,
             summary=self.regression_summary,
             report=self.regression_result,
             latest=self.latest_meta_info,
@@ -173,9 +178,16 @@ def insert_to_db(
         utc_naive = aware.astimezone(dt.timezone.utc).replace(tzinfo=None)
         last_record_ts = utc_naive.strftime("%Y-%m-%d %H:%M:%S")
 
-        report_json = json.dumps(
-            self.report_data, ensure_ascii=False, separators=(",", ":"), default=str
-        )
+        try:
+            report_json = json.dumps(
+                self.report_data, ensure_ascii=False, separators=(",", ":"), default=str
+            )
+        except Exception:
+            logger.exception(
+                "[%s] failed to serialize report data to json",
+                self.config_id,
+            )
+            raise
 
         params = {
             "id": str(self.id),
@@ -195,7 +207,6 @@ def insert_to_db(
             "repo": self.repo,
             "report_json": report_json,
         }
-
         logger.info(
             "[%s]inserting benchmark regression report(%s)", self.config_id, self.id
         )