POC byte based batching

exporter/exporterbatcher/config.go

@@ -31,6 +31,7 @@ type MinSizeConfig struct {
 	// sent regardless of the timeout. There is no guarantee that the batch size always greater than this value.
 	// This option requires the Request to implement RequestItemsCounter interface. Otherwise, it will be ignored.
 	MinSizeItems int `mapstructure:"min_size_items"`
+	MinSizeBytes int `mapstructure:"min_size_bytes"`
 }
 
 // MaxSizeConfig defines the configuration for the maximum number of items in a batch.
@@ -41,18 +42,32 @@ type MaxSizeConfig struct {
 	// If the batch size exceeds this value, it will be broken up into smaller batches if possible.
 	// Setting this value to zero disables the maximum size limit.
 	MaxSizeItems int `mapstructure:"max_size_items"`
+	MaxSizeBytes int `mapstructure:"max_size_bytes"`
 }
 
 func (c Config) Validate() error {
+	if c.MinSizeBytes != 0 && c.MinSizeItems != 0 || c.MinSizeBytes != 0 && c.MaxSizeItems != 0 || c.MinSizeItems != 0 && c.MaxSizeBytes != 0 {
+		return errors.New("size limit and bytes limit cannot be specified at the same time")
+	}
+
 	if c.MinSizeItems < 0 {
 		return errors.New("min_size_items must be greater than or equal to zero")
 	}
+	if c.MinSizeBytes < 0 {
+		return errors.New("min_size_bytes must be greater than or equal to zero")
+	}
 	if c.MaxSizeItems < 0 {
 		return errors.New("max_size_items must be greater than or equal to zero")
 	}
+	if c.MaxSizeBytes < 0 {
+		return errors.New("max_size_bytes must be greater than or equal to zero")
+	}
 	if c.MaxSizeItems != 0 && c.MaxSizeItems < c.MinSizeItems {
 		return errors.New("max_size_items must be greater than or equal to min_size_items")
 	}
+	if c.MaxSizeBytes != 0 && c.MaxSizeBytes < c.MinSizeBytes {
+		return errors.New("max_size_bytes must be greater than or equal to min_size_bytes")
+	}
 	if c.FlushTimeout <= 0 {
 		return errors.New("timeout must be greater than zero")
 	}

exporter/exporterhelper/internal/request.go

@@ -56,6 +56,10 @@ func (r *fakeRequest) ItemsCount() int {
 	return r.items
 }
 
+func (r *fakeRequest) ByteSize() int {
+	return r.items
+}
+
 func (r *fakeRequest) MergeSplit(_ context.Context, cfg exporterbatcher.MaxSizeConfig, r2 internal.Request) ([]internal.Request, error) {
 	if r.mergeErr != nil {
 		return nil, r.mergeErr

exporter/exporterhelper/internal/retry_sender_test.go

@@ -418,6 +418,10 @@ func (mer *mockErrorRequest) ItemsCount() int {
 	return 7
 }
 
+func (mer *mockErrorRequest) ByteSize() int {
+	return 7
+}
+
 func (mer *mockErrorRequest) MergeSplit(context.Context, exporterbatcher.MaxSizeConfig, internal.Request) ([]internal.Request, error) {
 	return nil, nil
 }
@@ -464,6 +468,10 @@ func (m *mockRequest) ItemsCount() int {
 	return m.cnt
 }
 
+func (m *mockRequest) ByteSize() int {
+	return m.cnt
+}
+
 func (m *mockRequest) MergeSplit(context.Context, exporterbatcher.MaxSizeConfig, internal.Request) ([]internal.Request, error) {
 	return nil, nil
 }

exporter/exporterhelper/logs.go

@@ -25,8 +25,9 @@ var (
 )
 
 type logsRequest struct {
-	ld     plog.Logs
-	pusher consumer.ConsumeLogsFunc
+	ld             plog.Logs
+	pusher         consumer.ConsumeLogsFunc
+	cachedByteSize int
 }
 
 func newLogsRequest(ld plog.Logs, pusher consumer.ConsumeLogsFunc) Request {
@@ -66,6 +67,21 @@ func (req *logsRequest) ItemsCount() int {
 	return req.ld.LogRecordCount()
 }
 
+func (req *logsRequest) ByteSize() int {
+	if req.cachedByteSize == 0 {
+		req.cachedByteSize = logsMarshaler.LogsSize(req.ld)
+	}
+	return req.cachedByteSize
+}
+
+func (req *logsRequest) invalidateCachedByteSize() {
+	req.cachedByteSize = 0
+}
+
+func (req *logsRequest) updateCachedByteSize(delta int) {
+	req.cachedByteSize += delta
+}
+
 type logsExporter struct {
 	*internal.BaseExporter
 	consumer.Logs

exporter/exporterhelper/logs_batch.go

@@ -6,6 +6,7 @@ package exporterhelper // import "go.opentelemetry.io/collector/exporter/exporte
 import (
 	"context"
 	"errors"
+	math_bits "math/bits"
 
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
 	"go.opentelemetry.io/collector/pdata/plog"
@@ -23,11 +24,144 @@ func (req *logsRequest) MergeSplit(_ context.Context, cfg exporterbatcher.MaxSiz
 		}
 	}
 
-	if cfg.MaxSizeItems == 0 {
-		req2.ld.ResourceLogs().MoveAndAppendTo(req.ld.ResourceLogs())
+	if cfg.MaxSizeItems == 0 && cfg.MaxSizeBytes == 0 {
+		if req2 != nil {
+			req2.ld.ResourceLogs().MoveAndAppendTo(req.ld.ResourceLogs())
+		}
 		return []Request{req}, nil
 	}
+	if cfg.MaxSizeBytes > 0 {
+		return req.mergeSplitBasedOnByteSize(cfg, req2)
+	}
+	return req.mergeSplitBasedOnItemCount(cfg, req2)
+}
 
+func (req *logsRequest) mergeSplitBasedOnByteSize(cfg exporterbatcher.MaxSizeConfig, req2 *logsRequest) ([]Request, error) {
+	var (
+		res          []Request
+		destReq      *logsRequest
+		capacityLeft = cfg.MaxSizeBytes
+	)
+	for _, srcReq := range []*logsRequest{req, req2} {
+		if srcReq == nil {
+			continue
+		}
+
+		ByteSize := srcReq.ByteSize()
+		if ByteSize <= capacityLeft {
+			if destReq == nil {
+				destReq = srcReq
+			} else {
+				destReq.updateCachedByteSize(ByteSize)
+				srcReq.ld.ResourceLogs().MoveAndAppendTo(destReq.ld.ResourceLogs())
+			}
+			capacityLeft -= ByteSize
+			continue
+		}
+
+		for {
+			srcReq.invalidateCachedByteSize()
+			extractedLogs, capacityReached := extractLogsBasedOnByteSize(srcReq.ld, capacityLeft)
+
+			if extractedLogs.LogRecordCount() == 0 {
+				break
+			}
+			if destReq == nil {
+				destReq = &logsRequest{
+					ld:     extractedLogs,
+					pusher: srcReq.pusher,
+				}
+			} else {
+				destReq.updateCachedByteSize(logsMarshaler.LogsSize(extractedLogs))
+				extractedLogs.ResourceLogs().MoveAndAppendTo(destReq.ld.ResourceLogs())
+			}
+			// Create new batch once capacity is reached.
+			if capacityReached {
+				res = append(res, destReq)
+				destReq = nil
+				capacityLeft = cfg.MaxSizeBytes
+			} else {
+				capacityLeft = cfg.MaxSizeBytes - destReq.ByteSize()
+			}
+		}
+	}
+
+	if destReq != nil {
+		res = append(res, destReq)
+	}
+	return res, nil
+}
+
+// extractLogs extracts logs from the input logs and returns a new logs with the specified number of log records.
+func extractLogsBasedOnByteSize(srcLogs plog.Logs, capacity int) (plog.Logs, bool) {
+	capacityReached := false
+	destLogs := plog.NewLogs()
+	capacityLeft := capacity - logsMarshaler.LogsSize(destLogs)
+	srcLogs.ResourceLogs().RemoveIf(func(srcRL plog.ResourceLogs) bool {
+		if capacityReached {
+			return false
+		}
+		needToExtract := logsMarshaler.ResourceLogsSize(srcRL) > capacityLeft
+		if needToExtract {
+			srcRL, capacityReached = extractResourceLogsBasedOnByteSize(srcRL, capacityLeft)
+			if srcRL.ScopeLogs().Len() == 0 {
+				return false
+			}
+		}
+		capacityLeft -= deltaCapacity(logsMarshaler.ResourceLogsSize(srcRL))
+		srcRL.MoveTo(destLogs.ResourceLogs().AppendEmpty())
+		return !needToExtract
+	})
+	return destLogs, capacityReached
+}
+
+// extractResourceLogs extracts resource logs and returns a new resource logs with the specified number of log records.
+func extractResourceLogsBasedOnByteSize(srcRL plog.ResourceLogs, capacity int) (plog.ResourceLogs, bool) {
+	capacityReached := false
+	destRL := plog.NewResourceLogs()
+	destRL.SetSchemaUrl(srcRL.SchemaUrl())
+	srcRL.Resource().CopyTo(destRL.Resource())
+	capacityLeft := capacity - logsMarshaler.ResourceLogsSize(destRL)
+	srcRL.ScopeLogs().RemoveIf(func(srcSL plog.ScopeLogs) bool {
+		if capacityReached {
+			return false
+		}
+		needToExtract := logsMarshaler.ScopeLogsSize(srcSL) > capacityLeft
+		if needToExtract {
+			srcSL, capacityReached = extractScopeLogsBasedOnByteSize(srcSL, capacityLeft)
+			if srcSL.LogRecords().Len() == 0 {
+				return false
+			}
+		}
+
+		capacityLeft -= deltaCapacity(logsMarshaler.ScopeLogsSize(srcSL))
+		srcSL.MoveTo(destRL.ScopeLogs().AppendEmpty())
+		return !needToExtract
+	})
+	return destRL, capacityReached
+}
+
+// extractScopeLogs extracts scope logs and returns a new scope logs with the specified number of log records.
+func extractScopeLogsBasedOnByteSize(srcSL plog.ScopeLogs, capacity int) (plog.ScopeLogs, bool) {
+	capacityReached := false
+	destSL := plog.NewScopeLogs()
+	destSL.SetSchemaUrl(srcSL.SchemaUrl())
+	srcSL.Scope().CopyTo(destSL.Scope())
+	capacityLeft := capacity - logsMarshaler.ScopeLogsSize(destSL)
+
+	srcSL.LogRecords().RemoveIf(func(srcLR plog.LogRecord) bool {
+		if capacityReached || logsMarshaler.LogRecordSize(srcLR) > capacityLeft {
+			capacityReached = true
+			return false
+		}
+		capacityLeft -= deltaCapacity(logsMarshaler.LogRecordSize(srcLR))
+		srcLR.MoveTo(destSL.LogRecords().AppendEmpty())
+		return true
+	})
+	return destSL, capacityReached
+}
+
+func (req *logsRequest) mergeSplitBasedOnItemCount(cfg exporterbatcher.MaxSizeConfig, req2 *logsRequest) ([]Request, error) {
 	var (
 		res          []Request
 		destReq      *logsRequest
@@ -75,6 +209,11 @@ func (req *logsRequest) MergeSplit(_ context.Context, cfg exporterbatcher.MaxSiz
 	return res, nil
 }
 
+// deltaCapacity() returns the delta size of a proto slice when a new item is added.
+func deltaCapacity(newItemSize int) int {
+	return 1 + newItemSize + int(math_bits.Len64(uint64(newItemSize|1)+6)/7)
+}
+
 // extractLogs extracts logs from the input logs and returns a new logs with the specified number of log records.
 func extractLogs(srcLogs plog.Logs, count int) plog.Logs {
 	destLogs := plog.NewLogs()