x-pack/filebeat/input/awss3: exclude backup-prefixed keys from polling

changelog/fragments/1783988471-exclude-backup-prefix-from-polling.yaml

@@ -0,0 +1,8 @@
+kind: bug-fix
+summary: Fix S3 polling input to exclude same-bucket backup objects from listing.
+description: |
+  When same-bucket backup is configured with the default empty
+  bucket_list_prefix, backup objects are listed alongside source objects
+  and reprocessed indefinitely. Exclude keys matching the backup prefix
+  from listing results when the backup destination is the same bucket.
+component: filebeat

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

@@ -190,6 +190,25 @@ func (c *backupConfig) GetBucketName() string {
 	return c.NonAWSBackupToBucketName
 }
 
+// backupPrefixToExclude returns the backup prefix that should be excluded from
+// listing results when the backup destination is the same bucket and the backup
+// prefix falls within the listing scope. Returns "" when no exclusion is needed.
+func (c *config) backupPrefixToExclude() string {
+	if c.BackupConfig.BackupToBucketPrefix == "" {
+		return ""
+	}
+	sameBucket := (c.BackupConfig.BackupToBucketArn != "" &&
+		(c.BackupConfig.BackupToBucketArn == c.BucketARN || c.BackupConfig.BackupToBucketArn == c.AccessPointARN)) ||
+		(c.BackupConfig.NonAWSBackupToBucketName != "" && c.BackupConfig.NonAWSBackupToBucketName == c.NonAWSBucketName)
+	if !sameBucket {
+		return ""
+	}
+	if !strings.HasPrefix(c.BackupConfig.BackupToBucketPrefix, c.BucketListPrefix) {
+		return ""
+	}
+	return c.BackupConfig.BackupToBucketPrefix
+}
+
 // fileSelectorConfig defines reader configuration that applies to a subset
 // of S3 objects whose URL matches the given regex.
 type fileSelectorConfig struct {

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

@@ -666,7 +666,7 @@
 			if tc.expectedCfg == nil {
 				t.Fatal("missing expected config in test case")
 			}
 			assert.EqualValues(t, tc.expectedCfg(tc.queueURL, tc.s3Bucket, tc.s3AccessPoint, tc.nonAWSS3Bucket), c)
 		})
 	}
 }
@@ -730,3 +730,99 @@
 		})
 	}
 }
+
+func TestBackupPrefixToExclude(t *testing.T) {
+	tests := []struct {
+		name string
+		cfg  config
+		want string
+	}{
+		{
+			name: "no backup configured",
+			cfg: config{
+				BucketARN:        "arn:aws:s3:::my-bucket",
+				BucketListPrefix: "",
+			},
+			want: "",
+		},
+		{
+			name: "different bucket",
+			cfg: config{
+				BucketARN:        "arn:aws:s3:::my-bucket",
+				BucketListPrefix: "",
+				BackupConfig: backupConfig{
+					BackupToBucketArn:    "arn:aws:s3:::other-bucket",
+					BackupToBucketPrefix: "processed/",
+				},
+			},
+			want: "",
+		},
+		{
+			name: "same bucket, empty list prefix",
+			cfg: config{
+				BucketARN:        "arn:aws:s3:::my-bucket",
+				BucketListPrefix: "",
+				BackupConfig: backupConfig{
+					BackupToBucketArn:    "arn:aws:s3:::my-bucket",
+					BackupToBucketPrefix: "processed/",
+				},
+			},
+			want: "processed/",
+		},
+		{
+			name: "same bucket, backup prefix within list prefix scope",
+			cfg: config{
+				BucketARN:        "arn:aws:s3:::my-bucket",
+				BucketListPrefix: "logs/",
+				BackupConfig: backupConfig{
+					BackupToBucketArn:    "arn:aws:s3:::my-bucket",
+					BackupToBucketPrefix: "logs/processed/",
+				},
+			},
+			want: "logs/processed/",
+		},
+		{
+			name: "same bucket, backup prefix outside list prefix scope",
+			cfg: config{
+				BucketARN:        "arn:aws:s3:::my-bucket",
+				BucketListPrefix: "logs/",
+				BackupConfig: backupConfig{
+					BackupToBucketArn:    "arn:aws:s3:::my-bucket",
+					BackupToBucketPrefix: "archived/",
+				},
+			},
+			want: "",
+		},
+		{
+			name: "same bucket via access point ARN",
+			cfg: config{
+				AccessPointARN:   "arn:aws:s3:us-east-1:123456789:accesspoint/ap",
+				BucketListPrefix: "",
+				BackupConfig: backupConfig{
+					BackupToBucketArn:    "arn:aws:s3:us-east-1:123456789:accesspoint/ap",
+					BackupToBucketPrefix: "done/",
+				},
+			},
+			want: "done/",
+		},
+		{
+			name: "same non-AWS bucket",
+			cfg: config{
+				NonAWSBucketName: "minio-bucket",
+				BucketListPrefix: "",
+				BackupConfig: backupConfig{
+					NonAWSBackupToBucketName: "minio-bucket",
+					BackupToBucketPrefix:     "backup/",
+				},
+			},
+			want: "backup/",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.cfg.backupPrefixToExclude()
+			assert.Equal(t, tt.want, got, "backupPrefixToExclude() mismatch")
+		})
+	}
+}

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

@@ -10,6 +10,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 
@@ -269,6 +270,7 @@ func (in *s3PollerInput) readerLoop(ctx context.Context, workChan chan<- state)
 	bucketName := getBucketNameFromARN(in.config.getBucketARN())
 
 	isStateValid := in.filterProvider.getApplierFunc()
+	excludePrefix := in.config.backupPrefixToExclude()
 
 	errorBackoff := backoff.NewEqualJitterBackoff(ctx.Done(), 1, 120)
 	circuitBreaker := 0
@@ -306,6 +308,10 @@ func (in *s3PollerInput) readerLoop(ctx context.Context, workChan chan<- state)
 		// Metrics
 		in.metrics.s3ObjectsListedTotal.Add(uint64(totListedObjects))
 		for _, object := range page.Contents {
+			if excludePrefix != "" && strings.HasPrefix(*object.Key, excludePrefix) {
+				continue
+			}
+
 			state := newState(bucketName, *object.Key, *object.ETag, *object.LastModified)
 
 			if in.strategy.ShouldSkipObject(state, isStateValid) {

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

@@ -273,6 +273,120 @@ func TestS3Poller(t *testing.T) {
 		waitForChannel(t, deleteDone, 3*testTimeout)
 	})
 
+	t.Run("Same-bucket backup objects are excluded from listing", 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{})
+
+		mockAPI.EXPECT().
+			ListObjectsPaginator(gomock.Eq(bucket), gomock.Eq(""), gomock.Any()).
+			Times(1).
+			DoAndReturn(func(_, _, _ string) s3Pager {
+				return mockPager
+			})
+
+		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("input.log"),
+							LastModified: aws.Time(time.Now()),
+						},
+						{
+							ETag:         aws.String("etag2"),
+							Key:          aws.String("processed/input.log"),
+							LastModified: aws.Time(time.Now()),
+						},
+						{
+							ETag:         aws.String("etag3"),
+							Key:          aws.String("processed/processed/input.log"),
+							LastModified: aws.Time(time.Now()),
+						},
+					},
+				}, nil
+			})
+
+		// Only the source object should be fetched.
+		mockAPI.EXPECT().
+			GetObject(gomock.Any(), gomock.Eq(""), gomock.Eq(bucket), gomock.Eq("input.log")).
+			Times(1).
+			DoAndReturn(func(_ context.Context, _ string, _ string, _ string) (*s3.GetObjectOutput, error) {
+				return &s3.GetObjectOutput{
+					Body: io.NopCloser(strings.NewReader("hello\n")),
+				}, nil
+			})
+
+		// Finalization copies the source object to the backup prefix.
+		mockAPI.EXPECT().
+			CopyObject(gomock.Any(), gomock.Eq(""), gomock.Eq(bucket), gomock.Eq(bucket), gomock.Eq("input.log"), gomock.Eq("processed/input.log")).
+			Times(1).
+			DoAndReturn(func(_ context.Context, _ string, _ string, _ string, _ string, _ string) (*s3.CopyObjectOutput, error) {
+				close(backupDone)
+				return &s3.CopyObjectOutput{}, nil
+			})
+
+		// The backup-prefixed keys must NOT trigger GetObject. gomock will
+		// fail the test if an unexpected GetObject call is made.
+
+		backupCfg := backupConfig{
+			BackupToBucketArn:    bucket,
+			BackupToBucketPrefix: "processed/",
+		}
+
+		cfg := config{
+			NumberOfWorkers:    numberOfWorkers,
+			BucketListInterval: pollInterval,
+			BucketARN:          bucket,
+			BucketListPrefix:   "",
+			BackupConfig:       backupCfg,
+		}
+		log := logptest.NewTestingLogger(t, inputName)
+
+		s3ObjProc := newS3ObjectProcessorFactory(nil, mockAPI, nil, backupCfg, logp.NewNopLogger())
+		registry, err := newStateRegistry(nil, store, "", 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)
+		waitForChannel(t, backupDone, 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

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

@@ -266,6 +266,7 @@ func (in *inputV2) runPolling(ctx context.Context, log *logp.Logger, st status.S
 		Status:          st,
 		BucketARN:       in.config.getBucketARN(),
 		ListPrefix:      in.config.BucketListPrefix,
+		ExcludePrefix:   in.config.backupPrefixToExclude(),
 		ListInterval:    in.config.BucketListInterval,
 		NumWorkers:      in.config.NumberOfWorkers,
 		Region:          awsCfg.Region,

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

@@ -8,6 +8,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 
@@ -34,6 +35,7 @@ type pollingDiscoveryV2 struct {
 	bucketARN      string
 	bucketName     string
 	listPrefix     string
+	excludePrefix  string
 	listInterval   time.Duration
 	numWorkers     int
 	region         string
@@ -52,6 +54,7 @@ type pollingDiscoveryV2Config struct {
 	Status          status.StatusReporter
 	BucketARN       string
 	ListPrefix      string
+	ExcludePrefix   string
 	ListInterval    time.Duration
 	NumWorkers      int
 	Region          string
@@ -71,6 +74,7 @@ func newPollingDiscoveryV2(cfg pollingDiscoveryV2Config) *pollingDiscoveryV2 {
 		bucketARN:      cfg.BucketARN,
 		bucketName:     getBucketNameFromARN(cfg.BucketARN),
 		listPrefix:     cfg.ListPrefix,
+		excludePrefix:  cfg.ExcludePrefix,
 		listInterval:   cfg.ListInterval,
 		numWorkers:     cfg.NumWorkers,
 		region:         cfg.Region,
@@ -222,6 +226,10 @@ func (p *pollingDiscoveryV2) listObjects(ctx context.Context, workChan chan<- st
 		p.metrics.s3ObjectsListedTotal.Add(uint64(len(page.Contents)))
 
 		for _, obj := range page.Contents {
+			if p.excludePrefix != "" && strings.HasPrefix(*obj.Key, p.excludePrefix) {
+				continue
+			}
+
 			st := newState(p.bucketName, *obj.Key, *obj.ETag, *obj.LastModified)
 
 			if p.strategy.ShouldSkipObject(st, isStateValid) {