benchmark: add benchmark processing large OTLP messages :dev:

Signed-off-by: Laurent Dufresne <laurent.dufresne@grafana.com>

Author: ldufr

pkg/distributor/otel_test.go

@@ -7,9 +7,11 @@ import (
 	"compress/gzip"
 	"context"
 	"encoding/json"
+	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"runtime"
 	"slices"
 	"strconv"
 	"strings"
@@ -1140,6 +1142,96 @@ func BenchmarkOTLPHandler(b *testing.B) {
 	})
 }
 
+func BenchmarkOTLPHandlerWithLargeMessage(b *testing.B) {
+	const NumberOfMetrics = 70
+	const NumberOfScopedMetrics = 25
+	const NumberOfResourceMetrics = 300
+	const stepDuration time.Duration = 10 * time.Second
+	const defaultDuration time.Duration = 60 * time.Second
+	now := time.Date(2020, time.October, 30, 23, 0, 0, 0, time.UTC)
+
+	createMetrics := func(mts pmetric.MetricSlice, count int, startTime time.Time, duration time.Duration) {
+		mts.EnsureCapacity(count)
+		for idx := range count {
+			mt := mts.AppendEmpty()
+			mt.SetName(fmt.Sprintf("metric-%d", idx))
+			datapoints := mt.SetEmptyGauge().DataPoints()
+
+			nDatapoints := int((int64(duration) + (int64(stepDuration) - 1)) / int64(stepDuration))
+			datapoints.EnsureCapacity(nDatapoints)
+
+			sampleTime := startTime
+			for range nDatapoints {
+				datapoint := datapoints.AppendEmpty()
+				datapoint.SetTimestamp(pcommon.NewTimestampFromTime(sampleTime))
+				attrs := datapoint.Attributes()
+				attrs.PutStr("route", "/hello")
+				attrs.PutStr("status", "200")
+				sampleTime = sampleTime.Add(stepDuration)
+			}
+		}
+	}
+
+	createManyScopedMetrics := func(rm pmetric.ResourceMetrics, count int) {
+		sms := rm.ScopeMetrics()
+		sms.EnsureCapacity(count)
+		for idx := range count {
+			scopeName := fmt.Sprintf("scope-%d", idx)
+
+			sm := sms.AppendEmpty()
+			scope := sm.Scope()
+			scope.SetName(scopeName)
+			attrs := scope.Attributes()
+			attrs.PutStr("package", scopeName)
+
+			metrics := sm.Metrics()
+			createMetrics(metrics, NumberOfMetrics, now, defaultDuration)
+		}
+	}
+
+	createResourceMetrics := func(md pmetric.Metrics, count int) {
+		rms := md.ResourceMetrics()
+		rms.EnsureCapacity(count)
+		for idx := range count {
+			rm := rms.AppendEmpty()
+			attrs := rm.Resource().Attributes()
+			attrs.PutStr("env", "dev")
+			attrs.PutStr("region", "us-east-1")
+			attrs.PutStr("pod", fmt.Sprintf("pod-%d", idx))
+			createManyScopedMetrics(rm, NumberOfScopedMetrics)
+		}
+	}
+
+	pushFunc := func(_ context.Context, pushReq *Request) error {
+		if _, err := pushReq.WriteRequest(); err != nil {
+			return err
+		}
+
+		pushReq.CleanUp()
+		return nil
+	}
+	limits := validation.MockDefaultOverrides()
+	handler := OTLPHandler(
+		200000000, nil, nil, limits, nil, nil,
+		RetryConfig{}, nil, pushFunc, nil, nil, log.NewNopLogger(),
+	)
+
+	b.Run("protobuf", func(b *testing.B) {
+		md := pmetric.NewMetrics()
+		createResourceMetrics(md, NumberOfResourceMetrics)
+		exportReq := pmetricotlp.NewExportRequestFromMetrics(md)
+		req := createOTLPProtoRequest(b, exportReq, "")
+
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			resp := httptest.NewRecorder()
+			handler.ServeHTTP(resp, req)
+			require.Equal(b, http.StatusOK, resp.Code)
+			req.Body.(*reusableReader).Reset()
+		}
+	})
+}
+
 func createOTLPProtoRequest(tb testing.TB, metricRequest pmetricotlp.ExportRequest, compression string) *http.Request {
 	tb.Helper()