[receiver/awss3] use object tags to checkpoint processed objects (#47397)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
introduce a `skip_ingesting_tagged_objects` to the s3 receiver. This
builds on `tag_object_after_ingestion` to checkpoint the progress of the
receiver and avoid reprocessing objects
<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes #47396

<!--Describe what testing was performed and which tests were added.-->
#### Testing
unit tests
<!--Describe the documentation added.-->
#### Documentation
updated README
<!--Please delete paragraphs that you did not use before submitting.-->

Author: tommyzli

.chloggen/s3-skip-reingestion.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: receiver/awss3
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: add a flag to the s3 receiver to skip objects previously tagged as ingested.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [47396]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

receiver/awss3receiver/README.md

@@ -52,6 +52,7 @@ The following exporter configuration parameters are supported.
 | `notifications:`        |                                                                                                                                            |             |          |
 | `opampextension`        | Name of the OpAMP Extension to use to send ingest progress notifications.                                                                  |             |          |
 | `tag_object_after_ingestion`        | If enabled the receiver will attempt to tag the object after successfully ingesting it.                                                                  | false       | Optional |
+| `skip_ingesting_tagged_objects`        | If enabled the receiver will skip objects tagged by `tag_object_after_ingestion`. This can be used as a checkpointing mechanism, and requires an additional `s3:GetObjectTagging` permission                                                                  | false       | Optional |
 
 There are two modes of operation:
 

receiver/awss3receiver/config.go

@@ -27,6 +27,7 @@ type S3DownloaderConfig struct {
 	EndpointPartitionID            string `mapstructure:"endpoint_partition_id"`
 	S3ForcePathStyle               bool   `mapstructure:"s3_force_path_style"`
 	TagObjectAfterIngestion        bool   `mapstructure:"tag_object_after_ingestion"`
+	SkipIngestingTaggedObjects     bool   `mapstructure:"skip_ingesting_tagged_objects"`
 }
 
 // SQSConfig holds SQS queue configuration for receiving object change notifications.

receiver/awss3receiver/config.schema.yaml

@@ -40,6 +40,8 @@ $defs:
         type: string
       s3_prefix:
         type: string
+      skip_ingesting_tagged_objects:
+        type: boolean
       tag_object_after_ingestion:
         type: boolean
   sqs_config:

receiver/awss3receiver/config_test.go

@@ -37,6 +37,7 @@ func TestConfig_Validate_Valid(t *testing.T) {
 				EndpointPartitionID:            "aws",
 				S3ForcePathStyle:               false,
 				TagObjectAfterIngestion:        true,
+				SkipIngestingTaggedObjects:     false,
 			},
 			StartTime: "2024-01-01",
 			EndTime:   "2024-01-01",
@@ -59,6 +60,7 @@ func TestConfig_Validate_Valid(t *testing.T) {
 				EndpointPartitionID:            "aws",
 				S3ForcePathStyle:               false,
 				TagObjectAfterIngestion:        true,
+				SkipIngestingTaggedObjects:     false,
 			},
 			SQS: &SQSConfig{
 				QueueURL: "https://sqs.us-east-1.amazonaws.com/123456789012/test-queue",
@@ -96,6 +98,7 @@ func TestLoadConfig(t *testing.T) {
 					FilePrefixIncludeTelemetryType: true,
 					EndpointPartitionID:            "aws",
 					TagObjectAfterIngestion:        false,
+					SkipIngestingTaggedObjects:     false,
 				},
 				StartTime: "2024-01-31 15:00",
 				EndTime:   "2024-02-03",
@@ -113,6 +116,7 @@ func TestLoadConfig(t *testing.T) {
 					FilePrefixIncludeTelemetryType: false,
 					EndpointPartitionID:            "aws",
 					TagObjectAfterIngestion:        false,
+					SkipIngestingTaggedObjects:     false,
 				},
 				StartTime: "2024-01-31 15:00",
 				EndTime:   "2024-02-03",
@@ -141,6 +145,7 @@ func TestLoadConfig(t *testing.T) {
 					FilePrefixIncludeTelemetryType: true,
 					EndpointPartitionID:            "aws",
 					TagObjectAfterIngestion:        true,
+					SkipIngestingTaggedObjects:     true,
 				},
 				StartTime: "2024-01-31T15:00:00Z",
 				EndTime:   "2024-02-03T00:00:00Z",
@@ -156,6 +161,7 @@ func TestLoadConfig(t *testing.T) {
 					FilePrefixIncludeTelemetryType: true,
 					EndpointPartitionID:            "aws",
 					TagObjectAfterIngestion:        false,
+					SkipIngestingTaggedObjects:     false,
 				},
 				SQS: &SQSConfig{
 					QueueURL: "https://sqs.us-east-1.amazonaws.com/123456789012/test-queue",

receiver/awss3receiver/s3intf.go

@@ -30,6 +30,7 @@ type ListObjectsAPI interface {
 
 type SingleObjectAPI interface {
 	GetObject(ctx context.Context, params *s3.GetObjectInput, optFns ...func(*s3.Options)) (*s3.GetObjectOutput, error)
+	GetObjectTagging(ctx context.Context, params *s3.GetObjectTaggingInput, optFns ...func(*s3.Options)) (*s3.GetObjectTaggingOutput, error)
 	PutObjectTagging(ctx context.Context, params *s3.PutObjectTaggingInput, optFns ...func(*s3.Options)) (*s3.PutObjectTaggingOutput, error)
 }
 
@@ -101,3 +102,23 @@ func tagS3Object(ctx context.Context, client SingleObjectAPI, bucket, key string
 	_, err := client.PutObjectTagging(ctx, &params)
 	return err
 }
+
+// hasIngestedTag checks if an S3 object has the ingested tag set
+func hasIngestedTag(ctx context.Context, client SingleObjectAPI, bucket, key string) (bool, error) {
+	params := s3.GetObjectTaggingInput{
+		Bucket: &bucket,
+		Key:    &key,
+	}
+	output, err := client.GetObjectTagging(ctx, &params)
+	if err != nil {
+		return false, err
+	}
+
+	for _, tag := range output.TagSet {
+		if tag.Key != nil && *tag.Key == ingestedTag &&
+			tag.Value != nil && *tag.Value == ingestedStatus {
+			return true, nil
+		}
+	}
+	return false, nil
+}

receiver/awss3receiver/s3reader.go

@@ -29,6 +29,7 @@ type s3TimeBasedReader struct {
 	endTime                        time.Time
 	notifier                       statusNotifier
 	tagObjectAfterIngestion        bool
+	skipIngestingTaggedObjects     bool
 }
 
 func newS3TimeBasedReader(ctx context.Context, notifier statusNotifier, logger *zap.Logger, cfg *Config) (*s3TimeBasedReader, error) {
@@ -69,6 +70,7 @@ func newS3TimeBasedReader(ctx context.Context, notifier statusNotifier, logger *
 		endTime:                        endTime,
 		notifier:                       notifier,
 		tagObjectAfterIngestion:        cfg.S3Downloader.TagObjectAfterIngestion,
+		skipIngestingTaggedObjects:     cfg.S3Downloader.SkipIngestingTaggedObjects,
 	}, nil
 }
 
@@ -146,6 +148,21 @@ func (s3Reader *s3TimeBasedReader) readTelemetryForTime(ctx context.Context, t t
 			s3Reader.logger.Info("No telemetry found for time", zap.String("prefix", prefix), zap.Time("time", t))
 		} else {
 			for _, obj := range page.Contents {
+				if s3Reader.skipIngestingTaggedObjects {
+					var hasTag bool
+					hasTag, err = hasIngestedTag(ctx, s3Reader.singleObjectClient, s3Reader.s3Bucket, *obj.Key)
+					if err != nil {
+						s3Reader.logger.Error("Failed to check object tags",
+							zap.String("key", *obj.Key),
+							zap.Error(err))
+						return err
+					} else if hasTag {
+						s3Reader.logger.Info("Skipping already ingested object",
+							zap.String("key", *obj.Key))
+						continue
+					}
+				}
+
 				data, err := retrieveS3Object(ctx, s3Reader.singleObjectClient, s3Reader.s3Bucket, *obj.Key)
 				if err != nil {
 					return err

receiver/awss3receiver/s3reader_test.go

@@ -12,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/stretchr/testify/require"
@@ -185,6 +186,7 @@ func Test_s3Reader_getObjectPrefixForTime(t *testing.T) {
 
 type mockSingleObjectAPI struct {
 	getObjectFunc        func(ctx context.Context, params *s3.GetObjectInput, optFns ...func(*s3.Options)) (*s3.GetObjectOutput, error)
+	getObjectTaggingFunc func(ctx context.Context, params *s3.GetObjectTaggingInput, optFns ...func(*s3.Options)) (*s3.GetObjectTaggingOutput, error)
 	putObjectTaggingFunc func(ctx context.Context, params *s3.PutObjectTaggingInput, optFns ...func(*s3.Options)) (*s3.PutObjectTaggingOutput, error)
 }
 
@@ -195,6 +197,14 @@ func (m *mockSingleObjectAPI) GetObject(ctx context.Context, params *s3.GetObjec
 	return nil, errors.New("GetObject not mocked")
 }
 
+func (m *mockSingleObjectAPI) GetObjectTagging(ctx context.Context, params *s3.GetObjectTaggingInput, optFns ...func(*s3.Options)) (*s3.GetObjectTaggingOutput, error) {
+	if m.getObjectTaggingFunc != nil {
+		return m.getObjectTaggingFunc(ctx, params, optFns...)
+	}
+	// Default to empty tag set if no mock function provided
+	return &s3.GetObjectTaggingOutput{TagSet: []types.Tag{}}, nil
+}
+
 func (m *mockSingleObjectAPI) PutObjectTagging(ctx context.Context, params *s3.PutObjectTaggingInput, optFns ...func(*s3.Options)) (*s3.PutObjectTaggingOutput, error) {
 	if m.putObjectTaggingFunc != nil {
 		return m.putObjectTaggingFunc(ctx, params, optFns...)
@@ -294,6 +304,93 @@ func Test_readTelemetryForTime(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func Test_readTelemetryForTime_skipTaggedObjects(t *testing.T) {
+	testKey1 := "year=2021/month=02/day=01/hour=17/minute=32/traces_1"
+	testKey2 := "year=2021/month=02/day=01/hour=17/minute=32/traces_2"
+	testKey3 := "year=2021/month=02/day=01/hour=17/minute=32/traces_3"
+	reader := s3TimeBasedReader{
+		listObjectsClient: mockListObjectsAPI(func(params *s3.ListObjectsV2Input) ListObjectsV2Pager {
+			t.Helper()
+			require.Equal(t, "bucket", *params.Bucket)
+			require.Equal(t, "year=2021/month=02/day=01/hour=17/minute=32/traces_", *params.Prefix)
+
+			return &mockListObjectsV2Pager{
+				Pages: []*s3.ListObjectsV2Output{
+					{
+						Contents: []types.Object{
+							{
+								// Not tagged
+								Key: &testKey1,
+							},
+						},
+					},
+					{
+						Contents: []types.Object{
+							{
+								// Already tagged by the receiver as ingested
+								Key: &testKey2,
+							},
+						},
+					},
+					{
+						Contents: []types.Object{
+							{
+								// Has a tag, but not one set by the receiver
+								Key: &testKey3,
+							},
+						},
+					},
+				},
+			}
+		}),
+		singleObjectClient: &mockSingleObjectAPI{
+			getObjectFunc: func(_ context.Context, params *s3.GetObjectInput, _ ...func(*s3.Options)) (*s3.GetObjectOutput, error) {
+				t.Helper()
+				require.Equal(t, "bucket", *params.Bucket)
+				// testKey2 should not be fetched because it has the ingested tag
+				require.Contains(t, []string{testKey1, testKey3}, *params.Key)
+				return &s3.GetObjectOutput{
+					Body: io.NopCloser(bytes.NewReader([]byte("this is the body of the object"))),
+				}, nil
+			},
+			getObjectTaggingFunc: func(_ context.Context, params *s3.GetObjectTaggingInput, _ ...func(*s3.Options)) (*s3.GetObjectTaggingOutput, error) {
+				t.Helper()
+				require.Equal(t, "bucket", *params.Bucket)
+				require.Contains(t, []string{testKey1, testKey2, testKey3}, *params.Key)
+				var tagSet []types.Tag
+				switch *params.Key {
+				case testKey2:
+					tagSet = []types.Tag{{Key: aws.String(ingestedTag), Value: aws.String(ingestedStatus)}}
+				case testKey3:
+					tagSet = []types.Tag{{Key: aws.String("env"), Value: aws.String("dev")}}
+				}
+				return &s3.GetObjectTaggingOutput{TagSet: tagSet}, nil
+			},
+		},
+		logger:                         zap.NewNop(),
+		s3Bucket:                       "bucket",
+		s3PartitionFormat:              s3PartitionFormatDefault,
+		S3PartitionTimeLocation:        time.UTC,
+		s3Prefix:                       "",
+		filePrefix:                     "",
+		filePrefixIncludeTelemetryType: true,
+		startTime:                      testTime,
+		endTime:                        testTime.Add(time.Minute),
+		skipIngestingTaggedObjects:     true,
+	}
+
+	dataCallbackKeys := make([]string, 0)
+
+	err := reader.readTelemetryForTime(t.Context(), testTime, "traces", func(_ context.Context, key string, data []byte) error {
+		t.Helper()
+		require.Equal(t, "this is the body of the object", string(data))
+		dataCallbackKeys = append(dataCallbackKeys, key)
+		return nil
+	})
+	require.Equal(t, []string{testKey1, testKey3}, dataCallbackKeys)
+	require.NoError(t, err)
+}
+
 func Test_readTelemetryForTime_GetObjectError(t *testing.T) {
 	testKey := "year=2021/month=02/day=01/hour=17/minute=32/traces_1"
 	testError := errors.New("test error")
@@ -757,6 +854,65 @@ func Test_readTelemetryForTime_TagFailure(t *testing.T) {
 	require.Equal(t, []string{testKey}, dataCallbackKeys, "Data should still be processed")
 }
 
+func Test_readTelemetryForTime_GetTagError(t *testing.T) {
+	testKey := "year=2023/month=01/day=02/hour=03/minute=04/traces_test"
+
+	reader := &s3TimeBasedReader{
+		listObjectsClient: mockListObjectsAPI(func(params *s3.ListObjectsV2Input) ListObjectsV2Pager {
+			require.Equal(t, "bucket", *params.Bucket)
+			require.Equal(t, "year=2023/month=01/day=02/hour=03/minute=04/traces_", *params.Prefix)
+
+			return &mockListObjectsV2Pager{
+				Pages: []*s3.ListObjectsV2Output{
+					{
+						Contents: []types.Object{
+							{Key: &testKey},
+						},
+					},
+				},
+			}
+		}),
+		singleObjectClient: &mockSingleObjectAPI{
+			getObjectFunc: func(_ context.Context, params *s3.GetObjectInput, _ ...func(*s3.Options)) (*s3.GetObjectOutput, error) {
+				t.Helper()
+				require.Equal(t, "bucket", *params.Bucket)
+				require.Equal(t, testKey, *params.Key)
+				return &s3.GetObjectOutput{
+					Body: io.NopCloser(bytes.NewReader([]byte("this is the body of the object"))),
+				}, nil
+			},
+			getObjectTaggingFunc: func(_ context.Context, params *s3.GetObjectTaggingInput, _ ...func(*s3.Options)) (*s3.GetObjectTaggingOutput, error) {
+				t.Helper()
+				require.Equal(t, "bucket", *params.Bucket)
+				require.Equal(t, testKey, *params.Key)
+				return nil, errors.New("failed to get object tags")
+			},
+		},
+		logger:                         zap.NewNop(),
+		s3Bucket:                       "bucket",
+		s3PartitionFormat:              s3PartitionFormatDefault,
+		S3PartitionTimeLocation:        time.UTC,
+		filePrefix:                     "",
+		filePrefixIncludeTelemetryType: true,
+		tagObjectAfterIngestion:        true,
+		skipIngestingTaggedObjects:     true,
+	}
+
+	testTime, err := time.Parse(time.RFC3339, "2023-01-02T03:04:05Z")
+	require.NoError(t, err)
+
+	dataCallbackKeys := make([]string, 0)
+	err = reader.readTelemetryForTime(t.Context(), testTime, "traces", func(_ context.Context, key string, _ []byte) error {
+		t.Helper()
+		dataCallbackKeys = append(dataCallbackKeys, key)
+		return nil
+	})
+
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "failed to get object tags")
+	require.Empty(t, dataCallbackKeys, "No data should be processed when GetObjectTagging fails")
+}
+
 func Test_determineTimestep(t *testing.T) {
 	tests := []struct {
 		name           string