[exporter/splunkhec] Implement maxEventSize on exporter side (#20752)

Added a new config to apply the maxEventSize on our data.
- It applies this limit on uncompressed data

Link to tracking Issue: #18066

---------

Co-authored-by: Antoine Toulme <[email protected]>

Author: VihasMakwana

.chloggen/max_event_size.yaml

@@ -0,0 +1,16 @@
+# 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: splunkhecexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Apply a new config `maxEventSize` on our events, it would break down the events based on this. This basically implements HEC's `maxEventSize` on our exporter. It is implemented on uncompressed data.
+
+# One or more tracking issues related to the change
+issues: [20290]
+
+# (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:

exporter/splunkhecexporter/buffer.go

@@ -43,19 +43,22 @@ type bufferState struct {
 	compressionAvailable bool
 	compressionEnabled   bool
 	bufferMaxLen         uint
+	maxEventLength       uint
 	writer               io.Writer
 	buf                  *bytes.Buffer
 	bufFront             *index
 	resource             int
 	library              int
 	containsData         bool
+	rawLength            int
 }
 
 func (b *bufferState) reset() {
 	b.buf.Reset()
 	b.compressionEnabled = false
 	b.writer = &cancellableBytesWriter{innerWriter: b.buf, maxCapacity: b.bufferMaxLen}
 	b.containsData = false
+	b.rawLength = 0
 }
 
 func (b *bufferState) Read(p []byte) (n int, err error) {
@@ -71,6 +74,9 @@ func (b *bufferState) Close() error {
 
 // accept returns true if data is accepted by the buffer
 func (b *bufferState) accept(data []byte) (bool, error) {
+	if len(data)+b.rawLength > int(b.maxEventLength) {
+		return false, nil
+	}
 	_, err := b.writer.Write(data)
 	overCapacity := errors.Is(err, errOverCapacity)
 	bufLen := b.buf.Len()
@@ -111,13 +117,15 @@ func (b *bufferState) accept(data []byte) (bool, error) {
 			}
 
 		}
+		b.rawLength += len(data)
 		b.containsData = true
 		return true, nil
 	}
 	if overCapacity {
 		return false, nil
 	}
 	b.containsData = true
+	b.rawLength += len(data)
 	return true, err
 }
 
@@ -180,7 +188,7 @@ func (c *cancellableGzipWriter) close() error {
 	return c.innerWriter.Close()
 }
 
-func makeBlankBufferState(bufCap uint, compressionAvailable bool) *bufferState {
+func makeBlankBufferState(bufCap uint, compressionAvailable bool, maxEventLength uint) *bufferState {
 	// Buffer of JSON encoded Splunk events, last record is expected to overflow bufCap, hence the padding
 	buf := bytes.NewBuffer(make([]byte, 0, bufCap+bufCapPadding))
 
@@ -190,6 +198,7 @@ func makeBlankBufferState(bufCap uint, compressionAvailable bool) *bufferState {
 		writer:               &cancellableBytesWriter{innerWriter: buf, maxCapacity: bufCap},
 		buf:                  buf,
 		bufferMaxLen:         bufCap,
+		maxEventLength:       maxEventLength,
 		bufFront:             nil, // Index of the log record of the first unsent event in buffer.
 		resource:             0,   // Index of currently processed Resource
 		library:              0,   // Index of currently processed Library

exporter/splunkhecexporter/client.go

@@ -124,8 +124,8 @@ func (c *client) pushLogDataInBatches(ctx context.Context, ld plog.Logs, headers
 		profilingLocalHeaders[k] = v
 	}
 
-	var bufState = makeBlankBufferState(c.config.MaxContentLengthLogs, !c.config.DisableCompression)
-	var profilingBufState = makeBlankBufferState(c.config.MaxContentLengthLogs, !c.config.DisableCompression)
+	var bufState = makeBlankBufferState(c.config.MaxContentLengthLogs, !c.config.DisableCompression, c.config.MaxEventSize)
+	var profilingBufState = makeBlankBufferState(c.config.MaxContentLengthLogs, !c.config.DisableCompression, c.config.MaxEventSize)
 	var permanentErrors []error
 
 	var rls = ld.ResourceLogs()
@@ -395,7 +395,7 @@ func (c *client) pushTracesData(ctx context.Context, tds ptrace.ResourceSpansSli
 // The batch content length is restricted to MaxContentLengthMetrics.
 // md metrics are parsed to Splunk events.
 func (c *client) pushMetricsDataInBatches(ctx context.Context, md pmetric.Metrics, headers map[string]string) error {
-	var bufState = makeBlankBufferState(c.config.MaxContentLengthMetrics, !c.config.DisableCompression)
+	var bufState = makeBlankBufferState(c.config.MaxContentLengthMetrics, !c.config.DisableCompression, c.config.MaxEventSize)
 	var permanentErrors []error
 
 	var rms = md.ResourceMetrics()
@@ -430,7 +430,7 @@ func (c *client) pushMetricsDataInBatches(ctx context.Context, md pmetric.Metric
 // The batch content length is restricted to MaxContentLengthMetrics.
 // td traces are parsed to Splunk events.
 func (c *client) pushTracesDataInBatches(ctx context.Context, td ptrace.Traces, headers map[string]string) error {
-	bufState := makeBlankBufferState(c.config.MaxContentLengthTraces, !c.config.DisableCompression)
+	bufState := makeBlankBufferState(c.config.MaxContentLengthTraces, !c.config.DisableCompression, c.config.MaxEventSize)
 	var permanentErrors []error
 
 	var rts = td.ResourceSpans()

exporter/splunkhecexporter/client_test.go

@@ -694,6 +694,54 @@ func TestReceiveLogs(t *testing.T) {
 				wantErr:    "timeout", // our server will time out waiting for the data.
 			},
 		},
+		{
+			name: "two events with 2000 bytes, one with 2000 bytes, then one with 20000 bytes",
+			logs: func() plog.Logs {
+				firstLog := createLogData(1, 1, 3)
+				firstLog.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0).Body().SetStr(repeatableString(2000))
+				firstLog.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(1).Body().SetStr(repeatableString(2000))
+				firstLog.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(2).Body().SetStr(repeatableString(20000))
+				return firstLog
+			}(),
+			conf: func() *Config {
+				cfg := NewFactory().CreateDefaultConfig().(*Config)
+				cfg.MaxEventSize = 20000 // small so we can reproduce without allocating big logs.
+				cfg.DisableCompression = true
+				return cfg
+			}(),
+			want: wantType{
+				batches: [][]string{
+					{`"otel.log.name":"0_0_0"`, `"otel.log.name":"0_0_1"`},
+				},
+				numBatches: 1,
+			},
+		},
+		{
+			name: "two events with 2000 bytes, one with 1000 bytes, then one with 4200 bytes",
+			logs: func() plog.Logs {
+				firstLog := createLogData(1, 1, 5)
+				firstLog.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0).Body().SetStr(repeatableString(2000))
+				firstLog.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(1).Body().SetStr(repeatableString(2000))
+				firstLog.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(2).Body().SetStr(repeatableString(1000))
+				firstLog.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(3).Body().SetStr(repeatableString(4200))
+				return firstLog
+			}(),
+			conf: func() *Config {
+				cfg := NewFactory().CreateDefaultConfig().(*Config)
+				cfg.MaxEventSize = 10000 // small so we can reproduce without allocating big logs.
+				cfg.MaxContentLengthLogs = 5000
+				cfg.DisableCompression = true
+				return cfg
+			}(),
+			want: wantType{
+				batches: [][]string{
+					{`"otel.log.name":"0_0_0"`, `"otel.log.name":"0_0_1"`},
+					{`"otel.log.name":"0_0_2"`},
+					{`"otel.log.name":"0_0_3"`, `"otel.log.name":"0_0_4"`},
+				},
+				numBatches: 3,
+			},
+		},
 	}
 
 	for _, test := range tests {
@@ -1559,7 +1607,7 @@ func BenchmarkPushLogRecords(b *testing.B) {
 		hecWorker: &mockHecWorker{},
 	}
 
-	state := makeBlankBufferState(4096, true)
+	state := makeBlankBufferState(4096, true, 4096)
 	for n := 0; n < b.N; n++ {
 		permanentErrs, sendingErr := c.pushLogRecords(context.Background(), logs.ResourceLogs(), state, map[string]string{})
 		assert.NoError(b, sendingErr)

exporter/splunkhecexporter/config.go

@@ -33,9 +33,11 @@ const (
 	defaultContentLengthLogsLimit    = 2 * 1024 * 1024
 	defaultContentLengthMetricsLimit = 2 * 1024 * 1024
 	defaultContentLengthTracesLimit  = 2 * 1024 * 1024
+	defaultMaxEventSize              = 5 * 1024 * 1024
 	maxContentLengthLogsLimit        = 800 * 1024 * 1024
 	maxContentLengthMetricsLimit     = 800 * 1024 * 1024
 	maxContentLengthTracesLimit      = 800 * 1024 * 1024
+	maxMaxEventSize                  = 800 * 1024 * 1024
 )
 
 // OtelToHecFields defines the mapping of attributes to HEC fields
@@ -90,6 +92,10 @@ type Config struct {
 	// Maximum allowed value is 838860800 (~ 800 MB).
 	MaxContentLengthTraces uint `mapstructure:"max_content_length_traces"`
 
+	// Maximum payload size, raw uncompressed. Default value is 5242880 bytes (5MiB).
+	// Maximum allowed value is 838860800 (~ 800 MB).
+	MaxEventSize uint `mapstructure:"max_event_size"`
+
 	// App name is used to track telemetry information for Splunk App's using HEC by App name. Defaults to "OpenTelemetry Collector Contrib".
 	SplunkAppName string `mapstructure:"splunk_app_name"`
 
@@ -153,6 +159,11 @@ func (cfg *Config) Validate() error {
 	if cfg.MaxContentLengthTraces > maxContentLengthTracesLimit {
 		return fmt.Errorf(`requires "max_content_length_traces" <= %d`, maxContentLengthTracesLimit)
 	}
+
+	if cfg.MaxEventSize > maxMaxEventSize {
+		return fmt.Errorf(`requires "max_event_size" <= %d`, maxMaxEventSize)
+	}
+
 	if err := cfg.QueueSettings.Validate(); err != nil {
 		return fmt.Errorf("sending_queue settings has invalid configuration: %w", err)
 	}

exporter/splunkhecexporter/config_test.go

@@ -65,6 +65,7 @@ func TestLoadConfig(t *testing.T) {
 				LogDataEnabled:          true,
 				ProfilingDataEnabled:    true,
 				ExportRaw:               true,
+				MaxEventSize:            5 * 1024 * 1024,
 				MaxContentLengthLogs:    2 * 1024 * 1024,
 				MaxContentLengthMetrics: 2 * 1024 * 1024,
 				MaxContentLengthTraces:  2 * 1024 * 1024,
@@ -190,6 +191,17 @@ func TestConfig_Validate(t *testing.T) {
 			}(),
 			wantErr: "requires \"max_content_length_traces\" <= 838860800",
 		},
+		{
+			name: "max default event-size",
+			cfg: func() *Config {
+				cfg := createDefaultConfig().(*Config)
+				cfg.HTTPClientSettings.Endpoint = "http://foo_bar.com"
+				cfg.MaxEventSize = maxMaxEventSize + 1
+				cfg.Token = "foo"
+				return cfg
+			}(),
+			wantErr: "requires \"max_event_size\" <= 838860800",
+		},
 	}
 
 	for _, tt := range tests {

exporter/splunkhecexporter/factory.go

@@ -80,6 +80,7 @@ func createDefaultConfig() component.Config {
 		MaxContentLengthLogs:    defaultContentLengthLogsLimit,
 		MaxContentLengthMetrics: defaultContentLengthMetricsLimit,
 		MaxContentLengthTraces:  defaultContentLengthTracesLimit,
+		MaxEventSize:            defaultMaxEventSize,
 		HecToOtelAttrs: splunk.HecToOtelAttrs{
 			Source:     splunk.DefaultSourceLabel,
 			SourceType: splunk.DefaultSourceTypeLabel,