[exporter/elasticsearch] Feed the symbolization queues

exporter/elasticsearchexporter/exporter.go

@@ -589,6 +589,9 @@ func (e *elasticsearchExporter) pushProfileRecord(
 			return eventsSession.Add(ctx, index, docID, "", buf, nil, docappender.ActionCreate)
 		case otelserializer.ExecutablesIndex:
 			return executablesSession.Add(ctx, index, docID, "", buf, nil, docappender.ActionUpdate)
+		case otelserializer.ExecutablesSymQueueIndex, otelserializer.LeafFramesSymQueueIndex:
+			// These regular indices have a low write-frequency and can share the executablesSession.
+			return executablesSession.Add(ctx, index, docID, "", buf, nil, docappender.ActionCreate)
 		default:
 			return defaultSession.Add(ctx, index, docID, "", buf, nil, docappender.ActionCreate)
 		}

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

@@ -18,6 +18,9 @@ const (
 	StackTraceIndex  = "profiling-stacktraces"
 	StackFrameIndex  = "profiling-stackframes"
 	ExecutablesIndex = "profiling-executables"
+
+	ExecutablesSymQueueIndex = "profiling-sq-executables"
+	LeafFramesSymQueueIndex  = "profiling-sq-leafframes"
 )
 
 // SerializeProfile serializes a profile and calls the `pushData` callback for each generated document.
@@ -64,6 +67,18 @@ func SerializeProfile(resource pcommon.Resource, scope pcommon.InstrumentationSc
 				return err
 			}
 		}
+
+		for _, frame := range payload.UnsymbolizedLeafFrames {
+			if err = pushDataAsJSON(frame, frame.DocID, LeafFramesSymQueueIndex); err != nil {
+				return err
+			}
+		}
+
+		for _, executable := range payload.UnsymbolizedExecutables {
+			if err = pushDataAsJSON(executable, executable.DocID, ExecutablesSymQueueIndex); err != nil {
+				return err
+			}
+		}
 	}
 	return nil
 }

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

@@ -19,6 +19,8 @@ import (
 )
 
 func TestSerializeProfile(t *testing.T) {
+	nowTime := time.Now()
+	nowTimeStr := nowTime.Format(time.RFC3339Nano)
 	tests := []struct {
 		name              string
 		profileCustomizer func(resource pcommon.Resource, scope pcommon.InstrumentationScope, record pprofile.Profile)
@@ -80,6 +82,20 @@ func TestSerializeProfile(t *testing.T) {
 					"scripted_upsert": true,
 					"upsert":          map[string]any{},
 				},
+				{
+					"Stacktrace.frame.id":     []any{"YA3K_koRAADyvzjEk_X7kgAAAAAAAABv"},
+					"Symbolization.retries":   json.Number("0"),
+					"Symbolization.time.next": nowTimeStr,
+					"Time.created":            nowTimeStr,
+					"ecs.version":             serializeprofiles.EcsVersionString,
+				},
+				{
+					"Executable.file.id":      []any{"YA3K_koRAADyvzjEk_X7kg"},
+					"Symbolization.retries":   json.Number("0"),
+					"Symbolization.time.next": nowTimeStr,
+					"Time.created":            nowTimeStr,
+					"ecs.version":             serializeprofiles.EcsVersionString,
+				},
 				{
 					"@timestamp":          "1970-01-01T00:00:00Z",
 					"Stacktrace.count":    json.Number("1"),
@@ -100,6 +116,14 @@ func TestSerializeProfile(t *testing.T) {
 			tt.profileCustomizer(resource.Resource(), scope.Scope(), profile)
 			profiles.MarkReadOnly()
 
+			nowOld := serializeprofiles.Now
+			t.Cleanup(func() {
+				serializeprofiles.Now = nowOld
+			})
+			serializeprofiles.Now = func() time.Time {
+				return nowTime
+			}
+
 			buf := []*bytes.Buffer{}
 			err := SerializeProfile(resource.Resource(), scope.Scope(), profile, func(b *bytes.Buffer, _ string, _ string) error {
 				buf = append(buf, b)

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

@@ -4,7 +4,7 @@
 package serializeprofiles // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/elasticsearchexporter/internal/serializer/otelserializer/serializeprofiles"
 
 import (
-	"go.opentelemetry.io/ebpf-profiler/libpf"
+	"time"
 )
 
 // EcsVersionString is the value for the `ecs.version` metrics field.
@@ -27,7 +27,8 @@ type StackPayload struct {
 	StackFrames     []StackFrame
 	Executables     []ExeMetadata
 
-	UnsymbolizedLeafFrames []libpf.FrameID
+	UnsymbolizedLeafFrames  []UnsymbolizedLeafFrame
+	UnsymbolizedExecutables []UnsymbolizedExecutable
 }
 
 // StackTraceEvent represents a stacktrace event serializable into ES.
@@ -130,3 +131,25 @@ func NewExeMetadata(docID string, lastSeen uint32, buildID, fileName string) Exe
 		},
 	}
 }
+
+// UnsymbolizedExecutable represents an array of executable FileIDs written into the
+// executable symbolization queue index.
+type UnsymbolizedExecutable struct {
+	EcsVersion
+	DocID   string    `json:"-"`
+	FileID  []string  `json:"Executable.file.id"`
+	Created time.Time `json:"Time.created"`
+	Next    time.Time `json:"Symbolization.time.next"`
+	Retries int       `json:"Symbolization.retries"`
+}
+
+// UnsymbolizedLeafFrame represents an array of frame IDs written into the
+// leaf frame symbolization queue index.
+type UnsymbolizedLeafFrame struct {
+	EcsVersion
+	DocID   string    `json:"-"`
+	FrameID []string  `json:"Stacktrace.frame.id"`
+	Created time.Time `json:"Time.created"`
+	Next    time.Time `json:"Symbolization.time.next"`
+	Retries int       `json:"Symbolization.retries"`
+}

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

@@ -18,6 +18,8 @@ import (
 	semconv "go.opentelemetry.io/otel/semconv/v1.25.0"
 )
 
+var Now = time.Now
+
 // Transform transforms a [pprofile.Profile] into our own
 // representation, for ingestion into Elasticsearch
 func Transform(resource pcommon.Resource, scope pcommon.InstrumentationScope, profile pprofile.Profile) ([]StackPayload, error) {
@@ -77,7 +79,8 @@ func checkProfileType(profile pprofile.Profile) error {
 // stackPayloads creates a slice of StackPayloads from the given ResourceProfiles,
 // ScopeProfiles, and ProfileContainer.
 func stackPayloads(resource pcommon.Resource, scope pcommon.InstrumentationScope, profile pprofile.Profile) ([]StackPayload, error) {
-	unsymbolizedLeafFrames := make([]libpf.FrameID, 0, profile.Sample().Len())
+	unsymbolizedLeafFramesSet := make(map[libpf.FrameID]struct{}, profile.Sample().Len())
+	unsymbolizedExecutablesSet := make(map[libpf.FileID]struct{})
 	stackPayload := make([]StackPayload, 0, profile.Sample().Len())
 
 	hostMetadata := newHostMetadata(resource, scope, profile)
@@ -108,7 +111,23 @@ func stackPayloads(resource pcommon.Resource, scope pcommon.InstrumentationScope
 		})
 
 		if !isFrameSymbolized(frames[len(frames)-1]) && leafFrame != nil {
-			unsymbolizedLeafFrames = append(unsymbolizedLeafFrames, *leafFrame)
+			unsymbolizedLeafFramesSet[*leafFrame] = struct{}{}
+		}
+
+		for j := range frames {
+			if frameTypes[j].IsError() {
+				// Artificial error frames can't be symbolized.
+				continue
+			}
+			if isFrameSymbolized(frames[j]) {
+				// Skip interpreted frames and already symbolized native frames (kernel, Golang is planned).
+				continue
+			}
+			frameID, err := libpf.NewFrameIDFromString(frames[j].DocID)
+			if err != nil {
+				return nil, fmt.Errorf("stackPayloads: %w", err)
+			}
+			unsymbolizedExecutablesSet[frameID.FileID()] = struct{}{}
 		}
 
 		// Add one event per timestamp and its count value.
@@ -138,12 +157,45 @@ func stackPayloads(resource pcommon.Resource, scope pcommon.InstrumentationScope
 
 			stackPayload[0].Executables = exeMetadata
 		}
-		stackPayload[0].UnsymbolizedLeafFrames = unsymbolizedLeafFrames
+		stackPayload[0].UnsymbolizedLeafFrames = unsymbolizedLeafFrames(unsymbolizedLeafFramesSet)
+		stackPayload[0].UnsymbolizedExecutables = unsymbolizedExecutables(unsymbolizedExecutablesSet)
 	}
 
 	return stackPayload, nil
 }
 
+func unsymbolizedExecutables(executables map[libpf.FileID]struct{}) []UnsymbolizedExecutable {
+	nowTime := Now()
+	unsymbolized := make([]UnsymbolizedExecutable, 0, len(executables))
+	for fileID := range executables {
+		unsymbolized = append(unsymbolized, UnsymbolizedExecutable{
+			EcsVersion: EcsVersion{V: EcsVersionString},
+			DocID:      fileID.Base64(),
+			FileID:     []string{fileID.Base64()},
+			Created:    nowTime,
+			Next:       nowTime,
+			Retries:    0,
+		})
+	}
+	return unsymbolized
+}
+
+func unsymbolizedLeafFrames(frameIDs map[libpf.FrameID]struct{}) []UnsymbolizedLeafFrame {
+	nowTime := Now()
+	unsymbolized := make([]UnsymbolizedLeafFrame, 0, len(frameIDs))
+	for frameID := range frameIDs {
+		unsymbolized = append(unsymbolized, UnsymbolizedLeafFrame{
+			EcsVersion: EcsVersion{V: EcsVersionString},
+			DocID:      frameID.String(),
+			FrameID:    []string{frameID.String()},
+			Created:    nowTime,
+			Next:       nowTime,
+			Retries:    0,
+		})
+	}
+	return unsymbolized
+}
+
 // symbolizedFrames returns a slice of StackFrames that have symbols.
 func symbolizedFrames(frames []StackFrame) []StackFrame {
 	framesWithSymbols := make([]StackFrame, 0, len(frames))