From 3900b9b6938422b4237c5f5b4da4d701103ee641 Mon Sep 17 00:00:00 2001 From: Nikola Grcevski Date: Tue, 28 Oct 2025 17:02:33 -0400 Subject: [PATCH 01/10] improve gibberish detection --- .../transform/route/clusterurl/cluster.go | 24 +++++++++++++++++-- .../route/clusterurl/cluster_test.go | 6 +++++ 2 files changed, 28 insertions(+), 2 deletions(-) diff --git a/pkg/internal/transform/route/clusterurl/cluster.go b/pkg/internal/transform/route/clusterurl/cluster.go index c63b4d52ee..800bc1a772 100644 --- a/pkg/internal/transform/route/clusterurl/cluster.go +++ b/pkg/internal/transform/route/clusterurl/cluster.go @@ -52,6 +52,8 @@ func NewClusterURLClassifier(config *Config) (*ClusterURLClassifier, error) { validCharTable[c] = true } + validCharTable[' '] = true + return &ClusterURLClassifier{ classifier: classifier, cache: cache, @@ -162,8 +164,26 @@ func (csf *ClusterURLClassifier) okWord(w string) bool { if ok { return ok } - if gibberish.IsGibberish(w, csf.classifier) { - return false + + start := 0 + for i, c := range w { + if c == '-' || c == '_' || c == '.' { + if i == start { + return false + } + + if gibberish.IsGibberish(w[start:i], csf.classifier) { + return false + } + + start = i + 1 + } + } + + if start < len(w) { + if gibberish.IsGibberish(w[start:], csf.classifier) { + return false + } } csf.cache.Add(w, true) diff --git a/pkg/internal/transform/route/clusterurl/cluster_test.go b/pkg/internal/transform/route/clusterurl/cluster_test.go index 2bc83f44da..a978dd590c 100644 --- a/pkg/internal/transform/route/clusterurl/cluster_test.go +++ b/pkg/internal/transform/route/clusterurl/cluster_test.go @@ -13,7 +13,9 @@ import ( func TestClusterURL(t *testing.T) { csf, err := NewClusterURLClassifier(DefaultConfig()) assert.NoError(t, err) + assert.Equal(t, "*", csf.ClusterURL("registry-apjkmyp")) assert.Empty(t, csf.ClusterURL("")) + assert.Equal(t, "*", csf.ClusterURL("apjkmyp")) assert.Equal(t, "/users/*/j4elk/*/job/*", csf.ClusterURL("/users/fdklsd/j4elk/23993/job/2")) assert.Equal(t, "*", csf.ClusterURL("123")) assert.Equal(t, "/*", csf.ClusterURL("/123")) @@ -51,6 +53,10 @@ func TestClusterURL(t *testing.T) { assert.Equal(t, "HTTP GET", csf.ClusterURL("HTTP GET")) assert.Equal(t, "GET /api/cart", csf.ClusterURL("GET /api/cart?sessionId=55f4e5ea-5d6d-482a-80c4-799e3c72dfb0¤cyCode=USD")) assert.Equal(t, "/getquote", csf.ClusterURL("/getquote")) + assert.Equal(t, "PUT /bar/test/test/*/files/*/test/*", csf.ClusterURL("PUT /bar/test/test/bar-attach-generic-product-apjkmyp/files/multi-test-version-jwbCm/test/some-file.txt")) + assert.Equal(t, "PUT /test/bar/test/test/*/files/*/test/*", csf.ClusterURL("PUT /test/bar/test/test/bar-attach-generic-registry-apjkmyp/files/push-metrics-test-OYboK/test/README.md")) + assert.Equal(t, "PUT /test/bar/test_plus/test.now/*/files/*/test/*", csf.ClusterURL("PUT /test/bar/test_plus/test.now/bar-attach-generic-registry-apjkmyp/files/push-metrics-test-OYboK/test/README.md")) + assert.Equal(t, "PUT /bar/test/test/*/files/*/test/*", csf.ClusterURL("PUT /bar/test/test/a----/files/-a-a-a--/test/some-file.txt")) assert.Equal(t, "", csf.ClusterURL("?")) assert.Equal(t, "*", csf.ClusterURL("attach12?")) assert.Equal(t, "*", csf.ClusterURL("1?")) From 82e584ed2133d52053ef027e629cfe6413a32a64 Mon Sep 17 00:00:00 2001 From: Nikola Grcevski Date: Tue, 28 Oct 2025 20:03:55 -0400 Subject: [PATCH 02/10] secondary clustering per service --- pkg/appolly/app/svc/svc.go | 2 + pkg/appolly/discover/matcher_test.go | 9 +- pkg/appolly/discover/typer.go | 7 +- pkg/appolly/discover/typer_test.go | 5 +- .../transform/route/clusterurl/trie.go | 230 ++++++++++++++++++ .../transform/route/clusterurl/trie_test.go | 153 ++++++++++++ pkg/obi/config.go | 5 +- pkg/obi/config_test.go | 5 +- pkg/transform/routes.go | 6 + pkg/transform/routes_test.go | 61 +++++ 10 files changed, 471 insertions(+), 12 deletions(-) create mode 100644 pkg/internal/transform/route/clusterurl/trie.go create mode 100644 pkg/internal/transform/route/clusterurl/trie_test.go diff --git a/pkg/appolly/app/svc/svc.go b/pkg/appolly/app/svc/svc.go index f61e249ff7..be4c9dcd37 100644 --- a/pkg/appolly/app/svc/svc.go +++ b/pkg/appolly/app/svc/svc.go @@ -10,6 +10,7 @@ import ( "go.opentelemetry.io/obi/pkg/appolly/services" attr "go.opentelemetry.io/obi/pkg/export/attributes/names" "go.opentelemetry.io/obi/pkg/internal/transform/route" + "go.opentelemetry.io/obi/pkg/internal/transform/route/clusterurl" ) type InstrumentableType int @@ -117,6 +118,7 @@ type Attrs struct { CustomInRouteMatcher route.Matcher CustomOutRouteMatcher route.Matcher HarvestedRouteMatcher route.Matcher + PathTrie *clusterurl.PathTrie } func (i *Attrs) GetUID() UID { diff --git a/pkg/appolly/discover/matcher_test.go b/pkg/appolly/discover/matcher_test.go index 9da2d2180d..2ce7533578 100644 --- a/pkg/appolly/discover/matcher_test.go +++ b/pkg/appolly/discover/matcher_test.go @@ -14,6 +14,7 @@ import ( "go.opentelemetry.io/obi/pkg/internal/testutil" "go.opentelemetry.io/obi/pkg/obi" "go.opentelemetry.io/obi/pkg/pipe/msg" + "go.opentelemetry.io/obi/pkg/transform" ) func testMatch(t *testing.T, m Event[ProcessMatch], name string, @@ -551,7 +552,7 @@ func TestCriteriaMatcher_Granular(t *testing.T) { require.Len(t, planetMatch.Criteria, 2) - planetAttrs := makeServiceAttrs(&planetMatch) + planetAttrs := makeServiceAttrs(&planetMatch, &transform.RoutesConfig{}) assert.True(t, planetAttrs.ExportModes.CanExportTraces()) assert.False(t, planetAttrs.ExportModes.CanExportMetrics()) @@ -562,7 +563,7 @@ func TestCriteriaMatcher_Granular(t *testing.T) { require.Len(t, satelliteMatch.Criteria, 2) - satelliteAttrs := makeServiceAttrs(&satelliteMatch) + satelliteAttrs := makeServiceAttrs(&satelliteMatch, &transform.RoutesConfig{}) assert.False(t, satelliteAttrs.ExportModes.CanExportTraces()) assert.False(t, satelliteAttrs.ExportModes.CanExportMetrics()) @@ -572,7 +573,7 @@ func TestCriteriaMatcher_Granular(t *testing.T) { require.Len(t, starMatch.Criteria, 2) - starAttrs := makeServiceAttrs(&starMatch) + starAttrs := makeServiceAttrs(&starMatch, &transform.RoutesConfig{}) assert.False(t, starAttrs.ExportModes.CanExportTraces()) assert.True(t, starAttrs.ExportModes.CanExportMetrics()) @@ -582,7 +583,7 @@ func TestCriteriaMatcher_Granular(t *testing.T) { require.Len(t, asteroidMatch.Criteria, 2) - asteroidAttrs := makeServiceAttrs(&asteroidMatch) + asteroidAttrs := makeServiceAttrs(&asteroidMatch, &transform.RoutesConfig{}) assert.True(t, asteroidAttrs.ExportModes.CanExportTraces()) assert.True(t, asteroidAttrs.ExportModes.CanExportMetrics()) diff --git a/pkg/appolly/discover/typer.go b/pkg/appolly/discover/typer.go index d3b3966f15..e31e6dfd0c 100644 --- a/pkg/appolly/discover/typer.go +++ b/pkg/appolly/discover/typer.go @@ -21,11 +21,13 @@ import ( "go.opentelemetry.io/obi/pkg/export/imetrics" "go.opentelemetry.io/obi/pkg/internal/goexec" "go.opentelemetry.io/obi/pkg/internal/procs" + "go.opentelemetry.io/obi/pkg/internal/transform/route/clusterurl" "go.opentelemetry.io/obi/pkg/kube" "go.opentelemetry.io/obi/pkg/obi" "go.opentelemetry.io/obi/pkg/pipe/msg" "go.opentelemetry.io/obi/pkg/pipe/swarm" "go.opentelemetry.io/obi/pkg/pipe/swarm/swarms" + "go.opentelemetry.io/obi/pkg/transform" ) type instrumentedExecutable struct { @@ -87,7 +89,7 @@ func samplerFromConfig(s *services.SamplerConfig) trace.Sampler { return nil } -func makeServiceAttrs(processMatch *ProcessMatch) svc.Attrs { +func makeServiceAttrs(processMatch *ProcessMatch, routesCfg *transform.RoutesConfig) svc.Attrs { var name string var namespace string exportModes := services.ExportModeUnset @@ -124,6 +126,7 @@ func makeServiceAttrs(processMatch *ProcessMatch) svc.Attrs { ProcPID: processMatch.Process.Pid, ExportModes: exportModes, Sampler: samplerFromConfig(samplerConfig), + PathTrie: clusterurl.NewPathTrie(routesCfg.MaxPathSegmentCardinality), } if routesConfig != nil { @@ -146,7 +149,7 @@ func (t *typer) FilterClassify(evs []Event[ProcessMatch]) []Event[ebpf.Instrumen ev := &evs[i] switch evs[i].Type { case EventCreated: - svcID := makeServiceAttrs(&ev.Obj) + svcID := makeServiceAttrs(&ev.Obj, t.cfg.Routes) if elfFile, err := findExecElf(ev.Obj.Process, svcID, t.k8sInformer.IsKubeEnabled()); err != nil { t.log.Debug("error finding process ELF. Ignoring", "error", err) diff --git a/pkg/appolly/discover/typer_test.go b/pkg/appolly/discover/typer_test.go index 462351d441..90292575b8 100644 --- a/pkg/appolly/discover/typer_test.go +++ b/pkg/appolly/discover/typer_test.go @@ -10,6 +10,7 @@ import ( "github.com/stretchr/testify/assert" "go.opentelemetry.io/obi/pkg/appolly/services" + "go.opentelemetry.io/obi/pkg/transform" ) type dummyCriterion struct { @@ -41,7 +42,7 @@ func TestMakeServiceAttrs(t *testing.T) { dummyCriterion{name: "svc1", namespace: "ns1", export: services.ExportModeUnset}, }, } - attrs := makeServiceAttrs(proc) + attrs := makeServiceAttrs(proc, &transform.RoutesConfig{}) assert.Equal(t, "svc1", attrs.UID.Name) assert.Equal(t, "ns1", attrs.UID.Namespace) assert.Equal(t, int32(1234), attrs.ProcPID) @@ -60,7 +61,7 @@ func TestMakeServiceAttrs(t *testing.T) { dummyCriterion{sampler: sampler, routes: routes}, }, } - attrs2 := makeServiceAttrs(proc2) + attrs2 := makeServiceAttrs(proc2, &transform.RoutesConfig{}) assert.NotNil(t, attrs2.Sampler) assert.NotNil(t, attrs2.CustomInRouteMatcher) assert.NotNil(t, attrs2.CustomOutRouteMatcher) diff --git a/pkg/internal/transform/route/clusterurl/trie.go b/pkg/internal/transform/route/clusterurl/trie.go new file mode 100644 index 0000000000..3183ec6b86 --- /dev/null +++ b/pkg/internal/transform/route/clusterurl/trie.go @@ -0,0 +1,230 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package clusterurl + +import ( + "strings" + "sync" +) + +// PathNode represents a node in the path trie +type PathNode struct { + // segment is the path component (e.g., "test", "files") + segment string + + // children maps segment values to their nodes + // e.g., children["bar-attach-generic-product-apjkmyp"] = &PathNode{...} + children map[string]*PathNode + + // collapsed indicates if this node has been collapsed to "*" + collapsed bool + + // cardinality tracks how many unique children this node has + cardinality int + + // isWildcard indicates if this represents a "*" wildcard + isWildcard bool +} + +// PathTrie manages the dynamic collapsing trie structure +type PathTrie struct { + root *PathNode + maxCardinality int + mu sync.RWMutex +} + +// NewPathTrie creates a new path trie with the given max cardinality +func NewPathTrie(maxCardinality int) *PathTrie { + return &PathTrie{ + root: &PathNode{ + segment: "", + children: make(map[string]*PathNode), + }, + maxCardinality: maxCardinality, + } +} + +// Insert adds a path to the trie and returns the normalized path +// If a segment exceeds maxCardinality, it collapses to "*" +func (pt *PathTrie) Insert(path string) string { + pt.mu.Lock() + defer pt.mu.Unlock() + + segments := strings.Split(strings.Trim(path, "/"), "/") + if len(segments) == 0 || (len(segments) == 1 && segments[0] == "") { + return path + } + + return pt.insertSegments(segments) +} + +func (pt *PathTrie) insertSegments(segments []string) string { + current := pt.root + result := make([]string, 0, len(segments)) + + for _, segment := range segments { + if segment == "" { + result = append(result, segment) + continue + } + + // If current node is already collapsed, all children become wildcards + if current.collapsed { + result = append(result, "*") + // Continue with the wildcard child + if current.children["*"] == nil { + current.children["*"] = &PathNode{ + segment: "*", + children: make(map[string]*PathNode), + isWildcard: true, + } + } + current = current.children["*"] + continue + } + + // Check if this segment already exists + child, exists := current.children[segment] + + if !exists { + // New segment - check if we need to collapse + if current.cardinality >= pt.maxCardinality { + // Collapse this level + pt.collapseNode(current) + result = append(result, "*") + current = current.children["*"] + continue + } + + // Create new child + child = &PathNode{ + segment: segment, + children: make(map[string]*PathNode), + } + current.children[segment] = child + current.cardinality++ + + // Check if we just hit the threshold + if current.cardinality > pt.maxCardinality { + pt.collapseNode(current) + result = append(result, "*") + current = current.children["*"] + continue + } + } + + result = append(result, segment) + current = child + } + + return "/" + strings.Join(result, "/") +} + +// collapseNode collapses a node by replacing all children with a single wildcard +// and merging their children into the wildcard node +func (pt *PathTrie) collapseNode(node *PathNode) { + if node.collapsed { + return + } + + node.collapsed = true + + // Create or get wildcard node + wildcardNode, hasWildcard := node.children["*"] + if !hasWildcard { + wildcardNode = &PathNode{ + segment: "*", + children: make(map[string]*PathNode), + isWildcard: true, + } + } + + // Merge all children into the wildcard node + for segment, child := range node.children { + if segment == "*" { + continue // Skip the wildcard itself + } + pt.mergeChildren(wildcardNode, child) + } + + // Replace all children with just the wildcard + node.children = map[string]*PathNode{ + "*": wildcardNode, + } + node.cardinality = 1 + + // Recursively check if wildcard node needs collapsing + if wildcardNode.cardinality > pt.maxCardinality { + pt.collapseNode(wildcardNode) + } +} + +// mergeChildren merges children from source into target +// This is called during collapse to combine all child paths +func (pt *PathTrie) mergeChildren(target, source *PathNode) { + for segment, child := range source.children { + if existing, exists := target.children[segment]; exists { + // Child already exists, recursively merge their children + pt.mergeChildren(existing, child) + } else { + // New child, add it + target.children[segment] = child + target.cardinality++ + } + } +} + +// Lookup returns the normalized path for a given input path +// This is used to query existing paths without modifying the trie +func (pt *PathTrie) Lookup(path string) string { + pt.mu.RLock() + defer pt.mu.RUnlock() + + segments := strings.Split(strings.Trim(path, "/"), "/") + if len(segments) == 0 || (len(segments) == 1 && segments[0] == "") { + return path + } + + return pt.lookupSegments(segments) +} + +func (pt *PathTrie) lookupSegments(segments []string) string { + current := pt.root + result := make([]string, 0, len(segments)) + + for _, segment := range segments { + if segment == "" { + result = append(result, segment) + continue + } + + // If node is collapsed, use wildcard + if current.collapsed { + result = append(result, "*") + current = current.children["*"] + continue + } + + // Try to find exact match + child, exists := current.children[segment] + if !exists { + // No exact match, check for wildcard + if wildcardChild, hasWildcard := current.children["*"]; hasWildcard { + result = append(result, "*") + current = wildcardChild + continue + } + // Not found at all, return segment as-is and stop traversing + result = append(result, segment) + // Can't traverse further, append remaining segments + result = append(result, segments[len(result):]...) + break + } + + result = append(result, segment) + current = child + } + + return "/" + strings.Join(result, "/") +} diff --git a/pkg/internal/transform/route/clusterurl/trie_test.go b/pkg/internal/transform/route/clusterurl/trie_test.go new file mode 100644 index 0000000000..a593ec3679 --- /dev/null +++ b/pkg/internal/transform/route/clusterurl/trie_test.go @@ -0,0 +1,153 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package clusterurl + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestPathTrie_BasicInsertAndLookup(t *testing.T) { + trie := NewPathTrie(2) + + // Insert first path + result := trie.Insert("test/bar-attach-generic-product-apjkmyp/files/multi-test-version-jwbCm/test") + assert.Equal(t, "/test/bar-attach-generic-product-apjkmyp/files/multi-test-version-jwbCm/test", result) + + // Insert second path with different second segment + result = trie.Insert("test/apjkmyp/files/jwbCm/test") + assert.Equal(t, "/test/apjkmyp/files/jwbCm/test", result) + + // Insert third path - should trigger collapse at second segment (cardinality > 2) + result = trie.Insert("test/xyz/files/abc/test") + assert.Equal(t, "/test/*/files/*/test", result) + + // Lookup should now return collapsed path + result = trie.Lookup("test/anything-new/files/something/test") + assert.Equal(t, "/test/*/files/*/test", result) +} + +func TestPathTrie_CardinalityThreshold(t *testing.T) { + trie := NewPathTrie(3) + + // Add paths up to threshold + assert.Equal(t, "/api/v1/users", trie.Insert("api/v1/users")) + assert.Equal(t, "/api/v2/users", trie.Insert("api/v2/users")) + assert.Equal(t, "/api/v3/users", trie.Insert("api/v3/users")) + + // Next insert should trigger collapse + assert.Equal(t, "/api/*/users", trie.Insert("api/v4/users")) + + // Verify lookup uses collapsed path + assert.Equal(t, "/api/*/users", trie.Lookup("api/v999/users")) +} + +func TestPathTrie_CascadingCollapse(t *testing.T) { + trie := NewPathTrie(2) + + // Build tree: /root/child1/grandchild1 + // /root/child1/grandchild2 + // /root/child2/grandchild3 + trie.Insert("root/child1/grandchild1") + trie.Insert("root/child1/grandchild2") + trie.Insert("root/child2/grandchild3") + + // This should trigger collapse at "child" level + // which should cascade to grandchildren + result := trie.Insert("root/child3/grandchild4") + + // After collapse, all should be wildcards + assert.Equal(t, "/root/*/*", result) +} + +func TestPathTrie_EmptyPath(t *testing.T) { + trie := NewPathTrie(2) + + assert.Empty(t, trie.Insert("")) + assert.Empty(t, trie.Lookup("")) +} + +func TestPathTrie_SingleSegment(t *testing.T) { + trie := NewPathTrie(2) + + result := trie.Insert("test") + assert.Equal(t, "/test", result) + + result = trie.Lookup("test") + assert.Equal(t, "/test", result) +} + +func TestPathTrie_PreserveExistingPaths(t *testing.T) { + trie := NewPathTrie(2) + + // Insert paths + trie.Insert("api/users/123") + trie.Insert("api/users/456") + + // Before collapse, lookups should return exact matches + assert.Equal(t, "/api/users/123", trie.Lookup("api/users/123")) + assert.Equal(t, "/api/users/456", trie.Lookup("api/users/456")) + + // Trigger collapse + trie.Insert("api/users/789") + + // After collapse, all should use wildcard + assert.Equal(t, "/api/users/*", trie.Lookup("api/users/123")) + assert.Equal(t, "/api/users/*", trie.Lookup("api/users/999")) +} + +func TestPathTrie_ComplexPaths(t *testing.T) { + trie := NewPathTrie(3) + + paths := []string{ + "bar/test/test/bar-attach-generic-product-apjkmyp/files/multi-test-version-jwbCm/test", + "bar/test/test/bar-attach-generic-registry-apjkmyp/files/push-metrics-test-OYboK/test", + "bar/test/test/another-product-xyz/files/version-abc/test", + } + + for _, path := range paths { + trie.Insert(path) + } + + // Should not collapse yet (cardinality = 3, threshold = 3) + result := trie.Lookup(paths[0]) + assert.Contains(t, result, "bar-attach-generic-product-apjkmyp") + + // Fourth path should trigger collapse + trie.Insert("bar/test/test/fourth-product/files/version-def/test") + + result = trie.Lookup("bar/test/test/any-product/files/any-version/test") + assert.Equal(t, "/bar/test/test/*/files/*/test", result) +} + +func BenchmarkPathTrie_Insert(b *testing.B) { + trie := NewPathTrie(10) + + paths := []string{ + "api/v1/users/123/posts/456", + "api/v1/users/789/posts/012", + "api/v2/products/abc/reviews/def", + "api/v3/orders/xyz/items/pqr", + } + + b.ResetTimer() + for i := 0; i < b.N; i++ { + trie.Insert(paths[i%len(paths)]) + } +} + +func BenchmarkPathTrie_Lookup(b *testing.B) { + trie := NewPathTrie(10) + + // Pre-populate trie + for i := 0; i < 100; i++ { + trie.Insert("api/v1/users/123/posts/456") + } + + b.ResetTimer() + for i := 0; i < b.N; i++ { + trie.Lookup("api/v1/users/999/posts/888") + } +} diff --git a/pkg/obi/config.go b/pkg/obi/config.go index df114a96f0..f01ffd41d6 100644 --- a/pkg/obi/config.go +++ b/pkg/obi/config.go @@ -174,8 +174,9 @@ var DefaultConfig = Config{ MetricSpanNameAggregationLimit: 100, }, Routes: &transform.RoutesConfig{ - Unmatch: transform.UnmatchDefault, - WildcardChar: "*", + Unmatch: transform.UnmatchDefault, + WildcardChar: "*", + MaxPathSegmentCardinality: 10, }, NetworkFlows: DefaultNetworkConfig, Discovery: services.DiscoveryConfig{ diff --git a/pkg/obi/config_test.go b/pkg/obi/config_test.go index 90f66f018d..babe5257a0 100644 --- a/pkg/obi/config_test.go +++ b/pkg/obi/config_test.go @@ -221,8 +221,9 @@ discovery: MetricSpanNameAggregationLimit: 100, }, Routes: &transform.RoutesConfig{ - Unmatch: transform.UnmatchHeuristic, - WildcardChar: "*", + Unmatch: transform.UnmatchHeuristic, + WildcardChar: "*", + MaxPathSegmentCardinality: 10, }, NameResolver: &transform.NameResolverConfig{ Sources: []string{"k8s", "dns"}, diff --git a/pkg/transform/routes.go b/pkg/transform/routes.go index 833dd9848f..afc569bc44 100644 --- a/pkg/transform/routes.go +++ b/pkg/transform/routes.go @@ -60,6 +60,8 @@ type RoutesConfig struct { IgnoredEvents IgnoreMode `yaml:"ignore_mode"` // Character that will be used to replace route segments WildcardChar string `yaml:"wildcard_char,omitempty"` + // Max allowed path segment cardinality (per service) for the heuristic matcher + MaxPathSegmentCardinality int `yaml:"max_path_segment_cardinality"` } func RoutesProvider(rc *RoutesConfig, input, output *msg.Queue[[]request.Span]) swarm.InstanceFunc { @@ -203,6 +205,10 @@ func setUnmatchToPath(_ *routerNode, str *request.Span) { func classifyFromPath(rc *routerNode, s *request.Span) { if s.Route == "" && s.IsHTTPSpan() { s.Route = rc.classifier.ClusterURL(s.Path) + if s.Service.PathTrie != nil { + s.Service.PathTrie.Insert(s.Route) + s.Route = s.Service.PathTrie.Lookup(s.Route) + } } } diff --git a/pkg/transform/routes_test.go b/pkg/transform/routes_test.go index f8b52ba6ec..5a48cd590f 100644 --- a/pkg/transform/routes_test.go +++ b/pkg/transform/routes_test.go @@ -11,7 +11,9 @@ import ( "github.com/stretchr/testify/require" "go.opentelemetry.io/obi/pkg/appolly/app/request" + "go.opentelemetry.io/obi/pkg/appolly/app/svc" "go.opentelemetry.io/obi/pkg/internal/testutil" + "go.opentelemetry.io/obi/pkg/internal/transform/route/clusterurl" "go.opentelemetry.io/obi/pkg/pipe/msg" ) @@ -121,6 +123,65 @@ func TestUnmatchedAuto(t *testing.T) { } } +func TestUnmatchedAutoSuper(t *testing.T) { + trie := clusterurl.NewPathTrie(3) + for _, tc := range []UnmatchType{UnmatchHeuristic} { + t.Run(string(tc), func(t *testing.T) { + input := msg.NewQueue[[]request.Span](msg.ChannelBufferLen(10)) + output := msg.NewQueue[[]request.Span](msg.ChannelBufferLen(10)) + router, err := RoutesProvider(&RoutesConfig{Unmatch: tc, WildcardChar: "*"}, + input, output)(t.Context()) + require.NoError(t, err) + out := output.Subscribe() + defer input.Close() + go router(t.Context()) + input.Send([]request.Span{{Path: "/v1/user/1234", Type: request.EventTypeHTTP, Service: svc.Attrs{PathTrie: trie}}}) + s := testutil.ReadChannel(t, out, testTimeout) + // Heuristic only detects the last component as an ID, 1234 -> * + assert.Equal(t, "/v1/user/1234", s[0].Path) + assert.Equal(t, "/v1/user/*", s[0].Route) + input.Send([]request.Span{{Path: "/v2/user/1234", Type: request.EventTypeHTTP, Service: svc.Attrs{PathTrie: trie}}}) + // Heuristic only detects the last component as an ID, 1234 -> * + s = testutil.ReadChannel(t, out, testTimeout) + assert.Equal(t, "/v2/user/1234", s[0].Path) + assert.Equal(t, "/v2/user/*", s[0].Route) + input.Send([]request.Span{{Path: "/v3/user/1234", Type: request.EventTypeHTTP, Service: svc.Attrs{PathTrie: trie}}}) + // Heuristic only detects the last component as an ID, 1234 -> * + s = testutil.ReadChannel(t, out, testTimeout) + assert.Equal(t, "/v3/user/1234", s[0].Path) + assert.Equal(t, "/v3/user/*", s[0].Route) + input.Send([]request.Span{{Path: "/v4/user/1234", Type: request.EventTypeHTTPClient, Service: svc.Attrs{PathTrie: trie}}}) + // We finally blow the cardinality of the first path segment, v4 -> * + s = testutil.ReadChannel(t, out, testTimeout) + assert.Equal(t, "/v4/user/1234", s[0].Path) + assert.Equal(t, "/*/user/*", s[0].Route) + input.Send([]request.Span{{Path: "/v1/user/1234", Type: request.EventTypeHTTPClient, Service: svc.Attrs{PathTrie: trie}}}) + // From now on, even previously matched routes are collapsed + s = testutil.ReadChannel(t, out, testTimeout) + assert.Equal(t, "/v1/user/1234", s[0].Path) + assert.Equal(t, "/*/user/*", s[0].Route) + // let's blow cardinality of the second path component, "user" + input.Send([]request.Span{{Path: "/v1/user-one/1234", Type: request.EventTypeHTTPClient, Service: svc.Attrs{PathTrie: trie}}}) + s = testutil.ReadChannel(t, out, testTimeout) + assert.Equal(t, "/v1/user-one/1234", s[0].Path) + assert.Equal(t, "/*/user-one/*", s[0].Route) + input.Send([]request.Span{{Path: "/v1/user-two/1234", Type: request.EventTypeHTTPClient, Service: svc.Attrs{PathTrie: trie}}}) + s = testutil.ReadChannel(t, out, testTimeout) + assert.Equal(t, "/v1/user-two/1234", s[0].Path) + assert.Equal(t, "/*/user-two/*", s[0].Route) + input.Send([]request.Span{{Path: "/v1/user-three/1234", Type: request.EventTypeHTTPClient, Service: svc.Attrs{PathTrie: trie}}}) + s = testutil.ReadChannel(t, out, testTimeout) + assert.Equal(t, "/v1/user-three/1234", s[0].Path) + assert.Equal(t, "/*/*/*", s[0].Route) + input.Send([]request.Span{{Path: "/v1/user/1234", Type: request.EventTypeHTTPClient, Service: svc.Attrs{PathTrie: trie}}}) + // From now on, even previously matched routes are collapsed + s = testutil.ReadChannel(t, out, testTimeout) + assert.Equal(t, "/v1/user/1234", s[0].Path) + assert.Equal(t, "/*/*/*", s[0].Route) + }) + } +} + func TestIgnoreRoutes(t *testing.T) { input := msg.NewQueue[[]request.Span](msg.ChannelBufferLen(10)) output := msg.NewQueue[[]request.Span](msg.ChannelBufferLen(10)) From 0715232354163b7283dce020eb893adf53e69087 Mon Sep 17 00:00:00 2001 From: Nikola Grcevski Date: Fri, 7 Nov 2025 12:22:11 -0500 Subject: [PATCH 03/10] merge children once one path has cardinality --- .../transform/route/clusterurl/trie.go | 3 ++ .../transform/route/clusterurl/trie_test.go | 30 +++++++++++++++++++ 2 files changed, 33 insertions(+) diff --git a/pkg/internal/transform/route/clusterurl/trie.go b/pkg/internal/transform/route/clusterurl/trie.go index 3183ec6b86..7c02a7c270 100644 --- a/pkg/internal/transform/route/clusterurl/trie.go +++ b/pkg/internal/transform/route/clusterurl/trie.go @@ -170,6 +170,9 @@ func (pt *PathTrie) mergeChildren(target, source *PathNode) { } else { // New child, add it target.children[segment] = child + if segment == "*" { + target.cardinality = pt.maxCardinality + } target.cardinality++ } } diff --git a/pkg/internal/transform/route/clusterurl/trie_test.go b/pkg/internal/transform/route/clusterurl/trie_test.go index a593ec3679..7aa33152c9 100644 --- a/pkg/internal/transform/route/clusterurl/trie_test.go +++ b/pkg/internal/transform/route/clusterurl/trie_test.go @@ -44,6 +44,36 @@ func TestPathTrie_CardinalityThreshold(t *testing.T) { assert.Equal(t, "/api/*/users", trie.Lookup("api/v999/users")) } +func TestPathTrie_CardinalitySecondaryThreshold(t *testing.T) { + trie := NewPathTrie(3) + + // Add paths up to threshold + assert.Equal(t, "/api/v1/items/teddy_bear", trie.Insert("api/v1/items/teddy_bear")) + assert.Equal(t, "/api/v1/items/sports_car", trie.Insert("api/v1/items/sports_car")) + assert.Equal(t, "/api/v1/items/t-shirt", trie.Insert("api/v1/items/t-shirt")) + + assert.Equal(t, "/api/v1/items/t-shirt", trie.Lookup("api/v1/items/t-shirt")) + + // Add paths up to threshold + assert.Equal(t, "/api/v1/customers", trie.Insert("api/v1/customers")) + assert.Equal(t, "/api/v1/admin", trie.Insert("api/v1/admin")) + + // Next insert should trigger collapse + assert.Equal(t, "/api/v1/*", trie.Insert("api/v1/users")) + + // Let's trigger the secondary collapse now + assert.Equal(t, "/api/v2/items", trie.Insert("api/v2/items")) + assert.Equal(t, "/api/v3/items", trie.Insert("api/v3/items")) + + assert.Equal(t, "/api/*/*", trie.Insert("api/v4/items")) + assert.Equal(t, "/api/*/*/t-shirt", trie.Lookup("api/v4/items/t-shirt")) + + // trigger the third level collapse + assert.Equal(t, "/api/*/*/*", trie.Insert("api/v1/customers/list")) + + assert.Equal(t, "/api/*/*/*", trie.Lookup("api/v4/items/t-shirt")) +} + func TestPathTrie_CascadingCollapse(t *testing.T) { trie := NewPathTrie(2) From b7ff5e8aabb70c68c84da779ba100e2e5b210f02 Mon Sep 17 00:00:00 2001 From: Nikola Grcevski Date: Fri, 7 Nov 2025 13:10:38 -0500 Subject: [PATCH 04/10] allow higher cardinality on higher levels, introduce new mode --- .../transform/route/clusterurl/trie.go | 5 +-- .../transform/route/clusterurl/trie_test.go | 28 +++++++++----- pkg/transform/routes.go | 37 ++++++++++++++++--- pkg/transform/routes_test.go | 4 +- 4 files changed, 53 insertions(+), 21 deletions(-) diff --git a/pkg/internal/transform/route/clusterurl/trie.go b/pkg/internal/transform/route/clusterurl/trie.go index 7c02a7c270..05a6a012b9 100644 --- a/pkg/internal/transform/route/clusterurl/trie.go +++ b/pkg/internal/transform/route/clusterurl/trie.go @@ -14,7 +14,7 @@ type PathNode struct { segment string // children maps segment values to their nodes - // e.g., children["bar-attach-generic-product-apjkmyp"] = &PathNode{...} + // e.g., children["bar/attach/generic-product-apjkmyp"] = &PathNode{...} children map[string]*PathNode // collapsed indicates if this node has been collapsed to "*" @@ -170,9 +170,6 @@ func (pt *PathTrie) mergeChildren(target, source *PathNode) { } else { // New child, add it target.children[segment] = child - if segment == "*" { - target.cardinality = pt.maxCardinality - } target.cardinality++ } } diff --git a/pkg/internal/transform/route/clusterurl/trie_test.go b/pkg/internal/transform/route/clusterurl/trie_test.go index 7aa33152c9..578c30dfd6 100644 --- a/pkg/internal/transform/route/clusterurl/trie_test.go +++ b/pkg/internal/transform/route/clusterurl/trie_test.go @@ -4,6 +4,7 @@ package clusterurl import ( + "strconv" "testing" "github.com/stretchr/testify/assert" @@ -65,7 +66,11 @@ func TestPathTrie_CardinalitySecondaryThreshold(t *testing.T) { assert.Equal(t, "/api/v2/items", trie.Insert("api/v2/items")) assert.Equal(t, "/api/v3/items", trie.Insert("api/v3/items")) - assert.Equal(t, "/api/*/*", trie.Insert("api/v4/items")) + assert.Equal(t, "/api/*/items", trie.Insert("api/v4/items")) + for i := range 3 { + trie.Insert("api/v4/items" + strconv.Itoa(i)) + } + assert.Equal(t, "/api/*/*", trie.Lookup("api/v4/items")) assert.Equal(t, "/api/*/*/t-shirt", trie.Lookup("api/v4/items/t-shirt")) // trigger the third level collapse @@ -156,15 +161,20 @@ func BenchmarkPathTrie_Insert(b *testing.B) { trie := NewPathTrie(10) paths := []string{ - "api/v1/users/123/posts/456", - "api/v1/users/789/posts/012", - "api/v2/products/abc/reviews/def", - "api/v3/orders/xyz/items/pqr", + "/users/fdklsd/j4elk/23993/job/2", + "/v1/products/22", + "/products/1/org/3", + "/attach?session_id=ddfsdsf&track_id=sjdklnfldsn", + "GET /user_space/", + "/api/hello.world", + "123/ljgdflgjf", + "", } - b.ResetTimer() - for i := 0; i < b.N; i++ { - trie.Insert(paths[i%len(paths)]) + for b.Loop() { + for i := 0; i < len(paths); i++ { + trie.Insert(paths[i]) + } } } @@ -173,7 +183,7 @@ func BenchmarkPathTrie_Lookup(b *testing.B) { // Pre-populate trie for i := 0; i < 100; i++ { - trie.Insert("api/v1/users/123/posts/456") + trie.Insert("api/v1/users/" + strconv.Itoa(i) + "/posts/456") } b.ResetTimer() diff --git a/pkg/transform/routes.go b/pkg/transform/routes.go index afc569bc44..da1fcf4d9b 100644 --- a/pkg/transform/routes.go +++ b/pkg/transform/routes.go @@ -29,6 +29,9 @@ const ( UnmatchWildcard = UnmatchType("wildcard") // UnmatchHeuristic detects the route field using a heuristic UnmatchHeuristic = UnmatchType("heuristic") + // UnmatchLowCardinality uses the same classifier as the Heuristic, but + // it also has a second level Trie based cache to cap the max cardinality + UnmatchLowCardinality = UnmatchType("low-cardinality") UnmatchDefault = UnmatchHeuristic ) @@ -145,6 +148,19 @@ func (rn *routerNode) provideRoutes(_ context.Context) (swarm.RunFunc, error) { }, nil } +func makeHeuristicClassifier(rc *RoutesConfig) (*clusterurl.ClusterURLClassifier, error) { + classifierCfg := clusterurl.DefaultConfig() + if rc.WildcardChar != "" { + classifierCfg.ReplaceWith = rc.WildcardChar[0] + } + classifier, err := clusterurl.NewClusterURLClassifier(classifierCfg) + if err != nil { + return nil, fmt.Errorf("chooseUnmatchPolicy: unable to create cluster URL classifier: %w", err) + } + + return classifier, nil +} + func chooseUnmatchPolicy(rn *routerNode) (func(rn *routerNode, span *request.Span), error) { var unmatchAction func(rn *routerNode, span *request.Span) rc := rn.config @@ -168,16 +184,19 @@ func chooseUnmatchPolicy(rn *routerNode) (func(rn *routerNode, span *request.Spa case UnmatchPath: unmatchAction = setUnmatchToPath case UnmatchHeuristic: - classifierCfg := clusterurl.DefaultConfig() - if rc.WildcardChar != "" { - classifierCfg.ReplaceWith = rc.WildcardChar[0] - } - classifier, err := clusterurl.NewClusterURLClassifier(classifierCfg) + classifier, err := makeHeuristicClassifier(rc) if err != nil { - return nil, fmt.Errorf("chooseUnmatchPolicy: unable to create cluster URL classifier: %w", err) + return nil, err } rn.classifier = classifier unmatchAction = classifyFromPath + case UnmatchLowCardinality: + classifier, err := makeHeuristicClassifier(rc) + if err != nil { + return nil, err + } + rn.classifier = classifier + unmatchAction = classifyFromPathWithCappedCardinality default: slog.With("component", "RoutesProvider"). Warn("invalid 'unmatch' value in configuration, defaulting to '"+string(UnmatchDefault)+"'", @@ -203,6 +222,12 @@ func setUnmatchToPath(_ *routerNode, str *request.Span) { } func classifyFromPath(rc *routerNode, s *request.Span) { + if s.Route == "" && s.IsHTTPSpan() { + s.Route = rc.classifier.ClusterURL(s.Path) + } +} + +func classifyFromPathWithCappedCardinality(rc *routerNode, s *request.Span) { if s.Route == "" && s.IsHTTPSpan() { s.Route = rc.classifier.ClusterURL(s.Path) if s.Service.PathTrie != nil { diff --git a/pkg/transform/routes_test.go b/pkg/transform/routes_test.go index 5a48cd590f..9185bf7fca 100644 --- a/pkg/transform/routes_test.go +++ b/pkg/transform/routes_test.go @@ -123,9 +123,9 @@ func TestUnmatchedAuto(t *testing.T) { } } -func TestUnmatchedAutoSuper(t *testing.T) { +func TestUnmatchedAutoLowCardinality(t *testing.T) { trie := clusterurl.NewPathTrie(3) - for _, tc := range []UnmatchType{UnmatchHeuristic} { + for _, tc := range []UnmatchType{UnmatchLowCardinality} { t.Run(string(tc), func(t *testing.T) { input := msg.NewQueue[[]request.Span](msg.ChannelBufferLen(10)) output := msg.NewQueue[[]request.Span](msg.ChannelBufferLen(10)) From 252fc542bf69e01c34a15ef834077b733a850cc3 Mon Sep 17 00:00:00 2001 From: Nikola Grcevski Date: Fri, 7 Nov 2025 13:15:19 -0500 Subject: [PATCH 05/10] refactor lookup to be test method only --- .../transform/route/clusterurl/trie.go | 54 ------------ .../transform/route/clusterurl/trie_test.go | 86 +++++++++++++++---- pkg/transform/routes.go | 3 +- 3 files changed, 72 insertions(+), 71 deletions(-) diff --git a/pkg/internal/transform/route/clusterurl/trie.go b/pkg/internal/transform/route/clusterurl/trie.go index 05a6a012b9..9285178829 100644 --- a/pkg/internal/transform/route/clusterurl/trie.go +++ b/pkg/internal/transform/route/clusterurl/trie.go @@ -174,57 +174,3 @@ func (pt *PathTrie) mergeChildren(target, source *PathNode) { } } } - -// Lookup returns the normalized path for a given input path -// This is used to query existing paths without modifying the trie -func (pt *PathTrie) Lookup(path string) string { - pt.mu.RLock() - defer pt.mu.RUnlock() - - segments := strings.Split(strings.Trim(path, "/"), "/") - if len(segments) == 0 || (len(segments) == 1 && segments[0] == "") { - return path - } - - return pt.lookupSegments(segments) -} - -func (pt *PathTrie) lookupSegments(segments []string) string { - current := pt.root - result := make([]string, 0, len(segments)) - - for _, segment := range segments { - if segment == "" { - result = append(result, segment) - continue - } - - // If node is collapsed, use wildcard - if current.collapsed { - result = append(result, "*") - current = current.children["*"] - continue - } - - // Try to find exact match - child, exists := current.children[segment] - if !exists { - // No exact match, check for wildcard - if wildcardChild, hasWildcard := current.children["*"]; hasWildcard { - result = append(result, "*") - current = wildcardChild - continue - } - // Not found at all, return segment as-is and stop traversing - result = append(result, segment) - // Can't traverse further, append remaining segments - result = append(result, segments[len(result):]...) - break - } - - result = append(result, segment) - current = child - } - - return "/" + strings.Join(result, "/") -} diff --git a/pkg/internal/transform/route/clusterurl/trie_test.go b/pkg/internal/transform/route/clusterurl/trie_test.go index 578c30dfd6..78702b24ec 100644 --- a/pkg/internal/transform/route/clusterurl/trie_test.go +++ b/pkg/internal/transform/route/clusterurl/trie_test.go @@ -5,6 +5,7 @@ package clusterurl import ( "strconv" + "strings" "testing" "github.com/stretchr/testify/assert" @@ -26,7 +27,7 @@ func TestPathTrie_BasicInsertAndLookup(t *testing.T) { assert.Equal(t, "/test/*/files/*/test", result) // Lookup should now return collapsed path - result = trie.Lookup("test/anything-new/files/something/test") + result = trie.lookup("test/anything-new/files/something/test") assert.Equal(t, "/test/*/files/*/test", result) } @@ -42,7 +43,7 @@ func TestPathTrie_CardinalityThreshold(t *testing.T) { assert.Equal(t, "/api/*/users", trie.Insert("api/v4/users")) // Verify lookup uses collapsed path - assert.Equal(t, "/api/*/users", trie.Lookup("api/v999/users")) + assert.Equal(t, "/api/*/users", trie.lookup("api/v999/users")) } func TestPathTrie_CardinalitySecondaryThreshold(t *testing.T) { @@ -53,7 +54,7 @@ func TestPathTrie_CardinalitySecondaryThreshold(t *testing.T) { assert.Equal(t, "/api/v1/items/sports_car", trie.Insert("api/v1/items/sports_car")) assert.Equal(t, "/api/v1/items/t-shirt", trie.Insert("api/v1/items/t-shirt")) - assert.Equal(t, "/api/v1/items/t-shirt", trie.Lookup("api/v1/items/t-shirt")) + assert.Equal(t, "/api/v1/items/t-shirt", trie.lookup("api/v1/items/t-shirt")) // Add paths up to threshold assert.Equal(t, "/api/v1/customers", trie.Insert("api/v1/customers")) @@ -70,13 +71,13 @@ func TestPathTrie_CardinalitySecondaryThreshold(t *testing.T) { for i := range 3 { trie.Insert("api/v4/items" + strconv.Itoa(i)) } - assert.Equal(t, "/api/*/*", trie.Lookup("api/v4/items")) - assert.Equal(t, "/api/*/*/t-shirt", trie.Lookup("api/v4/items/t-shirt")) + assert.Equal(t, "/api/*/*", trie.lookup("api/v4/items")) + assert.Equal(t, "/api/*/*/t-shirt", trie.lookup("api/v4/items/t-shirt")) // trigger the third level collapse assert.Equal(t, "/api/*/*/*", trie.Insert("api/v1/customers/list")) - assert.Equal(t, "/api/*/*/*", trie.Lookup("api/v4/items/t-shirt")) + assert.Equal(t, "/api/*/*/*", trie.lookup("api/v4/items/t-shirt")) } func TestPathTrie_CascadingCollapse(t *testing.T) { @@ -101,7 +102,7 @@ func TestPathTrie_EmptyPath(t *testing.T) { trie := NewPathTrie(2) assert.Empty(t, trie.Insert("")) - assert.Empty(t, trie.Lookup("")) + assert.Empty(t, trie.lookup("")) } func TestPathTrie_SingleSegment(t *testing.T) { @@ -110,7 +111,7 @@ func TestPathTrie_SingleSegment(t *testing.T) { result := trie.Insert("test") assert.Equal(t, "/test", result) - result = trie.Lookup("test") + result = trie.lookup("test") assert.Equal(t, "/test", result) } @@ -122,15 +123,15 @@ func TestPathTrie_PreserveExistingPaths(t *testing.T) { trie.Insert("api/users/456") // Before collapse, lookups should return exact matches - assert.Equal(t, "/api/users/123", trie.Lookup("api/users/123")) - assert.Equal(t, "/api/users/456", trie.Lookup("api/users/456")) + assert.Equal(t, "/api/users/123", trie.lookup("api/users/123")) + assert.Equal(t, "/api/users/456", trie.lookup("api/users/456")) // Trigger collapse trie.Insert("api/users/789") // After collapse, all should use wildcard - assert.Equal(t, "/api/users/*", trie.Lookup("api/users/123")) - assert.Equal(t, "/api/users/*", trie.Lookup("api/users/999")) + assert.Equal(t, "/api/users/*", trie.lookup("api/users/123")) + assert.Equal(t, "/api/users/*", trie.lookup("api/users/999")) } func TestPathTrie_ComplexPaths(t *testing.T) { @@ -147,13 +148,13 @@ func TestPathTrie_ComplexPaths(t *testing.T) { } // Should not collapse yet (cardinality = 3, threshold = 3) - result := trie.Lookup(paths[0]) + result := trie.lookup(paths[0]) assert.Contains(t, result, "bar-attach-generic-product-apjkmyp") // Fourth path should trigger collapse trie.Insert("bar/test/test/fourth-product/files/version-def/test") - result = trie.Lookup("bar/test/test/any-product/files/any-version/test") + result = trie.lookup("bar/test/test/any-product/files/any-version/test") assert.Equal(t, "/bar/test/test/*/files/*/test", result) } @@ -188,6 +189,61 @@ func BenchmarkPathTrie_Lookup(b *testing.B) { b.ResetTimer() for i := 0; i < b.N; i++ { - trie.Lookup("api/v1/users/999/posts/888") + trie.lookup("api/v1/users/999/posts/888") } } + +// lookup returns the normalized path for a given input path +// This is used to query existing paths without modifying the trie +// At the moment this is only used in testing. +func (pt *PathTrie) lookup(path string) string { + pt.mu.RLock() + defer pt.mu.RUnlock() + + segments := strings.Split(strings.Trim(path, "/"), "/") + if len(segments) == 0 || (len(segments) == 1 && segments[0] == "") { + return path + } + + return pt.lookupSegments(segments) +} + +func (pt *PathTrie) lookupSegments(segments []string) string { + current := pt.root + result := make([]string, 0, len(segments)) + + for _, segment := range segments { + if segment == "" { + result = append(result, segment) + continue + } + + // If node is collapsed, use wildcard + if current.collapsed { + result = append(result, "*") + current = current.children["*"] + continue + } + + // Try to find exact match + child, exists := current.children[segment] + if !exists { + // No exact match, check for wildcard + if wildcardChild, hasWildcard := current.children["*"]; hasWildcard { + result = append(result, "*") + current = wildcardChild + continue + } + // Not found at all, return segment as-is and stop traversing + result = append(result, segment) + // Can't traverse further, append remaining segments + result = append(result, segments[len(result):]...) + break + } + + result = append(result, segment) + current = child + } + + return "/" + strings.Join(result, "/") +} diff --git a/pkg/transform/routes.go b/pkg/transform/routes.go index da1fcf4d9b..c186590786 100644 --- a/pkg/transform/routes.go +++ b/pkg/transform/routes.go @@ -231,8 +231,7 @@ func classifyFromPathWithCappedCardinality(rc *routerNode, s *request.Span) { if s.Route == "" && s.IsHTTPSpan() { s.Route = rc.classifier.ClusterURL(s.Path) if s.Service.PathTrie != nil { - s.Service.PathTrie.Insert(s.Route) - s.Route = s.Service.PathTrie.Lookup(s.Route) + s.Route = s.Service.PathTrie.Insert(s.Route) } } } From 5443102960f264469f88fc15323f5362e045597c Mon Sep 17 00:00:00 2001 From: Nikola Grcevski Date: Fri, 7 Nov 2025 13:38:15 -0500 Subject: [PATCH 06/10] add cleanup --- .../transform/route/clusterurl/trie.go | 27 +++++++++++++++++++ .../transform/route/clusterurl/trie_test.go | 27 +++++++++---------- 2 files changed, 40 insertions(+), 14 deletions(-) diff --git a/pkg/internal/transform/route/clusterurl/trie.go b/pkg/internal/transform/route/clusterurl/trie.go index 9285178829..a3efa65757 100644 --- a/pkg/internal/transform/route/clusterurl/trie.go +++ b/pkg/internal/transform/route/clusterurl/trie.go @@ -45,12 +45,39 @@ func NewPathTrie(maxCardinality int) *PathTrie { } } +func isHTTPOp(op string) bool { + return op == "GET" || op == "POST" || op == "PATCH" || op == "DELETE" || op == "OPTIONS" || op == "HEAD" +} + +func (pt *PathTrie) cleanup(path string) string { + i := strings.Index(path, "?") + if i >= 0 { + path = path[:i] + } + + if path == "" || path[0] == '/' { + return path + } + + i = strings.Index(path, " ") + if i > 0 { + op := path[:i] + if isHTTPOp(op) && i < len(path) { + return path[i+1:] + } + } + + return path +} + // Insert adds a path to the trie and returns the normalized path // If a segment exceeds maxCardinality, it collapses to "*" func (pt *PathTrie) Insert(path string) string { pt.mu.Lock() defer pt.mu.Unlock() + path = pt.cleanup(path) + segments := strings.Split(strings.Trim(path, "/"), "/") if len(segments) == 0 || (len(segments) == 1 && segments[0] == "") { return path diff --git a/pkg/internal/transform/route/clusterurl/trie_test.go b/pkg/internal/transform/route/clusterurl/trie_test.go index 78702b24ec..3ecd7bde43 100644 --- a/pkg/internal/transform/route/clusterurl/trie_test.go +++ b/pkg/internal/transform/route/clusterurl/trie_test.go @@ -158,6 +158,19 @@ func TestPathTrie_ComplexPaths(t *testing.T) { assert.Equal(t, "/bar/test/test/*/files/*/test", result) } +func TestPathTrie_Weird(t *testing.T) { + trie := NewPathTrie(100) + + // In case we get paths without cleaned up HTTP path + assert.Equal(t, "/attach", trie.Insert("/attach?session_id=ddfsdsf&track_id=sjdklnfldsn")) + assert.Equal(t, "/user_space", trie.Insert("GET /user_space?kernel_space")) + + // Non-HTTP + assert.Equal(t, "/MET /user_space", trie.Insert("MET /user_space?kernel_space")) + assert.Equal(t, "/MET ", trie.Insert("MET ")) + assert.Equal(t, "/MET", trie.Insert("MET")) +} + func BenchmarkPathTrie_Insert(b *testing.B) { trie := NewPathTrie(10) @@ -179,20 +192,6 @@ func BenchmarkPathTrie_Insert(b *testing.B) { } } -func BenchmarkPathTrie_Lookup(b *testing.B) { - trie := NewPathTrie(10) - - // Pre-populate trie - for i := 0; i < 100; i++ { - trie.Insert("api/v1/users/" + strconv.Itoa(i) + "/posts/456") - } - - b.ResetTimer() - for i := 0; i < b.N; i++ { - trie.lookup("api/v1/users/999/posts/888") - } -} - // lookup returns the normalized path for a given input path // This is used to query existing paths without modifying the trie // At the moment this is only used in testing. From bca1044d449222a8283ea6aa43aeaf2070058f0f Mon Sep 17 00:00:00 2001 From: Nikola Grcevski Date: Fri, 7 Nov 2025 13:53:04 -0500 Subject: [PATCH 07/10] add integration test --- .../configs/obi-config-no-route-lc.yml | 13 +++++ internal/test/integration/red_test.go | 47 +++++++++++++++++++ internal/test/integration/suites_test.go | 10 ++++ 3 files changed, 70 insertions(+) create mode 100644 internal/test/integration/configs/obi-config-no-route-lc.yml diff --git a/internal/test/integration/configs/obi-config-no-route-lc.yml b/internal/test/integration/configs/obi-config-no-route-lc.yml new file mode 100644 index 0000000000..f2e6695822 --- /dev/null +++ b/internal/test/integration/configs/obi-config-no-route-lc.yml @@ -0,0 +1,13 @@ +routes: + ignored_patterns: + - /metrics + unmatched: low-cardinality + max_path_segment_cardinality: 3 +otel_metrics_export: + endpoint: http://otelcol:4318 +otel_traces_export: + endpoint: http://jaeger:4318 +attributes: + select: + "*": + include: ["*"] diff --git a/internal/test/integration/red_test.go b/internal/test/integration/red_test.go index f2f15b08ca..d9b6348329 100644 --- a/internal/test/integration/red_test.go +++ b/internal/test/integration/red_test.go @@ -881,6 +881,42 @@ func testREDMetricsForHTTPLibraryNoRoute(t *testing.T, url, svcName string) { require.Empty(t, results) } +func testREDMetricsForHTTPLibraryNoRouteLowCardinality(t *testing.T, url, svcName string) { + validNames := []string{"user", "customer", "test", "option", "metric"} + + // Call 3 times the instrumented service, forcing it to: + // - take at least 30ms to respond + // - returning a 404 code + for i := 0; i < 3; i++ { + for _, s := range validNames { + ti.DoHTTPGet(t, url+"/api/"+s+"?delay=30ms&status=404", 404) + } + } + + // Eventually, Prometheus would make this query visible + pq := prom.Client{HostPort: prometheusHostPort} + var results []prom.Result + test.Eventually(t, testTimeout, func(t require.TestingT) { + var err error + results, err = pq.Query(`http_server_request_duration_seconds_count{` + + `http_request_method="GET",` + + `http_response_status_code="404",` + + `service_namespace="integration-test",` + + `service_name="` + svcName + `",` + + `http_route="/api/*"}`) + require.NoError(t, err) + // check duration_count has 3 calls and all the arguments + enoughPromResults(t, results) + val := totalPromCount(t, results) + assert.LessOrEqual(t, 3, val) + if len(results) > 0 { + res := results[0] + addr := res.Metric["client_address"] + assert.NotNil(t, addr) + } + }) +} + func testREDMetricsHTTPNoRoute(t *testing.T) { for _, testCaseURL := range []string{ instrumentedServiceGorillaURL, @@ -892,6 +928,17 @@ func testREDMetricsHTTPNoRoute(t *testing.T) { } } +func testREDMetricsHTTPNoRouteLowCardinality(t *testing.T) { + for _, testCaseURL := range []string{ + instrumentedServiceStdURL, + } { + t.Run(testCaseURL, func(t *testing.T) { + waitForTestComponents(t, testCaseURL) + testREDMetricsForHTTPLibraryNoRouteLowCardinality(t, testCaseURL, "testserver") + }) + } +} + func testREDMetricsUnsupportedHTTP(t *testing.T) { for _, testCaseURL := range []string{ instrumentedServiceStdURL, diff --git a/internal/test/integration/suites_test.go b/internal/test/integration/suites_test.go index 71fbfeaa77..d52461ca29 100644 --- a/internal/test/integration/suites_test.go +++ b/internal/test/integration/suites_test.go @@ -625,6 +625,16 @@ func TestSuiteNoRoutes(t *testing.T) { require.NoError(t, compose.Close()) } +func TestSuiteNoRoutesLowCardinality(t *testing.T) { + compose, err := docker.ComposeSuite("docker-compose.yml", path.Join(pathOutput, "test-suite-no-routes-low-cardinality.log")) + require.NoError(t, err) + + compose.Env = append(compose.Env, "INSTRUMENTER_CONFIG_SUFFIX=-no-route-lc") + require.NoError(t, compose.Up()) + t.Run("RED metrics", testREDMetricsHTTPNoRouteLowCardinality) + require.NoError(t, compose.Close()) +} + func TestSuite_Elixir(t *testing.T) { compose, err := docker.ComposeSuite("docker-compose-elixir.yml", path.Join(pathOutput, "test-suite-elixir.log")) require.NoError(t, err) From fc1088b31fa2740cc7a8a463f6d4a80f08bd2f42 Mon Sep 17 00:00:00 2001 From: Nikola Grcevski Date: Fri, 7 Nov 2025 14:00:12 -0500 Subject: [PATCH 08/10] add the integration test routes to show validity with cluster url --- pkg/internal/transform/route/clusterurl/cluster_test.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/pkg/internal/transform/route/clusterurl/cluster_test.go b/pkg/internal/transform/route/clusterurl/cluster_test.go index a978dd590c..34789bf8ef 100644 --- a/pkg/internal/transform/route/clusterurl/cluster_test.go +++ b/pkg/internal/transform/route/clusterurl/cluster_test.go @@ -66,6 +66,11 @@ func TestClusterURL(t *testing.T) { assert.Equal(t, "/*", csf.ClusterURL("/1#")) assert.Equal(t, "a", csf.ClusterURL("a#")) assert.Equal(t, "/a/b/c/d/e/f/g/h/i", csf.ClusterURL("/a/b/c/d/e/f/g/h/i/j")) + assert.Equal(t, "/api/user", csf.ClusterURL("/api/user")) + assert.Equal(t, "/api/customer", csf.ClusterURL("/api/customer")) + assert.Equal(t, "/api/test", csf.ClusterURL("/api/test")) + assert.Equal(t, "/api/option", csf.ClusterURL("/api/option")) + assert.Equal(t, "/api/metric", csf.ClusterURL("/api/metric")) } func BenchmarkClusterURLWithCache(b *testing.B) { From 3c4ed91405d7dd0eb36a8dbc1060114aac61f325 Mon Sep 17 00:00:00 2001 From: Nikola Grcevski Date: Mon, 10 Nov 2025 19:23:01 -0500 Subject: [PATCH 09/10] allow the wildcard to be configurable --- pkg/appolly/discover/typer.go | 7 ++++- .../transform/route/clusterurl/cluster.go | 2 -- .../transform/route/clusterurl/trie.go | 30 ++++++++++--------- .../transform/route/clusterurl/trie_test.go | 20 ++++++------- pkg/transform/routes_test.go | 2 +- 5 files changed, 33 insertions(+), 28 deletions(-) diff --git a/pkg/appolly/discover/typer.go b/pkg/appolly/discover/typer.go index e31e6dfd0c..e8017cdd1b 100644 --- a/pkg/appolly/discover/typer.go +++ b/pkg/appolly/discover/typer.go @@ -118,6 +118,11 @@ func makeServiceAttrs(processMatch *ProcessMatch, routesCfg *transform.RoutesCon } } + var wildcard = byte('*') + if routesCfg.WildcardChar != "" { + wildcard = routesCfg.WildcardChar[0] + } + s := svc.Attrs{ UID: svc.UID{ Name: name, @@ -126,7 +131,7 @@ func makeServiceAttrs(processMatch *ProcessMatch, routesCfg *transform.RoutesCon ProcPID: processMatch.Process.Pid, ExportModes: exportModes, Sampler: samplerFromConfig(samplerConfig), - PathTrie: clusterurl.NewPathTrie(routesCfg.MaxPathSegmentCardinality), + PathTrie: clusterurl.NewPathTrie(routesCfg.MaxPathSegmentCardinality, wildcard), } if routesConfig != nil { diff --git a/pkg/internal/transform/route/clusterurl/cluster.go b/pkg/internal/transform/route/clusterurl/cluster.go index 800bc1a772..805ca886ba 100644 --- a/pkg/internal/transform/route/clusterurl/cluster.go +++ b/pkg/internal/transform/route/clusterurl/cluster.go @@ -52,8 +52,6 @@ func NewClusterURLClassifier(config *Config) (*ClusterURLClassifier, error) { validCharTable[c] = true } - validCharTable[' '] = true - return &ClusterURLClassifier{ classifier: classifier, cache: cache, diff --git a/pkg/internal/transform/route/clusterurl/trie.go b/pkg/internal/transform/route/clusterurl/trie.go index a3efa65757..593fad56ad 100644 --- a/pkg/internal/transform/route/clusterurl/trie.go +++ b/pkg/internal/transform/route/clusterurl/trie.go @@ -32,16 +32,18 @@ type PathTrie struct { root *PathNode maxCardinality int mu sync.RWMutex + replaceWith string } // NewPathTrie creates a new path trie with the given max cardinality -func NewPathTrie(maxCardinality int) *PathTrie { +func NewPathTrie(maxCardinality int, replacement byte) *PathTrie { return &PathTrie{ root: &PathNode{ segment: "", children: make(map[string]*PathNode), }, maxCardinality: maxCardinality, + replaceWith: string(replacement), } } @@ -98,16 +100,16 @@ func (pt *PathTrie) insertSegments(segments []string) string { // If current node is already collapsed, all children become wildcards if current.collapsed { - result = append(result, "*") + result = append(result, pt.replaceWith) // Continue with the wildcard child - if current.children["*"] == nil { - current.children["*"] = &PathNode{ - segment: "*", + if current.children[pt.replaceWith] == nil { + current.children[pt.replaceWith] = &PathNode{ + segment: pt.replaceWith, children: make(map[string]*PathNode), isWildcard: true, } } - current = current.children["*"] + current = current.children[pt.replaceWith] continue } @@ -119,8 +121,8 @@ func (pt *PathTrie) insertSegments(segments []string) string { if current.cardinality >= pt.maxCardinality { // Collapse this level pt.collapseNode(current) - result = append(result, "*") - current = current.children["*"] + result = append(result, pt.replaceWith) + current = current.children[pt.replaceWith] continue } @@ -135,8 +137,8 @@ func (pt *PathTrie) insertSegments(segments []string) string { // Check if we just hit the threshold if current.cardinality > pt.maxCardinality { pt.collapseNode(current) - result = append(result, "*") - current = current.children["*"] + result = append(result, pt.replaceWith) + current = current.children[pt.replaceWith] continue } } @@ -158,10 +160,10 @@ func (pt *PathTrie) collapseNode(node *PathNode) { node.collapsed = true // Create or get wildcard node - wildcardNode, hasWildcard := node.children["*"] + wildcardNode, hasWildcard := node.children[pt.replaceWith] if !hasWildcard { wildcardNode = &PathNode{ - segment: "*", + segment: pt.replaceWith, children: make(map[string]*PathNode), isWildcard: true, } @@ -169,7 +171,7 @@ func (pt *PathTrie) collapseNode(node *PathNode) { // Merge all children into the wildcard node for segment, child := range node.children { - if segment == "*" { + if segment == pt.replaceWith { continue // Skip the wildcard itself } pt.mergeChildren(wildcardNode, child) @@ -177,7 +179,7 @@ func (pt *PathTrie) collapseNode(node *PathNode) { // Replace all children with just the wildcard node.children = map[string]*PathNode{ - "*": wildcardNode, + pt.replaceWith: wildcardNode, } node.cardinality = 1 diff --git a/pkg/internal/transform/route/clusterurl/trie_test.go b/pkg/internal/transform/route/clusterurl/trie_test.go index 3ecd7bde43..640bbefe7a 100644 --- a/pkg/internal/transform/route/clusterurl/trie_test.go +++ b/pkg/internal/transform/route/clusterurl/trie_test.go @@ -12,7 +12,7 @@ import ( ) func TestPathTrie_BasicInsertAndLookup(t *testing.T) { - trie := NewPathTrie(2) + trie := NewPathTrie(2, '*') // Insert first path result := trie.Insert("test/bar-attach-generic-product-apjkmyp/files/multi-test-version-jwbCm/test") @@ -32,7 +32,7 @@ func TestPathTrie_BasicInsertAndLookup(t *testing.T) { } func TestPathTrie_CardinalityThreshold(t *testing.T) { - trie := NewPathTrie(3) + trie := NewPathTrie(3, '*') // Add paths up to threshold assert.Equal(t, "/api/v1/users", trie.Insert("api/v1/users")) @@ -47,7 +47,7 @@ func TestPathTrie_CardinalityThreshold(t *testing.T) { } func TestPathTrie_CardinalitySecondaryThreshold(t *testing.T) { - trie := NewPathTrie(3) + trie := NewPathTrie(3, '*') // Add paths up to threshold assert.Equal(t, "/api/v1/items/teddy_bear", trie.Insert("api/v1/items/teddy_bear")) @@ -81,7 +81,7 @@ func TestPathTrie_CardinalitySecondaryThreshold(t *testing.T) { } func TestPathTrie_CascadingCollapse(t *testing.T) { - trie := NewPathTrie(2) + trie := NewPathTrie(2, '*') // Build tree: /root/child1/grandchild1 // /root/child1/grandchild2 @@ -99,14 +99,14 @@ func TestPathTrie_CascadingCollapse(t *testing.T) { } func TestPathTrie_EmptyPath(t *testing.T) { - trie := NewPathTrie(2) + trie := NewPathTrie(2, '*') assert.Empty(t, trie.Insert("")) assert.Empty(t, trie.lookup("")) } func TestPathTrie_SingleSegment(t *testing.T) { - trie := NewPathTrie(2) + trie := NewPathTrie(2, '*') result := trie.Insert("test") assert.Equal(t, "/test", result) @@ -116,7 +116,7 @@ func TestPathTrie_SingleSegment(t *testing.T) { } func TestPathTrie_PreserveExistingPaths(t *testing.T) { - trie := NewPathTrie(2) + trie := NewPathTrie(2, '*') // Insert paths trie.Insert("api/users/123") @@ -135,7 +135,7 @@ func TestPathTrie_PreserveExistingPaths(t *testing.T) { } func TestPathTrie_ComplexPaths(t *testing.T) { - trie := NewPathTrie(3) + trie := NewPathTrie(3, '*') paths := []string{ "bar/test/test/bar-attach-generic-product-apjkmyp/files/multi-test-version-jwbCm/test", @@ -159,7 +159,7 @@ func TestPathTrie_ComplexPaths(t *testing.T) { } func TestPathTrie_Weird(t *testing.T) { - trie := NewPathTrie(100) + trie := NewPathTrie(100, '*') // In case we get paths without cleaned up HTTP path assert.Equal(t, "/attach", trie.Insert("/attach?session_id=ddfsdsf&track_id=sjdklnfldsn")) @@ -172,7 +172,7 @@ func TestPathTrie_Weird(t *testing.T) { } func BenchmarkPathTrie_Insert(b *testing.B) { - trie := NewPathTrie(10) + trie := NewPathTrie(10, '*') paths := []string{ "/users/fdklsd/j4elk/23993/job/2", diff --git a/pkg/transform/routes_test.go b/pkg/transform/routes_test.go index 9185bf7fca..c9347ef0b3 100644 --- a/pkg/transform/routes_test.go +++ b/pkg/transform/routes_test.go @@ -124,7 +124,7 @@ func TestUnmatchedAuto(t *testing.T) { } func TestUnmatchedAutoLowCardinality(t *testing.T) { - trie := clusterurl.NewPathTrie(3) + trie := clusterurl.NewPathTrie(3, '*') for _, tc := range []UnmatchType{UnmatchLowCardinality} { t.Run(string(tc), func(t *testing.T) { input := msg.NewQueue[[]request.Span](msg.ChannelBufferLen(10)) From f86621885e2a51f55842a57dbc765e7f44d69cc2 Mon Sep 17 00:00:00 2001 From: Nikola Grcevski Date: Mon, 10 Nov 2025 19:24:38 -0500 Subject: [PATCH 10/10] fix format --- internal/tools/tools.go | 3 ++- pkg/appolly/discover/typer.go | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/internal/tools/tools.go b/internal/tools/tools.go index da18df77cc..99ad13ef31 100644 --- a/internal/tools/tools.go +++ b/internal/tools/tools.go @@ -11,8 +11,9 @@ import ( _ "github.com/google/go-licenses/v2" _ "github.com/grafana/go-offsets-tracker/cmd/go-offsets-tracker" _ "github.com/onsi/ginkgo/v2/ginkgo" - _ "go.opentelemetry.io/build-tools/multimod" _ "gotest.tools/gotestsum" _ "sigs.k8s.io/controller-runtime/tools/setup-envtest" _ "sigs.k8s.io/kind" + + _ "go.opentelemetry.io/build-tools/multimod" ) diff --git a/pkg/appolly/discover/typer.go b/pkg/appolly/discover/typer.go index e8017cdd1b..3960b83251 100644 --- a/pkg/appolly/discover/typer.go +++ b/pkg/appolly/discover/typer.go @@ -118,7 +118,7 @@ func makeServiceAttrs(processMatch *ProcessMatch, routesCfg *transform.RoutesCon } } - var wildcard = byte('*') + wildcard := byte('*') if routesCfg.WildcardChar != "" { wildcard = routesCfg.WildcardChar[0] }