feat: add RCAEval RE2/RE3 as a downloadable eval corpus (#78)

pyproject.toml

@@ -51,6 +51,7 @@ dev = [
     "moto[logs]>=5.0.0",
     "ruff>=0.6.0",
     "PyYAML>=6.0",
+    "huggingface-hub>=0.24.0",
 ]
 
 [project.scripts]

scripts/eval_data.py

@@ -1,29 +1,94 @@
 #!/usr/bin/env python3
-"""Download external eval corpora into tests/eval/cases/.
+"""Download external eval corpora and convert them into harness cases.
 
-The harness ships with only the small, committed cases (the sample-incident
-canary and negative cases). Large labeled corpora are downloaded here rather
-than committed. The corpora themselves arrive in follow-up issues:
+Currently wires up **RCAEval RE2/RE3** (https://github.com/phamquiluan/RCAEval,
+MIT) — 360 labeled failure cases whose ``inject_time.txt`` is exactly the
+trigger label the harness needs. The raw data (~3.4 GB) and the converted cases
+are written to gitignored ``data/`` and are never committed.
 
-  - RCAEval RE2/RE3  (issue #78)
-  - OpenTelemetry-demo + Chaos Mesh generated incidents (issue #79)
-  - Loghub-2.0 normalization ground truth (issue #80)
+Usage:
 
-Until a downloader lands, this is a no-op that documents the contract: each
-corpus writes case directories under tests/eval/cases/ in the format described
-in tests/eval/README.md, then `make eval` picks them up automatically.
+    python scripts/eval_data.py                 # RE2 + RE3
+    python scripts/eval_data.py --suite re2     # one suite
+    python scripts/eval_data.py --limit 20      # a quick subset per suite
+
+Then score each suite separately (RE2 = resource/network faults that don't
+announce themselves; RE3 = code-level faults visible as stack traces):
+
+    raglogs eval --cases data/eval-cases/rcaeval/re2 --json eval_re2.json
+    raglogs eval --cases data/eval-cases/rcaeval/re3 --json eval_re3.json
+
+Other corpora (OTel-demo #79, Loghub-2.0 #80) will plug in here the same way.
 """
 
 from __future__ import annotations
 
+import argparse
 import sys
+from pathlib import Path
 
+RAW_DIR = Path("data/rcaeval")
+OUT_DIR = Path("data/eval-cases/rcaeval")
+REPO_ID = "phamquiluan/RCAEval"
 
-def main() -> int:
-    print(
-        "No external corpora are wired up yet — see issues #78/#79/#80.\n"
-        "The committed cases under tests/eval/cases/ already run via `make eval`."
+
+def _download(suite: str) -> Path:
+    try:
+        from huggingface_hub import snapshot_download
+    except ImportError:
+        print(
+            "huggingface_hub is required to download RCAEval.\n"
+            "Install with: pip install 'raglogs[dev]'  (or: pip install huggingface_hub)",
+            file=sys.stderr,
+        )
+        raise SystemExit(1)
+
+    print(f"Downloading RCAEval {suite} into {RAW_DIR} (this is large)...")
+    local = snapshot_download(
+        repo_id=REPO_ID,
+        repo_type="dataset",
+        allow_patterns=f"{suite}*",
+        local_dir=str(RAW_DIR),
     )
+    return Path(local)
+
+
+def _convert_suite(suite: str, limit: int | None) -> int:
+    from src.eval.rcaeval import convert_case
+
+    root = _download(suite)
+    # Case directories are named like re2ob_adservice_cpu_1; find any dir that
+    # contains the required files, regardless of nesting.
+    case_dirs = sorted(
+        p.parent for p in root.rglob("inject_time.txt") if (p.parent / "logs.csv").exists()
+    )
+    if limit is not None:
+        case_dirs = case_dirs[:limit]
+
+    converted = 0
+    for src in case_dirs:
+        out = OUT_DIR / suite / src.name
+        try:
+            if convert_case(src, out):
+                converted += 1
+        except Exception as exc:  # keep going; one bad case shouldn't stop the run
+            print(f"  skipped {src.name}: {exc}", file=sys.stderr)
+    print(f"{suite}: converted {converted}/{len(case_dirs)} cases into {OUT_DIR / suite}")
+    return converted
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Download + convert eval corpora")
+    parser.add_argument("--suite", choices=["re2", "re3", "all"], default="all")
+    parser.add_argument("--limit", type=int, default=None, help="Max cases per suite")
+    args = parser.parse_args()
+
+    suites = ["re2", "re3"] if args.suite == "all" else [args.suite]
+    total = sum(_convert_suite(s, args.limit) for s in suites)
+    if total == 0:
+        print("No cases converted.", file=sys.stderr)
+        return 1
+    print(f"\nDone. Run: raglogs eval --cases {OUT_DIR}/re2  (and .../re3)")
     return 0
 
 

src/eval/rcaeval.py

@@ -0,0 +1,238 @@
+"""Convert the RCAEval RE2/RE3 corpus into eval-harness cases.
+
+RCAEval (https://github.com/phamquiluan/RCAEval, MIT) ships one directory per
+failure case named ``{benchmark}{system}_{service}_{fault}_{instance}`` (e.g.
+``re2ob_adservice_cpu_1``), each containing:
+
+- ``inject_time.txt`` — the fault-injection Unix timestamp (the trigger label).
+- ``logs.csv`` — ``time, service, message`` rows.
+
+The case id encodes the ground-truth root-cause **service** and **fault type**,
+so we derive labels from the directory name + ``inject_time.txt`` and never need
+the (Parquet) metadata index. Conversion is pure and unit-tested; the download
+(~3.4 GB) lives in ``scripts/eval_data.py`` and is never committed.
+"""
+
+from __future__ import annotations
+
+import csv
+import io
+import json
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+# Window around the injection timestamp. RCAEval doesn't state a fixed
+# observation duration, so this is a documented starting heuristic (refine
+# empirically): a few minutes before to give a baseline, ~10 minutes after to
+# capture the fault's effects.
+DEFAULT_PRE_SECONDS = 300
+DEFAULT_POST_SECONDS = 600
+
+# RCAEval fault families -> the harness trigger taxonomy
+# (deploy|config|dependency|resource|code|none). RE2 faults are resource/network
+# stressors; RE3 faults are code-level. None of them announce themselves in the
+# log stream — that is the point of the dataset.
+_FAULT_TYPE = {
+    "cpu": "resource",
+    "mem": "resource",
+    "memory": "resource",
+    "disk": "resource",
+    "io": "resource",
+    "socket": "resource",
+    "delay": "dependency",
+    "latency": "dependency",
+    "loss": "dependency",
+    "network": "dependency",
+    "code": "code",
+}
+
+
+@dataclass
+class RCACase:
+    case_id: str
+    suite: str  # "re2" | "re3"
+    system: str  # ob | ss | tt
+    service: str
+    fault: str
+    instance: str
+
+
+def parse_case_dir_name(name: str) -> RCACase:
+    """Parse ``re2ob_adservice_cpu_1`` into its ground-truth components."""
+    parts = name.split("_")
+    if len(parts) < 4:
+        raise ValueError(f"unexpected RCAEval case name: {name!r}")
+    benchmark = parts[0]
+    suite = benchmark[:3]
+    system = benchmark[3:]
+    if suite not in ("re2", "re3"):
+        raise ValueError(f"unexpected RCAEval benchmark prefix in {name!r}")
+    instance = parts[-1]
+    fault = parts[-2]
+    service = "_".join(parts[1:-2])
+    if not service:
+        raise ValueError(f"could not parse service from {name!r}")
+    return RCACase(
+        case_id=name,
+        suite=suite,
+        system=system,
+        service=service,
+        fault=fault,
+        instance=instance,
+    )
+
+
+def trigger_type_for_fault(fault: str) -> str:
+    """Map an RCAEval fault name to the harness trigger taxonomy."""
+    key = fault.lower()
+    for token, kind in _FAULT_TYPE.items():
+        if token in key:
+            return kind
+    return "resource"
+
+
+def parse_inject_time(text: str) -> datetime:
+    """Parse ``inject_time.txt`` — a Unix timestamp (seconds, possibly float)."""
+    value = float(text.strip())
+    # Guard against millisecond timestamps.
+    if value > 1e12:
+        value /= 1000.0
+    return datetime.fromtimestamp(value, tz=timezone.utc)
+
+
+def _parse_log_time(raw: str) -> datetime | None:
+    raw = raw.strip()
+    if not raw:
+        return None
+    try:
+        value = float(raw)
+        if value > 1e12:
+            value /= 1000.0
+        return datetime.fromtimestamp(value, tz=timezone.utc)
+    except ValueError:
+        pass
+    from src.utils.time import parse_iso
+
+    try:
+        return parse_iso(raw)
+    except Exception:
+        return None
+
+
+def _infer_level(message: str) -> str:
+    m = message.lower()
+    if any(k in m for k in ("error", "exception", "traceback", "fatal", "panic")):
+        return "error"
+    if "warn" in m:
+        return "warn"
+    return "info"
+
+
+def convert_logs_csv(text: str) -> list[dict]:
+    """Convert RCAEval ``logs.csv`` text into raglogs JSONL ingest records.
+
+    Reads ``time, service, message`` by header when present, else positionally.
+    A ``level`` is inferred from the message so error lines feed clustering and
+    the baseline arm. Rows with an unparseable timestamp are skipped.
+    """
+    reader = csv.reader(io.StringIO(text))
+    rows = list(reader)
+    if not rows:
+        return []
+
+    header = [c.strip().lower() for c in rows[0]]
+    if {"time", "service", "message"} <= set(header):
+        ti, si, mi = header.index("time"), header.index("service"), header.index("message")
+        body = rows[1:]
+    else:
+        ti, si, mi = 0, 1, 2
+        body = rows
+
+    records: list[dict] = []
+    for row in body:
+        if len(row) <= max(ti, si, mi):
+            continue
+        ts = _parse_log_time(row[ti])
+        if ts is None:
+            continue
+        message = row[mi].strip()
+        records.append(
+            {
+                "timestamp": ts.isoformat(),
+                "service": row[si].strip() or None,
+                "message": message,
+                "level": _infer_level(message),
+            }
+        )
+    return records
+
+
+def window_around(
+    inject_time: datetime,
+    pre_seconds: int = DEFAULT_PRE_SECONDS,
+    post_seconds: int = DEFAULT_POST_SECONDS,
+) -> tuple[datetime, datetime]:
+    return (
+        inject_time - timedelta(seconds=pre_seconds),
+        inject_time + timedelta(seconds=post_seconds),
+    )
+
+
+def build_case_yaml(
+    meta: RCACase,
+    inject_time: datetime,
+    window: tuple[datetime, datetime],
+) -> dict:
+    """Build the harness ``case.yaml`` dict for one RCAEval case."""
+    start, end = window
+    # No `logs` key: the loader defaults to logs.jsonl in the case dir, which is
+    # exactly where convert_case writes it.
+    return {
+        "id": meta.case_id,
+        "window": {"start": start.isoformat(), "end": end.isoformat()},
+        "root_cause": {"service": meta.service},
+        "trigger": {
+            "timestamp": inject_time.isoformat(),
+            "type": trigger_type_for_fault(meta.fault),
+        },
+        "expect_explanation": True,
+        "notes": f"RCAEval {meta.suite} {meta.system} fault={meta.fault} instance={meta.instance}",
+    }
+
+
+def convert_case(
+    src_dir: Path,
+    out_dir: Path,
+    pre_seconds: int = DEFAULT_PRE_SECONDS,
+    post_seconds: int = DEFAULT_POST_SECONDS,
+) -> bool:
+    """Convert one RCAEval case directory into an eval-harness case directory.
+
+    Writes ``case.yaml`` + ``logs.jsonl`` under ``out_dir``. Returns False (and
+    writes nothing) when the source is missing required files.
+    """
+    import yaml
+
+    src_dir = Path(src_dir)
+    logs_csv = src_dir / "logs.csv"
+    inject_txt = src_dir / "inject_time.txt"
+    if not logs_csv.exists() or not inject_txt.exists():
+        return False
+
+    meta = parse_case_dir_name(src_dir.name)
+    inject_time = parse_inject_time(inject_txt.read_text())
+    records = convert_logs_csv(logs_csv.read_text())
+    if not records:
+        return False
+
+    window = window_around(inject_time, pre_seconds, post_seconds)
+    out_dir = Path(out_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    with (out_dir / "logs.jsonl").open("w") as f:
+        for rec in records:
+            f.write(json.dumps(rec) + "\n")
+    (out_dir / "case.yaml").write_text(
+        yaml.safe_dump(build_case_yaml(meta, inject_time, window), sort_keys=False)
+    )
+    return True

tests/eval/README.md

@@ -70,5 +70,28 @@ team could learn, and the harness makes it impossible to avoid.
 - `002-quiet-healthy`, `003-steady-state` — negative cases where the correct
   answer is "nothing significant happened".
 
-Large labeled corpora (RCAEval #78, OTel-demo #79, Loghub-2.0 #80) arrive via
-`make eval-data` and drop into this directory in the same format.
+## External corpora
+
+Large labeled corpora are downloaded (never committed) and converted into the
+same case format under gitignored `data/eval-cases/`.
+
+### RCAEval RE2/RE3 (`make eval-data`)
+
+[RCAEval](https://github.com/phamquiluan/RCAEval) (MIT) ships 360 labeled
+failure cases whose `inject_time.txt` is exactly the trigger label. `make
+eval-data` downloads it (~3.4 GB) and converts each case — deriving the
+root-cause service and fault type from the directory name and the trigger from
+`inject_time.txt` — into `data/eval-cases/rcaeval/{re2,re3}/`. Then:
+
+```bash
+raglogs eval --cases data/eval-cases/rcaeval/re2 --json eval_re2.json
+raglogs eval --cases data/eval-cases/rcaeval/re3 --json eval_re3.json
+```
+
+Score **RE2 and RE3 separately**: RE2 faults (CPU/memory/disk/network stress)
+don't announce themselves in the logs, so raglogs' pattern-based trigger
+detection is expected to score near-zero there — that gap is the finding, and it
+motivates the trigger-redesign work (#82). RE3 (code-level faults, visible as
+stack traces) is where raglogs should be competitive.
+
+OTel-demo (#79) and Loghub-2.0 (#80) will plug in here the same way.