[autorevert] Add event logging for autorevert actions (#6997)

Core changes:

* add clickHouse event logging: log detection, restart outcomes, and
secondary c
onfirmations to `misc.autorevert_events`
* define schema for `misc.autorevert_events`

Misc:
* allow `--dry-run` after the `autorevert-checker` subcommand.


# Testing

locally:
```
 python -m pytorch_auto_revert autorevert-checker inductor --hours 164 --do-restart --do-revert --dry-run
 ```
patterns were detected, events were logged

---------

Co-authored-by: Jean Schmidt <[email protected]>

Author: izaitsevfb

aws/lambda/pytorch-auto-revert/pytorch_auto_revert/__main__.py

@@ -179,7 +179,7 @@ def main(*args, **kwargs) -> None:
                 "linux-binary-manywheel",
             ],
             do_restart=True,
-            do_revert=False,
+            do_revert=True,
             hours=2,
             verbose=True,
             dry_run=opts.dry_run,

aws/lambda/pytorch-auto-revert/pytorch_auto_revert/event_logger.py

@@ -0,0 +1,103 @@
+import logging
+from dataclasses import dataclass
+from typing import Optional
+
+from .clickhouse_client_helper import CHCliFactory
+
+
+def log_autorevert_event(
+    *,
+    workflow: str,
+    action: str,
+    first_failing_sha: str,
+    previous_sha: str,
+    failure_rule: str,
+    job_name_base: str,
+    second_failing_sha: Optional[str] = None,
+    dry_run: bool = False,
+    notes: str = "",
+) -> None:
+    """Insert a single autorevert event row into ClickHouse.
+
+    Uses the misc.autorevert_events table. Best-effort: logs and continues on failure.
+    """
+    try:
+        client = CHCliFactory().client
+        columns = [
+            "workflow",
+            "action",
+            "first_failing_sha",
+            "previous_sha",
+            "second_failing_sha",
+            "failure_rule",
+            "job_name_base",
+            "dry_run",
+            "notes",
+        ]
+        data = [
+            [
+                workflow,
+                action,
+                first_failing_sha,
+                previous_sha,
+                second_failing_sha,
+                failure_rule,
+                job_name_base,
+                1 if dry_run else 0,
+                notes or "",
+            ]
+        ]
+        # Specify database explicitly since this table lives in 'misc'
+        client.insert(
+            table="autorevert_events",
+            data=data,
+            column_names=columns,
+            database="misc",
+        )
+    except Exception as e:
+        logging.warning(f"Failed to log autorevert event to ClickHouse: {e}")
+
+
+@dataclass
+class AutoRevertEvent:
+    workflow: str
+    first_failing_sha: str
+    previous_sha: str
+    second_failing_sha: Optional[str]
+    failure_rule: str
+    job_name_base: str
+    dry_run: bool = False
+    notes: str = ""
+
+    def send(self, action: str, notes: Optional[str] = None) -> None:
+        """Send this event with the given action; optional notes override."""
+        log_autorevert_event(
+            workflow=self.workflow,
+            action=action,
+            first_failing_sha=self.first_failing_sha,
+            previous_sha=self.previous_sha,
+            second_failing_sha=self.second_failing_sha,
+            failure_rule=self.failure_rule,
+            job_name_base=self.job_name_base,
+            dry_run=self.dry_run,
+            notes=(notes if notes is not None else self.notes),
+        )
+
+    def send_restart_outcome(
+        self, *, already_count: int, success_count: int, failure_count: int
+    ) -> None:
+        """Consolidated restart outcome sender with minimal branching."""
+        if self.dry_run or (
+            already_count > 0 and success_count == 0 and failure_count == 0
+        ):
+            self.send(
+                "restart_skipped",
+                notes=(
+                    "dry_run" if self.dry_run else f"already_restarted={already_count}"
+                ),
+            )
+            return
+
+        action = "restart_dispatched" if success_count > 0 else "restart_failed"
+        notes = f"success={success_count}, failures={failure_count}, already={already_count}"
+        self.send(action, notes=notes)

aws/lambda/pytorch-auto-revert/pytorch_auto_revert/testers/autorevert.py

@@ -2,6 +2,7 @@
 from collections import defaultdict
 
 from ..autorevert_checker import AutorevertPatternChecker
+from ..event_logger import AutoRevertEvent
 from ..workflow_checker import WorkflowRestartChecker
 
 
@@ -121,6 +122,19 @@ def autorevert_checker(
             previous_commit = pattern[
                 "older_commit"
             ]  # previously successful commit for the matched job
+            # Prepare base DTO for logging events for this pattern
+            evt = AutoRevertEvent(
+                workflow=workflow_name,
+                first_failing_sha=first_failing,
+                previous_sha=previous_commit,
+                second_failing_sha=(pattern["newer_commits"][0]),
+                failure_rule=pattern["failure_rule"],
+                job_name_base=pattern.get("job_name_base", ""),
+                dry_run=dry_run,
+            )
+
+            # Log the detection event
+            evt.send("detected")
             revert_result = revert_checker.is_commit_reverted(first_failing)
 
             if revert_result:
@@ -140,13 +154,17 @@ def autorevert_checker(
                 # Try to restart workflow if --do-restart flag is set and not already reverted
                 if do_restart and restart_checker:
                     # Restart the first failing (older failing) and the previous (successful) commit
+                    already_restarted = []
+                    successes = []
+                    failures = []
                     for target_commit in (first_failing, previous_commit):
                         if restart_checker.has_restarted_workflow(
                             workflow_name, target_commit
                         ):
                             print(
                                 f"  ⟳ ALREADY RESTARTED: {workflow_name} for {target_commit[:8]}"
                             )
+                            already_restarted.append(target_commit)
                             continue
                         if dry_run:
                             print(
@@ -162,15 +180,26 @@ def autorevert_checker(
                                     f"  ✓ RESTARTED: {workflow_name} for {target_commit[:8]}"
                                 )
                                 restarted_commits.append((workflow_name, target_commit))
+                                successes.append(target_commit)
                             else:
                                 print(
                                     f"  ✗ FAILED TO RESTART: {workflow_name} for {target_commit[:8]}"
                                 )
+                                failures.append(target_commit)
+
+                    # Log restart outcome once per pattern
+                    evt.send_restart_outcome(
+                        already_count=len(already_restarted),
+                        success_count=len(successes),
+                        failure_count=len(failures),
+                    )
 
                 # Secondary verification: compare first failing vs previous on restarted runs.
                 if do_revert:
                     try:
                         if checker.confirm_commit_caused_failure_on_restarted(pattern):
+                            # Log secondary confirmation
+                            evt.send("secondary_confirmed")
                             if dry_run:
                                 print(
                                     f"  ⚠ DRY RUN: Would record REVERT for {first_failing[:8]} ({workflow_name})"

clickhouse_db_schema/misc.autorevert_events/schema.sql

@@ -0,0 +1,18 @@
+CREATE TABLE misc.autorevert_events
+(
+    `ts` DateTime DEFAULT now(),
+    `repo` LowCardinality(String) DEFAULT 'pytorch/pytorch',
+    `workflow` LowCardinality(String),
+    `action` Enum8('detected' = 1, 'restart_dispatched' = 2, 'restart_skipped' = 3, 'restart_failed' = 4, 'secondary_confirmed' = 5),
+    `first_failing_sha` FixedString(40),
+    `previous_sha` FixedString(40),
+    `second_failing_sha` Nullable(FixedString(40)) DEFAULT NULL,
+    `failure_rule` LowCardinality(String),
+    `job_name_base` String,
+    `dry_run` UInt8 DEFAULT 0,
+    `notes` String DEFAULT '',
+    `version` UInt64 MATERIALIZED toUInt64(toUnixTimestamp(ts))
+)
+    ENGINE = SharedReplacingMergeTree('/clickhouse/tables/{uuid}/{shard}', '{replica}', version)
+        ORDER BY (repo, workflow, action, first_failing_sha, dry_run)
+        SETTINGS index_granularity = 8192