Add otel-specific outputController to libbeat pipeline

libbeat/otel/otelconsumer/otelconsumer.go

@@ -31,10 +31,8 @@ import (
 	"github.com/elastic/beats/v7/libbeat/otel/otelmap"
 	"github.com/elastic/beats/v7/libbeat/outputs"
 	"github.com/elastic/beats/v7/libbeat/publisher"
-	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/mapstr"
-	"github.com/elastic/elastic-agent-libs/paths"
 
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumererror"
@@ -49,10 +47,6 @@ const (
 	esDocumentIDAttribute = "elasticsearch.document_id"
 )
 
-func init() {
-	outputs.RegisterType("otelconsumer", makeOtelConsumer)
-}
-
 type otelConsumer struct {
 	observer       outputs.Observer
 	logsConsumer   consumer.Logs
@@ -61,12 +55,7 @@ type otelConsumer struct {
 	isReceiverTest bool // whether we are running in receivertest context
 }
 
-func makeOtelConsumer(_ outputs.IndexManager, beat beat.Info, observer outputs.Observer, cfg *config.C, beatPaths *paths.Path) (outputs.Group, error) {
-	ocConfig := defaultConfig()
-	if err := cfg.Unpack(&ocConfig); err != nil {
-		return outputs.Fail(err)
-	}
-
+func MakeOtelConsumer(beat beat.Info, observer outputs.Observer) (outputs.Group, error) {
 	isReceiverTest := os.Getenv("OTELCONSUMER_RECEIVERTEST") == "1"
 
 	// Default to runtime.NumCPU() workers
@@ -81,7 +70,7 @@ func makeOtelConsumer(_ outputs.IndexManager, beat beat.Info, observer outputs.O
 		})
 	}
 
-	return outputs.Success(ocConfig.Queue, -1, 0, nil, beat.Logger, beatPaths, clients...)
+	return outputs.Group{Clients: clients}, nil
 }
 
 // Close is a noop for otelconsumer

libbeat/publisher/includes/includes.go

@@ -19,7 +19,6 @@ package includes
 
 import (
 	// import queue types
-	_ "github.com/elastic/beats/v7/libbeat/otel/otelconsumer"
 	_ "github.com/elastic/beats/v7/libbeat/outputs/codec/format"
 	_ "github.com/elastic/beats/v7/libbeat/outputs/codec/json"
 	_ "github.com/elastic/beats/v7/libbeat/outputs/console"

libbeat/publisher/pipeline/output_otel.go

@@ -0,0 +1,138 @@
+// 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 pipeline
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/otel/otelconsumer"
+	"github.com/elastic/beats/v7/libbeat/outputs"
+	"github.com/elastic/beats/v7/libbeat/publisher"
+	"github.com/elastic/beats/v7/libbeat/publisher/queue"
+	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/monitoring"
+)
+
+type otelOutputController struct {
+	beatInfo beat.Info
+	logger   *logp.Logger
+	monitors Monitors
+	queue    queue.Queue[publisher.Event]
+
+	// consumer is a helper goroutine that reads event batches from the queue
+	// and sends them to workerChan for an output worker to process.
+	consumer *eventConsumer
+
+	// Each worker is a goroutine that will read batches from workerChan and
+	// send them to the output.
+	workers    []outputWorker
+	workerChan chan publisher.Batch
+}
+
+func newOTelOutputController(
+	beatInfo beat.Info,
+	monitors Monitors,
+	retryObserver retryObserver,
+	queueFactory queue.QueueFactory[publisher.Event],
+) (*otelOutputController, error) {
+
+	// Queue metrics are reported under the pipeline namespace
+	var pipelineMetrics *monitoring.Registry
+	if monitors.Metrics != nil {
+		pipelineMetrics = monitors.Metrics.GetOrCreateRegistry("pipeline")
+	}
+	queueObserver := queue.NewQueueObserver(pipelineMetrics)
+
+	// Create the queue
+	queue, err := queueFactory(monitors.Logger, queueObserver, 0, nil)
+	if err != nil {
+		return nil, fmt.Errorf("queue creation failed: %w", err)
+	}
+
+	// Initialize output group
+	out, err := loadOutput(monitors, func(outStats outputs.Observer) (string, outputs.Group, error) {
+		out, err := otelconsumer.MakeOtelConsumer(beatInfo, outStats)
+		return "otelconsumer", out, err
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	// Create output workers
+	workerChan := make(chan publisher.Batch)
+	workers := make([]outputWorker, len(out.Clients))
+	logger := beatInfo.Logger.Named("otel_output_worker")
+	for i, client := range out.Clients {
+		workers[i] = makeClientWorker(workerChan, client, logger, monitors.Tracer)
+	}
+
+	// Create an event consumer pulling batches from the queue and sending
+	// them to the output worker channel.
+	consumer := newEventConsumer(monitors.Logger, retryObserver)
+	consumer.setTarget(
+		consumerTarget{
+			queue:      queue,
+			ch:         workerChan,
+			batchSize:  out.BatchSize,
+			timeToLive: out.Retry + 1,
+		})
+
+	return &otelOutputController{
+		beatInfo:   beatInfo,
+		logger:     beatInfo.Logger.Named("otelOutputController"),
+		monitors:   monitors,
+		queue:      queue,
+		consumer:   consumer,
+		workers:    workers,
+		workerChan: workerChan,
+	}, nil
+}
+
+func (c *otelOutputController) waitClose(ctx context.Context, _ bool) error {
+	// First: signal the queue that we're shutting down, and allow it to drain
+	// and process ACKs until the given context terminates.
+	c.logger.Infof("Output shutdown started. Waiting for enqueued events to be published.")
+	c.queue.Close(false)
+	select {
+	case <-c.queue.Done():
+		c.logger.Infof("Continue shutdown: All enqueued events have been published.")
+	case <-ctx.Done():
+		c.logger.Infof("Continue shutdown: Time out waiting for events to be published.")
+		c.queue.Close(true)
+		<-c.queue.Done()
+	}
+
+	// We've drained the queue as much as we can, signal eventConsumer to
+	// close, and wait for it to finish. After consumer.close returns,
+	// there will be no more writes to c.workerChan, so it is safe to close.
+	c.consumer.close()
+	close(c.workerChan)
+
+	// Signal the output workers to close.
+	for _, out := range c.workers {
+		out.Close()
+	}
+
+	return nil
+}
+
+func (c *otelOutputController) queueProducer(config queue.ProducerConfig) queue.Producer[publisher.Event] {
+	return c.queue.Producer(config)
+}

libbeat/publisher/pipeline/output_otel_test.go

@@ -0,0 +1,55 @@
+// 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 pipeline
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/publisher"
+	"github.com/elastic/beats/v7/libbeat/publisher/queue/memqueue"
+	"github.com/elastic/elastic-agent-libs/logp/logptest"
+	"github.com/elastic/elastic-agent-libs/monitoring"
+)
+
+func TestOTelQueueMetrics(t *testing.T) {
+	// More thorough testing of queue metrics are in the queue package,
+	// here we just want to make sure that they appear under the right
+	// monitoring namespace.
+	reg := monitoring.NewRegistry()
+	logger := logptest.NewTestingLogger(t, "")
+	controller, err := newOTelOutputController(
+		beat.Info{Logger: logger},
+		Monitors{
+			Logger:  logger,
+			Metrics: reg,
+		},
+		nilObserver,
+		memqueue.FactoryForSettings[publisher.Event](memqueue.Settings{Events: 1000}))
+	require.NoError(t, err, "creating OTel output controller should succeed")
+	defer controller.waitClose(context.Background(), true)
+	entry := reg.Get("pipeline.queue.max_events")
+	require.NotNil(t, entry, "pipeline.queue.max_events must exist")
+	value, ok := entry.(*monitoring.Uint)
+	require.True(t, ok, "pipeline.queue.max_events must be a *monitoring.Uint")
+	assert.Equal(t, uint64(1000), value.Get(), "pipeline.queue.max_events should match the events configuration key")
+}

libbeat/publisher/pipeline/controller.go → libbeat/publisher/pipeline/output_process.go

@@ -32,11 +32,6 @@ import (
 	"github.com/elastic/elastic-agent-libs/monitoring"
 )
 
-type outputController interface {
-	waitClose(ctx context.Context, force bool) error
-	queueProducer(config queue.ProducerConfig) queue.Producer[publisher.Event]
-}
-
 // processOutputController manages the pipelines output capabilities, like:
 // - start
 // - stop
@@ -98,7 +93,7 @@ func newProcessOutputController(
 ) (*processOutputController, error) {
 	controller := &processOutputController{
 		beat:           beat,
-		logger:         beat.Logger.Named("outputController"),
+		logger:         beat.Logger.Named("processOutputController"),
 		monitors:       monitors,
 		queueFactory:   queueFactory,
 		workerChan:     make(chan publisher.Batch),
@@ -195,8 +190,8 @@ func (c *processOutputController) Reload(
 	return nil
 }
 
-// Close the queue, waiting up to the specified timeout for pending events
-// to complete.
+// Close the queue and output, waiting for pending events until all are
+// acknowledged or the provided context expires.
 func (c *processOutputController) closeQueue(ctx context.Context, force bool) {
 	c.queueLock.Lock()
 	defer c.queueLock.Unlock()

libbeat/publisher/pipeline/pipeline.go

@@ -88,6 +88,7 @@
 	// When and how WaitClose is applied depends on WaitCloseMode.
 	WaitClose time.Duration
 
+	// This field has no effect when running as a Beats receiver.
 	WaitCloseMode WaitCloseMode
 
 	Processors processing.Supporter
@@ -118,6 +119,21 @@
 	WaitOnPipelineCloseThenForce
 )
 
+// outputController is the interface between the Pipeline and the output,
+// which may be either the legacy Beats output pipeline (under the process
+// runtime) or a bridge to the OTel Collector (when running as a Beats
+// receiver under the otel runtime).
+type outputController interface {
+	// queueProducer creates a queue producer with the given config, blocking
+	// until the queue is created if it does not yet exist.
+	queueProducer(config queue.ProducerConfig) queue.Producer[publisher.Event]
+
+	// Close the queue and output, waiting for pending events until all are
+	// acknowledged or the provided context expires.
+	// The force parameter has no effect when running as a Beats receiver.
+	waitClose(ctx context.Context, force bool) error
+}
+
 // OutputReloader interface, that can be queried from an active publisher pipeline.
 // The output reloader can be used to change the active output.
 type OutputReloader interface {
@@ -138,7 +154,7 @@
 	settings Settings,
 ) (*Pipeline, error) {
 	if monitors.Logger == nil {
 		monitors.Logger = logp.NewLogger("publish")
 	}
 
 	p := &Pipeline{
@@ -149,13 +165,6 @@
 		processors:       settings.Processors,
 		paths:            settings.Paths,
 	}
-	switch settings.WaitCloseMode {
-	case WaitOnPipelineClose, WaitOnPipelineCloseThenForce:
-		if settings.WaitClose > 0 {
-			p.waitCloseTimeout = settings.WaitClose
-		}
-	default:
-	}
 
 	p.forceCloseQueue = settings.WaitCloseMode == WaitOnPipelineCloseThenForce
 
@@ -185,6 +194,41 @@
 	return p, nil
 }
 
+func NewForReceiver(
+	beatInfo beat.Info,
+	monitors Monitors,
+	userQueueConfig conf.Namespace,
+	settings Settings,
+) (*Pipeline, error) {
+	p := &Pipeline{
+		beatInfo:         beatInfo,
+		monitors:         monitors,
+		observer:         newMetricsObserver(monitors.Metrics),
+		waitCloseTimeout: settings.WaitClose,
+		processors:       settings.Processors,
+		paths:            settings.Paths,
+	}
+
+	// Convert the raw queue config to a parsed Settings object that will
+	// be used during queue creation. This lets us fail immediately on startup
+	// if there's a configuration problem.
+	queueType := defaultQueueType
+	if b := userQueueConfig.Name(); b != "" {
+		queueType = b
+	}
+	queueFactory, err := queueFactoryForUserConfig(queueType, userQueueConfig.Config(), settings.Paths)
+	if err != nil {
+		return nil, err
+	}
+
+	p.outputController, err = newOTelOutputController(beatInfo, monitors, p.observer, queueFactory)
+	if err != nil {
+		return nil, err
+	}
+
+	return p, nil
+}
+
 // Close stops the pipeline, outputs and queue.
 // If WaitClose with WaitOnPipelineClose mode is configured, Close will block
 // for a duration of WaitClose, if there are still active events in the pipeline.