[exporter/elasticsearch] Profiling: Store sampling frequency (#41017)

#### Description
Add support for profiles variable sampling frequencies by storing the
frequency with every profiling event.

#### Link to tracking issue
Fixes #40115

Author: rockdaboot

.chloggen/elasticsearchexporter-add-sampling-frequency.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: Support profiles variable sampling frequency.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [40115]
+
+# (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/internal/serializer/otelserializer/profile_test.go

@@ -104,6 +104,7 @@ func TestSerializeProfile(t *testing.T) {
 				pt := profile.PeriodType()
 				pt.SetTypeStrindex(2)
 				pt.SetUnitStrindex(3)
+				profile.SetPeriod(1e9 / 20)
 
 				profile.AttributeIndices().Append(2)
 
@@ -119,12 +120,13 @@ func TestSerializeProfile(t *testing.T) {
 					"ecs.version":            "1.12.0",
 				},
 				{
-					"@timestamp":          "1970-01-01T00:00:00Z",
-					"Stacktrace.count":    json.Number("1"),
-					"Stacktrace.id":       "02VzuClbpt_P3xxwox83Ng",
-					"ecs.version":         "1.12.0",
-					"host.id":             "localhost",
-					"process.thread.name": "",
+					"@timestamp":                    "1970-01-01T00:00:00Z",
+					"Stacktrace.count":              json.Number("1"),
+					"Stacktrace.sampling_frequency": json.Number("20"),
+					"Stacktrace.id":                 "02VzuClbpt_P3xxwox83Ng",
+					"ecs.version":                   "1.12.0",
+					"host.id":                       "localhost",
+					"process.thread.name":           "",
 				},
 				{
 					"script": map[string]any{

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

@@ -46,6 +46,7 @@ type StackTraceEvent struct {
 	ContainerName    string `json:"container.name,omitempty"`
 	K8sNamespaceName string `json:"k8s.namespace.name,omitempty"`
 	ThreadName       string `json:"process.thread.name"`
+	Frequency        int64  `json:"Stacktrace.sampling_frequency"`
 	Count            uint16 `json:"Stacktrace.count"`
 }
 

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

@@ -7,6 +7,7 @@ import (
 	"bytes"
 	"fmt"
 	"hash/fnv"
+	"math"
 	"strconv"
 	"strings"
 	"time"
@@ -83,6 +84,12 @@ func stackPayloads(dic pprofile.ProfilesDictionary, resource pcommon.Resource, s
 
 	hostMetadata := newHostMetadata(dic, resource, scope, profile)
 
+	frequency := int64(math.Round(1e9 / float64(profile.Period())))
+	if frequency <= 0 {
+		// The lowest sensical frequency is 1Hz.
+		frequency = 1
+	}
+
 	for i := 0; i < profile.Sample().Len(); i++ {
 		sample := profile.Sample().At(i)
 
@@ -99,7 +106,7 @@ func stackPayloads(dic pprofile.ProfilesDictionary, resource pcommon.Resource, s
 			return nil, fmt.Errorf("failed to create stacktrace ID: %w", err)
 		}
 
-		event := stackTraceEvent(dic, traceID, sample, hostMetadata)
+		event := stackTraceEvent(dic, traceID, sample, frequency, hostMetadata)
 
 		// Set the stacktrace and stackframes to the payload.
 		// The docs only need to be written once.
@@ -209,12 +216,13 @@ func isFrameSymbolized(frame StackFrame) bool {
 	return len(frame.FileName) > 0 || len(frame.FunctionName) > 0
 }
 
-func stackTraceEvent(dic pprofile.ProfilesDictionary, traceID string, sample pprofile.Sample, hostMetadata map[string]string) StackTraceEvent {
+func stackTraceEvent(dic pprofile.ProfilesDictionary, traceID string, sample pprofile.Sample, frequency int64, hostMetadata map[string]string) StackTraceEvent {
 	event := StackTraceEvent{
 		EcsVersion:   EcsVersion{V: EcsVersionString},
 		HostID:       hostMetadata[string(semconv.HostIDKey)],
 		StackTraceID: traceID,
 		Count:        1, // TODO: Check whether count can be dropped with nanosecond timestamps
+		Frequency:    frequency,
 	}
 
 	// Store event-specific attributes.

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

@@ -192,6 +192,7 @@ func TestTransform(t *testing.T) {
 
 				sp := rp.ScopeProfiles().AppendEmpty()
 				p := sp.Profiles().AppendEmpty()
+				p.SetPeriod(1e9 / 20)
 
 				st := p.SampleType().AppendEmpty()
 				st.SetTypeStrindex(2)
@@ -260,6 +261,7 @@ func TestTransform(t *testing.T) {
 						EcsVersion:   EcsVersion{V: EcsVersionString},
 						TimeStamp:    42000000000,
 						StackTraceID: wantedTraceID,
+						Frequency:    20,
 						Count:        1,
 					},
 				},
@@ -337,6 +339,7 @@ func TestStackPayloads(t *testing.T) {
 
 				sp := rp.ScopeProfiles().AppendEmpty()
 				p := sp.Profiles().AppendEmpty()
+				p.SetPeriod(1e9 / 20)
 
 				s := p.Sample().AppendEmpty()
 				s.TimestampsUnixNano().Append(1)
@@ -398,6 +401,7 @@ func TestStackPayloads(t *testing.T) {
 						EcsVersion:   EcsVersion{V: EcsVersionString},
 						TimeStamp:    1000000000,
 						StackTraceID: wantedTraceID,
+						Frequency:    20,
 						Count:        1,
 					},
 				},
@@ -442,6 +446,7 @@ func TestStackPayloads(t *testing.T) {
 
 				sp := rp.ScopeProfiles().AppendEmpty()
 				p := sp.Profiles().AppendEmpty()
+				p.SetPeriod(1e9 / 20)
 
 				s := p.Sample().AppendEmpty()
 				s.TimestampsUnixNano().Append(1)
@@ -503,6 +508,7 @@ func TestStackPayloads(t *testing.T) {
 						EcsVersion:   EcsVersion{V: EcsVersionString},
 						TimeStamp:    1000000000,
 						StackTraceID: wantedTraceID,
+						Frequency:    20,
 						Count:        2,
 					},
 				},
@@ -558,6 +564,7 @@ func TestStackTraceEvent(t *testing.T) {
 			wantEvent: StackTraceEvent{
 				EcsVersion:   EcsVersion{V: EcsVersionString},
 				StackTraceID: stacktraceIDBase64,
+				Frequency:    20,
 				Count:        1,
 			},
 		},
@@ -584,6 +591,7 @@ func TestStackTraceEvent(t *testing.T) {
 				EcsVersion:   EcsVersion{V: EcsVersionString},
 				TimeStamp:    1000000000000000000,
 				StackTraceID: stacktraceIDBase64,
+				Frequency:    20,
 				Count:        1,
 			},
 		},
@@ -608,6 +616,7 @@ func TestStackTraceEvent(t *testing.T) {
 			wantEvent: StackTraceEvent{
 				EcsVersion:   EcsVersion{V: EcsVersionString},
 				StackTraceID: stacktraceIDBase64,
+				Frequency:    20,
 				Count:        1,
 			},
 		},
@@ -649,6 +658,7 @@ func TestStackTraceEvent(t *testing.T) {
 				ContainerName: "my_container",
 				ThreadName:    "my_thread",
 				StackTraceID:  stacktraceIDBase64,
+				Frequency:     20,
 				Count:         1,
 			},
 		},
@@ -659,7 +669,7 @@ func TestStackTraceEvent(t *testing.T) {
 			p := rp.ScopeProfiles().At(0).Profiles().At(0)
 			s := p.Sample().At(0)
 
-			event := stackTraceEvent(dic, stacktraceIDBase64, s, map[string]string{})
+			event := stackTraceEvent(dic, stacktraceIDBase64, s, 20, map[string]string{})
 			event.TimeStamp = newUnixTime64(tt.timestamp)
 
 			assert.Equal(t, tt.wantEvent, event)