AGTMETRICS-398 Fix intake v3 validation request id header (#45566)

### What does this PR do?

Fix metrics v3 payloads validation request id header. It should have the same value for both v2 and v3 payloads that carry identical metrics.

### Motivation

Bugfix.

### Describe how you validated your changes

Tests.

### Additional Notes


Co-authored-by: vikentiy.fesunov <vikentiy.fesunov@datadoghq.com>

Author: vickenty

pkg/serializer/internal/metrics/pipeline.go

@@ -11,8 +11,6 @@ import (
 	"slices"
 	"strconv"
 
-	"github.com/google/uuid"
-
 	"github.com/DataDog/datadog-agent/comp/forwarder/defaultforwarder/resolver"
 	"github.com/DataDog/datadog-agent/comp/forwarder/defaultforwarder/transaction"
 )
@@ -75,21 +73,16 @@ type PipelineConfig struct {
 
 // PipelineDestination describes how to deliver a payload to the intake.
 type PipelineDestination struct {
-	Resolver             resolver.DomainResolver
-	Endpoint             transaction.Endpoint
-	AddValidationHeaders bool
+	Resolver          resolver.DomainResolver
+	Endpoint          transaction.Endpoint
+	ValidationBatchID string
 }
 
 type forwarder interface {
 	SubmitTransaction(*transaction.HTTPTransaction) error
 }
 
 func (dest *PipelineDestination) send(payloads transaction.BytesPayloads, forwarder forwarder, headers http.Header) error {
-	batchID, err := dest.maybeMakeBatchID()
-	if err != nil {
-		return err
-	}
-
 	domain := dest.Resolver.Resolve(dest.Endpoint)
 	for _, auth := range dest.Resolver.GetAuthorizers() {
 		for seq, payload := range payloads {
@@ -100,8 +93,8 @@ func (dest *PipelineDestination) send(payloads transaction.BytesPayloads, forwar
 			for key := range headers {
 				txn.Headers.Set(key, headers.Get(key))
 			}
-			if dest.AddValidationHeaders {
-				txn.Headers.Set("X-Metrics-Request-ID", batchID)
+			if dest.ValidationBatchID != "" {
+				txn.Headers.Set("X-Metrics-Request-ID", dest.ValidationBatchID)
 				txn.Headers.Set("X-Metrics-Request-Seq", strconv.Itoa(seq))
 				txn.Headers.Set("X-Metrics-Request-Len", strconv.Itoa(len(payloads)))
 			}
@@ -116,17 +109,6 @@ func (dest *PipelineDestination) send(payloads transaction.BytesPayloads, forwar
 	return nil
 }
 
-func (dest *PipelineDestination) maybeMakeBatchID() (string, error) {
-	if dest.AddValidationHeaders {
-		uuid, err := uuid.NewV7()
-		if err != nil {
-			return "", err
-		}
-		return uuid.String(), nil
-	}
-	return "", nil
-}
-
 // PipelineContext holds information needed during and after pipeline execution.
 type PipelineContext struct {
 	Destinations []PipelineDestination

pkg/serializer/internal/metrics/pipeline_test.go

@@ -46,13 +46,13 @@ func TestPipelineSendValidate(t *testing.T) {
 
 	ctx := PipelineContext{
 		Destinations: []PipelineDestination{{
-			Resolver:             res1,
-			Endpoint:             endpoints.SeriesEndpoint,
-			AddValidationHeaders: true,
+			Resolver:          res1,
+			Endpoint:          endpoints.SeriesEndpoint,
+			ValidationBatchID: "123",
 		}, {
-			Resolver:             res2,
-			Endpoint:             endpoints.SeriesEndpoint,
-			AddValidationHeaders: false,
+			Resolver:          res2,
+			Endpoint:          endpoints.SeriesEndpoint,
+			ValidationBatchID: "",
 		}}}
 
 	ctx.addPayload(transaction.NewBytesPayload([]byte{1}, 1))
@@ -68,14 +68,14 @@ func TestPipelineSendValidate(t *testing.T) {
 		func(t require.TestingT, _ int, txn *transaction.HTTPTransaction) {
 			require.Equal(t, "http://example.test", txn.Domain)
 			require.Equal(t, []byte{1}, txn.Payload.GetContent())
-			require.NotEmpty(t, txn.Headers.Get("x-metrics-request-id"))
+			require.Equal(t, "123", txn.Headers.Get("x-metrics-request-id"))
 			require.Equal(t, "0", txn.Headers.Get("x-metrics-request-seq"))
 			require.Equal(t, "2", txn.Headers.Get("x-metrics-request-len"))
 		},
 		func(t require.TestingT, _ int, txn *transaction.HTTPTransaction) {
 			require.Equal(t, "http://example.test", txn.Domain)
 			require.Equal(t, []byte{2}, txn.Payload.GetContent())
-			require.NotEmpty(t, txn.Headers.Get("x-metrics-request-id"))
+			require.Equal(t, "123", txn.Headers.Get("x-metrics-request-id"))
 			require.Equal(t, "1", txn.Headers.Get("x-metrics-request-seq"))
 			require.Equal(t, "2", txn.Headers.Get("x-metrics-request-len"))
 		},

pkg/serializer/metrics.go

@@ -9,6 +9,8 @@ import (
 	"fmt"
 	"slices"
 
+	"github.com/google/uuid"
+
 	"github.com/DataDog/datadog-agent/comp/core/config"
 	"github.com/DataDog/datadog-agent/comp/forwarder/defaultforwarder/endpoints"
 	"github.com/DataDog/datadog-agent/comp/forwarder/defaultforwarder/resolver"
@@ -71,11 +73,15 @@ func (s *Serializer) buildPipelines(kind metricsKind) metrics.PipelineSet {
 	for _, resolver := range s.Forwarder.GetDomainResolvers() {
 		useV3 := metricsUseV3(resolver, s.config, kind)
 		validateV3 := useV3 && validateV3
+		batchID := ""
+		if validateV3 {
+			batchID = s.genUUID()
+		}
 
 		dest := metrics.PipelineDestination{
-			Resolver:             resolver,
-			Endpoint:             metricsEndpointFor(kind, useV3),
-			AddValidationHeaders: validateV3,
+			Resolver:          resolver,
+			Endpoint:          metricsEndpointFor(kind, useV3),
+			ValidationBatchID: batchID,
 		}
 
 		switch {
@@ -110,9 +116,9 @@ func (s *Serializer) buildPipelines(kind metricsKind) metrics.PipelineSet {
 					V3:     false,
 				}
 				vdest := metrics.PipelineDestination{
-					Resolver:             resolver,
-					Endpoint:             metricsEndpointFor(kind, false),
-					AddValidationHeaders: true,
+					Resolver:          resolver,
+					Endpoint:          metricsEndpointFor(kind, false),
+					ValidationBatchID: batchID,
 				}
 				pipelines.Add(vconf, vdest)
 			}
@@ -121,3 +127,12 @@ func (s *Serializer) buildPipelines(kind metricsKind) metrics.PipelineSet {
 
 	return pipelines
 }
+
+func (s *Serializer) genUUID() string {
+	uuid, err := uuid.NewV7()
+	if err != nil {
+		s.logger.Warnf("failed to generate payload batch id: %v", err)
+		return ""
+	}
+	return uuid.String()
+}

pkg/serializer/metrics_test.go

@@ -379,6 +379,18 @@ func TestPipelinesWithV3Validate(t *testing.T) {
 
 	pipelines := s.buildPipelines(metricsKindSeries)
 
+	batchID := ""
+outer:
+	for _, ctx := range pipelines {
+		for _, d := range ctx.Destinations {
+			if d.ValidationBatchID != "" {
+				batchID = d.ValidationBatchID
+				break outer
+			}
+		}
+	}
+	require.NotEmpty(t, batchID)
+
 	testutil.ElementsMatchFn(t, maps.All(pipelines),
 		// v3 pipeline has one destination...
 		func(t require.TestingT, conf metrics.PipelineConfig, ctx *metrics.PipelineContext) {
@@ -389,7 +401,7 @@ func TestPipelinesWithV3Validate(t *testing.T) {
 				func(t require.TestingT, _ int, dest metrics.PipelineDestination) {
 					require.Equal(t, "http://example.test", dest.Resolver.GetConfigName())
 					require.Equal(t, endpoints.V3SeriesEndpoint, dest.Endpoint)
-					require.True(t, dest.AddValidationHeaders)
+					require.Equal(t, batchID, dest.ValidationBatchID)
 				})
 		},
 		// v2 pipeline has two destinations...
@@ -401,13 +413,13 @@ func TestPipelinesWithV3Validate(t *testing.T) {
 				func(t require.TestingT, _ int, dest metrics.PipelineDestination) {
 					require.Equal(t, "http://example.test", dest.Resolver.GetConfigName())
 					require.Equal(t, endpoints.SeriesEndpoint, dest.Endpoint)
-					require.True(t, dest.AddValidationHeaders)
+					require.Equal(t, batchID, dest.ValidationBatchID)
 				},
 				// ... to the alternative domain without validation headers
 				func(t require.TestingT, _ int, dest metrics.PipelineDestination) {
 					require.Equal(t, "http://another.test", dest.Resolver.GetConfigName())
 					require.Equal(t, endpoints.SeriesEndpoint, dest.Endpoint)
-					require.False(t, dest.AddValidationHeaders)
+					require.Empty(t, dest.ValidationBatchID)
 				},
 			)
 		},