add github notification

ghstack-source-id: d2e6db6
Pull-Request: #7096

Author: yangw-dev

aws/lambda/benchmark_regression_summary_report/common/config.py

@@ -15,7 +15,6 @@
 # their own benchmark regression config, currenlty place
 # here for lambda
 
-
 COMPILER_BENCHMARK_CONFIG = BenchmarkConfig(
     name="Compiler Benchmark Regression",
     id="compiler_regression",
@@ -67,7 +66,7 @@
             "compression_ratio": RegressionPolicy(
                 name="compression_ratio",
                 condition="greater_equal",
-                threshold=0.9,
+                threshold=0.95,
                 baseline_aggregation="max",
             ),
         },

aws/lambda/benchmark_regression_summary_report/common/config_model.py

@@ -3,8 +3,9 @@
 import json
 from dataclasses import dataclass, field
 from datetime import timedelta
-from typing import Any, Dict, Literal, Optional
+from typing import Any, ClassVar, Dict, Literal, Optional
 
+import requests
 from jinja2 import Environment, meta, Template
 
 
@@ -154,13 +155,60 @@ def is_violation(self, value: float, baseline: float) -> bool:
 
 
 @dataclass
-class Policy:
-    frequency: Frequency
-    range: RangeConfig
-    metrics: Dict[str, RegressionPolicy]
+class BaseNotificationConfig:
+    # subclasses override this
+    type_tag: ClassVar[str] = ""
+
+    @classmethod
+    def matches(cls, d: Dict[str, Any]) -> bool:
+        return d.get("type") == cls.type_tag
+
+
+@dataclass
+class GitHubNotificationConfig(BaseNotificationConfig):
+    type_tag: ClassVar[str] = "github"
+
+    # actual fields
+    type: str = "github"
+    repo: str = ""  # e.g. "owner/repo"
+    issue_number: str = ""  # store as str for simplicity
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "GitHubNotificationConfig":
+        # support 'issue' alias
+        issue = d.get("issue_number") or d.get("issue") or ""
+        return cls(
+            type="github",
+            repo=d.get("repo", ""),
+            issue_number=str(issue),
+        )
+
+    def create_github_comment(self, body: str, github_token: str) -> Dict[str, Any]:
+        url = f"https://api.github.com/repos/{self.repo}/issues/{self.issue_number}/comments"
+        headers = {
+            "Authorization": f"token {github_token}",
+            "Accept": "application/vnd.github+json",
+            "User-Agent": "bench-reporter/1.0",
+        }
+        resp = requests.post(url, headers=headers, json={"body": body})
+        resp.raise_for_status()
+        return resp.json()
+
 
-    # TODO(elainewy): add notification config
-    notification_config: Optional[Dict[str, Any]] = None
+@dataclass
+class Policy:
+    frequency: "Frequency"
+    range: "RangeConfig"
+    metrics: Dict[str, "RegressionPolicy"]
+
+    notification_config: Optional[dict[str, Any]] = None
+
+    def get_github_notification_config(self) -> Optional[GitHubNotificationConfig]:
+        if not self.notification_config:
+            return None
+        if self.notification_config.get("type") != "github":
+            return None
+        return GitHubNotificationConfig.from_dict(self.notification_config)
 
 
 # -------- Top-level benchmark regression config --------

aws/lambda/benchmark_regression_summary_report/common/report_manager.py

@@ -71,12 +71,36 @@ def __init__(
         self.db_table_name = db_table_name
         self.id = str(uuid.uuid4())
 
-    def run(self, cc: clickhouse_connect.driver.client.Client) -> None:
+    def run(
+        self, cc: clickhouse_connect.driver.client.Client, github_token: str
+    ) -> None:
         try:
             self.insert_to_db(cc)
         except Exception as e:
             logger.error(f"failed to insert report to db, error: {e}")
             raise
+        self.notify_github_comment(github_token)
+
+    def notify_github_comment(self, github_token: str):
+        if self.status != "regression":
+            logger.info(
+                "[%s] no regression found, skip notification",
+                self.config_id,
+            )
+            return
+
+        github_notification = self.config.policy.get_github_notification_config()
+        if not github_notification:
+            logger.info(
+                "[%s] no github notification config found, skip notification",
+                self.config_id,
+            )
+            return
+        logger.info("[%s] prepareing content", self.config_id)
+        content = self._to_markdoown()
+        logger.info("[%s] create comment to github issue", self.config_id)
+        github_notification.create_github_comment(content, github_token)
+        logger.info("[%s] done. comment is sent to github", self.config_id)
 
     def _to_markdoown(self):
         md = Template(REPORT_MD_TEMPLATE, trim_blocks=True, lstrip_blocks=True).render(
@@ -108,21 +132,19 @@ def insert_to_db(
         # ---- 转 UTC，并格式成 ClickHouse 友好的 'YYYY-MM-DD HH:MM:SS' ----
         aware = dt.datetime.fromisoformat(latest_ts_str.replace("Z", "+00:00"))
         utc_naive = aware.astimezone(dt.timezone.utc).replace(tzinfo=None)
-        last_record_ts = utc_naive.strftime(
-            "%Y-%m-%d %H:%M:%S"
-        )  # 给 {DateTime64(0)} 用
+        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
         )
 
         params = {
-            "id": str(self.id),  # 列是 UUID，用 {id:UUID}
+            "id": str(self.id),
             "report_id": self.config_id,
             "type": self.type,
             "status": self.status,
             "last_record_commit": self.latest_meta_info.get("commit", ""),
-            "last_record_ts": last_record_ts,  # 已是 UTC，无时区
+            "last_record_ts": last_record_ts,
             "regression_count": int(self.regression_summary.get("regression_count", 0)),
             "insufficient_data_count": int(
                 self.regression_summary.get("insufficient_data_count", 0)
@@ -139,7 +161,7 @@ def insert_to_db(
             "[%s]inserting benchmark regression report(%s)", self.config_id, self.id
         )
 
-        # 纯 INSERT ... SELECT ... FROM system.one + NOT EXISTS 保护
+        # INSERT ... SELECT ... FROM system.one + NOT EXISTS protection
         cc.query(
             f"""
             INSERT INTO {table} (

aws/lambda/benchmark_regression_summary_report/lambda_function.py

@@ -31,7 +31,7 @@
 logger.setLevel("INFO")
 
 ENVS = {
-    "GITHUB_ACCESS_TOKEN": os.getenv("GITHUB_ACCESS_TOKEN", ""),
+    "GITHUB_TOKEN": os.getenv("GITHUB_TOKEN", ""),
     "CLICKHOUSE_ENDPOINT": os.getenv("CLICKHOUSE_ENDPOINT", ""),
     "CLICKHOUSE_PASSWORD": os.getenv("CLICKHOUSE_PASSWORD", ""),
     "CLICKHOUSE_USERNAME": os.getenv("CLICKHOUSE_USERNAME", ""),
@@ -49,6 +49,7 @@ def get_clickhouse_client(
     host: str, user: str, password: str
 ) -> clickhouse_connect.driver.client.Client:
     # for local testing only, disable SSL verification
+    logger.info("get_clickhouse_client ...")
     return clickhouse_connect.get_client(
         host=host, user=user, password=password, secure=True, verify=False
     )
@@ -115,12 +116,13 @@ def log_error(msg: str):
 
         # check if the current time is > policy's time_delta + previous record_ts from summary_table
         report_freq = config.policy.frequency
+
         should_generate = self._should_generate_report(
             cc, end_time, config_id, report_freq
         )
         if not should_generate:
             log_info(
-                f"Skip generate report for time:{end_time} with frequency {report_freq.get_text()}, no data found",
+                "Skip generate report",
             )
             return
         else:
@@ -210,7 +212,8 @@ def get_baseline(self, config: BenchmarkConfig, end_time: dt.datetime):
         )
 
         logger.info(
-            "found %s # of data, with time range %s",
+            "[%s] found %s # of data, with time range %s",
+            config.id,
             len(raw_data.time_series),
             raw_data.time_range,
         )
@@ -325,6 +328,7 @@ def _get_latest_record_ts(
                 """,
                 parameters={"config_id": config_id},
             )
+
             if not res.result_rows or res.result_rows[0][0] is None:
                 return None
             latest: dt.datetime = res.result_rows[0][
@@ -337,16 +341,42 @@ def _get_latest_record_ts(
 
         freq_delta = f.to_timedelta()
         latest_record_ts = _get_latest_record_ts(cc, config_id)
-
         # No report exists yet, generate
         if not latest_record_ts:
+            logger.info("[%s] no latest record ts from db for the config_id", config_id)
             return True
+        logger.info(
+            "[%s] found latest record ts from db %s", config_id, latest_record_ts
+        )
         end_utc = (
             end_time if end_time.tzinfo else end_time.replace(tzinfo=dt.timezone.utc)
         )
         end_utc = end_utc.astimezone(dt.timezone.utc)
-        cutoff = end_time - freq_delta
-        return latest_record_ts < cutoff
+        time_boundary = latest_record_ts + freq_delta
+        should_generate = end_time > time_boundary
+
+        if not should_generate:
+            logger.info(
+                "[%s][frequency(%s)] skip generate report. end_time(%s) must greater than"
+                "time_boundary(%s) based on latest_record_ts(%s)",
+                config_id,
+                f.get_text(),
+                end_time,
+                time_boundary,
+                latest_record_ts,
+            )
+        else:
+            logger.info(
+                "[%s][frequency(%s)] plan to generate report. end_time(%s) is greater than"
+                "time_boundary(%s) based on latest_record_ts(%s)",
+                config_id,
+                f.get_text(),
+                end_time,
+                time_boundary,
+                latest_record_ts,
+            )
+
+        return should_generate
 
 
 class WorkerPoolHandler:
@@ -416,7 +446,7 @@ def main(
        2. call WorkerPoolHandler to geneterate and write histogram data for each interval in parallel
     """
     if not github_access_token:
-        raise ValueError("Missing environment variable GITHUB_ACCESS_TOKEN")
+        raise ValueError("Missing environment variable GITHUB_TOKEN")
 
     # get time intervals.
     logger.info("[Main] start work ....")
@@ -435,7 +465,7 @@ def lambda_handler(event: Any, context: Any) -> None:
     """
     main(
         None,
-        github_access_token=ENVS["GITHUB_ACCESS_TOKEN"],
+        github_access_token=ENVS["GITHUB_TOKEN"],
     )
     return
 
@@ -467,7 +497,7 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument(
         "--github-access-token",
         type=str,
-        default=ENVS["GITHUB_ACCESS_TOKEN"],
+        default=ENVS["GITHUB_TOKEN"],
         help="the github access token to access github api",
     )
     parser.add_argument(