[processor/spanmetrics] use LRU cache for metricKeyToDimensions

CHANGELOG.md

@@ -17,6 +17,8 @@
 
 - `lokiexporter`: add complete log record to body (#6619)
 
+- `spanmetricproccessor`: use an LRU cache for the cached Dimensions key-value pairs (#2179)
+
 ## v0.41.0
 
 ## 🛑 Breaking changes 🛑

processor/spanmetricsprocessor/README.md

@@ -53,7 +53,9 @@ The following settings can be optionally configured:
 
   If the `name`d attribute is missing in the span, the optional provided `default` is used.
 
-  If no `default` is provided, this dimension will be **omitted** from the metric. 
+  If no `default` is provided, this dimension will be **omitted** from the metric.
+- `dimensions_cache_size`: the max items number of `metric_key_to_dimensions_cache`. If not provided, will
+  use default value size `1000`.
 - `aggregation_temporality`: Defines the aggregation temporality of the generated metrics. 
   One of either `AGGREGATION_TEMPORALITY_CUMULATIVE` or `AGGREGATION_TEMPORALITY_DELTA`.
   - Default: `AGGREGATION_TEMPORALITY_CUMULATIVE`
@@ -93,6 +95,7 @@ processors:
       - name: http.method
         default: GET
       - name: http.status_code
+    dimensions_cache_size: 1000
     aggregation_temporality: "AGGREGATION_TEMPORALITY_DELTA"     
 
 exporters:

processor/spanmetricsprocessor/config.go

@@ -52,6 +52,11 @@ type Config struct {
 	// https://github.com/open-telemetry/opentelemetry-collector/blob/main/model/semconv/opentelemetry.go.
 	Dimensions []Dimension `mapstructure:"dimensions"`
 
+	// DimensionsCacheSize defines the size of cache for storing Dimensions, which helps to avoid cache memory growing
+	// indefinitely over the lifetime of the collector.
+	// Optional. See defaultDimensionsCacheSize in processor.go for the default value.
+	DimensionsCacheSize int `mapstructure:"dimensions_cache_size"`
+
 	AggregationTemporality string `mapstructure:"aggregation_temporality"`
 }
 

processor/spanmetricsprocessor/config_test.go

@@ -41,10 +41,21 @@ func TestLoadConfig(t *testing.T) {
 		wantMetricsExporter         string
 		wantLatencyHistogramBuckets []time.Duration
 		wantDimensions              []Dimension
+		wantDimensionsCacheSize     int
 		wantAggregationTemporality  string
 	}{
-		{configFile: "config-2-pipelines.yaml", wantMetricsExporter: "prometheus", wantAggregationTemporality: cumulative},
-		{configFile: "config-3-pipelines.yaml", wantMetricsExporter: "otlp/spanmetrics", wantAggregationTemporality: cumulative},
+		{
+			configFile:                 "config-2-pipelines.yaml",
+			wantMetricsExporter:        "prometheus",
+			wantAggregationTemporality: cumulative,
+			wantDimensionsCacheSize:    500,
+		},
+		{
+			configFile:                 "config-3-pipelines.yaml",
+			wantMetricsExporter:        "otlp/spanmetrics",
+			wantAggregationTemporality: cumulative,
+			wantDimensionsCacheSize:    defaultDimensionsCacheSize,
+		},
 		{
 			configFile:          "config-full.yaml",
 			wantMetricsExporter: "otlp/spanmetrics",
@@ -61,6 +72,7 @@ func TestLoadConfig(t *testing.T) {
 				{"http.method", &defaultMethod},
 				{"http.status_code", nil},
 			},
+			wantDimensionsCacheSize:    1500,
 			wantAggregationTemporality: delta,
 		},
 	}
@@ -92,6 +104,7 @@ func TestLoadConfig(t *testing.T) {
 					MetricsExporter:         tc.wantMetricsExporter,
 					LatencyHistogramBuckets: tc.wantLatencyHistogramBuckets,
 					Dimensions:              tc.wantDimensions,
+					DimensionsCacheSize:     tc.wantDimensionsCacheSize,
 					AggregationTemporality:  tc.wantAggregationTemporality,
 				},
 				cfg.Processors[config.NewComponentID(typeStr)],

processor/spanmetricsprocessor/factory.go

@@ -41,6 +41,7 @@ func createDefaultConfig() config.Processor {
 	return &Config{
 		ProcessorSettings:      config.NewProcessorSettings(config.NewComponentID(typeStr)),
 		AggregationTemporality: "AGGREGATION_TEMPORALITY_CUMULATIVE",
+		DimensionsCacheSize:    defaultDimensionsCacheSize,
 	}
 }
 

processor/spanmetricsprocessor/go.mod

@@ -3,6 +3,7 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanm
 go 1.17
 
 require (
+	github.com/hashicorp/golang-lru v0.5.4
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/jaegerexporter v0.41.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusexporter v0.41.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/jaegerreceiver v0.41.0

processor/spanmetricsprocessor/internal/cache/cache.go

@@ -0,0 +1,79 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cache // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanmetricsprocessor/internal/cache"
+
+import (
+	"sync"
+
+	lru "github.com/hashicorp/golang-lru"
+)
+
+// Cache consists of an LRU cache and the evicted items from the LRU cache.
+// This data structure makes sure all the cached items can be retrieved either from the LRU cache or the evictedItems
+// map. In spanmetricsprocessor's use case, we need to hold all the items during the current processing step for
+// building the metrics. The evicted items can/should be safely removed once the metrics are built from the current
+// batch of spans.
+type Cache struct {
+	*lru.Cache
+	rw           *sync.RWMutex
+	evictedItems map[interface{}]interface{}
+}
+
+// NewCache creates a Cache.
+func NewCache(size int) (*Cache, error) {
+	evictedItems := make(map[interface{}]interface{})
+	rw := new(sync.RWMutex)
+	lruCache, err := lru.NewWithEvict(size, func(key interface{}, value interface{}) {
+		rw.Lock()
+		evictedItems[key] = value
+		rw.Unlock()
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &Cache{
+		Cache:        lruCache,
+		evictedItems: evictedItems,
+		rw:           rw,
+	}, nil
+}
+
+// RemoveEvictedItems cleans all the evicted items.
+func (c *Cache) RemoveEvictedItems() {
+	c.rw.Lock()
+	// we need to keep the original pointer to evictedItems map as it is used in the closure of lru.NewWithEvict
+	for k := range c.evictedItems {
+		delete(c.evictedItems, k)
+	}
+	c.rw.Unlock()
+}
+
+// Get retrieves an item from the LRU cache or evicted items.
+func (c *Cache) Get(key interface{}) (interface{}, bool) {
+	if val, ok := c.Cache.Get(key); ok {
+		return val, ok
+	}
+	c.rw.RLock()
+	val, ok := c.evictedItems[key]
+	c.rw.RUnlock()
+	return val, ok
+}
+
+// Purge removes all the items from the LRU cache and evicted items.
+func (c *Cache) Purge() {
+	c.Cache.Purge()
+	c.RemoveEvictedItems()
+}

processor/spanmetricsprocessor/internal/cache/cache_test.go

@@ -0,0 +1,212 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cache
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewCache(t *testing.T) {
+	type args struct {
+		size int
+	}
+	tests := []struct {
+		name    string
+		args    args
+		wantErr bool
+	}{
+		{
+			name: "create a new Cache with length 10",
+			args: args{
+				size: 10,
+			},
+			wantErr: false,
+		},
+		{
+			name: "create a new Cache with length -1",
+			args: args{
+				size: -1,
+			},
+			wantErr: true,
+		},
+		{
+			name: "create a new Cache with length 0",
+			args: args{
+				size: 0,
+			},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			_, err := NewCache(tt.args.size)
+			if tt.wantErr {
+				assert.Error(t, err)
+				return
+			}
+			assert.NoError(t, err)
+		})
+	}
+}
+
+func TestCache_Get(t *testing.T) {
+	tests := []struct {
+		name         string
+		lruCache     func() *Cache
+		evictedItems map[interface{}]interface{}
+		key          interface{}
+		wantValue    interface{}
+		wantOk       bool
+	}{
+		{
+			name: "if key is not found in LRUCache, will get key from evictedItems",
+			lruCache: func() *Cache {
+				cache, _ := NewCache(1)
+				cache.evictedItems["key"] = "val"
+				return cache
+			},
+			key:       "key",
+			wantValue: "val",
+			wantOk:    true,
+		},
+		{
+			name: "if key is found in LRUCache, return the found item",
+			lruCache: func() *Cache {
+				cache, _ := NewCache(1)
+				cache.Add("key", "val_from_LRU")
+				cache.evictedItems["key"] = "val_from_evicted_items"
+				return cache
+			},
+			key:       "key",
+			wantValue: "val_from_LRU",
+			wantOk:    true,
+		},
+		{
+			name: "if key is not found either in LRUCache or evicted items, return nothing",
+			lruCache: func() *Cache {
+				cache, _ := NewCache(1)
+				return cache
+			},
+			key:       "key",
+			wantValue: nil,
+			wantOk:    false,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			c := tt.lruCache()
+			gotValue, gotOk := c.Get(tt.key)
+			if !assert.Equal(t, gotValue, tt.wantValue) {
+				t.Errorf("Get() gotValue = %v, want %v", gotValue, tt.wantValue)
+			}
+			if gotOk != tt.wantOk {
+				t.Errorf("Get() gotOk = %v, want %v", gotOk, tt.wantOk)
+			}
+		})
+	}
+}
+
+func TestCache_RemoveEvictedItems(t *testing.T) {
+	tests := []struct {
+		name     string
+		lruCache func() (*Cache, error)
+	}{
+		{
+			name: "no panic when there is no evicted item to remove",
+			lruCache: func() (*Cache, error) {
+				return NewCache(1)
+			},
+		},
+		{
+			name: "evicted items should be removed",
+			lruCache: func() (*Cache, error) {
+				cache, err := NewCache(1)
+				if err != nil {
+					return nil, err
+				}
+				cache.evictedItems["key0"] = "val0"
+				cache.evictedItems["key1"] = "val1"
+				return cache, nil
+			},
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			cache, err := tt.lruCache()
+			assert.NoError(t, err)
+			cache.RemoveEvictedItems()
+			assert.Empty(t, cache.evictedItems)
+		})
+	}
+}
+
+func TestCache_PurgeItems(t *testing.T) {
+	tests := []struct {
+		name     string
+		lruCache func() (*Cache, error)
+	}{
+		{
+			name: "no panic when there is no item to remove",
+			lruCache: func() (*Cache, error) {
+				return NewCache(1)
+			},
+		},
+		{
+			name: "remove items from the lru cache",
+			lruCache: func() (*Cache, error) {
+				cache, err := NewCache(1)
+				if err != nil {
+					return nil, err
+				}
+				cache.evictedItems["key0"] = "val0"
+				cache.evictedItems["key1"] = "val1"
+				return cache, nil
+			},
+		},
+		{
+			name: "remove all the items from lru cache and the evicted items",
+			lruCache: func() (*Cache, error) {
+				cache, err := NewCache(10)
+				if err != nil {
+					return nil, err
+				}
+				cache.Add("key", "val")
+				cache.Add("key2", "val2")
+				cache.evictedItems["key0"] = "val0"
+				cache.evictedItems["key1"] = "val1"
+				return cache, nil
+			},
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			cache, err := tt.lruCache()
+			assert.NoError(t, err)
+			cache.Purge()
+			assert.Zero(t, cache.Len())
+			assert.Empty(t, cache.evictedItems)
+		})
+	}
+}