lessismore/bench_real.py

"""bench_real.py — measure lessismore on REAL public log corpora, not synthetic ones.

Three benchmarks, all free, all with a deterministic metric (no model calls):

  loghub     10 LogHub sample logs (HDFS, BGL, OpenSSH, ...) — token ratios on
             real system logs. Downloads ~2.8 MB from GitHub on first run.
             https://github.com/logpai/loghub (CC-BY-4.0)

  logchunks  797 Travis CI failure logs with HUMAN-labeled failure-explaining
             chunks (MSR 2020). Metric: do the labeled chunk lines survive?
                 curl -L -o LogChunks.zip \\
                   "https://zenodo.org/api/records/3632351/files/LogChunks.zip/content"
                 unzip LogChunks.zip
                 python3 bench_real.py logchunks ./LogChunks

  logdx      LogDx-CI: 35 real GitHub Actions failures with per-case
             required-signal ground truth (arXiv 2605.28876, CC-BY-4.0).
             Metric: are the critical diagnostic signals still present?
                 pip install huggingface_hub
                 hf download eyuansu71/logdx-ci --repo-type dataset --local-dir ./logdx-ci
                 python3 bench_real.py logdx ./logdx-ci

Chunk/signal matching normalizes whitespace and ANSI on both sides (the labels
themselves contain terminal color remnants), nothing else.
"""
import json
import re
import sys
import urllib.request
import xml.etree.ElementTree as ET
from pathlib import Path

from lessismore import compress, count_tokens

_CSI = re.compile(r"\x1b?\[[0-9;]*[A-Za-z]")  # ANSI with or without the ESC byte


def norm(s):
    return re.sub(r"\s+", " ", _CSI.sub("", s)).strip()


LOGHUB = ["HDFS", "BGL", "OpenStack", "Zookeeper", "Apache",
          "OpenSSH", "Spark", "Thunderbird", "Linux", "HealthApp"]


def bench_loghub(cache=Path("loghub_samples")):
    cache.mkdir(exist_ok=True)
    tot_b = tot_a = 0
    print(f"{'dataset':14} {'raw tok':>8} {'small':>8} {'ratio':>6}")
    for name in LOGHUB:
        f = cache / f"{name}_2k.log"
        if not f.exists():
            url = f"https://raw.githubusercontent.com/logpai/loghub/master/{name}/{name}_2k.log"
            f.write_bytes(urllib.request.urlopen(url, timeout=60).read())
        text = f.read_text(encoding="utf-8", errors="replace")
        b, a = count_tokens(text), count_tokens(compress(text, 2))
        tot_b += b; tot_a += a
        print(f"{name:14} {b:>8,} {a:>8,} {b/a:>5.1f}x")
    print(f"{'TOTAL':14} {tot_b:>8,} {tot_a:>8,} {tot_b/tot_a:>5.1f}x")


def bench_logchunks(root):
    root = Path(root)
    tot_b = tot_a = n = lv = la = 0
    full_v = 0
    for xf in sorted(root.glob("build-failure-reason/*/*.xml")):
        try:
            tree = ET.parse(xf)
        except ET.ParseError:
            continue
        for ex in tree.getroot().iter("Example"):
            p = root / "logs" / ex.findtext("Log", "").strip()
            chunk = ex.findtext("Chunk", "")
            if not (p.exists() and chunk.strip()):
                continue
            out = compress(p.read_text(encoding="utf-8", errors="replace"), 2)
            tot_b += count_tokens(p.read_text(encoding="utf-8", errors="replace"))
            tot_a += count_tokens(out)
            n += 1
            nout = norm(out)
            clines = [norm(l) for l in chunk.split("\n") if norm(l)]
            v = sum(1 for l in clines if l in nout)
            lv += v; la += len(clines); full_v += v == len(clines)
    print(f"cases: {n}   tokens {tot_b:,} -> {tot_a:,} ({tot_b/tot_a:.1f}x, {1-tot_a/tot_b:.0%} saved)")
    print(f"labeled chunk lines retained verbatim: {lv}/{la} ({lv/la:.1%})")
    print(f"chunks fully verbatim: {full_v}/{n} ({full_v/n:.1%})")
    print("(the rest is mostly similar-line collapse: the message survives once "
          "with a value summary — see README)")


def bench_logdx(root):
    root = Path(root)
    gts = sorted(root.glob("cases/**/ground_truth.json"))
    tot_b = tot_a = ck = ca = 0
    for gt_path in gts:
        raw_p = gt_path.parent / "raw.log"
        if not raw_p.exists():
            continue
        raw = raw_p.read_text(encoding="utf-8", errors="replace")
        out = compress(raw, 2)
        tot_b += count_tokens(raw); tot_a += count_tokens(out)
        nout = norm(out)
        for sig in json.loads(gt_path.read_text()).get("required_signals", []):
            if sig.get("importance") != "critical":
                continue
            texts = [t for t in [sig.get("value"), sig.get("file")] + sig.get("aliases", []) if t]
            ca += 1
            ck += any(norm(t) in nout for t in texts)
    print(f"cases: {len(gts)}   tokens {tot_b:,} -> {tot_a:,} ({tot_b/tot_a:.1f}x, {1-tot_a/tot_b:.0%} saved)")
    print(f"critical diagnostic signals retained: {ck}/{ca} ({ck/ca:.1%})")


def main():
    if len(sys.argv) < 2 or sys.argv[1] not in ("loghub", "logchunks", "logdx"):
        sys.exit(__doc__)
    if sys.argv[1] == "loghub":
        bench_loghub()
    elif sys.argv[1] == "logchunks":
        if len(sys.argv) < 3:
            sys.exit("usage: bench_real.py logchunks <path-to-extracted-LogChunks>")
        bench_logchunks(sys.argv[2])
    else:
        if len(sys.argv) < 3:
            sys.exit("usage: bench_real.py logdx <path-to-logdx-ci-dataset>")
        bench_logdx(sys.argv[2])


if __name__ == "__main__":
    main()