fix(byoo): cap chunked log exporter batches

src/compute-plane-services/byoo-otel-collector/VERSION

@@ -1 +1 @@
-0.157.0
+0.157.1

src/compute-plane-services/byoo-otel-collector/internal/otelconfig/render.go

@@ -68,6 +68,9 @@ type OpenTelemetryConfig struct {
 const (
 	defaultLogChunkMaxPayloadBytes       = 262144
 	minConfiguredLogChunkMaxPayloadBytes = 4
+	defaultLogExporterBatchFlushTimeout  = "200ms"
+	// Leave 100 KB below the 1 MB receiver limit for the export envelope.
+	defaultLogExporterBatchSizeBytes = int64(900_000)
 )
 
 const (
@@ -267,6 +270,18 @@ func logExporterSendingQueue() map[string]interface{} {
 	}
 }
 
+func enableChunkedLogExporterBatching(otelConfig *OpenTelemetryConfig, exporterID string) {
+	exporter := otelConfig.Exporters[exporterID]
+	queue := mapFromInterface(exporter["sending_queue"])
+	queue["batch"] = map[string]interface{}{
+		"flush_timeout": defaultLogExporterBatchFlushTimeout,
+		"sizer":         "bytes",
+		"min_size":      defaultLogExporterBatchSizeBytes,
+		"max_size":      defaultLogExporterBatchSizeBytes,
+	}
+	exporter["sending_queue"] = queue
+}
+
 func exporterLogs(config TelemetryConfig, otelConfig *OpenTelemetryConfig) (exporterId string, err error) {
 	var exporterType, exporterName string
 	var exporterCredential interface{}
@@ -857,6 +872,9 @@ func generateExportersAndService(config TelemetryConfig, otelConfig *OpenTelemet
 				"dry_run":           logChunking.DryRun,
 			}
 			logPipeline.Processors = append(logPipeline.Processors, "logchunk/byoo")
+			if !logChunking.DryRun {
+				enableChunkedLogExporterBatching(otelConfig, exporterId)
+			}
 		}
 		logPipeline.Processors = append(logPipeline.Processors, "batch")
 		otelConfig.Service.Pipelines["logs"] = logPipeline

src/compute-plane-services/byoo-otel-collector/internal/otelconfig/render_test.go

@@ -482,8 +482,8 @@ func TestGenerateExportersAndServiceAddsLogChunkDefaultsWhenEnabled(t *testing.T
 			Logs: &Telemetry{
 				Name:     "example-logs",
 				Protocol: ProtocolHTTP,
-				Provider: ProviderSplunk,
-				Endpoint: "https://splunk.example.invalid",
+				Provider: ProviderKratosLogs,
+				Endpoint: "https://kratos.example.invalid",
 			},
 		},
 	}
@@ -502,8 +502,18 @@ func TestGenerateExportersAndServiceAddsLogChunkDefaultsWhenEnabled(t *testing.T
 		"max_payload_bytes": defaultLogChunkMaxPayloadBytes,
 		"dry_run":           false,
 	}, otelConfig.Processors["logchunk/byoo"])
-	exporter := otelConfig.Exporters["splunk_hec/SPLUNK-example-logs-logs"]
-	assert.NotContains(t, exporter["sending_queue"].(map[string]interface{}), "batch")
+	exporter := otelConfig.Exporters["otlp_http/KRATOS-example-logs-logs"]
+	assert.Equal(t, map[string]interface{}{
+		"enabled":       true,
+		"num_consumers": 10,
+		"queue_size":    1000,
+		"batch": map[string]interface{}{
+			"flush_timeout": defaultLogExporterBatchFlushTimeout,
+			"sizer":         "bytes",
+			"min_size":      defaultLogExporterBatchSizeBytes,
+			"max_size":      defaultLogExporterBatchSizeBytes,
+		},
+	}, exporter["sending_queue"])
 }
 
 func TestGenerateExportersAndServiceUsesExporterHelperQueueBatchConfig(t *testing.T) {