feat: add status reporting to S3 (#45748)

* Add health status reporting

* Move status reporter helper from CloudWatch status reporting to libbeat

* Enhance status reporter helper

* Fix linting issues, esp related to logging

Author: joecompute

CHANGELOG.next.asciidoc

@@ -510,6 +510,7 @@ otherwise no tag is added. {issue}42208[42208] {pull}42403[42403]
 - Add status reporting support for AWS CloudWatch input. {pull}45679[45679]
 - Add mechanism to allow HTTP JSON templates to terminate without logging an error. {issue}45664[45664] {pull}45810[45810]
 - Improve error reporting for schemeless URLs in HTTP JSON input. {pull}45953[45953]
+- Add status reporting support for AWS S3 input. {pull}45748[45748]
 
 *Auditbeat*
 

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

@@ -21,6 +21,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/management/status"
 	"github.com/elastic/beats/v7/libbeat/statestore"
 	awscommon "github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
+	"github.com/elastic/beats/v7/x-pack/libbeat/statusreporterhelper"
 	conf "github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/go-concert/unison"
@@ -94,7 +95,7 @@ func (in *cloudwatchInput) Run(inputContext v2.Context, pipeline beat.Pipeline)
 	log := inputContext.Logger
 
 	// setup status reporter
-	in.status = newCWStateReporter(inputContext, log)
+	in.status = statusreporterhelper.New(inputContext.StatusReporter, log, "CloudWatch")
 
 	defer in.status.UpdateStatus(status.Stopped, "")
 	in.status.UpdateStatus(status.Starting, "Input starting")

x-pack/filebeat/input/awscloudwatch/state_reporter.go

@@ -223,6 +223,7 @@ func benchmarkInputSQS(t *testing.T, workerCount int) testing.BenchmarkResult {
 		config.NumberOfWorkers = workerCount
 		sqsReader := newSQSReaderInput(config, aws.Config{})
 		sqsReader.log = log.Named("sqs")
+		sqsReader.status = &statusReporterHelperMock{}
 		sqsReader.pipeline = newFakePipeline()
 		sqsReader.metrics = newInputMetrics(monitoring.NewRegistry(), workerCount)
 		sqsReader.sqs, err = newConstantSQS()
@@ -235,6 +236,7 @@ func benchmarkInputSQS(t *testing.T, workerCount int) testing.BenchmarkResult {
 		b.Cleanup(cancel)
 
 		go func() {
+			//nolint:gosec // not going to have anywhere near uint64 overflow number of received messages
 			for sqsReader.metrics.sqsMessagesReceivedTotal.Get() < uint64(b.N) {
 				time.Sleep(5 * time.Millisecond)
 			}
@@ -262,8 +264,6 @@ func benchmarkInputSQS(t *testing.T, workerCount int) testing.BenchmarkResult {
 }
 
 func TestBenchmarkInputSQS(t *testing.T) {
-	err := logp.TestingSetup(logp.WithLevel(logp.InfoLevel))
-	require.NoError(t, err)
 
 	results := []testing.BenchmarkResult{
 		benchmarkInputSQS(t, 1),
@@ -352,6 +352,7 @@ func benchmarkInputS3(t *testing.T, numberOfWorkers int) testing.BenchmarkResult
 					states:          states,
 					provider:        "provider",
 					filterProvider:  newFilterProvider(&config),
+					status:          &statusReporterHelperMock{},
 				}
 
 				s3Poller.run(ctx)
@@ -392,8 +393,6 @@ func benchmarkInputS3(t *testing.T, numberOfWorkers int) testing.BenchmarkResult
 }
 
 func TestBenchmarkInputS3(t *testing.T) {
-	err := logp.TestingSetup(logp.WithLevel(logp.InfoLevel))
-	require.NoError(t, err)
 
 	results := []testing.BenchmarkResult{
 		benchmarkInputS3(t, 1),

x-pack/filebeat/input/awscloudwatch/state_reporter_test.go

@@ -6,17 +6,50 @@ package awss3
 
 import (
 	"errors"
+	"sync"
 	"testing"
 
 	awssdk "github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/stretchr/testify/assert"
 
 	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
+	"github.com/elastic/beats/v7/libbeat/management/status"
 	awscommon "github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/monitoring"
 )
 
+// statusReporterHelperMock is a thread-safe mock of a status reporter that
+// behaves like StatusReporterHelper
+type statusReporterHelperMock struct {
+	mu       sync.Mutex
+	statuses []mgmtStatusUpdate
+	current  status.Status
+}
+
+type mgmtStatusUpdate struct {
+	status status.Status
+	msg    string
+}
+
+func (r *statusReporterHelperMock) getStatuses() []mgmtStatusUpdate {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	s := make([]mgmtStatusUpdate, len(r.statuses))
+	copy(s, r.statuses)
+	return s
+}
+
+func (r *statusReporterHelperMock) UpdateStatus(s status.Status, msg string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	// Imitate behavior of statusReporterHelper. Only record if the new status is different.
+	if s != r.current {
+		r.current = s
+		r.statuses = append(r.statuses, mgmtStatusUpdate{status: s, msg: msg})
+	}
+}
+
 func TestGetProviderFromDomain(t *testing.T) {
 	tests := []struct {
 		endpoint string
@@ -155,6 +188,7 @@ func TestRegionSelection(t *testing.T) {
 				MetricsRegistry: monitoring.NewRegistry(),
 			}
 
+			in.status = &statusReporterHelperMock{}
 			// Run setup and verify that it put the correct region in awsConfig.Region
 			err := in.setup(inputCtx, &fakePipeline{})
 			in.cleanup()

x-pack/filebeat/input/awss3/input_benchmark_test.go

@@ -16,7 +16,9 @@ import (
 	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/common/backoff"
+	"github.com/elastic/beats/v7/libbeat/management/status"
 	"github.com/elastic/beats/v7/libbeat/statestore"
+	"github.com/elastic/beats/v7/x-pack/libbeat/statusreporterhelper"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/go-concert/timed"
 )
@@ -37,6 +39,9 @@ type s3PollerInput struct {
 	s3ObjectHandler s3ObjectHandlerFactory
 	states          *states
 	filterProvider  *filterProvider
+
+	// health status reporting
+	status status.StatusReporter
 }
 
 func newS3PollerInput(
@@ -62,21 +67,32 @@ func (in *s3PollerInput) Run(
 	inputContext v2.Context,
 	pipeline beat.Pipeline,
 ) error {
+
 	in.log = inputContext.Logger.Named("s3")
+
+	in.status = statusreporterhelper.New(inputContext.StatusReporter, in.log, "S3")
+	defer in.status.UpdateStatus(status.Stopped, "")
+	in.status.UpdateStatus(status.Starting, "Input starting")
+
 	in.pipeline = pipeline
 	var err error
 
 	// Load the persistent S3 polling state.
 	in.states, err = newStates(in.log, in.store, in.config.BucketListPrefix)
 	if err != nil {
-		return fmt.Errorf("can not start persistent store: %w", err)
+		err = fmt.Errorf("can not start persistent store: %w", err)
+		in.status.UpdateStatus(status.Failed, fmt.Sprintf("Setup failure: %s", err.Error()))
+		return err
 	}
 	defer in.states.Close()
 
 	ctx := v2.GoContextFromCanceler(inputContext.Cancelation)
+	in.status.UpdateStatus(status.Configuring, "Configuring input")
 	in.s3, err = in.createS3API(ctx)
 	if err != nil {
-		return fmt.Errorf("failed to create S3 API: %w", err)
+		err = fmt.Errorf("failed to create S3 API for bucket ARN '%s': Error: %w", in.config.getBucketARN(), err)
+		in.status.UpdateStatus(status.Failed, fmt.Sprintf("Setup failure: %s", err.Error()))
+		return err
 	}
 
 	in.metrics = newInputMetrics(inputContext.MetricsRegistry, in.config.NumberOfWorkers)
@@ -97,6 +113,7 @@ func (in *s3PollerInput) Run(
 func (in *s3PollerInput) run(ctx context.Context) {
 	// Scan the bucket in a loop, delaying by the configured interval each
 	// iteration.
+	in.status.UpdateStatus(status.Running, "Input is running")
 	for ctx.Err() == nil {
 		in.runPoll(ctx)
 		_ = timed.Wait(ctx, in.config.BucketListInterval)
@@ -129,6 +146,7 @@ func (in *s3PollerInput) runPoll(ctx context.Context) {
 	err := in.states.CleanUp(ids)
 	if err != nil {
 		in.log.Errorf("failed to cleanup states: %v", err.Error())
+		in.status.UpdateStatus(status.Degraded, fmt.Sprintf("Input state cleanup failure: %s", err.Error()))
 	}
 }
 
@@ -138,13 +156,13 @@ func (in *s3PollerInput) workerLoop(ctx context.Context, workChan <-chan state)
 	client, err := createPipelineClient(in.pipeline, acks)
 	if err != nil {
 		in.log.Errorf("failed to create pipeline client: %v", err.Error())
+		in.status.UpdateStatus(status.Degraded, fmt.Sprintf("A worker's pipeline client setup failed, error: %s", err.Error()))
 		return
 	}
 	defer client.Close()
 	defer acks.Close()
 
 	rateLimitWaiter := backoff.NewEqualJitterBackoff(ctx.Done(), 1, 120)
-
 	for _state := range workChan {
 		state := _state
 		event := in.s3EventForState(state)
@@ -166,6 +184,9 @@ func (in *s3PollerInput) workerLoop(ctx context.Context, workChan <-chan state)
 		if errors.Is(err, errS3DownloadFailed) {
 			// Download errors are ephemeral. Add a backoff delay, then skip to the
 			// next iteration so we don't mark the object as permanently failed.
+			in.status.UpdateStatus(status.Degraded,
+				fmt.Sprintf("S3 download failure for object key '%s' in bucket '%s': %s",
+					state.Key, state.Bucket, err.Error()))
 			rateLimitWaiter.Wait()
 			continue
 		}
@@ -176,7 +197,10 @@ func (in *s3PollerInput) workerLoop(ctx context.Context, workChan <-chan state)
 		if err != nil {
 			in.log.Errorf("failed processing S3 event for object key %q in bucket %q: %v",
 				state.Key, state.Bucket, err.Error())
-
+			in.status.UpdateStatus(status.Degraded,
+				fmt.Sprintf(
+					"S3 object processing failure for object key '%s' in bucket '%s': %s",
+					state.Key, state.Bucket, err.Error()))
 			// Non-retryable error.
 			state.Failed = true
 		} else {
@@ -188,6 +212,9 @@ func (in *s3PollerInput) workerLoop(ctx context.Context, workChan <-chan state)
 			err := in.states.AddState(state)
 			if err != nil {
 				in.log.Errorf("saving completed object state: %v", err.Error())
+				in.status.UpdateStatus(status.Degraded, fmt.Sprintf("Failure checkpointing (saving completed object state): %s", err.Error()))
+			} else {
+				in.status.UpdateStatus(status.Running, "Input is running")
 			}
 
 			// Metrics
@@ -213,12 +240,14 @@ func (in *s3PollerInput) readerLoop(ctx context.Context, workChan chan<- state)
 
 		if err != nil {
 			in.log.Warnw("Error when paginating listing.", "error", err)
+			in.status.UpdateStatus(status.Degraded, fmt.Sprintf("S3 pagination error: %s", err.Error()))
 			// QuotaExceededError is client-side rate limiting in the AWS sdk,
 			// don't include it in the circuit breaker count
 			if !errors.As(err, &ratelimit.QuotaExceededError{}) {
 				circuitBreaker++
 				if circuitBreaker >= readerLoopMaxCircuitBreaker {
 					in.log.Warnw(fmt.Sprintf("%d consecutive error when paginating listing, breaking the circuit.", circuitBreaker), "error", err)
+					in.status.UpdateStatus(status.Degraded, fmt.Sprintf("Too many consecutive errors (%d) in S3 pagination. Error: %s", circuitBreaker, err.Error()))
 					return nil, false
 				}
 			}