x-pack/filebeat/input/lumberjack: add fleet status update functionality (#44339)

Author: efd6

CHANGELOG.next.asciidoc

@@ -431,7 +431,7 @@ otherwise no tag is added. {issue}42208[42208] {pull}42403[42403]
 - Add milliseconds to document timestamp from awscloudwatch Filebeat input {pull}44306[44306]
 - Add support to the Active Directory entity analytics provider for device entities. {pull}44309[44309]
 - Add support for OPTIONS request to HTTP Endpoint input. {issue}43930[43930] {pull}44387[44387]
-
+- Add Fleet status update functionality to lumberjack input. {issue}44283[44283] {pull}44339[44339]
 
 *Auditbeat*
 

x-pack/filebeat/input/lumberjack/input.go

@@ -10,6 +10,7 @@ import (
 	inputv2 "github.com/elastic/beats/v7/filebeat/input/v2"
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/feature"
+	"github.com/elastic/beats/v7/libbeat/management/status"
 	conf "github.com/elastic/elastic-agent-libs/config"
 )
 
@@ -49,23 +50,27 @@ func newLumberjackInput(lumberjackConfig config) (*lumberjackInput, error) {
 func (i *lumberjackInput) Name() string { return inputName }
 
 func (i *lumberjackInput) Test(inputCtx inputv2.TestContext) error {
-	s, err := newServer(i.config, inputCtx.Logger, nil, nil)
+	s, err := newServer(i.config, inputCtx.Logger, nil, nil, nil)
 	if err != nil {
 		return err
 	}
 	return s.Close()
 }
 
 func (i *lumberjackInput) Run(inputCtx inputv2.Context, pipeline beat.Pipeline) error {
+	inputCtx.UpdateStatus(status.Starting, "")
 	inputCtx.Logger.Info("Starting " + inputName + " input")
 	defer inputCtx.Logger.Info(inputName + " input stopped")
 
+	inputCtx.UpdateStatus(status.Configuring, "")
 	// Create client for publishing events and receive notification of their ACKs.
 	client, err := pipeline.ConnectWith(beat.ClientConfig{
 		EventListener: newEventACKHandler(),
 	})
 	if err != nil {
-		return fmt.Errorf("failed to create pipeline client: %w", err)
+		err := fmt.Errorf("failed to create pipeline client: %w", err)
+		inputCtx.UpdateStatus(status.Failed, err.Error())
+		return err
 	}
 	defer client.Close()
 
@@ -74,7 +79,7 @@ func (i *lumberjackInput) Run(inputCtx inputv2.Context, pipeline beat.Pipeline)
 	metrics := newInputMetrics(inputCtx.ID, nil)
 	defer metrics.Close()
 
-	s, err := newServer(i.config, inputCtx.Logger, client.Publish, metrics)
+	s, err := newServer(i.config, inputCtx.Logger, client.Publish, inputCtx.StatusReporter, metrics)
 	if err != nil {
 		return err
 	}
@@ -83,6 +88,7 @@ func (i *lumberjackInput) Run(inputCtx inputv2.Context, pipeline beat.Pipeline)
 	// Shutdown the server when cancellation is signaled.
 	go func() {
 		<-inputCtx.Cancelation.Done()
+		inputCtx.UpdateStatus(status.Stopping, "")
 		s.Close()
 	}()
 

x-pack/filebeat/input/lumberjack/server.go

@@ -14,6 +14,7 @@ import (
 	"golang.org/x/net/netutil"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/management/status"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/monitoring"
 	"github.com/elastic/elastic-agent-libs/transport/tlscommon"
@@ -23,6 +24,7 @@ import (
 
 type server struct {
 	config         config
+	status         status.StatusReporter
 	log            *logp.Logger
 	publish        func(beat.Event)
 	metrics        *inputMetrics
@@ -31,9 +33,13 @@ type server struct {
 	bindAddress    string
 }
 
-func newServer(c config, log *logp.Logger, pub func(beat.Event), metrics *inputMetrics) (*server, error) {
+func newServer(c config, log *logp.Logger, pub func(beat.Event), stat status.StatusReporter, metrics *inputMetrics) (*server, error) {
+	if stat == nil {
+		stat = noopReporter{}
+	}
 	ljSvr, bindAddress, err := newLumberjack(c)
 	if err != nil {
+		stat.UpdateStatus(status.Failed, "failed to start lumberjack server: "+err.Error())
 		return nil, err
 	}
 
@@ -50,6 +56,7 @@ func newServer(c config, log *logp.Logger, pub func(beat.Event), metrics *inputM
 
 	return &server{
 		config:      c,
+		status:      stat,
 		log:         log,
 		publish:     pub,
 		metrics:     metrics,
@@ -58,20 +65,27 @@ func newServer(c config, log *logp.Logger, pub func(beat.Event), metrics *inputM
 	}, nil
 }
 
+type noopReporter struct{}
+
+func (noopReporter) UpdateStatus(status.Status, string) {}
+
 func (s *server) Close() error {
+	s.status.UpdateStatus(status.Stopping, "")
 	var err error
 	s.ljSvrCloseOnce.Do(func() {
 		err = s.ljSvr.Close()
 	})
+	s.status.UpdateStatus(status.Stopped, "")
 	return err
 }
 
 func (s *server) Run() error {
 	// Process batches until the input is stopped.
+	s.status.UpdateStatus(status.Running, "")
 	for batch := range s.ljSvr.ReceiveChan() {
 		s.processBatch(batch)
 	}
-
+	s.status.UpdateStatus(status.Stopped, "")
 	return nil
 }
 

x-pack/filebeat/input/lumberjack/server_test.go

@@ -78,7 +78,7 @@ func testSendReceive(t testing.TB, c config, numberOfEvents int, clientTLSConfig
 	collect := newEventCollector(ctx, numberOfEvents)
 
 	// Start server.
-	s, err := newServer(c, log, collect.Publish, nil)
+	s, err := newServer(c, log, collect.Publish, nil, nil)
 	require.NoError(t, err)
 	go func() {
 		<-ctx.Done()