Limit cardinality of Span Metrics (open-telemetry#545)

* first untested implementation of spannamelimiter

* add testable LRU cache

* expirable LRU: extra tests + eviction callback

* unit tested implementation

* some improvements in thelimiter. pipe tests not yet working

* Fixed pipeline and some changes ready for review

* Update pkg/components/helpers/cache/expirable_lru_test.go

Co-authored-by: Tyler Yahn <[email protected]>

* Update pkg/components/helpers/cache/expirable_lru_test.go

Co-authored-by: Tyler Yahn <[email protected]>

* Update pkg/components/helpers/cache/expirable_lru_test.go

Co-authored-by: Tyler Yahn <[email protected]>

* Update pkg/components/helpers/cache/expirable_lru_test.go

Co-authored-by: Tyler Yahn <[email protected]>

* Update pkg/components/helpers/cache/expirable_lru_test.go

Co-authored-by: Tyler Yahn <[email protected]>

* Update pkg/components/helpers/cache/expirable_lru_test.go

Co-authored-by: Tyler Yahn <[email protected]>

* Update pkg/components/helpers/cache/expirable_lru.go

Co-authored-by: Tyler Yahn <[email protected]>

* Update pkg/components/helpers/cache/expirable_lru.go

Co-authored-by: Tyler Yahn <[email protected]>

* Update pkg/app/request/span.go

Co-authored-by: Tyler Yahn <[email protected]>

* Update pkg/components/helpers/cache/expirable_lru.go

Co-authored-by: Tyler Yahn <[email protected]>

* fix linter complaint

* Fix pipeline

---------

Co-authored-by: Tyler Yahn <[email protected]>

Author: mariomac

devdocs/pipeline-map.md

@@ -32,10 +32,14 @@ flowchart TD
         KD:::optional --> NR
         NR(Name resolver):::optional --> AF
         
-        AF(Attributes filter):::optional --> OTELM(OTEL<br/> metrics<br/> exporter):::optional
-        AF --> OTELT(OTEL<br/> traces<br/> exporter):::optional
-        AF --> PROM(Prometheus<br/>HTTP<br/>endpoint):::optional
-        AF --> ALLOY(Alloy<br/>connector):::optional
+        AF(Attributes filter):::optional --> OTELT(OTEL/ALLOY<br/> traces<br/> exporter):::optional
+
+        
+        AF --> IPD(Unknown IP<br/>dropper):::optional
+        IPD --> SNCL(Span Name<br/>cardinality<br/>limiter)
+        SNCL --> OTELRM(OTEL<br/>RED metrics<br/> exporter):::optional
+        SNCL --> OTELSM(OTEL<br/>span/svc graph<br/>metrics<br/> exporter):::optional
+        SNCL --> PROM(Prometheus<br/>HTTP<br/>endpoint):::optional
     end
     CU -.-> |New PIDs| KDB
     KDB(KubeDatabase):::optional <-.- | Aggregated & indexed Pod info | KD

pkg/app/request/span.go

@@ -181,6 +181,10 @@ type Span struct {
 	SQLCommand     string         `json:"-"`
 	SQLError       *SQLError      `json:"-"`
 	MessagingInfo  *MessagingInfo `json:"-"`
+
+	// OverrideTraceName is set under some conditions, like spanmetrics reaching the maximum
+	// cardinality for trace names.
+	OverrideTraceName string `json:"-"`
 }
 
 func (s *Span) Inside(parent *Span) bool {
@@ -533,6 +537,9 @@ func (s *Span) ServiceGraphKind() string {
 }
 
 func (s *Span) TraceName() string {
+	if s.OverrideTraceName != "" {
+		return s.OverrideTraceName
+	}
 	switch s.Type {
 	case EventTypeHTTP, EventTypeHTTPClient:
 		name := s.Method
@@ -561,11 +568,11 @@ func (s *Span) TraceName() string {
 		if s.Path == "" {
 			return s.Method
 		}
-		return fmt.Sprintf("%s %s", s.Method, s.Path)
+		return s.Method + " " + s.Path
 	case EventTypeMongoClient:
 		if s.Path != "" && s.Method != "" {
 			// TODO for database operations like listCollections, we need to use s.DbNamespace instead of s.Path
-			return fmt.Sprintf("%s %s", s.Method, s.Path)
+			return s.Method + " " + s.Path
 		}
 		if s.Path != "" {
 			return s.Path

pkg/components/helpers/cache/expirable_lru.go

@@ -0,0 +1,109 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package cache
+
+import (
+	"container/list"
+	"time"
+)
+
+// ExpirableLRU cache. It is not safe for concurrent access.
+type ExpirableLRU[K comparable, V any] struct {
+	ttl           time.Duration
+	ll            *list.List
+	cache         map[K]*list.Element
+	evictCallBack func(K, V)
+}
+
+type expirableEntry[K comparable, V any] struct {
+	key        K
+	value      V
+	lastAccess time.Time
+}
+
+// ExpirableCacheOpt defines a functional option for configuring an ExpirableLRU cache.
+type ExpirableCacheOpt[K comparable, V any] func(*ExpirableLRU[K, V])
+
+// WithEvictCallBack sets a callback function to be called whenever an entry is evicted
+// from the [ExpirableLRU] cache.
+func WithEvictCallBack[K comparable, V any](cb func(K, V)) ExpirableCacheOpt[K, V] {
+	return func(c *ExpirableLRU[K, V]) {
+		c.evictCallBack = cb
+	}
+}
+
+// NewExpirableLRU creates a new [ExpirableLRU] whose entries older than the provided TTL are removed
+// only when the ExpireAll method is explicitly called.
+func NewExpirableLRU[K comparable, V any](ttl time.Duration, opts ...ExpirableCacheOpt[K, V]) *ExpirableLRU[K, V] {
+	lru := &ExpirableLRU[K, V]{
+		ttl:           ttl,
+		ll:            list.New(),
+		cache:         map[K]*list.Element{},
+		evictCallBack: func(K, V) {},
+	}
+	for _, opt := range opts {
+		opt(lru)
+	}
+	return lru
+}
+
+// Put a value into the cache.
+func (c *ExpirableLRU[K, V]) Put(key K, value V) {
+	timeNow := time.Now()
+	if ee, ok := c.cache[key]; ok {
+		c.ll.MoveToFront(ee)
+		entry := ee.Value.(*expirableEntry[K, V])
+		entry.value = value
+		entry.lastAccess = timeNow
+		return
+	}
+	ele := c.ll.PushFront(&expirableEntry[K, V]{key: key, value: value, lastAccess: timeNow})
+	c.cache[key] = ele
+}
+
+// Get looks up a key's value from the cache.
+func (c *ExpirableLRU[K, V]) Get(key K) (value V, ok bool) {
+	if ele, hit := c.cache[key]; hit {
+		c.ll.MoveToFront(ele)
+		ele.Value.(*expirableEntry[K, V]).lastAccess = time.Now()
+		return ele.Value.(*expirableEntry[K, V]).value, true
+	}
+	return
+}
+
+// Remove the provided key from the cache.
+func (c *ExpirableLRU[K, V]) Remove(key K) {
+	if ele, hit := c.cache[key]; hit {
+		c.removeElement(ele)
+	}
+}
+
+func (c *ExpirableLRU[K, V]) removeElement(e *list.Element) {
+	kv := e.Value.(*expirableEntry[K, V])
+	c.evictCallBack(kv.key, kv.value)
+	c.ll.Remove(e)
+	delete(c.cache, kv.key)
+}
+
+// ExpireAll removes all the entries that are older than the cache TTL.
+// Returns the number of entries removed.
+func (c *ExpirableLRU[K, V]) ExpireAll() int {
+	now := time.Now()
+	removed := 0
+	for older := c.ll.Back(); c.expired(older, now); older = c.ll.Back() {
+		removed++
+		c.removeElement(older)
+	}
+	return removed
+}
+
+func (c *ExpirableLRU[K, V]) expired(elem *list.Element, now time.Time) bool {
+	return elem != nil &&
+		elem.Value.(*expirableEntry[K, V]).lastAccess.Add(c.ttl).Before(now)
+}
+
+// Len returns the number of elements in the cache.
+func (c *ExpirableLRU[K, V]) Len() int {
+	return c.ll.Len()
+}

pkg/components/helpers/cache/expirable_lru_test.go

@@ -0,0 +1,196 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package cache
+
+import (
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestExpirableLRU_PutGetRemove(t *testing.T) {
+	lru := NewExpirableLRU[string, string](5 * time.Minute)
+
+	lru.Put("foo", "bar")
+	v, ok := lru.Get("foo")
+	assert.True(t, ok)
+	assert.Equal(t, "bar", v)
+	assert.Equal(t, 1, lru.Len())
+
+	v, ok = lru.Get("baz")
+	assert.False(t, ok)
+	assert.Empty(t, v)
+
+	lru.Put("baz", "bae")
+	v, ok = lru.Get("baz")
+	assert.True(t, ok)
+	assert.Equal(t, "bae", v)
+	assert.Equal(t, 2, lru.Len())
+
+	lru.Remove("foo")
+	_, ok = lru.Get("foo")
+	assert.False(t, ok)
+	assert.Equal(t, 1, lru.Len())
+}
+
+func TestExpirableLRU_ExpireAll(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		cache := NewExpirableLRU[string, string](5 * time.Minute)
+
+		// Add items at different times
+		cache.Put("key1", "value1")
+		// Advance time by 2 minutes
+		time.Sleep(2 * time.Minute)
+		cache.Put("key2", "value2")
+		// Advance time by 2 more minutes (key1 is now 4 minutes old, key2 is 2 minutes old)
+		time.Sleep(2 * time.Minute)
+		cache.Put("key3", "value3")
+
+		assert.Zero(t, cache.ExpireAll())
+
+		// All items should be there(none have exceeded 5-minute TTL)
+		assert.Equal(t, 3, cache.Len())
+
+		v, ok := cache.Get("key1")
+		require.True(t, ok, "Expected to find key1")
+		assert.Equal(t, "value1", v)
+		time.Sleep(2 * time.Minute)
+
+		v, ok = cache.Get("key2")
+		require.True(t, ok, "Expected to find key2")
+		assert.Equal(t, "value2", v)
+		time.Sleep(2 * time.Minute)
+
+		v, ok = cache.Get("key3")
+		require.True(t, ok, "Expected to find key3")
+		assert.Equal(t, "value3", v)
+		time.Sleep(2 * time.Minute)
+
+		assert.Equal(t, 1, cache.ExpireAll())
+
+		// key1 is 6 minutes old, should expire while others are still valid
+		assert.Equal(t, 2, cache.Len())
+
+		_, ok = cache.Get("key1")
+		assert.False(t, ok, "Expected key1 to be expired")
+		v, ok = cache.Get("key2")
+		require.True(t, ok, "Expected to find key2")
+		assert.Equal(t, "value2", v)
+		v, ok = cache.Get("key3")
+		require.True(t, ok, "Expected to find key3")
+		assert.Equal(t, "value3", v)
+
+		// Advance age by two minute:
+		//   - "key1": removed (expired)
+		//   - "key2": 6 minutes old
+		//   - "key3": 4 minutes old
+		time.Sleep(2 * time.Minute)
+
+		// add key 4
+		cache.Put("key4", "value4")
+		// Advance age by four minute:
+		//   - "key1": removed (expired)
+		//   - "key2": 10 minutes old
+		//   - "key3": 8 minutes old
+		//   - "key4": 2 minutes old
+		time.Sleep(4 * time.Minute)
+
+		// all the keys but key4 should be expired
+		assert.Equal(t, 2, cache.ExpireAll())
+		assert.Equal(t, 1, cache.Len())
+
+		_, ok = cache.Get("key1")
+		assert.False(t, ok, "Expected key1 to be expired")
+		_, ok = cache.Get("key2")
+		assert.False(t, ok, "Expected key2 to be expired")
+		_, ok = cache.Get("key3")
+		assert.False(t, ok, "Expected key3 to be expired")
+		v, ok = cache.Get("key4")
+		require.True(t, ok, "Expected to find key4")
+		assert.Equal(t, "value4", v)
+
+		// Advance age by four minute:
+		//   - "key1": removed (expired)
+		//   - "key2": removed (expired)
+		//   - "key3": removed (expired)
+		//   - "key4": 6 minutes old
+		time.Sleep(4 * time.Minute)
+
+		// a re-added key should not expire
+		cache.Put("key1", "value1")
+		// Advance age by two minute:
+		//   - "key1": two minutes old
+		//   - "key2": removed (expired)
+		//   - "key3": removed (expired)
+		//   - "key4": 10 minutes old
+		time.Sleep(2 * time.Minute)
+
+		assert.Equal(t, 1, cache.ExpireAll())
+
+		assert.Equal(t, 1, cache.Len())
+		v, ok = cache.Get("key1")
+		require.True(t, ok, "Expected to find key1")
+		assert.Equal(t, "value1", v)
+		_, ok = cache.Get("key2")
+		assert.False(t, ok, "Expected key2 to be expired")
+		_, ok = cache.Get("key3")
+		assert.False(t, ok, "expected key 3 to be expired")
+		_, ok = cache.Get("key4")
+		assert.False(t, ok, "expected key 4 to be expired")
+	})
+}
+
+func TestWithEvictCallBack(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		var evictedKeys []int
+		var evictedVals []string
+		cache := NewExpirableLRU[int, string](5*time.Minute,
+			WithEvictCallBack(func(k int, v string) {
+				evictedKeys = append(evictedKeys, k)
+				evictedVals = append(evictedVals, v)
+			}))
+
+		cache.Put(1, "one")
+		time.Sleep(1 * time.Minute)
+		cache.Put(2, "two")
+		time.Sleep(2 * time.Minute)
+		cache.Put(3, "three")
+
+		assert.Zero(t, cache.ExpireAll())
+		assert.Empty(t, evictedKeys)
+		assert.Empty(t, evictedVals)
+
+		time.Sleep(4 * time.Minute)
+
+		assert.Equal(t, 2, cache.ExpireAll())
+		assert.Equal(t, []int{1, 2}, evictedKeys)
+		assert.Equal(t, []string{"one", "two"}, evictedVals)
+	})
+}
+
+func TestPutAlsoUpdatesTTL(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		cache := NewExpirableLRU[int, string](5 * time.Minute)
+		cache.Put(1, "one")
+
+		time.Sleep(3 * time.Minute)
+		cache.Put(2, "two")
+
+		assert.Zero(t, cache.ExpireAll())
+
+		time.Sleep(3 * time.Minute)
+		cache.Put(1, "another")
+		time.Sleep(3 * time.Minute)
+
+		assert.Equal(t, 1, cache.ExpireAll())
+		assert.Equal(t, 1, cache.Len())
+		_, ok := cache.Get(1)
+		assert.True(t, ok, "Expected to find key1")
+		_, ok = cache.Get(2)
+		assert.False(t, ok, "Expected key2 to be expired")
+	})
+}