pkg/trace/api: limit simultaneous otlp requests, do not drop payloads (…

…#23085) pkg/trace/api: limit simultaneous otlp requests, do not drop payloads Co-authored-by: dineshg13 <[email protected]>
DataDog · Feb 28, 2024 · f6750c4 · f6750c4
1 parent 5b3c5cc
commit f6750c4
Show file tree

Hide file tree

Showing 6 changed files with 87 additions and 7 deletions.
diff --git a/comp/otelcol/otlp/map_provider_config_not_serverless.go b/comp/otelcol/otlp/map_provider_config_not_serverless.go
@@ -24,6 +24,8 @@ exporters:
     tls:
       insecure: true
     compression: none
+    sending_queue:
+      enabled: false
 
 service:
   telemetry:

diff --git a/comp/otelcol/otlp/map_provider_not_serverless_test.go b/comp/otelcol/otlp/map_provider_not_serverless_test.go
@@ -55,6 +55,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 				},
 				"processors": map[string]interface{}{
@@ -117,6 +120,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 					"serializer": map[string]interface{}{
 						"metrics": map[string]interface{}{
@@ -191,6 +197,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 					"serializer": map[string]interface{}{
 						"metrics": map[string]interface{}{
@@ -257,6 +266,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 				},
 				"service": map[string]interface{}{
@@ -364,6 +376,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 					"logging": map[string]interface{}{
 						"loglevel": "info",
@@ -485,6 +500,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 					"serializer": map[string]interface{}{
 						"metrics": map[string]interface{}{
@@ -546,6 +564,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 					"logsagent": interface{}(nil),
 				},
@@ -615,6 +636,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 					"serializer": map[string]interface{}{
 						"metrics": map[string]interface{}{
@@ -696,6 +720,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 					"serializer": map[string]interface{}{
 						"metrics": map[string]interface{}{
@@ -769,6 +796,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 					"logsagent": interface{}(nil),
 				},
@@ -890,6 +920,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 					"logging": map[string]interface{}{
 						"loglevel": "info",
@@ -1025,6 +1058,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 					"serializer": map[string]interface{}{
 						"metrics": map[string]interface{}{

diff --git a/comp/otelcol/otlp/map_provider_serverless_test.go b/comp/otelcol/otlp/map_provider_serverless_test.go
@@ -51,6 +51,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 				},
 				"service": map[string]interface{}{
@@ -102,6 +105,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 					"serializer": map[string]interface{}{
 						"metrics": map[string]interface{}{
@@ -169,6 +175,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 					"serializer": map[string]interface{}{
 						"metrics": map[string]interface{}{
@@ -228,6 +237,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 				},
 				"service": map[string]interface{}{
@@ -323,6 +335,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 					"logging": map[string]interface{}{
 						"loglevel": "info",
@@ -432,6 +447,9 @@ func TestNewMap(t *testing.T) {
 						},
 						"compression": "none",
 						"endpoint":    "localhost:5003",
+						"sending_queue": map[string]interface{}{
+							"enabled": false,
+						},
 					},
 					"serializer": map[string]interface{}{
 						"metrics": map[string]interface{}{

diff --git a/pkg/trace/api/otlp.go b/pkg/trace/api/otlp.go
@@ -74,7 +74,10 @@ func (o *OTLPReceiver) Start() {
 		if err != nil {
 			log.Criticalf("Error starting OpenTelemetry gRPC server: %v", err)
 		} else {
-			o.grpcsrv = grpc.NewServer(grpc.MaxRecvMsgSize(10 * 1024 * 1024))
+			o.grpcsrv = grpc.NewServer(
+				grpc.MaxRecvMsgSize(10*1024*1024),
+				grpc.MaxConcurrentStreams(1), // Each payload must be sent to processing stage before we decode the next.
+			)
 			ptraceotlp.RegisterGRPCServer(o.grpcsrv, o)
 			o.wg.Add(1)
 			go func() {
@@ -305,12 +308,8 @@ func (o *OTLPReceiver) ReceiveResourceSpans(ctx context.Context, rspans ptrace.R
 			tagContainersTags: payloadTags.String(),
 		}
 	}
-	select {
-	case o.out <- &p:
-		// success
-	default:
-		log.Warn("Payload in channel full. Dropped 1 payload.")
-	}
+
+	o.out <- &p
 	return src
 }
 

diff --git a/pkg/trace/api/otlp_test.go b/pkg/trace/api/otlp_test.go
@@ -166,6 +166,18 @@ func TestOTLPMetrics(t *testing.T) {
 		},
 	}).Traces().ResourceSpans()
 
+	stop := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-out:
+			case <-stop:
+				return
+			}
+		}
+	}()
+	defer close(stop)
+
 	rcv.ReceiveResourceSpans(context.Background(), rspans.At(0), http.Header{})
 	rcv.ReceiveResourceSpans(context.Background(), rspans.At(1), http.Header{})
 

diff --git a/releasenotes/notes/apm-otel-receiver-backpressure-40e301d75d00804d.yaml b/releasenotes/notes/apm-otel-receiver-backpressure-40e301d75d00804d.yaml
@@ -0,0 +1,13 @@
+# Each section from every release note are combined when the
+# CHANGELOG.rst is rendered. So the text needs to be worded so that
+# it does not depend on any information only available in another
+# section. This may mean repeating some details, but each section
+# must be readable independently of the other.
+#
+# Each section note must be formatted as reStructuredText.
+---
+fixes:
+  - |
+    APM: Stop dropping incoming OTel payloads when the processing channel is full
+    and eliminate OOM issues in the trace agent and collector component in high
+    load scenarios, making the OTel pipeline more reliable.