From 266f6140b681dc0479990832be5dc17cb71d145c Mon Sep 17 00:00:00 2001 From: zuub-don Date: Mon, 1 Jun 2026 17:54:43 -0700 Subject: [PATCH] fix: timestamp columns are sequences; add journal example (1.1.1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds examples/journal_anomalies.py — anomalies in the systemd journal (point/ structural/collective within a window, or _SYSTEMD_UNIT/PRIORITY distributional drift between two windows via --baseline-since). Pipes journald JSON on stdin so it sniffs as `journal`, and maps findings back to timestamp/unit/message. Building it surfaced a real roles gap: Role::Sequence required STRICT monotonicity, but journald clock columns (__REALTIME_TIMESTAMP etc.) tie/regress, so they were classed measurements and coll.cusum/point flagged their time-advancing "level shift" as noise. Fix: a `timestamp`/`ts` name token now classifies a column as Sequence (skipped by value detectors), kept narrow so response_time-style measurements are unaffected. cadence is role-agnostic, so --cadence on a "timestamp" column still works. No config_version change (classifier refinement, like 1.0.1's procid); goldens unaffected. Verified: journal example now shows real CPU_USAGE_NSEC spikes (not timestamp noise); drift mode flags unit/priority shifts. Gate: cargo-mutants 0-missed on roles.rs (39 caught); full workspace green. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 28 +++++- Cargo.lock | 10 +- Cargo.toml | 8 +- crates/ax-core/src/roles.rs | 41 ++++++++ examples/journal_anomalies.py | 180 ++++++++++++++++++++++++++++++++++ 5 files changed, 255 insertions(+), 12 deletions(-) create mode 100755 examples/journal_anomalies.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 0349dce..b90bfd2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,14 +6,35 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +## [1.1.1] - 2026-06-01 + +### Fixed + +- **Timestamp columns are now recognized as sequences and skipped by the value + detectors.** `Role::Sequence` required *strict* monotonicity, but real clock + columns (journald's `__REALTIME_TIMESTAMP`/`__MONOTONIC_TIMESTAMP`, a pcap + `timestamp`) tie or regress just often enough to fail it — so they were treated + as measurements, and `coll.cusum` flagged their "level shift" (time advancing) + and `point` their jumps. A `timestamp` / `ts` name token now classifies a + column as `sequence`, kept deliberately narrow so `response_time`-style + *measurements* (which you do want outliers on) are unaffected. Surfaced by the + new journald example. No `config_version` change — a classifier refinement, + like 1.0.1's `procid`. + ### Examples +- **`examples/journal_anomalies.py`** — find anomalies in the systemd journal: + point / structural / collective within one capture (e.g. CPU-usage spikes per + unit), or distributional drift of `_SYSTEMD_UNIT` / `PRIORITY` between two + windows (`--baseline-since`). Pipes journald JSON on stdin (so it sniffs as + `journal`, not plain JSON) and maps findings back to timestamp / unit / message. - **`examples/stock_anomalies.py`** — fetch a ticker's daily history from Yahoo Finance and find its anomalous trading days (point / multivariate / collective), or its distributional drift against another ticker (`--baseline`). A worked example of consuming the `tq1` envelope: it parses the dense JSON contract and - maps each finding's handle back to a calendar date. Outside the Cargo workspace, - so it doesn't affect the build or gates. + maps each finding's handle back to a calendar date. +- Both live outside the Cargo workspace (they shell out to the installed + binary), so they don't affect the build or gates. ## [1.1.0] - 2026-06-01 @@ -381,7 +402,8 @@ Initial release — a contract-first anomaly-detection CLI over arbitrary corpor gates on every push. - Dual-licensed under MIT OR Apache-2.0. -[Unreleased]: https://github.com/copyleftdev/anomalyx/compare/v1.1.0...HEAD +[Unreleased]: https://github.com/copyleftdev/anomalyx/compare/v1.1.1...HEAD +[1.1.1]: https://github.com/copyleftdev/anomalyx/compare/v1.1.0...v1.1.1 [1.1.0]: https://github.com/copyleftdev/anomalyx/compare/v1.0.1...v1.1.0 [1.0.1]: https://github.com/copyleftdev/anomalyx/compare/v1.0.0...v1.0.1 [1.0.0]: https://github.com/copyleftdev/anomalyx/compare/v0.9.0...v1.0.0 diff --git a/Cargo.lock b/Cargo.lock index b0b14fa..5c554ee 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -69,7 +69,7 @@ dependencies = [ [[package]] name = "anomalyx" -version = "1.1.0" +version = "1.1.1" dependencies = [ "anomalyx-core", "anomalyx-detect", @@ -80,7 +80,7 @@ dependencies = [ [[package]] name = "anomalyx-core" -version = "1.1.0" +version = "1.1.1" dependencies = [ "proptest", "serde", @@ -90,7 +90,7 @@ dependencies = [ [[package]] name = "anomalyx-detect" -version = "1.1.0" +version = "1.1.1" dependencies = [ "anomalyx-core", "proptest", @@ -101,7 +101,7 @@ dependencies = [ [[package]] name = "anomalyx-normalize" -version = "1.1.0" +version = "1.1.1" dependencies = [ "anomalyx-core", "apache-avro", @@ -129,7 +129,7 @@ dependencies = [ [[package]] name = "anomalyx-validate" -version = "1.1.0" +version = "1.1.1" dependencies = [ "anomalyx-core", "anomalyx-detect", diff --git a/Cargo.toml b/Cargo.toml index a3737d1..ff8c70f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,7 +9,7 @@ members = [ ] [workspace.package] -version = "1.1.0" +version = "1.1.1" edition = "2021" rust-version = "1.90" license = "MIT OR Apache-2.0" @@ -22,9 +22,9 @@ authors = ["copyleftdev"] # `ax-*` names were taken), but the import/extern name stays `ax_core` etc. via # the dependency key + `package` rename — so no source code changes are needed. # Keep versions in sync with workspace.package.version above. -ax-core = { path = "crates/ax-core", version = "1.1.0", package = "anomalyx-core" } -ax-normalize = { path = "crates/ax-normalize", version = "1.1.0", package = "anomalyx-normalize" } -ax-detect = { path = "crates/ax-detect", version = "1.1.0", package = "anomalyx-detect" } +ax-core = { path = "crates/ax-core", version = "1.1.1", package = "anomalyx-core" } +ax-normalize = { path = "crates/ax-normalize", version = "1.1.1", package = "anomalyx-normalize" } +ax-detect = { path = "crates/ax-detect", version = "1.1.1", package = "anomalyx-detect" } serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/crates/ax-core/src/roles.rs b/crates/ax-core/src/roles.rs index 0855495..bcaa837 100644 --- a/crates/ax-core/src/roles.rs +++ b/crates/ax-core/src/roles.rs @@ -125,6 +125,21 @@ pub fn name_is_identifier(name: &str) -> bool { .any(|t| ID_TOKENS.contains(&t.as_str())) } +/// Name tokens marking a clock/time column. A timestamp is a monotonic-ish clock +/// value, never a measurement to outlier-test — and real timestamps (journald's +/// `__REALTIME_TIMESTAMP`, a pcap `timestamp`) tie/regress just often enough to +/// fail strict-monotonic [`Role::Sequence`] detection, so we also catch them by +/// name. Kept narrow (`timestamp`/`ts`) to avoid `response_time`-style +/// measurements you *do* want outliers on. +const TIME_TOKENS: &[&str] = &["timestamp", "ts"]; + +/// Whether a column name reads as a timestamp/clock column. +pub fn name_is_timestamp(name: &str) -> bool { + name_tokens(name) + .iter() + .any(|t| TIME_TOKENS.contains(&t.as_str())) +} + /// A stable per-value key for distinct counting (NaN-safe via bit pattern). fn distinct_key(v: &Value) -> String { match v { @@ -173,6 +188,10 @@ impl Column { if name_is_identifier(&self.name) { return Role::Identifier; } + // A clock column is a sequence regardless of type or exact monotonicity. + if name_is_timestamp(&self.name) { + return Role::Sequence; + } if self.ty.is_numeric() { let xs = self.numeric(); if xs.len() >= SEQUENCE_MIN_LEN && is_strictly_monotonic(&xs) { @@ -284,6 +303,28 @@ mod tests { } } + #[test] + fn timestamp_named_columns_are_sequences() { + // journald's near-monotonic clocks tie/regress, so strict-monotonic + // detection misses them; the name catches them. Numeric or not. + for ts in [ + "__REALTIME_TIMESTAMP", + "__MONOTONIC_TIMESTAMP", + "timestamp", + "ts", + ] { + assert!(name_is_timestamp(ts), "{ts} should read as a timestamp"); + // a jittery (non-strictly-monotonic) clock column → Sequence by name + let jittery = ints(ts, &[100, 101, 101, 105, 104, 110, 130]); + assert_eq!(jittery.role(), Role::Sequence, "{ts}"); + } + // but a real measurement whose name merely contains "time" is NOT a + // timestamp — we still want outliers on it. + for m in ["response_time", "DAYS_LOST", "duration_ms", "fare"] { + assert!(!name_is_timestamp(m), "{m} must NOT read as a timestamp"); + } + } + #[test] fn constant_takes_precedence() { assert_eq!(ints("anything", &[5, 5, 5, 5]).role(), Role::Constant); diff --git a/examples/journal_anomalies.py b/examples/journal_anomalies.py new file mode 100755 index 0000000..f6397ef --- /dev/null +++ b/examples/journal_anomalies.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +""" +journal_anomalies.py — find anomalies in the systemd journal with anomalyx. + +Captures journald entries as JSON, scans them, and reports the anomalous log +entries — mapping each finding back to its timestamp / unit / message. A worked +example of the tq1 contract on *log telemetry* (Linux + systemd only). + +The journal is piped to `anomalyx` on **stdin** on purpose: the journal parser +is content-sniffed (a `.json` file would route to the plain-JSON parser). With +column roles on (the default), anomalyx skips journald's many id / counter / +timestamp fields automatically, so the findings are about content, not noise. + +Two modes: + * single window — point / structural / collective anomalies within one + capture (a rare PRIORITY level, a sparse field, a burst); + * `--baseline-since` — distributional drift of a recent window against an + earlier one: which units appeared or changed share, did the + error-level mix shift (`dist.chi2` on `_SYSTEMD_UNIT` / + `PRIORITY`). Scoped to those interpretable fields so the + free-text `MESSAGE` column doesn't drown it in noise. + +Usage: + cargo install anomalyx # or set $ANOMALYX to the binary path + python3 examples/journal_anomalies.py --lines 20000 + python3 examples/journal_anomalies.py --since "2 hours ago" --top 20 + python3 examples/journal_anomalies.py --since "1 hour ago" \ + --baseline-since "3 hours ago" --baseline-until "1 hour ago" + +Anything after the known flags passes through to `anomalyx scan`. +Requires: python3, journalctl, and the `anomalyx` binary (or `$ANOMALYX`). +Exit code mirrors anomalyx: 0 clean, 1 anomalies found, 2 error. +""" +from __future__ import annotations + +import argparse +import json +import os +import shutil +import subprocess +import sys +import tempfile + +# Drift is only interpretable on a few journald fields; the rest is id noise +# (skipped by roles anyway) or free-text MESSAGE (every line "new"). +DRIFT_COLUMNS = "PRIORITY,_SYSTEMD_UNIT" + + +def journalctl(selectors: list[str]) -> bytes: + """Capture journald entries as JSON bytes for the given selectors.""" + if shutil.which("journalctl") is None: + sys.exit("journalctl not found — this example needs Linux + systemd") + proc = subprocess.run( + ["journalctl", "--no-pager", "-o", "json", *selectors], + capture_output=True, + ) + if proc.returncode != 0: + sys.exit(f"journalctl failed: {proc.stderr.decode(errors='replace').strip()}") + if not proc.stdout.strip(): + sys.exit("journalctl returned no entries for that window") + return proc.stdout + + +def context(raw: bytes) -> list[tuple[str, str, str]]: + """Per-entry (timestamp, unit/comm, message) for mapping findings back.""" + out = [] + for line in raw.splitlines(): + if not line.strip(): + continue + try: + e = json.loads(line) + except json.JSONDecodeError: + out.append(("?", "?", "?")) + continue + us = e.get("__REALTIME_TIMESTAMP") + ts = "?" + if isinstance(us, str) and us.isdigit(): + import datetime + + ts = datetime.datetime.fromtimestamp( + int(us) / 1e6, datetime.timezone.utc + ).strftime("%Y-%m-%d %H:%M:%S") + who = e.get("_SYSTEMD_UNIT") or e.get("SYSLOG_IDENTIFIER") or e.get("_COMM") or "?" + msg = e.get("MESSAGE") or "" + if isinstance(msg, list): # journald can encode MESSAGE as a byte array + msg = "" + out.append((ts, str(who), str(msg)[:80])) + return out + + +def anomalyx_scan(stdin_bytes: bytes, extra_args: list[str]) -> dict: + exe = os.environ.get("ANOMALYX", "anomalyx") + if shutil.which(exe) is None and not os.path.exists(exe): + sys.exit(f"`{exe}` not found — run `cargo install anomalyx` or set $ANOMALYX") + proc = subprocess.run( + [exe, "scan", *extra_args], input=stdin_bytes, capture_output=True + ) + if proc.returncode == 2: + sys.exit(f"anomalyx error: {proc.stderr.decode(errors='replace').strip()}") + return json.loads(proc.stdout) + + +def describe_handle(handle: str, ctx: list[tuple[str, str, str]]) -> str: + parts = handle.split(":") + kind = parts[0] + if kind == "cell": # cell:COLUMN:row + ts, who, _ = ctx[int(parts[2])] + return f"{ts} [{who}] {parts[1]}" + if kind == "row": # row:index (multivariate) + ts, who, msg = ctx[int(parts[1])] + return f"{ts} [{who}] {msg}" + if kind == "range": # range:COLUMN:start:end + a, b = int(parts[2]), min(int(parts[3]), len(ctx) - 1) + return f"{parts[1]} {ctx[a][0]} -> {ctx[b][0]}" + if kind == "dist": # dist:COLUMN + return f"{parts[1]} (distribution)" + return handle + + +def report(env: dict, ctx: list[tuple[str, str, str]]) -> None: + dic = env["dict"] + summ = env["summary"] + print( + f"format={env['format']} rows={env['rows_scanned']} " + f"exit={env['exit']} detected={summ['total']} max_severity={summ.get('max_severity')}" + ) + roles = {c["role"] for c in env.get("roles", [])} + print(f"columns={len(env.get('roles', []))} roles_present={sorted(roles)}") + if scope := env.get("scope"): + print(f"scope: emitted {scope['emitted']} of {scope['detected']} (dropped {scope['dropped']})") + print() + for row in env["rows"]: + detector, severity = dic[row[0]], dic[row[4]] + print(f" [{severity:>8}] {detector:<15} {describe_handle(dic[row[2]], ctx)}") + print(f" {dic[row[6]]}") + if not env["rows"]: + print(" (no findings)") + + +def main() -> None: + ap = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + ap.add_argument("--since", help="journalctl --since (e.g. '2 hours ago')") + ap.add_argument("--lines", type=int, help="last N entries instead of --since") + ap.add_argument("--baseline-since", help="drift mode: baseline window start") + ap.add_argument("--baseline-until", help="drift mode: baseline window end") + args, scan_args = ap.parse_known_args() + + cur_sel = ["-n", str(args.lines)] if args.lines else ( + ["--since", args.since] if args.since else ["-n", "10000"] + ) + raw = journalctl(cur_sel) + ctx = context(raw) + + extra = list(scan_args) + if args.baseline_since: + bsel = ["--since", args.baseline_since] + if args.baseline_until: + bsel += ["--until", args.baseline_until] + braw = journalctl(bsel) + tmp = tempfile.mkdtemp(prefix="anomalyx-journal-") + # No `.json` extension → content-sniffed as `journal`, not plain JSON. + bpath = os.path.join(tmp, "baseline_journal") + with open(bpath, "wb") as f: + f.write(braw) + if "--columns" not in extra and "--exclude" not in extra: + extra = ["--columns", DRIFT_COLUMNS, *extra] + extra = ["--baseline", bpath, *extra] + print(f"# journal drift: current vs baseline since {args.baseline_since}\n") + else: + print("# journal: anomalous entries in the captured window\n") + + env = anomalyx_scan(raw, extra) + report(env, ctx) + sys.exit(0 if env["exit"] == 0 else 1) + + +if __name__ == "__main__": + main()