elasticsearchexporter: Introduce LRU cache for profiles

.chloggen/elasticsearch-lru.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: elasticsearchexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Introduce LRU cache for profiles
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38606]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/elasticsearchexporter/go.mod

@@ -4,8 +4,10 @@ go 1.23.0
 
 require (
 	github.com/cenkalti/backoff/v4 v4.3.0
+	github.com/cespare/xxhash v1.1.0
 	github.com/elastic/go-docappender/v2 v2.9.0
 	github.com/elastic/go-elasticsearch/v8 v8.17.1
+	github.com/elastic/go-freelru v0.16.0
 	github.com/elastic/go-structform v0.0.12
 	github.com/klauspost/compress v1.18.0
 	github.com/lestrrat-go/strftime v1.1.0
@@ -44,7 +46,6 @@ require (
 	github.com/cilium/ebpf v0.16.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/elastic/elastic-transport-go/v8 v8.6.1 // indirect
-	github.com/elastic/go-freelru v0.16.0 // indirect
 	github.com/elastic/go-sysinfo v1.15.3 // indirect
 	github.com/elastic/go-windows v1.0.2 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect

exporter/elasticsearchexporter/integrationtest/go.mod

@@ -43,6 +43,7 @@ require (
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/cenkalti/backoff/v5 v5.0.2 // indirect
 	github.com/census-instrumentation/opencensus-proto v0.4.1 // indirect
+	github.com/cespare/xxhash v1.1.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cilium/ebpf v0.16.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect

exporter/elasticsearchexporter/internal/lru/lruset.go

@@ -0,0 +1,74 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package lru // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/elasticsearchexporter/internal/lru"
+
+import (
+	"time"
+
+	"github.com/cespare/xxhash"
+	"github.com/elastic/go-freelru"
+	"go.opentelemetry.io/ebpf-profiler/libpf/xsync"
+)
+
+type void struct{}
+
+func stringHashFn(s string) uint32 {
+	return uint32(xxhash.Sum64String(s))
+}
+
+// LockedLRUSet is the interface provided to the LRUSet once a lock has been
+// acquired.
+type LockedLRUSet interface {
+	// CheckAndAdd checks whether the entry is already stored in the cache, and
+	// adds it.
+	// It returns whether the entry should be excluded, as it was already present
+	// in cache.
+	CheckAndAdd(entry string) bool
+}
+
+// LRUSet is an LRU cache implementation that allows acquiring a lock, and
+// checking whether specific keys have already been stored.
+type LRUSet struct {
+	syncMu *xsync.RWMutex[*freelru.LRU[string, void]]
+}
+
+func (l *LRUSet) WithLock(fn func(LockedLRUSet) error) error {
+	if l == nil || l.syncMu == nil {
+		return fn(nilLockedLRUSet{})
+	}
+
+	lru := l.syncMu.WLock()
+	defer l.syncMu.WUnlock(&lru)
+
+	return fn(lockedLRUSet{*lru})
+}
+
+func NewLRUSet(size uint32, rollover time.Duration) (*LRUSet, error) {
+	lru, err := freelru.New[string, void](size, stringHashFn)
+	if err != nil {
+		return nil, err
+	}
+	lru.SetLifetime(rollover)
+
+	syncMu := xsync.NewRWMutex(lru)
+	return &LRUSet{syncMu: &syncMu}, nil
+}
+
+type lockedLRUSet struct {
+	lru *freelru.LRU[string, void]
+}
+
+func (l lockedLRUSet) CheckAndAdd(entry string) (excluded bool) {
+	if _, exclude := (l.lru).Get(entry); exclude {
+		return true
+	}
+	(l.lru).Add(entry, void{})
+	return false
+}
+
+type nilLockedLRUSet struct{}
+
+func (l nilLockedLRUSet) CheckAndAdd(string) bool {
+	return false
+}

exporter/elasticsearchexporter/internal/lru/lruset_test.go

@@ -0,0 +1,102 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package lru
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLRUSet(t *testing.T) {
+	cache, err := NewLRUSet(5, time.Minute)
+	require.NoError(t, err)
+
+	err = cache.WithLock(func(lock LockedLRUSet) error {
+		assert.False(t, lock.CheckAndAdd("a"))
+		assert.True(t, lock.CheckAndAdd("a"))
+		assert.False(t, lock.CheckAndAdd("b"))
+
+		assert.InDelta(t, 0.0, testing.AllocsPerRun(5, func() {
+			_ = lock.CheckAndAdd("c")
+		}), 0)
+
+		return nil
+	})
+
+	assert.NoError(t, err)
+}
+
+func TestLRUSetLifeTime(t *testing.T) {
+	const lifetime = 100 * time.Millisecond
+	cache, err := NewLRUSet(5, lifetime)
+	require.NoError(t, err)
+
+	err = cache.WithLock(func(lock LockedLRUSet) error {
+		assert.False(t, lock.CheckAndAdd("a"))
+		assert.True(t, lock.CheckAndAdd("a"))
+		return nil
+	})
+	require.NoError(t, err)
+
+	// Wait until cache item is expired.
+	time.Sleep(lifetime)
+	err = cache.WithLock(func(lock LockedLRUSet) error {
+		assert.False(t, lock.CheckAndAdd("a"))
+		assert.True(t, lock.CheckAndAdd("a"))
+		return nil
+	})
+	require.NoError(t, err)
+
+	// Wait 50% of the lifetime, so the item is not expired.
+	time.Sleep(lifetime / 2)
+	err = cache.WithLock(func(lock LockedLRUSet) error {
+		assert.True(t, lock.CheckAndAdd("a"))
+		return nil
+	})
+	require.NoError(t, err)
+
+	// Wait another 50% of the lifetime, so the item should be expired.
+	time.Sleep(lifetime / 2)
+	err = cache.WithLock(func(lock LockedLRUSet) error {
+		assert.False(t, lock.CheckAndAdd("a"))
+		return nil
+	})
+	require.NoError(t, err)
+}
+
+func TestNilLRUSet(t *testing.T) {
+	cache := &LRUSet{}
+
+	err := cache.WithLock(func(lock LockedLRUSet) error {
+		assert.False(t, lock.CheckAndAdd("a"))
+		assert.False(t, lock.CheckAndAdd("a"))
+		assert.False(t, lock.CheckAndAdd("b"))
+
+		assert.InDelta(t, 0.0, testing.AllocsPerRun(5, func() {
+			_ = lock.CheckAndAdd("c")
+		}), 0)
+
+		return nil
+	})
+
+	assert.NoError(t, err)
+}
+
+func BenchmarkLRUSetCheck(b *testing.B) {
+	cache, err := NewLRUSet(5, time.Minute)
+	require.NoError(b, err)
+
+	_ = cache.WithLock(func(lock LockedLRUSet) error {
+		b.ReportAllocs()
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			lock.CheckAndAdd("a")
+		}
+
+		return nil
+	})
+}

exporter/elasticsearchexporter/internal/serializer/otelserializer/logs_test.go

@@ -9,6 +9,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
 
@@ -185,8 +186,9 @@ func TestSerializeLog(t *testing.T) {
 			logs.MarkReadOnly()
 
 			var buf bytes.Buffer
-			ser := New()
-			err := ser.SerializeLog(resourceLogs.Resource(), "", scopeLogs.Scope(), "", record, elasticsearch.Index{}, &buf)
+			ser, err := New()
+			require.NoError(t, err)
+			err = ser.SerializeLog(resourceLogs.Resource(), "", scopeLogs.Scope(), "", record, elasticsearch.Index{}, &buf)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("Log() error = %v, wantErr %v", err, tt.wantErr)
 			}

exporter/elasticsearchexporter/internal/serializer/otelserializer/metrics_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/elasticsearchexporter/internal/datapoints"
@@ -33,8 +34,9 @@ func TestSerializeMetricsConflict(t *testing.T) {
 
 	var validationErrors []error
 	var buf bytes.Buffer
-	ser := New()
-	_, err := ser.SerializeMetrics(resourceMetrics.Resource(), "", scopeMetrics.Scope(), "", dataPoints, &validationErrors, elasticsearch.Index{}, &buf)
+	ser, err := New()
+	require.NoError(t, err)
+	_, err = ser.SerializeMetrics(resourceMetrics.Resource(), "", scopeMetrics.Scope(), "", dataPoints, &validationErrors, elasticsearch.Index{}, &buf)
 	if err != nil {
 		t.Errorf("Metrics() error = %v", err)
 	}

exporter/elasticsearchexporter/internal/serializer/otelserializer/otelserializer.go

@@ -3,9 +3,34 @@
 
 package otelserializer // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/elasticsearchexporter/internal/serializer/otelserializer"
 
-type Serializer struct{}
+import (
+	"sync"
+	"time"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/elasticsearchexporter/internal/lru"
+)
+
+const (
+	knownExecutablesCacheSize             = 16 * 1024
+	knownFramesCacheSize                  = 128 * 1024
+	knownTracesCacheSize                  = 128 * 1024
+	knownUnsymbolizedFramesCacheSize      = 128 * 1024
+	knownUnsymbolizedExecutablesCacheSize = 16 * 1024
+
+	minILMRolloverTime = 3 * time.Hour
+)
+
+type Serializer struct {
+	// Data cache for profiles
+	loadLRUsOnce                 sync.Once
+	knownTraces                  *lru.LRUSet
+	knownFrames                  *lru.LRUSet
+	knownExecutables             *lru.LRUSet
+	knownUnsymbolizedFrames      *lru.LRUSet
+	knownUnsymbolizedExecutables *lru.LRUSet
+}
 
 // New builds a new Serializer
-func New() *Serializer {
-	return &Serializer{}
+func New() (*Serializer, error) {
+	return &Serializer{}, nil
 }