Emit WDL per-stage overall metrics for Manifold (facebookresearch#684)

Summary:

D108110315 teaches the benchpress perf hook to collect PMU + sysstat data in
per-WDL-sub-benchmark stage directories for WDL prod_set. The raw staged CSVs
are preserved by perfpub's recursive Manifold upload, but users still need a
single processed summary per sub-benchmark, analogous to the normal
`overall-metrics.csv` that perfpub emits for a single benchmark.

This change adds stage-aware summary generation to perfpub:

  benchmark_metrics_<run_id>/
    memcpy_benchmark/
      overall-metrics.csv              # NEW
      perf-stat.csv
      nv-perf-collector-summary.csv
      ...
    hash_hash_benchmark/
      overall-metrics.csv              # NEW
      ...
    wdl_stage_overall_metrics.csv      # NEW aggregate index
    wdl_stage_overall_metrics.json     # NEW aggregate index
    wdl_stage_perf_summary.csv         # generic numeric summary from prior patch
    wdl_stage_perf_summary.json

Implementation details:
- For each immediate stage subdir that contains CSVs, perfpub temporarily
  chdirs into that subdir and reuses the same reader functions as the top-level
  path (`read_mpstat`, `read_memstat`, `read_cpufreq_*`, `read_perfstat`,
  `read_nv_perf_collector`, `read_arm_perf_collector`, etc.). This gives each
  stage the same processed metric lines as a normal single benchmark's
  `overall-metrics.csv`.
- Adds the WDL sub-benchmark score at the top when the score exists in the
  parent benchmark metrics JSON.
- Writes per-stage `overall-metrics.csv`, plus top-level CSV/JSON indices for
  easy discovery in Manifold.
- No XDB/dashboard changes. The goal is Manifold artifact usability.
- `sample_avg_from_csv()` now tolerates partial CSV schemas by selecting only
  requested metric columns that exist and warning about missing ones. This is
  useful for stage dirs where a monitor didn't emit the full standard set of
  columns.

Differential Revision: D108195608

Author: charles-typ

perfpub/utils.py

@@ -194,6 +194,320 @@ def find_closest_timestamp_index(metric_times, target_datetime):
     return closest_idx
 
 
+STAGE_PERF_SUMMARY_CSV = "stage_perf_summary.csv"
+STAGE_PERF_SUMMARY_JSON = "stage_perf_summary.json"
+
+
+def _numeric_series(series):
+    """Coerce a pandas Series to numeric values, tolerating comma separators."""
+    if series.dtype == object:
+        series = series.astype(str).str.replace(",", "", regex=False)
+    return pd.to_numeric(series, errors="coerce").dropna()
+
+
+def _summarize_numeric_series(series):
+    values = _numeric_series(series)
+    if values.empty:
+        return None
+    return {
+        "count": int(values.count()),
+        "mean": float(values.mean()),
+        "min": float(values.min()),
+        "p50": float(values.quantile(0.50)),
+        "p95": float(values.quantile(0.95)),
+        "max": float(values.max()),
+    }
+
+
+def generate_stage_perf_summary():
+    """Generate compact summary artifacts for per-stage perf data.
+
+    The benchpress perf hook's stage-aware mode writes monitor CSVs under
+    immediate subdirectories of ``benchmark_metrics_<run_id>/``:
+
+        benchmark_metrics_<run_id>/<stage>/perf-stat.csv
+        benchmark_metrics_<run_id>/<stage>/mpstat.csv
+        benchmark_metrics_<run_id>/<stage>/nv-perf-collector-summary.csv
+        ...
+
+    PerfPub already uploads the whole directory tree to Manifold via
+    ``manifold putr``. This helper adds two easy-to-discover top-level
+    artifacts before that upload happens:
+
+        stage_perf_summary.csv
+        stage_perf_summary.json
+
+    No XDB schema change is required. The summary is intentionally generic:
+    for every immediate subdirectory, every CSV file, every numeric column,
+    compute count/mean/min/p50/p95/max.
+
+    Returns:
+        List of generated summary filenames. Empty when no per-stage CSVs
+        are found.
+    """
+    rows = []
+    json_summary = {"stages": {}}
+
+    # Do not recurse into nested timestamp dirs produced by some monitors;
+    # stage-aware perf writes stage dirs directly under cwd.
+    for stage in sorted(
+        d for d in os.listdir(".") if os.path.isdir(d) and not d.startswith(".")
+    ):
+        csv_files = sorted(
+            f
+            for f in os.listdir(stage)
+            if f.endswith(".csv")
+            and f not in {STAGE_PERF_SUMMARY_CSV, STAGE_PERF_SUMMARY_JSON}
+        )
+        if not csv_files:
+            continue
+
+        stage_json = {"files": {}}
+        for csv_file in csv_files:
+            path = os.path.join(stage, csv_file)
+            try:
+                df = pd.read_csv(path)
+            except Exception as e:
+                print(f"Warning: failed to read stage perf CSV {path}: {e}")
+                continue
+
+            file_json = {"rows": int(len(df)), "metrics": {}}
+            for col in df.columns:
+                stats = _summarize_numeric_series(df[col])
+                if stats is None:
+                    continue
+                file_json["metrics"][col] = stats
+                rows.append(
+                    {
+                        "stage": stage,
+                        "file": csv_file,
+                        "metric": col,
+                        **stats,
+                    }
+                )
+            if file_json["metrics"]:
+                stage_json["files"][csv_file] = file_json
+
+        if stage_json["files"]:
+            json_summary["stages"][stage] = stage_json
+
+    if not rows:
+        return []
+
+    with open(STAGE_PERF_SUMMARY_JSON, "w") as f:
+        json.dump(json_summary, f, indent=2, sort_keys=True)
+
+    pd.DataFrame(rows).to_csv(STAGE_PERF_SUMMARY_CSV, index=False)
+    print(
+        f"Generated per-stage perf summaries: {STAGE_PERF_SUMMARY_CSV}, "
+        f"{STAGE_PERF_SUMMARY_JSON} ({len(json_summary['stages'])} stages, "
+        f"{len(rows)} numeric metrics)"
+    )
+    return [STAGE_PERF_SUMMARY_CSV, STAGE_PERF_SUMMARY_JSON]
+
+
+STAGE_OVERALL_METRICS_CSV = "stage_overall_metrics.csv"
+STAGE_OVERALL_METRICS_JSON = "stage_overall_metrics.json"
+STAGE_OVERALL_METRICS_FILENAME = "overall-metrics.csv"
+
+
+def _build_stage_overall_metrics_text(stage, bm_metrics, interval):
+    """Build an overall-metrics.csv-style text blob for one stage directory.
+
+    This mirrors the normal top-level overall-metrics.csv generated by
+    process_metrics(), but is intentionally limited to the perf/sysstat data
+    available inside one stage-aware benchmark stage directory. The
+    caller must chdir into the stage directory before calling this helper.
+    """
+    res = ""
+    # Include the stage score at the top when the benchmark parser
+    # reported one in the top-level metrics JSON.
+    try:
+        metrics = bm_metrics.get("metrics", {}) if bm_metrics else {}
+        if stage in metrics:
+            res += f'score,"{metrics[stage]}"\n'
+    except Exception:
+        pass
+
+    # Stage directories already represent the exact sub-benchmark window, so
+    # process the whole CSV (last_secs=0, skip_last_secs=0) instead of slicing
+    # by the parent benchmark's breakdown.csv window.
+    last_secs = 0
+    skip_last_secs = 0
+    bm_epoch = None
+    start_time = None
+    end_time = None
+
+    res += read_mpstat(
+        interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch
+    )
+    res += read_memstat(
+        interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch
+    )
+    res += read_cpufreq_scaling(
+        interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch
+    )
+    res += read_cpufreq_cpuinfo(
+        interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch
+    )
+    res += read_netstat(
+        interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch
+    )
+    res += read_perfstat(
+        interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch
+    )
+    res += read_amd_perf_collector(
+        interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch
+    )
+    res += read_amd_zen4_perf_collector(
+        interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch
+    )
+    res += read_amd_zen5_perf_collector(
+        interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch
+    )
+    res += read_nv_perf_collector(
+        interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch
+    )
+    res += read_arm_perf_collector(
+        interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch
+    )
+    res += read_neoversev3_perf_collector(
+        interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch
+    )
+    res += read_intel_perfspect(
+        interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch
+    )
+    # Include collector-provided PMU summaries (cache MPKI, TopDown %, BW,
+    # latency, etc.) such as nv-perf-collector-summary.csv.
+    res += _read_stage_summary_csvs()
+    return res
+
+
+def _read_stage_summary_csvs():
+    """Read *-summary.csv files in the current stage directory and emit
+    overall-metrics-style key,value lines.
+
+    Stage-aware WDL perf collection produces collector summaries such as
+    nv-perf-collector-summary.csv. Those files contain PMU-derived metrics like
+    cache MPKI, TopDown BackendBound %, memory bandwidth, etc. The existing
+    perfpub readers primarily consume timeseries files, so without this helper
+    the per-stage overall-metrics.csv misses the most useful PMU summary rows.
+
+    For each row in each summary CSV, emit only:
+      metric,<mean>
+
+    This matches the existing perfpub overall-metrics.csv convention (one
+    representative value per metric). The richer count/min/p50/p95/max
+    distribution remains available in the top-level stage_perf_summary
+    artifacts, not in each stage's overall-metrics.csv.
+    """
+    res = ""
+    summary_files = sorted(
+        f
+        for f in os.listdir(".")
+        if f.endswith("-summary.csv") or f.endswith("_summary.csv")
+    )
+    for summary_file in summary_files:
+        try:
+            df = pd.read_csv(summary_file)
+        except Exception as e:
+            print(f"Warning: failed to read stage summary CSV {summary_file}: {e}")
+            continue
+        if "metric" not in df.columns or "mean" not in df.columns:
+            continue
+        for _, row in df.iterrows():
+            metric = str(row.get("metric", "")).strip()
+            if not metric:
+                continue
+            # Preserve the simple key for the mean because that's what users
+            # expect in overall-metrics.csv.
+            mean = row.get("mean")
+            if pd.notna(mean):
+                res += f'{metric},"{mean}"\n'
+    return res
+
+
+def _parse_overall_metrics_text(text):
+    """Parse key,value lines from an overall-metrics.csv-style blob."""
+    out = {}
+    for line in text.splitlines():
+        if not line or "," not in line:
+            continue
+        key, value = line.split(",", 1)
+        key = key.strip()
+        value = value.strip().strip('"')
+        if not key:
+            continue
+        out[key] = value
+    return out
+
+
+def generate_stage_overall_metrics(bm_metrics=None, interval=5):
+    """Generate per-stage overall-metrics artifacts for WDL prod_set.
+
+    For every immediate subdirectory with perf/sysstat CSV files, write:
+
+        <stage>/overall-metrics.csv
+
+    and also create top-level aggregate indices:
+
+        stage_overall_metrics.csv
+        stage_overall_metrics.json
+
+    This gives Manifold users the same processed PMU/sysstat summary they get
+    from a normal single-benchmark perfpub run, but scoped to each WDL
+    sub-benchmark stage.
+
+    Returns:
+        List of generated files (top-level aggregate files plus per-stage
+        overall-metrics.csv paths). Empty when no stages are found.
+    """
+    generated = []
+    summary_rows = []
+    summary_json = {"stages": {}}
+    root = os.getcwd()
+
+    for stage in sorted(
+        d for d in os.listdir(".") if os.path.isdir(d) and not d.startswith(".")
+    ):
+        csv_files = [f for f in os.listdir(stage) if f.endswith(".csv")]
+        if not csv_files:
+            continue
+
+        try:
+            os.chdir(os.path.join(root, stage))
+            text = _build_stage_overall_metrics_text(stage, bm_metrics or {}, interval)
+        finally:
+            os.chdir(root)
+
+        if not text.strip():
+            continue
+
+        stage_overall_path = os.path.join(stage, STAGE_OVERALL_METRICS_FILENAME)
+        with open(stage_overall_path, "w") as f:
+            f.write(text)
+        generated.append(stage_overall_path)
+
+        metrics = _parse_overall_metrics_text(text)
+        summary_json["stages"][stage] = metrics
+        for metric, value in metrics.items():
+            summary_rows.append({"stage": stage, "metric": metric, "value": value})
+
+    if not summary_rows:
+        return generated
+
+    with open(STAGE_OVERALL_METRICS_JSON, "w") as f:
+        json.dump(summary_json, f, indent=2, sort_keys=True)
+    pd.DataFrame(summary_rows).to_csv(STAGE_OVERALL_METRICS_CSV, index=False)
+    generated.extend([STAGE_OVERALL_METRICS_CSV, STAGE_OVERALL_METRICS_JSON])
+    print(
+        f"Generated per-stage overall metrics: {STAGE_OVERALL_METRICS_CSV}, "
+        f"{STAGE_OVERALL_METRICS_JSON} ({len(summary_json['stages'])} stages, "
+        f"{len(summary_rows)} metrics)"
+    )
+    return generated
+
+
 def read_benchmark_metrics():
     metrics_jsons = glob.glob("*_metrics_*.json")
     if len(metrics_jsons) == 0:
@@ -392,7 +706,15 @@ def sample_avg_from_csv(
         return ""
     samples.to_csv(filename.split(".", maxsplit=1)[0] + ".sampled.csv")
     if metrics:
-        samples = samples[metrics]
+        present_metrics = [metric for metric in metrics if metric in samples.columns]
+        missing_metrics = [
+            metric for metric in metrics if metric not in samples.columns
+        ]
+        if missing_metrics:
+            print(f"Columns {missing_metrics} not found in {filename}")
+        if not present_metrics:
+            return ""
+        samples = samples[present_metrics]
     if exclude_columns:
         for excl in exclude_columns:
             if excl in samples:
@@ -659,12 +981,15 @@ def process_metrics(
     if os.path.exists(breakdown_path):
         start_time, end_time = parse_breakdown_csv(breakdown_path)
 
-    columns = "("
-    # values = "("
     db_fields = {}
     bm_metrics = read_benchmark_metrics()
     if not bm_metrics:
         return ""
+    # Generate compact per-stage perf summary artifacts before internal
+    # processing uploads the whole benchmark_metrics directory to Manifold.
+    # This is a no-op for benchmarks without stage-aware perf subdirectories.
+    generate_stage_perf_summary()
+    generate_stage_overall_metrics(bm_metrics, args.interval)
     bm_name = bm_metrics["benchmark_name"]
     db_fields["benchmark_name"] = f'"{bm_name}"'
     bm_epoch = datetime.fromtimestamp(bm_metrics["timestamp"])