From 4fc7c3b4732ac30a487bbc650b40815c8ceeb25c Mon Sep 17 00:00:00 2001 From: eitam Date: Fri, 10 Jul 2026 10:24:08 +0300 Subject: [PATCH 1/2] feat: runnable cascade demo over production-shaped logs --- .gitignore | 4 +- Makefile | 2 +- README.md | 29 ++++- docs/architecture.md | 5 +- examples/classmesh.yaml | 11 ++ examples/genprod.go | 119 ++++++++++++++++++ examples/mock.yml | 13 ++ examples/prod-rules.yml | 32 +++++ .../cli/internal/app/example_config_test.go | 114 +++++++++++++++++ shared/pkg/tokenizer/wordpiece/wordpiece.go | 8 +- 10 files changed, 324 insertions(+), 13 deletions(-) create mode 100644 examples/classmesh.yaml create mode 100644 examples/genprod.go create mode 100644 examples/mock.yml create mode 100644 examples/prod-rules.yml create mode 100644 services/cli/internal/app/example_config_test.go diff --git a/.gitignore b/.gitignore index 91692f0..e07a478 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,6 @@ temp/ bin/ coverage.out -.claude/ +prod.log +classified.jsonl +examples/review.jsonl diff --git a/Makefile b/Makefile index ba61924..56924c8 100644 --- a/Makefile +++ b/Makefile @@ -68,4 +68,4 @@ clean: ## bench: run benchmarks across workspace modules bench: - go test -bench=. -benchmem -run=^$$ ./shared/pkg/stage/rules/... ./shared/pkg/engine/... ./shared/pkg/tokenizer/wordpiece/... ./shared/pkg/sink/jsonl/... ./shared/pkg/source/textfile/... ./shared/pkg/source/jsonl/... + go test -bench=. -benchmem -run=^$$ ./shared/pkg/stage/rules/... ./shared/pkg/stage/schema/... ./shared/pkg/engine/... ./shared/pkg/classifier/... ./shared/pkg/logfields/... ./shared/pkg/tokenizer/wordpiece/... ./shared/pkg/sink/jsonl/... ./shared/pkg/source/textfile/... ./shared/pkg/source/jsonl/... diff --git a/README.md b/README.md index 3320d12..f080ad2 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ source -> [ stage 1: rules ] -> [ stage 2: model ] -> [ stage N ] -> sink └── exit early ───────┴── exit early uncertain -> review sink ``` -Everything is an interface: input sources, classification stages, and output sinks are pluggable modules. Today: text files and stdin. Tomorrow: whatever implements `Source`. +Everything is an interface: input sources, classification stages, and output sinks are pluggable modules. Today: text files, JSONL event streams, and stdin. Tomorrow: whatever implements `Source`. For working examples of each contract, see [`textfile`](shared/pkg/source/textfile) (a `Source`), [`rules`](shared/pkg/stage/rules) (a `Stage`), and [`jsonl`](shared/pkg/sink/jsonl) (a `Sink`). [`docs/architecture.md`](docs/architecture.md) explains how the pieces fit and why the core stays payload-agnostic. @@ -44,14 +44,32 @@ Each classified record is one JSON object on stdout with its category, confidence, the matched rule's reason, and (for events) the decoded fields. Records no rule matches are counted and reported on stderr. +The full multi-stage cascade runs from a fresh clone too. `genprod.go` writes +production-shaped logs (access lines, probes, app chatter, payments, auth +events, warns, errors, and a slice nothing recognizes; deterministic by seed), +and `classmesh.yaml` declares a two-tier cascade over them: rules first, a +gated model stand-in for the leftovers, review for what neither tier can +decide, health-check noise dropped by route: + +``` +go run examples/genprod.go -n 1000000 > prod.log +classmesh validate --config examples/classmesh.yaml +classmesh run --config examples/classmesh.yaml prod.log > classified.jsonl +``` + +With the default seed, the million lines classify in under two seconds: +the rules tier decides 88%, the model tier 6%, and 6% lands in +`examples/review.jsonl` (stats on stderr: +`processed=1000000 classified=940162 review=59838`). + ### Cascade config A whole multi-stage cascade can be declared in one versioned YAML file, checked with `validate` and run with `run --config`: ``` -classmesh validate --config classmesh.yaml # parse + validate only -classmesh run --config classmesh.yaml app.log # build and run it +classmesh validate --config examples/classmesh.yaml # parse + validate only +classmesh run --config examples/classmesh.yaml app.log # build and run it ``` ```yaml @@ -84,10 +102,11 @@ Measured on a single core (AMD Ryzen 7 3800X, `make bench`): |---|---|---|---| | Rules stage, first-rule hit | 46 ns | ~22M records/sec | 0 | | Rules stage, worst case (20-rule walk, regex-heavy) | ~6-7 µs | ~150k records/sec | 0 | -| Full pipeline (engine + rules + sink) | ~500 ns | ~2M records/sec | 0 | +| Full pipeline (engine + rules, discard sink) | ~500 ns | ~2M records/sec | 0 | Per-record cost depends on your ruleset: order rules by expected volume so the -hot path exits early. +hot path exits early. The pipeline row isolates engine + rules behind a discard +sink; the JSONL output path is benchmarked separately in `sink/jsonl`. The comparison that motivates the cascade: classifying 1M short log lines with a budget LLM API (~25 input + 5 output tokens each at $0.15/$0.60 per million diff --git a/docs/architecture.md b/docs/architecture.md index cc439f8..6ae1bba 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -33,8 +33,9 @@ logs, events, and structured records without a parallel type per payload. A pipeline is assembled from four interfaces, each in its own package: -- `source.Source` yields records until it is drained. Text files and stdin - implement it today; a CSV reader or a network stream could implement it next. +- `source.Source` yields records until it is drained. Text files, JSONL + streams, and stdin implement it today; a CSV reader or a network stream could + implement it next. - `stage.Stage` classifies a record or reports `ErrUnclassified`. Stages range from deterministic rule matching to in-process models to remote calls. - `sink.Sink` consumes a record together with its classification: stdout, a diff --git a/examples/classmesh.yaml b/examples/classmesh.yaml new file mode 100644 index 0000000..e583cdb --- /dev/null +++ b/examples/classmesh.yaml @@ -0,0 +1,11 @@ +# Production-log cascade demo: rules tier -> gated model stand-in -> review, +# noise dropped by route. Commands: README, Examples section. +version: 1 +input: { type: text } +stages: + - { id: rules, type: rules, path: prod-rules.yml, gate: 1.0 } + - { id: model, type: mock, path: mock.yml, gate: 0.7 } +sink: { type: jsonl, stream: stdout } +review: { type: jsonl, path: review.jsonl } +routes: + noise: { type: drop } diff --git a/examples/genprod.go b/examples/genprod.go new file mode 100644 index 0000000..7a6ba7c --- /dev/null +++ b/examples/genprod.go @@ -0,0 +1,119 @@ +//go:build ignore + +// genprod writes n production-shaped log lines to stdout (deterministic by +// seed): access logs, probes, app chatter, payments, auth events, warns, +// errors, and a slice nothing matches. go run examples/genprod.go -n 1000000 +package main + +import ( + "bufio" + "flag" + "fmt" + "math/rand" + "os" + "strings" + "time" +) + +func main() { + n := flag.Int("n", 1_000_000, "number of log lines to generate") + seed := flag.Int64("seed", 42, "PRNG seed; same seed, same output") + flag.Parse() + + w := bufio.NewWriterSize(os.Stdout, 1<<20) + r := rand.New(rand.NewSource(*seed)) + ts := time.Date(2026, 7, 10, 8, 0, 0, 0, time.UTC) + + paths := []string{"/api/v1/orders/%d", "/api/v1/users/%d", "/api/v1/search?q=q%d", "/api/v1/carts/%d/items", "/assets/app.js", "/login"} + methods := []string{"GET", "GET", "GET", "POST", "PUT"} + services := []string{"gateway", "orders", "users", "payments", "search", "carts"} + infos := []string{ + "info service=%s cache miss key=user:%d refill_ms=%d", + "info service=%s job completed name=reindex-%d duration_ms=%d", + "info service=%s connection pool size=%d idle=%d", + } + odd := []string{ + "tick %d drift %dms", + "peer gossip state=stable nodes=%d epoch=%d", + "compaction window advanced segment=%d ratio=0.%d", + } + + for i := 0; i < *n; i++ { + ts = ts.Add(time.Duration(r.Intn(4)) * time.Millisecond) + stamp := ts.Format(time.RFC3339Nano) + svc := services[r.Intn(len(services))] + draw := r.Float64() + switch { + case draw < 0.28: // http access, 2xx + fmt.Fprintf(w, "%s info service=%s http request method=%s path=%s status=%d latency_ms=%d ip=10.%d.%d.%d\n", + stamp, svc, methods[r.Intn(len(methods))], pick(r, paths), 200+[]int{0, 0, 0, 1, 4}[r.Intn(5)], 1+r.Intn(180), r.Intn(256), r.Intn(256), 1+r.Intn(254)) + case draw < 0.50: // probes: pure noise + if r.Intn(3) == 0 { + fmt.Fprintf(w, "GET /readiness 200\n") + } else { + fmt.Fprintf(w, "GET /healthz 200\n") + } + case draw < 0.62: // app info chatter + f := infos[r.Intn(len(infos))] + fmt.Fprintf(w, "%s "+f+"\n", stamp, svc, r.Intn(100000), r.Intn(900)) + case draw < 0.70: // payments + verb := []string{"succeeded", "succeeded", "declined", "failed"}[r.Intn(4)] + fmt.Fprintf(w, "%s info service=payments payment %s for order %d amount=%d.%02d currency=USD\n", + stamp, verb, r.Intn(200000), 1+r.Intn(400), r.Intn(100)) + case draw < 0.76: // auth signals + switch r.Intn(3) { + case 0: + fmt.Fprintf(w, "%s blocked: unauthorized token for user %d\n", stamp, r.Intn(9000)) + case 1: + fmt.Fprintf(w, "%s login failed user=u%d attempts=%d\n", stamp, r.Intn(9000), 1+r.Intn(5)) + default: + fmt.Fprintf(w, "%s rate limited ip=10.%d.%d.%d window=60s\n", stamp, r.Intn(256), r.Intn(256), 1+r.Intn(254)) + } + case draw < 0.82: // http client/server errors + fmt.Fprintf(w, "%s info service=%s http request method=%s path=%s status=%d latency_ms=%d ip=10.%d.%d.%d\n", + stamp, svc, methods[r.Intn(len(methods))], pick(r, paths), []int{404, 404, 429, 500, 503}[r.Intn(5)], 1+r.Intn(900), r.Intn(256), r.Intn(256), 1+r.Intn(254)) + case draw < 0.88: // warns: the model tier's job + switch r.Intn(3) { + case 0: + fmt.Fprintf(w, "%s warn service=%s slow query duration_ms=%d table=orders_%d\n", stamp, svc, 800+r.Intn(2400), 1+r.Intn(40)) + case 1: + fmt.Fprintf(w, "%s warn service=%s queue depth=%d threshold=500\n", stamp, svc, 500+r.Intn(2000)) + default: + fmt.Fprintf(w, "%s warn service=%s retrying request attempt=%d upstream=inventory\n", stamp, svc, 2+r.Intn(4)) + } + case draw < 0.94: // severe errors + switch r.Intn(4) { + case 0: + fmt.Fprintf(w, "%s error service=%s upstream timeout after %dms trace=%08x\n", stamp, svc, 1000+r.Intn(4000), r.Uint32()) + case 1: + fmt.Fprintf(w, "%s error service=%s request failed status=503 trace=%08x attempt=%d\n", stamp, svc, r.Uint32(), 1+r.Intn(3)) + case 2: + fmt.Fprintf(w, "%s error service=%s panic recovered in handler trace=%08x goroutine=%d\n", stamp, svc, r.Uint32(), r.Intn(90000)) + default: + fmt.Fprintf(w, "%s OOMKilled container=%s-%d restart_count=%d\n", stamp, svc, r.Intn(40), 1+r.Intn(6)) + } + case draw < 0.97: // business events: matched low-confidence by the model tier + if r.Intn(2) == 0 { + fmt.Fprintf(w, "%s shipment %d dispatched from tlv warehouse=%c\n", stamp, r.Intn(999999), 'A'+rune(r.Intn(4))) + } else { + fmt.Fprintf(w, "%s inventory sync started region=eu-%d items=%d\n", stamp, 1+r.Intn(3), r.Intn(50000)) + } + default: // chatter no tier recognizes: exercises the review path + f := odd[r.Intn(len(odd))] + fmt.Fprintf(w, "%s "+f+"\n", stamp, r.Intn(100000), 1+r.Intn(500)) + } + } + if err := w.Flush(); err != nil { + fmt.Fprintf(os.Stderr, "genprod: %v\n", err) + os.Exit(1) + } +} + +// pick returns a path template with its id filled in when it has one. +func pick(r *rand.Rand, paths []string) string { + p := paths[r.Intn(len(paths))] + if strings.Contains(p, "%d") { + return fmt.Sprintf(p, r.Intn(90000)) + } + return p +} diff --git a/examples/mock.yml b/examples/mock.yml new file mode 100644 index 0000000..88133d4 --- /dev/null +++ b/examples/mock.yml @@ -0,0 +1,13 @@ +# Model-tier stand-in: sub-1.0 confidences exercise the 0.7 gate in classmesh.yaml. +matchers: + - contains: ["warn "] + category: anomaly + confidence: 0.85 + + - contains: ["shipment ", "inventory "] + category: logistics + confidence: 0.6 + +default: + category: unknown + confidence: 0.4 diff --git a/examples/prod-rules.yml b/examples/prod-rules.yml new file mode 100644 index 0000000..7a60f03 --- /dev/null +++ b/examples/prod-rules.yml @@ -0,0 +1,32 @@ +# Ruleset for the genprod demo. Ordered by volume; first match wins. +rules: + - id: health-noise + category: noise + contains: ["/healthz", "/readiness"] + + - id: access-ok + category: traffic + contains: ["status=200", "status=201", "status=204"] + + - id: app-ops + category: ops + contains: ["cache miss", "job completed", "connection pool"] + + - id: payment-event + category: billing + regex: ["payment (succeeded|declined|failed)"] + + - id: auth-signal + category: auth + any: + - contains: "unauthorized" + - contains: "login failed" + - contains: "rate limited" + + - id: alert + category: alert + any: + - regex: "status=[45][0-9][0-9]" + - contains: "error " + - contains: "panic" + - contains: "OOMKilled" diff --git a/services/cli/internal/app/example_config_test.go b/services/cli/internal/app/example_config_test.go new file mode 100644 index 0000000..4495067 --- /dev/null +++ b/services/cli/internal/app/example_config_test.go @@ -0,0 +1,114 @@ +package app + +import ( + "bytes" + "context" + "os" + "path/filepath" + "strings" + "testing" +) + +// The committed cascade example must stay runnable from a fresh clone. +func TestExampleCascadeConfig(t *testing.T) { + t.Run("validate", func(t *testing.T) { + var out, errOut bytes.Buffer + err := Run(context.Background(), []string{"validate", "--config", examplePath(t, "classmesh.yaml")}, + Streams{In: strings.NewReader(""), Out: &out, Err: &errOut}) + if err != nil { + t.Fatalf("Run() error = %v, stderr=%s", err, errOut.String()) + } + if !strings.Contains(out.String(), "structurally valid") { + t.Fatalf("stdout = %q, want a validity confirmation", out.String()) + } + }) + + t.Run("run", func(t *testing.T) { + dir := t.TempDir() + for _, name := range []string{"classmesh.yaml", "prod-rules.yml", "mock.yml"} { + data, err := os.ReadFile(examplePath(t, name)) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, name), data, 0o644); err != nil { + t.Fatal(err) + } + } + + tests := []struct { + line string + category string // expected category for classified records + stage string // expected deciding stage + drop bool // routed to drop: classified but absent from stdout + review bool // neither tier decided: lands in review.jsonl + }{ + {line: "GET /healthz 200", category: "noise", drop: true}, + {line: "2026-07-10T08:00:00Z info service=gateway http request method=GET path=/api/v1/orders/1 status=200 latency_ms=5 ip=10.0.0.1", category: "traffic", stage: "rules"}, + {line: "2026-07-10T08:00:00Z info service=payments payment declined for order 7 amount=10.00 currency=USD", category: "billing", stage: "rules"}, + {line: "2026-07-10T08:00:01Z blocked: unauthorized token for user 3", category: "auth", stage: "rules"}, + {line: "2026-07-10T08:00:02Z error service=api upstream timeout after 3000ms trace=00000001", category: "alert", stage: "rules"}, + {line: "2026-07-10T08:00:03Z warn service=db slow query duration_ms=1900 table=orders_7", category: "anomaly", stage: "model"}, + {line: "2026-07-10T08:00:04Z shipment 84712 dispatched from tlv warehouse=A", review: true}, + {line: "tick 8912 drift 3ms", review: true}, + } + + var input strings.Builder + for _, tc := range tests { + input.WriteString(tc.line) + input.WriteByte('\n') + } + inPath := filepath.Join(dir, "input.log") + if err := os.WriteFile(inPath, []byte(input.String()), 0o644); err != nil { + t.Fatal(err) + } + + var out, errOut bytes.Buffer + err := Run(context.Background(), []string{"run", "--config", filepath.Join(dir, "classmesh.yaml"), inPath}, + Streams{In: strings.NewReader(""), Out: &out, Err: &errOut}) + if err != nil { + t.Fatalf("Run() error = %v, stderr=%s", err, errOut.String()) + } + if !strings.Contains(errOut.String(), "processed=8 classified=6 review=2") { + t.Fatalf("stats = %q, want processed=8 classified=6 review=2", errOut.String()) + } + + reviewData, err := os.ReadFile(filepath.Join(dir, "review.jsonl")) + if err != nil { + t.Fatalf("review sink: %v", err) + } + stdout := out.String() + for _, tc := range tests { + switch { + case tc.drop: + if strings.Contains(stdout, tc.line) { + t.Errorf("dropped line %q reached stdout", tc.line) + } + case tc.review: + if !strings.Contains(string(reviewData), tc.line) { + t.Errorf("line %q missing from review.jsonl", tc.line) + } + if strings.Contains(stdout, tc.line) { + t.Errorf("review line %q reached stdout", tc.line) + } + default: + rec := "" + for _, r := range strings.Split(strings.TrimSpace(stdout), "\n") { + if strings.Contains(r, tc.line) { + rec = r + break + } + } + if rec == "" { + t.Errorf("line %q missing from stdout", tc.line) + continue + } + if !strings.Contains(rec, `"category":"`+tc.category+`"`) { + t.Errorf("line %q: record %s, want category %q", tc.line, rec, tc.category) + } + if !strings.Contains(rec, `"stage":"`+tc.stage+`"`) { + t.Errorf("line %q: record %s, want stage %q", tc.line, rec, tc.stage) + } + } + } + }) +} diff --git a/shared/pkg/tokenizer/wordpiece/wordpiece.go b/shared/pkg/tokenizer/wordpiece/wordpiece.go index e7b68ba..150ebd0 100644 --- a/shared/pkg/tokenizer/wordpiece/wordpiece.go +++ b/shared/pkg/tokenizer/wordpiece/wordpiece.go @@ -1,13 +1,13 @@ // Package wordpiece implements BERT-style WordPiece tokenization in pure Go, // with no cgo and no external tokenizer runtime. It turns text into the token -// IDs an ONNX BERT/MiniLM classifier expects: basic tokenization (Unicode +// IDs a BERT-vocabulary model expects: basic tokenization (Unicode // cleanup, optional lowercasing and accent stripping, punctuation and CJK // splitting) followed by greedy longest-match-first subword segmentation // against a fixed vocabulary, wrapped with [CLS] and [SEP]. // -// It exists so the in-process model stage can stay a single static binary -// (CGO_ENABLED=0): the common alternative, the rust tokenizers library, needs -// cgo and a statically linked archive. We own this instead. +// It is intended to feed a future in-process model stage so the binary can +// stay CGO_ENABLED=0; no such stage is wired yet. The common alternative, the +// rust tokenizers library, needs cgo and a statically linked archive. package wordpiece import ( From e7b59c6422ec3896cb58ee4e205d97f903b45005 Mon Sep 17 00:00:00 2001 From: eitam Date: Fri, 10 Jul 2026 10:36:59 +0300 Subject: [PATCH 2/2] readme: quote the full stats line, note noise is counted then dropped --- README.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index f080ad2..f4c77ff 100644 --- a/README.md +++ b/README.md @@ -59,8 +59,10 @@ classmesh run --config examples/classmesh.yaml prod.log > classified.jsonl With the default seed, the million lines classify in under two seconds: the rules tier decides 88%, the model tier 6%, and 6% lands in -`examples/review.jsonl` (stats on stderr: -`processed=1000000 classified=940162 review=59838`). +`examples/review.jsonl`. The stderr stats line reads +`processed=1000000 classified=940162 review=59838 by_stage=map[model:60009 rules:880153]`; +classified counts the health-check records the noise route then discards, so +`classified.jsonl` holds 720,490 lines. ### Cascade config