From 2f1c3ff8486de7aae6aabaf0d5513e1ee9cca9e8 Mon Sep 17 00:00:00 2001 From: eitam Date: Sat, 11 Jul 2026 09:25:31 +0300 Subject: [PATCH] add integrated, scaling, and per-record benchmarks; correct readme perf claims --- README.md | 13 +- .../engine/parallel_perrecord_bench_test.go | 48 +++++ shared/pkg/engine/pipeline_bench_test.go | 164 ++++++++++++++++++ shared/pkg/stage/rules/rules_bench_test.go | 6 +- shared/pkg/stage/rules/scaling_bench_test.go | 86 +++++++++ 5 files changed, 312 insertions(+), 5 deletions(-) create mode 100644 shared/pkg/engine/parallel_perrecord_bench_test.go create mode 100644 shared/pkg/engine/pipeline_bench_test.go create mode 100644 shared/pkg/stage/rules/scaling_bench_test.go diff --git a/README.md b/README.md index 4f1c84b..bb5e086 100644 --- a/README.md +++ b/README.md @@ -106,12 +106,19 @@ Measured on a single core (AMD Ryzen 7 3800X, `make bench`): | Path | Per record | Throughput | Allocations | |---|---|---|---| | Rules stage, first-rule hit | 46 ns | ~22M records/sec | 0 | -| Rules stage, worst case (20-rule walk, regex-heavy) | 7-8 µs | ~130k records/sec | 0 | +| Rules stage, 20-rule regex-heavy miss (benchmark ruleset) | 7-8 µs | ~130k records/sec | 0 | | Full pipeline (engine + rules, discard sink) | 500-550 ns | ~1.8M records/sec | 0 | +| Integrated pipeline (text source -> rules -> JSONL sink) | ~600 ns | ~1.6M records/sec | 2 | Per-record cost depends on your ruleset: order rules by expected volume so the -hot path exits early. The pipeline row isolates engine + rules behind a discard -sink; the JSONL output path is benchmarked separately in `sink/jsonl`. +hot path exits early. The miss row is the benchmark ruleset's cost, not a +general bound: a pattern with no extractable required literal defeats the +prefilter and every such rule pays its full regex on a miss +(`BenchmarkClassifyRegexMissUnprefilterable`: ~101 µs at 20 rules on the same +machine). The pipeline row isolates engine + rules behind a discard sink; the +integrated row adds the source read and JSON encode every CLI run pays. The +structured path (JSONL source -> field rules -> structured output) is +benchmarked in `shared/pkg/engine` as well. 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/shared/pkg/engine/parallel_perrecord_bench_test.go b/shared/pkg/engine/parallel_perrecord_bench_test.go new file mode 100644 index 0000000..dae7525 --- /dev/null +++ b/shared/pkg/engine/parallel_perrecord_bench_test.go @@ -0,0 +1,48 @@ +package engine + +import ( + "context" + "fmt" + "testing" + + "github.com/ClassMesh/classmesh/shared/pkg/domain" + "github.com/ClassMesh/classmesh/shared/pkg/stage" +) + +func BenchmarkEnginePerRecordAllocs(b *testing.B) { + payload := []byte(`2026-06-12T10:00:00Z WARN payment declined order=84712 user=991 amount=49.90`) + cases := []struct { + name string + stage func() stage.Stage + }{ + {"cheap-stage", func() stage.Stage { return &spinStage{} }}, + {"cpu-heavy-stage", func() stage.Stage { return &spinStage{iters: 400} }}, + } + for _, tc := range cases { + for _, workers := range []int{0, 1, 2, 4, 8, 16} { + b.Run(fmt.Sprintf("%s/workers=%d", tc.name, workers), func(b *testing.B) { + src := &benchSource{record: domain.Record{ID: "bench", Data: payload}, n: b.N} + e, err := New(Deps{ + Source: src, + Stages: []stage.Stage{tc.stage()}, + Sink: discardSink{}, + Logger: discardLogger(), + Workers: workers, + }) + if err != nil { + b.Fatal(err) + } + b.SetBytes(int64(len(payload))) + b.ReportAllocs() + b.ResetTimer() + stats, err := e.Run(context.Background()) + if err != nil { + b.Fatal(err) + } + if stats.Processed != b.N { + b.Fatalf("processed = %d, want %d", stats.Processed, b.N) + } + }) + } + } +} diff --git a/shared/pkg/engine/pipeline_bench_test.go b/shared/pkg/engine/pipeline_bench_test.go new file mode 100644 index 0000000..bac72f4 --- /dev/null +++ b/shared/pkg/engine/pipeline_bench_test.go @@ -0,0 +1,164 @@ +package engine_test + +import ( + "bytes" + "context" + "io" + "os" + "path/filepath" + "testing" + + "github.com/ClassMesh/classmesh/shared/pkg/domain" + "github.com/ClassMesh/classmesh/shared/pkg/engine" + "github.com/ClassMesh/classmesh/shared/pkg/sink" + jsonlsink "github.com/ClassMesh/classmesh/shared/pkg/sink/jsonl" + "github.com/ClassMesh/classmesh/shared/pkg/source" + jsonlsrc "github.com/ClassMesh/classmesh/shared/pkg/source/jsonl" + "github.com/ClassMesh/classmesh/shared/pkg/source/textfile" + "github.com/ClassMesh/classmesh/shared/pkg/stage" + "github.com/ClassMesh/classmesh/shared/pkg/stage/rules" +) + +// loopReader replays its data forever so a bounded source can draw an +// arbitrary number of lines without holding them all in memory. +type loopReader struct { + data []byte + off int +} + +func (r *loopReader) Read(p []byte) (int, error) { + if r.off == len(r.data) { + r.off = 0 + } + n := copy(p, r.data[r.off:]) + r.off += n + return n, nil +} + +func lineReader(line string) *loopReader { + return &loopReader{data: bytes.Repeat([]byte(line+"\n"), 1024)} +} + +// boundedSource caps an underlying source at n records so engine.Run drains +// after exactly n records. +type boundedSource struct { + inner source.Source + n, i int +} + +func newBounded(inner source.Source, n int) *boundedSource { + return &boundedSource{inner: inner, n: n} +} + +func (s *boundedSource) Next(ctx context.Context) (domain.Record, error) { + if s.i >= s.n { + return domain.Record{}, source.ErrDrained + } + s.i++ + return s.inner.Next(ctx) +} + +func (s *boundedSource) Close() error { return s.inner.Close() } + +// realisticRules is the shape of a log-triage config: a health-check noise +// rule the text sample hits, plus regex rules a record walks past on a miss. +func realisticRules(tb testing.TB) *rules.Stage { + tb.Helper() + s, err := rules.New([]rules.Rule{ + {Category: "noise", Contains: []string{"healthz", "readiness", "liveness"}}, + {Category: "billing", Regex: []string{`payment (failed|declined)`}}, + {Category: "auth", Contains: []string{"login failed"}, Regex: []string{`(?i)unauthorized`}}, + {Category: "db", Regex: []string{`(connection refused|deadlock detected)`}}, + }) + if err != nil { + tb.Fatalf("rules.New() error = %v", err) + } + return s +} + +func runEngine(b *testing.B, src source.Source, st stage.Stage, out sink.Sink) { + b.Helper() + e, err := engine.New(engine.Deps{Source: src, Stages: []stage.Stage{st}, Sink: out}) + if err != nil { + b.Fatal(err) + } + stats, err := e.Run(context.Background()) + if err != nil { + b.Fatal(err) + } + if stats.Processed != b.N { + b.Fatalf("processed = %d, want %d", stats.Processed, b.N) + } +} + +// BenchmarkPipelineTextRulesJSONL is text source -> rules -> JSONL sink, the +// single-tier CLI path (./classmesh run --rules ... file.txt) end to end. The +// sample hits the first rule, so a classified record is encoded every time. +func BenchmarkPipelineTextRulesJSONL(b *testing.B) { + line := `10.2.3.4 - - [12/Jun/2026:10:00:00] "GET /healthz HTTP/1.1" 200 2 "-" "kube-probe/1.29"` + src := newBounded(textfile.New(lineReader(line), "bench"), b.N) + sink := jsonlsink.New(io.Discard) + defer func() { _ = sink.Close() }() + st := realisticRules(b) + + b.SetBytes(int64(len(line) + 1)) + b.ReportAllocs() + b.ResetTimer() + runEngine(b, src, st, sink) +} + +// BenchmarkPipelineJSONLFieldsJSONL is JSONL source -> field rules -> JSONL +// sink, the structured path (./classmesh run --input jsonl ...) end to end: +// decode into Fields, match on a field, re-encode with the decoded Fields. +func BenchmarkPipelineJSONLFieldsJSONL(b *testing.B) { + line := `{"level":"error","http":{"status":503},"msg":"upstream timeout","user_id":"u1234"}` + gte := 500.0 + st, err := rules.New([]rules.Rule{ + {Category: "alert", Fields: []rules.FieldMatcher{{Path: "http.status", Gte: >e}}}, + }) + if err != nil { + b.Fatal(err) + } + src := newBounded(jsonlsrc.New(lineReader(line), "bench"), b.N) + sink := jsonlsink.New(io.Discard) + defer func() { _ = sink.Close() }() + + b.SetBytes(int64(len(line) + 1)) + b.ReportAllocs() + b.ResetTimer() + runEngine(b, src, st, sink) +} + +// BenchmarkPipelineSink contrasts the same text pipeline draining into +// io.Discard against a real file. The file case measures buffered writes into +// the OS page cache, not durable disk latency: nothing calls fsync. +func BenchmarkPipelineSink(b *testing.B) { + line := `10.2.3.4 - - [12/Jun/2026:10:00:00] "GET /healthz HTTP/1.1" 200 2 "-" "kube-probe/1.29"` + cases := []struct { + name string + writer func(b *testing.B) io.Writer + }{ + {"discard", func(b *testing.B) io.Writer { return io.Discard }}, + {"file", func(b *testing.B) io.Writer { + f, err := os.Create(filepath.Join(b.TempDir(), "out.jsonl")) + if err != nil { + b.Fatal(err) + } + b.Cleanup(func() { _ = f.Close() }) + return f + }}, + } + for _, tc := range cases { + b.Run(tc.name, func(b *testing.B) { + src := newBounded(textfile.New(lineReader(line), "bench"), b.N) + sink := jsonlsink.New(tc.writer(b)) + defer func() { _ = sink.Close() }() + st := realisticRules(b) + + b.SetBytes(int64(len(line) + 1)) + b.ReportAllocs() + b.ResetTimer() + runEngine(b, src, st, sink) + }) + } +} diff --git a/shared/pkg/stage/rules/rules_bench_test.go b/shared/pkg/stage/rules/rules_bench_test.go index 9b614f2..fae0725 100644 --- a/shared/pkg/stage/rules/rules_bench_test.go +++ b/shared/pkg/stage/rules/rules_bench_test.go @@ -2,10 +2,12 @@ package rules import ( "context" + "errors" "fmt" "testing" "github.com/ClassMesh/classmesh/shared/pkg/domain" + "github.com/ClassMesh/classmesh/shared/pkg/stage" ) // benchStage builds a realistic 20-rule set: a mix of substring and regex @@ -42,7 +44,7 @@ func benchClassify(b *testing.B, payload string) { b.ResetTimer() for i := 0; i < b.N; i++ { _, err := s.Classify(ctx, r) - if err != nil && err.Error() != "stage: unclassified" { + if err != nil && !errors.Is(err, stage.ErrUnclassified) { b.Fatal(err) } } @@ -95,7 +97,7 @@ func BenchmarkClassifyMissLiteral(b *testing.B) { b.ResetTimer() for i := 0; i < b.N; i++ { _, err := s.Classify(ctx, r) - if err != nil && err.Error() != "stage: unclassified" { + if err != nil && !errors.Is(err, stage.ErrUnclassified) { b.Fatal(err) } } diff --git a/shared/pkg/stage/rules/scaling_bench_test.go b/shared/pkg/stage/rules/scaling_bench_test.go new file mode 100644 index 0000000..9826ede --- /dev/null +++ b/shared/pkg/stage/rules/scaling_bench_test.go @@ -0,0 +1,86 @@ +package rules + +import ( + "context" + "errors" + "fmt" + "testing" + + "github.com/ClassMesh/classmesh/shared/pkg/domain" + "github.com/ClassMesh/classmesh/shared/pkg/stage" +) + +func runClassify(b *testing.B, s *Stage, payload string) { + b.Helper() + r := domain.Record{ID: "bench", Data: []byte(payload)} + ctx := context.Background() + b.SetBytes(int64(len(payload))) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, err := s.Classify(ctx, r) + if err != nil && !errors.Is(err, stage.ErrUnclassified) { + b.Fatal(err) + } + } +} + +// containsRuleset builds n substring rules whose markers never appear in the +// benchmark payload, so a record walks the whole set before missing. +func containsRuleset(b *testing.B, n int) *Stage { + b.Helper() + rs := make([]Rule, 0, n) + for i := 0; i < n; i++ { + rs = append(rs, Rule{ + Category: fmt.Sprintf("svc-%d", i), + Contains: []string{fmt.Sprintf("service-%d-unique-marker", i)}, + }) + } + s, err := New(rs) + if err != nil { + b.Fatalf("New() error = %v", err) + } + return s +} + +// BenchmarkRulesetScaling shows the linear cost of the miss walk as the +// ruleset grows: order rules by expected volume so the hot path exits early. +func BenchmarkRulesetScaling(b *testing.B) { + payload := `2026-06-12T10:00:00Z INFO order shipped id=84712 warehouse=tlv carrier=ups` + for _, n := range []int{20, 100, 1000} { + b.Run(fmt.Sprintf("rules=%d", n), func(b *testing.B) { + runClassify(b, containsRuleset(b, n), payload) + }) + } +} + +// BenchmarkClassifyMiddleRuleHit fills the gap between first- and last-rule +// hit: the record matches rule 10 of the realistic 20-rule set. +func BenchmarkClassifyMiddleRuleHit(b *testing.B) { + runClassify(b, benchStage(b), `2026-06-12T10:00:00Z INFO service-4-marker request completed duration=12ms`) +} + +// unprefilterableRegexRuleset is 20 regex rules that all begin with an +// alternation or wildcard, so no literal prefix can be extracted and the +// prefilter cannot skip any of them: the true worst-case miss. +func unprefilterableRegexRuleset(b *testing.B) *Stage { + b.Helper() + rs := make([]Rule, 0, 20) + for i := 0; i < 20; i++ { + rs = append(rs, Rule{ + Category: fmt.Sprintf("re-%d", i), + Regex: []string{fmt.Sprintf(`(?i)(alpha%d|beta%d|gamma%d)\d+`, i, i, i)}, + }) + } + s, err := New(rs) + if err != nil { + b.Fatalf("New() error = %v", err) + } + return s +} + +// BenchmarkClassifyRegexMissUnprefilterable is the miss the literal prefilter +// cannot help: every regex is run against the payload. +func BenchmarkClassifyRegexMissUnprefilterable(b *testing.B) { + runClassify(b, unprefilterableRegexRuleset(b), `2026-06-12T10:00:00Z INFO order shipped id=84712 warehouse=tlv carrier=ups`) +}