Add docs / parameter checks

Author: faec

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/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{}
@@ -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()
 

libbeat/publisher/pipeline/controller.go

@@ -286,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
 

libbeat/publisher/queue/memqueue/broker.go

@@ -19,6 +19,7 @@ package memqueue
 
 import (
 	"context"
+	"errors"
 	"io"
 	"time"
 
@@ -142,7 +143,7 @@ func FactoryForSettings(settings Settings) queue.QueueFactory {
 		inputQueueSize int,
 		encoderFactory queue.EncoderFactory,
 	) (queue.Queue, error) {
-		return NewQueue(logger, observer, settings, inputQueueSize, encoderFactory), nil
+		return NewQueue(logger, observer, settings, inputQueueSize, encoderFactory)
 	}
 }
 
@@ -155,14 +156,16 @@ func NewQueue(
 	settings Settings,
 	inputQueueSize int,
 	encoderFactory queue.EncoderFactory,
-) *broker {
-	b := newQueue(logger, observer, settings, inputQueueSize, encoderFactory)
+) (*broker, error) {
+	b, err := newQueue(logger, observer, settings, inputQueueSize, encoderFactory)
 
-	// Start the queue workers
-	go b.runLoop.run()
-	go b.ackLoop.run()
+	if err == nil {
+		// Start the queue workers
+		go b.runLoop.run()
+		go b.ackLoop.run()
+	}
 
-	return b
+	return b, err
 }
 
 // newQueue does most of the work of creating a queue from the given
@@ -175,7 +178,7 @@ func newQueue(
 	settings Settings,
 	inputQueueSize int,
 	encoderFactory queue.EncoderFactory,
-) *broker {
+) (*broker, error) {
 	if observer == nil {
 		observer = queue.NewQueueObserver(nil)
 	}
@@ -190,6 +193,10 @@ func newQueue(
 		settings.MaxGetRequest = (settings.Events + 1) / 2
 	}
 
+	if settings.Bytes > 0 && encoderFactory == nil {
+		return nil, errors.New("queue.mem.bytes is set but the output doesn't support byte-based event buffers")
+	}
+
 	// Can't request more than the full queue
 	if settings.Events > 0 && settings.MaxGetRequest > settings.Events {
 		settings.MaxGetRequest = settings.Events
@@ -222,7 +229,7 @@ func newQueue(
 
 	observer.MaxEvents(settings.Events)
 
-	return b
+	return b, nil
 }
 
 func (b *broker) Close() error {

libbeat/publisher/queue/memqueue/queue_test.go

@@ -86,12 +86,13 @@ func TestProduceConsumer(t *testing.T) {
 // than 2 events to it, p.Publish will block, once we call q.Close,
 // we ensure the 3rd event was not successfully published.
 func TestProducerDoesNotBlockWhenQueueClosed(t *testing.T) {
-	q := NewQueue(nil, nil,
+	q, err := NewQueue(nil, nil,
 		Settings{
 			Events:        2, // Queue size
 			MaxGetRequest: 1, // make sure the queue won't buffer events
 			FlushTimeout:  time.Millisecond,
 		}, 0, nil)
+	require.NoError(t, err, "Queue creation must succeed")
 
 	p := q.Producer(queue.ProducerConfig{
 		// We do not read from the queue, so the callbacks are never called
@@ -156,12 +157,13 @@ func TestProducerClosePreservesEventCount(t *testing.T) {
 
 	var activeEvents atomic.Int64
 
-	q := NewQueue(nil, nil,
+	q, err := NewQueue(nil, nil,
 		Settings{
 			Events:        3, // Queue size
 			MaxGetRequest: 2,
 			FlushTimeout:  10 * time.Millisecond,
 		}, 1, nil)
+	require.NoError(t, err, "Queue creation must succeed")
 
 	p := q.Producer(queue.ProducerConfig{
 		ACK: func(count int) {
@@ -229,11 +231,12 @@ func TestProducerClosePreservesEventCount(t *testing.T) {
 
 func makeTestQueue(sz, minEvents int, flushTimeout time.Duration) queuetest.QueueFactory {
 	return func(_ *testing.T) queue.Queue {
-		return NewQueue(nil, nil, Settings{
+		q, _ := NewQueue(nil, nil, Settings{
 			Events:        sz,
 			MaxGetRequest: minEvents,
 			FlushTimeout:  flushTimeout,
 		}, 0, nil)
+		return q
 	}
 }
 

libbeat/publisher/queue/memqueue/runloop_test.go

@@ -38,7 +38,7 @@ func TestFlushSettingsDoNotBlockFullBatches(t *testing.T) {
 	// available. This test verifies that Get requests that can be completely
 	// filled do not wait for the flush timer.
 
-	broker := newQueue(
+	broker, err := newQueue(
 		logp.NewLogger("testing"),
 		nil,
 		Settings{
@@ -47,6 +47,7 @@ func TestFlushSettingsDoNotBlockFullBatches(t *testing.T) {
 			FlushTimeout:  10 * time.Second,
 		},
 		10, nil)
+	require.NoError(t, err, "Queue creation must succeed")
 
 	producer := newProducer(broker, nil, nil)
 	rl := broker.runLoop
@@ -77,7 +78,7 @@ func TestFlushSettingsBlockPartialBatches(t *testing.T) {
 	// there are enough events. This one uses the same setup to confirm that
 	// Get requests are delayed if there aren't enough events.
 
-	broker := newQueue(
+	broker, err := newQueue(
 		logp.NewLogger("testing"),
 		nil,
 		Settings{
@@ -86,6 +87,7 @@ func TestFlushSettingsBlockPartialBatches(t *testing.T) {
 			FlushTimeout:  10 * time.Second,
 		},
 		10, nil)
+	require.NoError(t, err, "Queue creation must succeed")
 
 	producer := newProducer(broker, nil, nil)
 	rl := broker.runLoop