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) 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/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..3960b83251 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 @@ -116,6 +118,11 @@ func makeServiceAttrs(processMatch *ProcessMatch) svc.Attrs { } } + wildcard := byte('*') + if routesCfg.WildcardChar != "" { + wildcard = routesCfg.WildcardChar[0] + } + s := svc.Attrs{ UID: svc.UID{ Name: name, @@ -124,6 +131,7 @@ func makeServiceAttrs(processMatch *ProcessMatch) svc.Attrs { ProcPID: processMatch.Process.Pid, ExportModes: exportModes, Sampler: samplerFromConfig(samplerConfig), + PathTrie: clusterurl.NewPathTrie(routesCfg.MaxPathSegmentCardinality, wildcard), } if routesConfig != nil { @@ -146,7 +154,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/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) { diff --git a/pkg/internal/transform/route/clusterurl/trie.go b/pkg/internal/transform/route/clusterurl/trie.go new file mode 100644 index 0000000000..593fad56ad --- /dev/null +++ b/pkg/internal/transform/route/clusterurl/trie.go @@ -0,0 +1,205 @@ +// 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 + replaceWith string +} + +// NewPathTrie creates a new path trie with the given max cardinality +func NewPathTrie(maxCardinality int, replacement byte) *PathTrie { + return &PathTrie{ + root: &PathNode{ + segment: "", + children: make(map[string]*PathNode), + }, + maxCardinality: maxCardinality, + replaceWith: string(replacement), + } +} + +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 + } + + 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, pt.replaceWith) + // Continue with the wildcard child + if current.children[pt.replaceWith] == nil { + current.children[pt.replaceWith] = &PathNode{ + segment: pt.replaceWith, + children: make(map[string]*PathNode), + isWildcard: true, + } + } + current = current.children[pt.replaceWith] + 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, pt.replaceWith) + current = current.children[pt.replaceWith] + 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, pt.replaceWith) + current = current.children[pt.replaceWith] + 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[pt.replaceWith] + if !hasWildcard { + wildcardNode = &PathNode{ + segment: pt.replaceWith, + children: make(map[string]*PathNode), + isWildcard: true, + } + } + + // Merge all children into the wildcard node + for segment, child := range node.children { + if segment == pt.replaceWith { + continue // Skip the wildcard itself + } + pt.mergeChildren(wildcardNode, child) + } + + // Replace all children with just the wildcard + node.children = map[string]*PathNode{ + pt.replaceWith: 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++ + } + } +} 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..640bbefe7a --- /dev/null +++ b/pkg/internal/transform/route/clusterurl/trie_test.go @@ -0,0 +1,248 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package clusterurl + +import ( + "strconv" + "strings" + "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_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/*/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 + 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, '*') + + // 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 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, '*') + + paths := []string{ + "/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", + "", + } + + for b.Loop() { + for i := 0; i < len(paths); i++ { + trie.Insert(paths[i]) + } + } +} + +// 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/obi/config.go b/pkg/obi/config.go index 18ffe481b3..a03cdc24be 100644 --- a/pkg/obi/config.go +++ b/pkg/obi/config.go @@ -181,8 +181,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 eeeb3c223d..2c28cde982 100644 --- a/pkg/obi/config_test.go +++ b/pkg/obi/config_test.go @@ -228,8 +228,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..c186590786 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 ) @@ -60,6 +63,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 { @@ -143,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 @@ -166,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)+"'", @@ -206,6 +227,15 @@ func classifyFromPath(rc *routerNode, s *request.Span) { } } +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.Route = s.Service.PathTrie.Insert(s.Route) + } + } +} + func setSpanIgnoreMode(mode IgnoreMode, s *request.Span) { switch mode { case IgnoreMetrics: diff --git a/pkg/transform/routes_test.go b/pkg/transform/routes_test.go index f8b52ba6ec..c9347ef0b3 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 TestUnmatchedAutoLowCardinality(t *testing.T) { + 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)) + 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))