diff --git a/docs/modules/components/pages/inputs/aws_s3.adoc b/docs/modules/components/pages/inputs/aws_s3.adoc index 9c9a62a6c8..cabb3d3154 100644 --- a/docs/modules/components/pages/inputs/aws_s3.adoc +++ b/docs/modules/components/pages/inputs/aws_s3.adoc @@ -107,6 +107,8 @@ If your notification events are being routed to SQS via an SNS topic then the ev When using SQS please make sure you have sensible values for `sqs.max_messages` and also the visibility timeout of the queue itself. When Redpanda Connect consumes an S3 object the SQS message that triggered it is not deleted until the S3 object has been sent onwards. This ensures at-least-once crash resiliency, but also means that if the S3 object takes longer to process than the visibility timeout of your queue then the same objects might be processed multiple times. +Amazon S3 sends an `s3:TestEvent` notification whenever a bucket's event configuration is saved, to verify the queue is reachable. Redpanda Connect detects these (including via an SNS envelope) and deletes them automatically. Any other message with no extractable target key, for example due to a misconfigured `sqs.key_path`/`sqs.bucket_path`, is logged as a warning and left on the queue instead. + == Download large files When downloading large files it's often necessary to process it in streamed parts in order to avoid loading the entire file in memory at a given time. In order to do this a <> can be specified that determines how to break the input into smaller individual messages. diff --git a/internal/impl/aws/s3/input.go b/internal/impl/aws/s3/input.go index 594b0c8d8a..2159ddc126 100644 --- a/internal/impl/aws/s3/input.go +++ b/internal/impl/aws/s3/input.go @@ -152,6 +152,8 @@ If your notification events are being routed to SQS via an SNS topic then the ev When using SQS please make sure you have sensible values for `+"`sqs.max_messages`"+` and also the visibility timeout of the queue itself. When Redpanda Connect consumes an S3 object the SQS message that triggered it is not deleted until the S3 object has been sent onwards. This ensures at-least-once crash resiliency, but also means that if the S3 object takes longer to process than the visibility timeout of your queue then the same objects might be processed multiple times. +Amazon S3 sends an `+"`s3:TestEvent`"+` notification whenever a bucket's event configuration is saved, to verify the queue is reachable. Redpanda Connect detects these (including via an SNS envelope) and deletes them automatically. Any other message with no extractable target key, for example due to a misconfigured `+"`sqs.key_path`"+`/`+"`sqs.bucket_path`"+`, is logged as a warning and left on the queue instead. + == Download large files When downloading large files it's often necessary to process it in streamed parts in order to avoid loading the entire file in memory at a given time. In order to do this a `+"<>"+` can be specified that determines how to break the input into smaller individual messages. @@ -472,20 +474,20 @@ func digStrsFromSlices(slice []any) []string { return strs } -func (s *sqsTargetReader) parseObjectPaths(sqsMsg *string) ([]s3ObjectTarget, error) { +func (s *sqsTargetReader) parseObjectPaths(sqsMsg *string) (*gabs.Container, []s3ObjectTarget, error) { gObj, err := gabs.ParseJSON([]byte(*sqsMsg)) if err != nil { - return nil, fmt.Errorf("parsing SQS message: %v", err) + return nil, nil, fmt.Errorf("parsing SQS message: %v", err) } if s.conf.SQS.EnvelopePath != "" { d := gObj.Path(s.conf.SQS.EnvelopePath).Data() if str, ok := d.(string); ok { if gObj, err = gabs.ParseJSON([]byte(str)); err != nil { - return nil, fmt.Errorf("parsing enveloped message: %v", err) + return nil, nil, fmt.Errorf("parsing enveloped message: %v", err) } } else { - return nil, fmt.Errorf("expected string at envelope path, found %T", d) + return nil, nil, fmt.Errorf("expected string at envelope path, found %T", d) } } @@ -510,14 +512,14 @@ func (s *sqsTargetReader) parseObjectPaths(sqsMsg *string) ([]s3ObjectTarget, er objects := make([]s3ObjectTarget, 0, len(keys)) for i, key := range keys { if key, err = url.QueryUnescape(key); err != nil { - return nil, fmt.Errorf("parsing key from SQS message: %v", err) + return nil, nil, fmt.Errorf("parsing key from SQS message: %v", err) } bucket := s.conf.Bucket if len(buckets) > i { bucket = buckets[i] } if bucket == "" { - return nil, errors.New("required bucket was not found in SQS message") + return nil, nil, errors.New("required bucket was not found in SQS message") } objects = append(objects, s3ObjectTarget{ key: key, @@ -525,7 +527,8 @@ func (s *sqsTargetReader) parseObjectPaths(sqsMsg *string) ([]s3ObjectTarget, er }) } - return objects, nil + // return gabs.Container to save reparsing + return gObj, objects, nil } func (s *sqsTargetReader) readSQSEvents(ctx context.Context) ([]*s3ObjectTarget, error) { @@ -570,15 +573,25 @@ func (s *sqsTargetReader) readSQSEvents(ctx context.Context) ([]*s3ObjectTarget, continue } - objects, err := s.parseObjectPaths(sqsMsg.Body) + gObj, objects, err := s.parseObjectPaths(sqsMsg.Body) if err != nil { addDudFn(sqsMsg) s.log.Errorf("SQS extract key error: %v", err) continue } if len(objects) == 0 { + if isS3TestEvent(gObj) { + s.log.Debugf("Received S3 test event, deleting: %s", *sqsMsg.Body) + if err := s.ackSQSMessage(ctx, sqsMsg); err != nil { + s.log.Errorf("Failed to delete SQS test event message: %v", err) + } + continue + } addDudFn(sqsMsg) - s.log.Debug("Extracted zero target keys from SQS message") + s.log.Warnf( + "Extracted zero target keys from SQS message using key_path %q (bucket_path %q) - this likely indicates a misconfigured key_path/bucket_path, or an unrecognised notification event type: %s", + s.conf.SQS.KeyPath, s.conf.SQS.BucketPath, *sqsMsg.Body, + ) continue } @@ -662,6 +675,14 @@ func (s *sqsTargetReader) ackSQSMessage(ctx context.Context, msg sqstypes.Messag return err } +func isS3TestEvent(gObj *gabs.Container) bool { + if gObj == nil { + return false + } + event, ok := gObj.Path("Event").Data().(string) + return ok && event == "s3:TestEvent" +} + //------------------------------------------------------------------------------ // AmazonS3 is a benthos reader.Type implementation that reads messages from an diff --git a/internal/impl/aws/s3/input_test.go b/internal/impl/aws/s3/input_test.go new file mode 100644 index 0000000000..319081cc01 --- /dev/null +++ b/internal/impl/aws/s3/input_test.go @@ -0,0 +1,93 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package s3 + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestParseObjectPathsSNSEnvelope(t *testing.T) { + tests := []struct { + name string + envelopePath string + body string + expectedKey string + expectedBucket string + expectTestEvent bool + }{ + { + name: "s3 test event", + body: `{"Service":"Amazon S3","Event":"s3:TestEvent","Time":"2025-08-19T17:34:58.550Z","Bucket":"bucket-test","RequestId":"N99ABJ6Q","HostId":"+3DhJHKGDGBwqSTufMSS1UgAMIoRovmGa9vkZwWIb1="}`, + expectTestEvent: true, + }, + { + name: "regular object created notification", + body: `{"Records":[{"eventName":"ObjectCreated:Put","s3":{"bucket":{"name":"bucket-test"},"object":{"key":"foo.txt"}}}]}`, + expectedKey: "foo.txt", + expectedBucket: "bucket-test", + }, + { + name: "Event field is not a string", + body: `{"Event":123}`, + }, + { + name: "Event field missing entirely", + body: `{"Service":"Amazon S3"}`, + }, + { + name: "SNS-enveloped test event", + envelopePath: "Message", + body: `{"Type":"Notification","MessageId":"abc","TopicArn":"arn:aws:sns:eu-west-1:000000000000:topic","Message":"{\"Service\":\"Amazon S3\",\"Event\":\"s3:TestEvent\",\"Time\":\"2025-08-19T17:34:58.550Z\",\"Bucket\":\"bucket-test\",\"RequestId\":\"N99ABJ6Q\",\"HostId\":\"+3DhJHKGDGBwqSTufMSS1UgAMIoRovmGa9vkZwWIb1=\"}"}`, + expectTestEvent: true, + }, + { + name: "SNS-enveloped object created notification", + envelopePath: "Message", + body: `{"Type":"Notification","MessageId":"abc","TopicArn":"arn:aws:sns:eu-west-1:000000000000:topic","Message":"{\"Records\":[{\"eventName\":\"ObjectCreated:Put\",\"s3\":{\"bucket\":{\"name\":\"bucket-test\"},\"object\":{\"key\":\"foo.txt\"}}}]}"}`, + expectedKey: "foo.txt", + expectedBucket: "bucket-test", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + reader := &sqsTargetReader{ + conf: s3iConfig{ + SQS: s3iSQSConfig{ + EnvelopePath: test.envelopePath, + KeyPath: "Records.*.s3.object.key", + BucketPath: "Records.*.s3.bucket.name", + }, + }, + } + + gObj, objects, err := reader.parseObjectPaths(&test.body) + require.NoError(t, err) + + if test.expectedKey == "" { + assert.Empty(t, objects) + } else { + require.Len(t, objects, 1) + assert.Equal(t, test.expectedKey, objects[0].key) + assert.Equal(t, test.expectedBucket, objects[0].bucket) + } + + assert.Equal(t, test.expectTestEvent, isS3TestEvent(gObj)) + }) + } +} diff --git a/internal/impl/aws/s3/integration_test.go b/internal/impl/aws/s3/integration_test.go index aa8b702307..dea660dbcb 100644 --- a/internal/impl/aws/s3/integration_test.go +++ b/internal/impl/aws/s3/integration_test.go @@ -15,9 +15,13 @@ package s3 import ( + "bytes" "context" "errors" "fmt" + "log/slog" + "strings" + "sync" "testing" "time" @@ -25,6 +29,8 @@ import ( awsconfig "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/credentials" "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/sqs" + sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -95,6 +101,196 @@ input: ) }) + t.Run("via_sqs_test_event_deleted", func(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + id := fmt.Sprintf("testevent%d", time.Now().UnixNano()) + require.NoError(t, awstest.CreateBucketQueue(ctx, lsPort, lsPort, id)) + + sqsClient := newLocalStackSQSClient(t, lsPort) + queueURL := fmt.Sprintf("http://localhost:%s/000000000000/queue-%s", lsPort, id) + + testEventBody := fmt.Sprintf( + `{"Service":"Amazon S3","Event":"s3:TestEvent","Time":"%s","Bucket":"bucket-%s","RequestId":"N99ABJ6Q","HostId":"+3DhJHKGDGBwqSTufMSS1UgAMIoRovmGa9vkZwWIb1="}`, + time.Now().UTC().Format(time.RFC3339), id, + ) + _, err := sqsClient.SendMessage(ctx, &sqs.SendMessageInput{ + QueueUrl: aws.String(queueURL), + MessageBody: aws.String(testEventBody), + }) + require.NoError(t, err) + + yaml := fmt.Sprintf(` +input: + aws_s3: + bucket: bucket-%s + endpoint: http://localhost:%s + force_path_style_urls: true + region: eu-west-1 + delete_objects: true + sqs: + url: %s + key_path: Records.*.s3.object.key + endpoint: http://localhost:%s + wait_time_seconds: 1 + credentials: + id: xxxxx + secret: xxxxx + token: xxxxx +`, id, lsPort, queueURL, lsPort) + + builder := service.NewStreamBuilder() + require.NoError(t, builder.SetYAML(yaml)) + + var ( + unexpectedMu sync.Mutex + unexpectedMsgs [][]byte + ) + require.NoError(t, builder.AddConsumerFunc(func(_ context.Context, m *service.Message) error { + b, err := m.AsBytes() + if err != nil { + b = fmt.Appendf(nil, "", err) + } + unexpectedMu.Lock() + unexpectedMsgs = append(unexpectedMsgs, b) + unexpectedMu.Unlock() + return nil + })) + + stream, err := builder.Build() + require.NoError(t, err) + + runErr := make(chan error, 1) + runCtx, runCancel := context.WithCancel(ctx) + defer runCancel() + go func() { runErr <- stream.Run(runCtx) }() + + // The test event should be deleted from the queue rather than left + // to be received again indefinitely. + assert.Eventually(t, func() bool { + out, err := sqsClient.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{ + QueueUrl: aws.String(queueURL), + AttributeNames: []sqstypes.QueueAttributeName{ + sqstypes.QueueAttributeNameApproximateNumberOfMessages, + sqstypes.QueueAttributeNameApproximateNumberOfMessagesNotVisible, + }, + }) + if err != nil { + return false + } + return out.Attributes[string(sqstypes.QueueAttributeNameApproximateNumberOfMessages)] == "0" && + out.Attributes[string(sqstypes.QueueAttributeNameApproximateNumberOfMessagesNotVisible)] == "0" + }, 15*time.Second, 500*time.Millisecond, "test event message should have been deleted from the queue") + + require.NoError(t, stream.StopWithin(10*time.Second)) + select { + case err := <-runErr: + if err != nil && !errors.Is(err, context.Canceled) { + require.NoError(t, err) + } + case <-time.After(10 * time.Second): + t.Fatal("stream did not exit after StopWithin returned") + } + + unexpectedMu.Lock() + assert.Empty(t, unexpectedMsgs, "did not expect any message to be emitted for an s3:TestEvent") + unexpectedMu.Unlock() + }) + + t.Run("via_sqs_zero_key_non_test_event_warns", func(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + id := fmt.Sprintf("zerokey%d", time.Now().UnixNano()) + require.NoError(t, awstest.CreateBucketQueue(ctx, lsPort, lsPort, id)) + + sqsClient := newLocalStackSQSClient(t, lsPort) + queueURL := fmt.Sprintf("http://localhost:%s/000000000000/queue-%s", lsPort, id) + + // Valid JSON, not an s3:TestEvent, but missing the object key at the + // configured key_path - simulates a key_path/bucket_path + // misconfiguration (or an unrecognised event type) rather than a + // test event, which should be handled differently: warned about and + // left on the queue, not silently deleted. + misconfiguredBody := fmt.Sprintf( + `{"Records":[{"eventName":"ObjectCreated:Put","s3":{"bucket":{"name":"bucket-%s"}}}]}`, id, + ) + _, err := sqsClient.SendMessage(ctx, &sqs.SendMessageInput{ + QueueUrl: aws.String(queueURL), + MessageBody: aws.String(misconfiguredBody), + }) + require.NoError(t, err) + + yaml := fmt.Sprintf(` +input: + aws_s3: + bucket: bucket-%s + endpoint: http://localhost:%s + force_path_style_urls: true + region: eu-west-1 + delete_objects: true + sqs: + url: %s + key_path: Records.*.s3.object.key + endpoint: http://localhost:%s + wait_time_seconds: 1 + credentials: + id: xxxxx + secret: xxxxx + token: xxxxx +`, id, lsPort, queueURL, lsPort) + + var logBuf syncBuffer + builder := service.NewStreamBuilder() + builder.SetLogger(slog.New(slog.NewTextHandler(&logBuf, &slog.HandlerOptions{Level: slog.LevelDebug}))) + + require.NoError(t, builder.SetYAML(yaml)) + require.NoError(t, builder.AddConsumerFunc(func(_ context.Context, m *service.Message) error { + b, _ := m.AsBytes() + t.Errorf("did not expect any message to be emitted for a zero-key notification, got: %s", b) + return nil + })) + + stream, err := builder.Build() + require.NoError(t, err) + + runErr := make(chan error, 1) + runCtx, runCancel := context.WithCancel(ctx) + defer runCancel() + go func() { runErr <- stream.Run(runCtx) }() + + // A non-test-event, zero-key message should be warned about, + // identifying the likely key_path misconfiguration, and - unlike an + // s3:TestEvent - left on the queue for redelivery rather than + // deleted. + assert.Eventually(t, func() bool { + return strings.Contains(logBuf.String(), "level=WARN") && strings.Contains(logBuf.String(), "key_path") + }, 15*time.Second, 500*time.Millisecond, "expected a WARN log identifying a key_path misconfiguration, got logs: %s", &logBuf) + + out, err := sqsClient.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{ + QueueUrl: aws.String(queueURL), + AttributeNames: []sqstypes.QueueAttributeName{ + sqstypes.QueueAttributeNameApproximateNumberOfMessages, + sqstypes.QueueAttributeNameApproximateNumberOfMessagesNotVisible, + }, + }) + require.NoError(t, err) + visible := out.Attributes[string(sqstypes.QueueAttributeNameApproximateNumberOfMessages)] + notVisible := out.Attributes[string(sqstypes.QueueAttributeNameApproximateNumberOfMessagesNotVisible)] + assert.False(t, visible == "0" && notVisible == "0", "misconfigured message should remain on the queue for redelivery, not be deleted") + + require.NoError(t, stream.StopWithin(10*time.Second)) + select { + case err := <-runErr: + if err != nil && !errors.Is(err, context.Canceled) { + require.NoError(t, err) + } + case <-time.After(10 * time.Second): + t.Fatal("stream did not exit after StopWithin returned") + } + }) + t.Run("via_sqs_lines", func(t *testing.T) { template := ` output: @@ -392,3 +588,34 @@ func newLocalStackS3Client(t *testing.T, port string) *s3.Client { o.UsePathStyle = true }) } + +// newLocalStackSQSClient builds an SQS client pointed at the LocalStack +// instance on the supplied port, with dummy credentials. +func newLocalStackSQSClient(t *testing.T, port string) *sqs.Client { + t.Helper() + endpoint := fmt.Sprintf("http://localhost:%s", port) + cfg, err := awsconfig.LoadDefaultConfig(t.Context(), + awsconfig.WithRegion("eu-west-1"), + awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("xxxxx", "xxxxx", "xxxxx")), + ) + require.NoError(t, err) + cfg.BaseEndpoint = &endpoint + return sqs.NewFromConfig(cfg) +} + +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *syncBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *syncBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +}