Fix: Implement backup_to_bucket_ and delete_after_backup for aws-s3 polling mode (#49734)

* Fix: Implement backup_to_bucket_ and delete_after_backup for aws-s3 polling mode

* add changelog

* fix linting errors

* Remove status update in every failure

Author: MichaelKatsoulis

changelog/fragments/1783515049-fix-s3-poller-backup-delete.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: Fix aws-s3 input not performing backup_to_bucket and delete_after_backup in polling mode
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: filebeat
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+issue: https://github.com/elastic/beats/issues/46672

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

@@ -235,8 +235,18 @@ func (in *s3PollerInput) workerLoop(ctx context.Context, workChan <-chan state)
 			state.Stored = true
 		}
 
+		finalize := objHandler.FinalizeS3Object
+
 		// Add the cleanup handling to the acks helper
 		acks.Add(publishCount, func() {
+			// Finalize the object (backup/delete) after all events are ACKed.
+			// Only successfully processed objects are finalized.
+			if state.Stored {
+				if err := finalize(); err != nil {
+					in.log.Errorf("failed finalizing S3 object key %q in bucket %q: %v", state.Key, state.Bucket, err)
+				}
+			}
+
 			err := in.registry.AddState(state)
 			if err != nil {
 				in.log.Errorf("saving completed object state: %v", err.Error())

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

@@ -6,11 +6,14 @@ package awss3
 
 import (
 	"context"
+	"io"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/logp/logptest"
 	"github.com/elastic/elastic-agent-libs/monitoring"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -139,7 +142,7 @@ func TestS3Poller(t *testing.T) {
 			BucketListPrefix:   listPrefix,
 			RegionName:         "region",
 		}
-		log := logp.NewLogger(inputName)
+		log := logptest.NewTestingLogger(t, inputName)
 		poller := &s3PollerInput{
 			log:             log,
 			config:          cfg,
@@ -156,6 +159,120 @@ func TestS3Poller(t *testing.T) {
 		poller.runPoll(ctx)
 	})
 
+	t.Run("Poll finalizes objects after ACK (backup + delete)", func(t *testing.T) {
+		store := openTestStatestore()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 3*testTimeout)
+		defer cancel()
+
+		ctrl, ctx := gomock.WithContext(ctx, t)
+		defer ctrl.Finish()
+
+		mockAPI := NewMockS3API(ctrl)
+		mockPager := NewMockS3Pager(ctrl)
+		pipeline := newFakePipeline()
+
+		backupDone := make(chan struct{})
+		deleteDone := make(chan struct{})
+
+		mockAPI.EXPECT().
+			ListObjectsPaginator(gomock.Eq(bucket), gomock.Eq(listPrefix), gomock.Any()).
+			Times(1).
+			DoAndReturn(func(_, _, _ string) s3Pager {
+				return mockPager
+			})
+
+		// Poller listing loop calls HasMorePages before and after NextPage.
+		hasMoreCalls := 0
+		mockPager.EXPECT().
+			HasMorePages().
+			Times(2).
+			DoAndReturn(func() bool {
+				hasMoreCalls++
+				return hasMoreCalls == 1
+			})
+
+		mockPager.EXPECT().
+			NextPage(gomock.Any()).
+			Times(1).
+			DoAndReturn(func(_ context.Context, optFns ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) {
+				return &s3.ListObjectsV2Output{
+					Contents: []types.Object{
+						{
+							ETag:         aws.String("etag1"),
+							Key:          aws.String("key1"),
+							LastModified: aws.Time(time.Now()),
+						},
+					},
+				}, nil
+			})
+
+		mockAPI.EXPECT().
+			GetObject(gomock.Any(), gomock.Eq(""), gomock.Eq(bucket), gomock.Eq("key1")).
+			Times(1).
+			DoAndReturn(func(_ context.Context, _ string, _ string, _ string) (*s3.GetObjectOutput, error) {
+				// A simple text payload that produces at least one event.
+				return &s3.GetObjectOutput{
+					Body: io.NopCloser(strings.NewReader("hello\n")),
+				}, nil
+			})
+
+		mockAPI.EXPECT().
+			CopyObject(gomock.Any(), gomock.Eq(""), gomock.Eq(bucket), gomock.Eq("backup-bucket"), gomock.Eq("key1"), gomock.Eq("processed/key1")).
+			Times(1).
+			DoAndReturn(func(_ context.Context, _ string, _ string, _ string, _ string, _ string) (*s3.CopyObjectOutput, error) {
+				close(backupDone)
+				return &s3.CopyObjectOutput{}, nil
+			})
+
+		mockAPI.EXPECT().
+			DeleteObject(gomock.Any(), gomock.Eq(""), gomock.Eq(bucket), gomock.Eq("key1")).
+			Times(1).
+			DoAndReturn(func(_ context.Context, _ string, _ string, _ string) (*s3.DeleteObjectOutput, error) {
+				close(deleteDone)
+				return &s3.DeleteObjectOutput{}, nil
+			})
+
+		backupCfg := backupConfig{
+			NonAWSBackupToBucketName: "backup-bucket",
+			BackupToBucketPrefix:     "processed/",
+			Delete:                   true,
+		}
+
+		cfg := config{
+			NumberOfWorkers:    1,
+			BucketListInterval: pollInterval,
+			BucketARN:          bucket,
+			BucketListPrefix:   listPrefix,
+			BackupConfig:       backupCfg,
+		}
+		log := logptest.NewTestingLogger(t, inputName)
+
+		s3ObjProc := newS3ObjectProcessorFactory(nil, mockAPI, nil, backupCfg, logp.NewNopLogger())
+		registry, err := newStateRegistry(nil, store, listPrefix, cfg.LexicographicalOrdering, cfg.LexicographicalLookbackKeys)
+		require.NoError(t, err, "registry creation must succeed")
+
+		poller := &s3PollerInput{
+			log:             log,
+			config:          cfg,
+			s3:              mockAPI,
+			pipeline:        pipeline,
+			s3ObjectHandler: s3ObjProc,
+			registry:        registry,
+			provider:        "provider",
+			metrics:         newInputMetrics(monitoring.NewRegistry(), 0, logp.NewNopLogger()),
+			filterProvider:  newFilterProvider(&cfg),
+			strategy:        newPollingStrategy(cfg.LexicographicalOrdering, log),
+			status:          &statusReporterHelperMock{},
+		}
+
+		poller.runPoll(ctx)
+
+		// Finalization happens asynchronously after ACK; wait until it completes.
+		waitForChannel(t, backupDone, 3*testTimeout)
+		waitForChannel(t, deleteDone, 3*testTimeout)
+	})
+
 	t.Run("restart bucket scan after paging errors", func(t *testing.T) {
 		// Change the restart limit to 2 consecutive errors, so the test doesn't
 		// take too long to run
@@ -284,7 +401,7 @@ func TestS3Poller(t *testing.T) {
 			RegionName:         "region",
 		}
 
-		log := logp.NewLogger(inputName)
+		log := logptest.NewTestingLogger(t, inputName)
 		poller := &s3PollerInput{
 			log: log,
 			config: config{
@@ -390,7 +507,7 @@ func TestS3Poller(t *testing.T) {
 			LexicographicalOrdering:     true,
 			LexicographicalLookbackKeys: 100,
 		}
-		log := logp.NewLogger(inputName)
+		log := logptest.NewTestingLogger(t, inputName)
 		poller := &s3PollerInput{
 			log:             log,
 			config:          cfg,
@@ -431,7 +548,7 @@ func TestS3Poller(t *testing.T) {
 			ListObjectsPaginator(gomock.Eq(bucket), gomock.Eq(""), gomock.Eq("")).
 			Times(1).
 			DoAndReturn(func(_, _, startAfterKey string) s3Pager {
-				require.Equal(t, "", startAfterKey)
+				require.Empty(t, startAfterKey)
 				return mockPager
 			})
 
@@ -479,7 +596,7 @@ func TestS3Poller(t *testing.T) {
 			LexicographicalOrdering:     true,
 			LexicographicalLookbackKeys: 100,
 		}
-		log := logp.NewLogger(inputName)
+		log := logptest.NewTestingLogger(t, inputName)
 		poller := &s3PollerInput{
 			log:             log,
 			config:          cfg,
@@ -517,7 +634,7 @@ func TestS3Poller(t *testing.T) {
 			ListObjectsPaginator(gomock.Eq(bucket), gomock.Eq(""), gomock.Eq("")).
 			Times(1).
 			DoAndReturn(func(_, _, startAfterKey string) s3Pager {
-				require.Equal(t, "", startAfterKey)
+				require.Empty(t, startAfterKey)
 				return mockPager
 			})
 
@@ -564,7 +681,7 @@ func TestS3Poller(t *testing.T) {
 			RegionName:              "region",
 			LexicographicalOrdering: false,
 		}
-		log := logp.NewLogger(inputName)
+		log := logptest.NewTestingLogger(t, inputName)
 		poller := &s3PollerInput{
 			log:             log,
 			config:          cfg,
@@ -702,7 +819,7 @@ func TestS3Poller(t *testing.T) {
 			BucketListPrefix:   "",
 			RegionName:         "region",
 		}
-		log := logp.NewLogger(inputName)
+		log := logptest.NewTestingLogger(t, inputName)
 		poller := &s3PollerInput{
 			log:             log,
 			config:          cfg,
@@ -728,9 +845,19 @@ func TestS3Poller(t *testing.T) {
 	})
 }
 
+func waitForChannel(t *testing.T, ch <-chan struct{}, timeout time.Duration) {
+	t.Helper()
+	select {
+	case <-ch:
+		return
+	case <-time.After(timeout):
+		t.Fatal("timed out waiting for async finalization")
+	}
+}
+
 func Test_S3StateHandling(t *testing.T) {
 	bucket := "bucket"
-	logger := logp.NewLogger(inputName)
+	logger := logptest.NewTestingLogger(t, inputName)
 	fixedTimeNow := time.Now()
 
 	tests := []struct {
@@ -933,6 +1060,11 @@ func Test_S3StateHandling(t *testing.T) {
 					eventCallback(beat.Event{})
 					return nil
 				})
+			// FinalizeS3Object is called inside the ACK callback for every successfully
+			// stored object (state.Stored == true). The tests here use an empty
+			// backupConfig so the real implementation is a no-op, but the mock still
+			// needs an expectation to avoid "unexpected call" panics.
+			mockS3ObjectHandler.EXPECT().FinalizeS3Object().AnyTimes().Return(nil)
 
 			store := openTestStatestore()
 			s3Registry, err := newStateRegistry(logger, store, "", false, 0)