Add byte-based ingestion limits to the queue and output

libbeat/common/fifo/fifo.go

@@ -0,0 +1,84 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package fifo
+
+type FIFO[T any] struct {
+	first *node[T]
+	last  *node[T]
+}
+
+type node[T any] struct {
+	next  *node[T]
+	value T
+}
+
+func (f *FIFO[T]) Add(value T) {
+	newNode := &node[T]{value: value}
+	if f.first == nil {
+		f.first = newNode
+	} else {
+		f.last.next = newNode
+	}
+	f.last = newNode
+}
+
+func (f *FIFO[T]) Empty() bool {
+	return f.first == nil
+}
+
+// Return the first value (if present) without removing it from the queue.
+// Returns a default value if the queue is empty. To recognize this case,
+// check (*FIFO).Empty().
+func (f *FIFO[T]) First() T {
+	if f.first == nil {
+		var none T
+		return none
+	}
+	return f.first.value
+}
+
+// Remove the first entry in this FIFO and return it.
+func (f *FIFO[T]) ConsumeFirst() T {
+	result := f.First()
+	f.Remove()
+	return result
+}
+
+// Append another FIFO queue to an existing one. Takes ownership of
+// the given FIFO's contents.
+func (f *FIFO[T]) Concat(list FIFO[T]) {
+	if list.Empty() {
+		return
+	}
+	if f.Empty() {
+		*f = list
+		return
+	}
+	f.last.next = list.first
+	f.last = list.last
+}
+
+// Remove the first entry in the queue. Does nothing if the FIFO is empty.
+func (f *FIFO[T]) Remove() {
+	if f.first != nil {
+		f.first = f.first.next
+		if f.first == nil {
+			f.last = nil
+		}
+	}
+}

libbeat/docs/queueconfig.asciidoc

@@ -67,6 +67,17 @@ queue.mem:
   flush.timeout: 5s
 ------------------------------------------------------------------------------
 
+Here is an alternate configuration that measures queue size in bytes rather
+than event count. In this case, the output must set `bulk_max_bytes`
+instead of `bulk_max_size` to control the batch size:
+
+[source,yaml]
+------------------------------------------------------------------------------
+queue.mem:
+  bytes: 32MB
+  flush.timeout: 10s
+------------------------------------------------------------------------------
+
 [float]
 === Configuration options
 
@@ -80,6 +91,16 @@ Number of events the queue can store.
 
 The default value is 3200 events.
 
+[float]
+[[queue-mem-bytes-option]]
+===== `bytes`
+
+Number of bytes the queue can store. This option is only available for outputs
+that support byte-based event buffers (currently just the Elasticsearch output).
+The queue should set either `events` or `bytes` but not both.
+
+The default is 0, indicating the queue should use the `events` limit instead.
+
 [float]
 [[queue-mem-flush-min-events-option]]
 ===== `flush.min_events`
@@ -92,6 +113,8 @@ publishing.
 If 0 or 1, sets the maximum number of events per batch to half the queue size, and sets
 the queue to synchronous mode (equivalent to `flush.timeout` of 0).
 
+This value is ignored when `bytes` is set.
+
 The default value is 1600.
 
 [float]

libbeat/monitoring/report/elasticsearch/elasticsearch.go

@@ -165,9 +165,9 @@ func makeReporter(beat beat.Info, settings report.Settings, cfg *conf.C) (report
 		},
 		queueConfig,
 		outputs.Group{
-			Clients:   []outputs.Client{outClient},
-			BatchSize: windowSize,
-			Retry:     0, // no retry. Drop event on error.
+			Clients:     []outputs.Client{outClient},
+			BatchEvents: windowSize,
+			Retry:       0, // no retry. Drop event on error.
 		},
 		pipeline.Settings{
 			WaitClose:     0,

libbeat/outputs/console/console.go

@@ -85,7 +85,7 @@ func makeConsole(
 		}
 	}
 
-	return outputs.Success(config.Queue, config.BatchSize, 0, nil, c)
+	return outputs.Success(config.Queue, config.BatchSize, 0, 0, nil, c)
 }
 
 func newConsole(index string, observer outputs.Observer, codec codec.Codec) (*console, error) {

libbeat/outputs/discard/discard.go

@@ -56,7 +56,7 @@ func makeDiscard(
 	// disable bulk support in publisher pipeline
 	_ = cfg.SetInt("bulk_max_size", -1, -1)
 	out.log.Infof("Initialized discard output")
-	return outputs.Success(doConfig.Queue, -1, 0, nil, out)
+	return outputs.Success(doConfig.Queue, -1, 0, 0, nil, out)
 }
 
 // Implement Outputer

libbeat/outputs/elasticsearch/config.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/elastic/beats/v7/libbeat/common/cfgtype"
 	"github.com/elastic/beats/v7/libbeat/common/transport/kerberos"
 	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/transport/httpcommon"
@@ -39,6 +40,7 @@ type elasticsearchConfig struct {
 	EscapeHTML         bool              `config:"escape_html"`
 	Kerberos           *kerberos.Config  `config:"kerberos"`
 	BulkMaxSize        int               `config:"bulk_max_size"`
+	BulkMaxBytes       cfgtype.ByteSize  `config:"bulk_max_bytes"`
 	MaxRetries         int               `config:"max_retries"`
 	Backoff            Backoff           `config:"backoff"`
 	NonIndexablePolicy *config.Namespace `config:"non_indexable_policy"`

libbeat/outputs/elasticsearch/elasticsearch.go

@@ -135,7 +135,14 @@ func makeES(
 		clients[i] = client
 	}
 
-	return outputs.SuccessNet(esConfig.Queue, esConfig.LoadBalance, esConfig.BulkMaxSize, esConfig.MaxRetries, encoderFactory, clients)
+	return outputs.SuccessNet(
+		esConfig.Queue,
+		esConfig.LoadBalance,
+		esConfig.BulkMaxSize,
+		int(esConfig.BulkMaxBytes),
+		esConfig.MaxRetries,
+		encoderFactory,
+		clients)
 }
 
 func buildSelectors(

libbeat/outputs/fileout/file.go

@@ -67,7 +67,7 @@ func makeFileout(
 		return outputs.Fail(err)
 	}
 
-	return outputs.Success(foConfig.Queue, -1, 0, nil, fo)
+	return outputs.Success(foConfig.Queue, -1, 0, 0, nil, fo)
 }
 
 func (out *fileOutput) init(beat beat.Info, c fileOutConfig) error {

libbeat/outputs/kafka/kafka.go

@@ -84,7 +84,7 @@ func makeKafka(
 	if kConfig.MaxRetries < 0 {
 		retry = -1
 	}
-	return outputs.Success(kConfig.Queue, kConfig.BulkMaxSize, retry, nil, client)
+	return outputs.Success(kConfig.Queue, kConfig.BulkMaxSize, 0, retry, nil, client)
 }
 
 // buildTopicSelector builds the topic selector for standalone Beat and when

libbeat/outputs/logstash/logstash.go

@@ -85,5 +85,5 @@ func makeLogstash(
 		clients[i] = client
 	}
 
-	return outputs.SuccessNet(lsConfig.Queue, lsConfig.LoadBalance, lsConfig.BulkMaxSize, lsConfig.MaxRetries, nil, clients)
+	return outputs.SuccessNet(lsConfig.Queue, lsConfig.LoadBalance, lsConfig.BulkMaxSize, 0, lsConfig.MaxRetries, nil, clients)
 }

libbeat/outputs/output_reg.go

@@ -56,7 +56,8 @@ type IndexSelector interface {
 // configuration into the outputs.
 type Group struct {
 	Clients      []Client
-	BatchSize    int
+	BatchEvents  int
+	BatchBytes   int
 	Retry        int
 	QueueFactory queue.QueueFactory
 

libbeat/outputs/redis/redis.go

@@ -165,7 +165,7 @@ func makeRedis(
 		clients[i] = newBackoffClient(client, rConfig.Backoff.Init, rConfig.Backoff.Max)
 	}
 
-	return outputs.SuccessNet(rConfig.Queue, rConfig.LoadBalance, rConfig.BulkMaxSize, rConfig.MaxRetries, nil, clients)
+	return outputs.SuccessNet(rConfig.Queue, rConfig.LoadBalance, rConfig.BulkMaxSize, 0, rConfig.MaxRetries, nil, clients)
 }
 
 func buildKeySelector(cfg *config.C) (outil.Selector, error) {

libbeat/outputs/util.go

@@ -35,7 +35,7 @@ func Fail(err error) (Group, error) { return Group{}, err }
 // instances.  The first argument is expected to contain a queue
 // config.Namespace.  The queue config is passed to assign the queue
 // factory when elastic-agent reloads the output.
-func Success(cfg config.Namespace, batchSize, retry int, encoderFactory queue.EncoderFactory, clients ...Client) (Group, error) {
+func Success(cfg config.Namespace, batchEvents, batchBytes, retry int, encoderFactory queue.EncoderFactory, clients ...Client) (Group, error) {
 	var q queue.QueueFactory
 	if cfg.IsSet() && cfg.Config().Enabled() {
 		switch cfg.Name() {
@@ -60,7 +60,8 @@ func Success(cfg config.Namespace, batchSize, retry int, encoderFactory queue.En
 	}
 	return Group{
 		Clients:        clients,
-		BatchSize:      batchSize,
+		BatchEvents:    batchEvents,
+		BatchBytes:     batchBytes,
 		Retry:          retry,
 		QueueFactory:   q,
 		EncoderFactory: encoderFactory,
@@ -80,12 +81,12 @@ func NetworkClients(netclients []NetworkClient) []Client {
 // The first argument is expected to contain a queue config.Namespace.
 // The queue config is passed to assign the queue factory when
 // elastic-agent reloads the output.
-func SuccessNet(cfg config.Namespace, loadbalance bool, batchSize, retry int, encoderFactory queue.EncoderFactory, netclients []NetworkClient) (Group, error) {
+func SuccessNet(cfg config.Namespace, loadbalance bool, batchEvents, batchBytes, retry int, encoderFactory queue.EncoderFactory, netclients []NetworkClient) (Group, error) {
 
 	if !loadbalance {
-		return Success(cfg, batchSize, retry, encoderFactory, NewFailoverClient(netclients))
+		return Success(cfg, batchEvents, batchBytes, retry, encoderFactory, NewFailoverClient(netclients))
 	}
 
 	clients := NetworkClients(netclients)
-	return Success(cfg, batchSize, retry, encoderFactory, clients...)
+	return Success(cfg, batchEvents, batchBytes, retry, encoderFactory, clients...)
 }

libbeat/publisher/pipeline/client.go

@@ -118,9 +118,9 @@ func (c *client) publish(e beat.Event) {
 
 	var published bool
 	if c.canDrop {
-		_, published = c.producer.TryPublish(pubEvent)
+		published = c.producer.TryPublish(pubEvent)
 	} else {
-		_, published = c.producer.Publish(pubEvent)
+		published = c.producer.Publish(pubEvent)
 	}
 
 	if published {

libbeat/publisher/pipeline/client_test.go

@@ -88,11 +88,12 @@ func TestClient(t *testing.T) {
 		l := logp.L()
 
 		// a small in-memory queue with a very short flush interval
-		q := memqueue.NewQueue(l, nil, memqueue.Settings{
+		q, err := memqueue.NewQueue(l, nil, memqueue.Settings{
 			Events:        5,
 			MaxGetRequest: 1,
 			FlushTimeout:  time.Millisecond,
 		}, 5, nil)
+		require.NoError(t, err, "Queue creation must succeed")
 
 		// model a processor that we're going to make produce errors after
 		p := &testProcessor{}
@@ -114,7 +115,7 @@ func TestClient(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
 			for {
-				batch, err := q.Get(2)
+				batch, err := q.Get(2, 0)
 				if errors.Is(err, io.EOF) {
 					break
 				}
@@ -201,7 +202,8 @@ func TestClientWaitClose(t *testing.T) {
 	}
 	logp.TestingSetup()
 
-	q := memqueue.NewQueue(logp.L(), nil, memqueue.Settings{Events: 1}, 0, nil)
+	q, err := memqueue.NewQueue(logp.L(), nil, memqueue.Settings{Events: 1}, 0, nil)
+	require.NoError(t, err, "Queue creation must succeed")
 	pipeline := makePipeline(Settings{}, q)
 	defer pipeline.Close()
 
@@ -306,8 +308,8 @@ func TestMonitoring(t *testing.T) {
 				})
 			}
 			return "output_name", outputs.Group{
-				BatchSize: batchSize,
-				Clients:   clients,
+				BatchEvents: batchSize,
+				Clients:     clients,
 			}, nil
 		},
 	)

libbeat/publisher/pipeline/consumer.go

@@ -58,10 +58,11 @@ type eventConsumer struct {
 // consumerTarget specifies the queue to read from, the parameters needed
 // to generate a batch, and the output channel to send batches to.
 type consumerTarget struct {
-	queue      queue.Queue
-	ch         chan publisher.Batch
-	timeToLive int
-	batchSize  int
+	queue       queue.Queue
+	ch          chan publisher.Batch
+	timeToLive  int
+	batchEvents int
+	batchBytes  int
 }
 
 // retryRequest is used by ttlBatch to add itself back to the eventConsumer
@@ -134,7 +135,8 @@ outerLoop:
 			c.queueReader.req <- queueReaderRequest{
 				queue:      target.queue,
 				retryer:    c,
-				batchSize:  target.batchSize,
+				eventCount: target.batchEvents,
+				byteCount:  target.batchBytes,
 				timeToLive: target.timeToLive,
 			}
 		}

libbeat/publisher/pipeline/controller.go

@@ -159,10 +159,11 @@ func (c *outputController) Set(outGrp outputs.Group) {
 	// Resume consumer targeting the new work queue
 	c.consumer.setTarget(
 		consumerTarget{
-			queue:      c.queue,
-			ch:         targetChan,
-			batchSize:  outGrp.BatchSize,
-			timeToLive: outGrp.Retry + 1,
+			queue:       c.queue,
+			ch:          targetChan,
+			batchEvents: outGrp.BatchEvents,
+			batchBytes:  outGrp.BatchBytes,
+			timeToLive:  outGrp.Retry + 1,
 		})
 }
 
@@ -285,7 +286,9 @@ func (c *outputController) createQueueIfNeeded(outGrp outputs.Group) {
 	if err != nil {
 		logger.Errorf("queue creation failed, falling back to default memory queue, check your queue configuration")
 		s, _ := memqueue.SettingsForUserConfig(nil)
-		queue = memqueue.NewQueue(logger, queueObserver, s, c.inputQueueSize, outGrp.EncoderFactory)
+		// Memqueue creation can only fail when it's configured for byte-based limits,
+		// so we don't need to handle the fallback error.
+		queue, _ = memqueue.NewQueue(logger, queueObserver, s, c.inputQueueSize, outGrp.EncoderFactory)
 	}
 	c.queue = queue
 
@@ -307,12 +310,12 @@ func (c *outputController) createQueueIfNeeded(outGrp outputs.Group) {
 // a producer for a nonexistent queue.
 type emptyProducer struct{}
 
-func (emptyProducer) Publish(_ queue.Entry) (queue.EntryID, bool) {
-	return 0, false
+func (emptyProducer) Publish(_ queue.Entry) bool {
+	return false
 }
 
-func (emptyProducer) TryPublish(_ queue.Entry) (queue.EntryID, bool) {
-	return 0, false
+func (emptyProducer) TryPublish(_ queue.Entry) bool {
+	return false
 }
 
 func (emptyProducer) Close() {