Repository: triggermesh/aws-event-sources (go)

Issue:
Allow customization of SQSSource visibility timeout
Currently when reading messages from SQS the visibility timeout is [set at a static 30 seconds](https://github.com/triggermesh/aws-event-sources/blob/main/pkg/adapter/awssqssource/receive.go#L43). However, depending on the type of sink configured this may be insufficient.

I have a long-running service as the sink in my configuration. The goal is to maintain SQS as the source of truth for whether or not a message has been processed, not deleting the message from the queue until the service returns successfully. This works today with the exception of the visibility timeout - it ends up in an infinite retry loop because the response from the sink takes >30 seconds even though each individual request is successful.

Ideally the visibility timeout should be configurable from the spec.

Pull request:
sqs: Add option for configuring visibility timeout
Adds a new option `receiveOptions.visibilityTimeout` which controls the [visibility timeout](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html) set on received messages:

```yaml
apiVersion: sources.triggermesh.io/v1alpha1
kind: AWSSQSSource
metadata:
  name: sample
spec:
  # ...
  receiveOptions:
    visibilityTimeout: <duration string; e.g. "1m30s">
```

The previously hardcoded value of `30s` has been removed and, when the user doesn't set a value at all, the overall value configured for the queue is used.

Closes #329

cc. @noahw3

Patch:
diff --git a/config/300-awssqs.yaml b/config/300-awssqs.yaml
index f9a4d54..5e7df61 100644
--- a/config/300-awssqs.yaml
+++ b/config/300-awssqs.yaml
@@ -57,6 +57,17 @@ spec:
                   https://docs.aws.amazon.com/IAM/latest/UserGuide/list_amazonsqs.html#amazonsqs-resources-for-iam-policies.
                 type: string
                 pattern: ^arn:aws(-cn|-us-gov)?:sqs:[a-z]{2}(-gov)?-[a-z]+-\d:\d{12}:.+$
+              receiveOptions:
+                description: Options that control the behavior of message receivers.
+                type: object
+                properties:
+                  visibilityTimeout:
+                    description: Period of time during which Amazon SQS prevents other consumers from receiving and
+                      processing a message that has been received via ReceiveMessage. Expressed as a duration string,
+                      which format is documented at https://pkg.go.dev/time#ParseDuration. If not defined, the overall
+                      visibility timeout for the queue is used. For more details, please refer to the Amazon SQS
+                      Developer Guide at https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html.
+                    type: string
               credentials:
                 description: Credentials to interact with the Amazon SQS API. For more information about AWS security
                   credentials, please refer to the AWS General Reference at
diff --git a/pkg/adapter/awssqssource/adapter.go b/pkg/adapter/awssqssource/adapter.go
index d54737d..1316b9d 100644
--- a/pkg/adapter/awssqssource/adapter.go
+++ b/pkg/adapter/awssqssource/adapter.go
@@ -59,6 +59,11 @@ type envConfig struct {
 	//
 	// Supported values: [ default s3 ]
 	MessageProcessor string `envconfig:"SQS_MESSAGE_PROCESSOR" default:"default"`
+
+	// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html
+	// Visibility timeout to set on all messages received by this event source.
+	// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html
+	VisibilityTimeout *time.Duration `envconfig:"SQS_VISIBILITY_TIMEOUT"`
 }
 
 // adapter implements the source's adapter.
@@ -75,6 +80,8 @@ type adapter struct {
 
 	msgPrcsr MessageProcessor
 
+	visibilityTimeoutSeconds *int64
+
 	processQueue chan *sqs.Message
 	deleteQueue  chan *sqs.Message
 
@@ -112,6 +119,16 @@ func NewAdapter(ctx context.Context, envAcc pkgadapter.EnvConfigAccessor, ceClie
 		panic("unsupported message processor " + strconv.Quote(env.MessageProcessor))
 	}
 
+	var visibilityTimeoutSeconds *int64
+	if vt := env.VisibilityTimeout; vt != nil {
+		if *vt < 0 || *vt > 12*time.Hour {
+			logger.Warn("Ignoring out of bounds visibility timeout (", *vt, ")")
+		} else {
+			vts := durationInSeconds(*vt)
+			visibilityTimeoutSeconds = &vts
+		}
+	}
+
 	cfg := session.Must(session.NewSession(aws.NewConfig().
 		WithRegion(arn.Region),
 	))
@@ -139,6 +156,8 @@ func NewAdapter(ctx context.Context, envAcc pkgadapter.EnvConfigAccessor, ceClie
 
 		msgPrcsr: msgPrcsr,
 
+		visibilityTimeoutSeconds: visibilityTimeoutSeconds,
+
 		processQueue: make(chan *sqs.Message, queueBufferSizeProcess),
 		deleteQueue:  make(chan *sqs.Message, queueBufferSizeDelete),
 
@@ -234,3 +253,11 @@ func prettifyBatchResultErrors(errs []*sqs.BatchResultErrorEntry) string {
 
 	return errStr.String()
 }
+
+// durationInSeconds returns a duration as a number of seconds truncated
+// towards zero.
+func durationInSeconds(d time.Duration) int64 {
+	// converting a floating-point number to an integer discards
+	// the fraction (truncation towards zero)
+	return int64(d.Seconds())
+}
diff --git a/pkg/adapter/awssqssource/receive.go b/pkg/adapter/awssqssource/receive.go
index 3197cfa..8f8531e 100644
--- a/pkg/adapter/awssqssource/receive.go
+++ b/pkg/adapter/awssqssource/receive.go
@@ -38,10 +38,6 @@ const (
 	// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-short-and-long-polling.html#sqs-long-polling
 	maxLongPollingWaitTimeSeconds = 20
 
-	// Visibility timeout to set on all messages received by this event source.
-	// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html
-	visibilityTimeoutSeconds = 30
-
 	// Duration between calls to ReceiveMessage when the previous call didn't return any message.
 	receiveMsgPeriod = 3 * time.Second
 )
@@ -57,7 +53,7 @@ func (a *adapter) runMessagesReceiver(ctx context.Context, queueURL string) {
 			return
 
 		case <-t.C:
-			messages, err := receiveMessages(ctx, a.sqsClient, queueURL)
+			messages, err := receiveMessages(ctx, a.sqsClient, queueURL, a.visibilityTimeoutSeconds)
 			if err != nil {
 				a.logger.Errorw("Failed to get messages from the SQS queue", zap.Error(err))
 				t.Reset(1 * time.Second)
@@ -99,13 +95,15 @@ func (ml messageList) MarshalLogArray(arr zapcore.ArrayEncoder) error {
 
 // receiveMessages returns a batch of messages read from the SQS queue, if any
 // is available.
-func receiveMessages(ctx context.Context, cli sqsiface.SQSAPI, queueURL string) ([]*sqs.Message, error) {
+func receiveMessages(ctx context.Context, cli sqsiface.SQSAPI,
+	queueURL string, visibilityTimeoutSeconds *int64) ([]*sqs.Message, error) {
+
 	resp, err := cli.ReceiveMessageWithContext(ctx, &sqs.ReceiveMessageInput{
 		AttributeNames:      aws.StringSlice([]string{sqs.QueueAttributeNameAll}),
 		QueueUrl:            &queueURL,
 		MaxNumberOfMessages: aws.Int64(maxReceiveMsgBatchSize),
 		WaitTimeSeconds:     aws.Int64(maxLongPollingWaitTimeSeconds),
-		VisibilityTimeout:   aws.Int64(visibilityTimeoutSeconds),
+		VisibilityTimeout:   visibilityTimeoutSeconds,
 	})
 	if err != nil {
 		return nil, err
diff --git a/pkg/apis/sources/v1alpha1/awssqs_types.go b/pkg/apis/sources/v1alpha1/awssqs_types.go
index 345b88b..b9a9f36 100644
--- a/pkg/apis/sources/v1alpha1/awssqs_types.go
+++ b/pkg/apis/sources/v1alpha1/awssqs_types.go
@@ -52,10 +52,30 @@ type AWSSQSSourceSpec struct {
 	// https://docs.aws.amazon.com/IAM/latest/UserGuide/list_amazonsqs.html#amazonsqs-resources-for-iam-policies
 	ARN apis.ARN `json:"arn"`
 
+	// Options that control the behavior of message receivers.
+	// +optional
+	ReceiveOptions *AWSSQSSourceReceiveOptions `json:"receiveOptions,omitempty"`
+
 	// Credentials to interact with the Amazon SQS API.
 	Credentials AWSSecurityCredentials `json:"credentials"`
 }
 
+// AWSSQSSourceReceiveOptions defines options that control the behavior of
+// Amazon SQS message receivers.
+type AWSSQSSourceReceiveOptions struct {
+	// Period of time during which Amazon SQS prevents other consumers from
+	// receiving and processing a message that has been received via ReceiveMessage.
+	// Expressed as a duration string, which format is documented at https://pkg.go.dev/time#ParseDuration.
+	//
+	// If not defined, the overall visibility timeout for the queue is used.
+	//
+	// For more details, please refer to the Amazon SQS Developer Guide at
+	// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html.
+	//
+	// +optional
+	VisibilityTimeout *apis.Duration `json:"visibilityTimeout,omitempty"`
+}
+
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
 // AWSSQSSourceList contains a list of event sources.
diff --git a/pkg/apis/sources/v1alpha1/deepcopy_generated.go b/pkg/apis/sources/v1alpha1/deepcopy_generated.go
index 0689f66..0adbd14 100644
--- a/pkg/apis/sources/v1alpha1/deepcopy_generated.go
+++ b/pkg/apis/sources/v1alpha1/deepcopy_generated.go
@@ -257,6 +257,11 @@ func (in *AWSCloudWatchSourceList) DeepCopyObject() runtime.Object {
 func (in *AWSCloudWatchSourceSpec) DeepCopyInto(out *AWSCloudWatchSourceSpec) {
 	*out = *in
 	in.SourceSpec.DeepCopyInto(&out.SourceSpec)
+	if in.PollingInterval != nil {
+		in, out := &in.PollingInterval, &out.PollingInterval
+		*out = new(apis.Duration)
+		**out = **in
+	}
 	if in.MetricQueries != nil {
 		in, out := &in.MetricQueries, &out.MetricQueries
 		*out = make([]AWSCloudWatchMetricQuery, len(*in))
@@ -264,11 +269,6 @@ func (in *AWSCloudWatchSourceSpec) DeepCopyInto(out *AWSCloudWatchSourceSpec) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
-	if in.PollingInterval != nil {
-		in, out := &in.PollingInterval, &out.PollingInterval
-		*out = new(apis.Duration)
-		**out = **in
-	}
 	in.Credentials.DeepCopyInto(&out.Credentials)
 	return
 }
@@ -1063,11 +1063,37 @@ func (in *AWSSQSSourceList) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *AWSSQSSourceReceiveOptions) DeepCopyInto(out *AWSSQSSourceReceiveOptions) {
+	*out = *in
+	if in.VisibilityTimeout != nil {
+		in, out := &in.VisibilityTimeout, &out.VisibilityTimeout
+		*out = new(apis.Duration)
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AWSSQSSourceReceiveOptions.
+func (in *AWSSQSSourceReceiveOptions) DeepCopy() *AWSSQSSourceReceiveOptions {
+	if in == nil {
+		return nil
+	}
+	out := new(AWSSQSSourceReceiveOptions)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *AWSSQSSourceSpec) DeepCopyInto(out *AWSSQSSourceSpec) {
 	*out = *in
 	in.SourceSpec.DeepCopyInto(&out.SourceSpec)
 	out.ARN = in.ARN
+	if in.ReceiveOptions != nil {
+		in, out := &in.ReceiveOptions, &out.ReceiveOptions
+		*out = new(AWSSQSSourceReceiveOptions)
+		(*in).DeepCopyInto(*out)
+	}
 	in.Credentials.DeepCopyInto(&out.Credentials)
 	return
 }


Test patch:
diff --git a/pkg/adapter/awssqssource/adapter_test.go b/pkg/adapter/awssqssource/adapter_test.go
index ad2420b..f152e98 100644
--- a/pkg/adapter/awssqssource/adapter_test.go
+++ b/pkg/adapter/awssqssource/adapter_test.go
@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/arn"
@@ -39,14 +40,22 @@ import (
 const (
 	tQueueArnResource = "MyQueue"
 	tQueueURL         = "https://sqs.us-fake-0.amazonaws.com/123456789012/MyQueue"
-	tMsgIDPrefix      = "00000000-0000-0000-0000-000000000" // + 3 digits appended for each msg
+
+	tMsgIDPrefix = "00000000-0000-0000-0000-000000000" // + 3 digits appended for each msg
+
+	tVisibilityTimeout = 120 // seconds
 )
 
 func TestAdapter(t *testing.T) {
-	// The test's data is pre-populated so the flow of messages is
-	// uninterrupted until every message has been retrieved. We can
-	// therefore affirm something went wrong if the receiveMsgPeriod timer
-	// happens during a test.
+	// When ReceiveMessage() returns at least one message, the next call to
+	// ReceiveMessage() is sent immediatly instead of waiting for
+	// 'receiveMsgPeriod' (burst mode).
+	// In these tests, data is pre-populated with 'numMsgs' messages to
+	// simulate a queue with a (potentially large) amount of unprocessed
+	// messages, which we expect to keep receivers in burst mode until all
+	// messages have been processed.
+	// For this reason, a test that would take more than 'receiveMsgPeriod'
+	// to complete would very likely indicate a flaw in the burst logic.
 	const testTimeout = receiveMsgPeriod
 
 	arn := makeARN(tQueueArnResource)
@@ -95,6 +104,8 @@ func TestAdapter(t *testing.T) {
 
 				msgPrcsr: &defaultMessageProcessor{ceSource: arn.String()},
 
+				visibilityTimeoutSeconds: aws.Int64(tVisibilityTimeout),
+
 				processQueue: make(chan *sqs.Message, tc.queueBufSize),
 				deleteQueue:  make(chan *sqs.Message, tc.queueBufSize),
 
@@ -144,16 +155,27 @@ func TestAdapter(t *testing.T) {
 				assert.NoError(t, err)
 			}
 
-			// asserting a single event suffices since the entire data set is mocked
-			ev := ceCli.Sent()[0]
-			assert.Equal(t, ev.Type(), "com.amazon.sqs.message")
-			assert.Equal(t, "arn:aws:sqs:us-fake-0:123456789012:MyQueue", ev.Source())
-			assert.Contains(t, ev.ID(), tMsgIDPrefix)
-
-			// final assertions
-			assert.Len(t, ceCli.Sent(), tc.numMsgs, "Received more events than expected")
-			assert.Equal(t, tc.numMsgs, sqsCli.totalDeleted, "Not all processed messages were deleted")
-			assert.Empty(t, sqsCli.inFlightMsgs, "Found unprocessed in-flight messages")
+			assrt := assert.New(t)
+			reqr := require.New(t)
+
+			// assertions on sent events
+			// (asserting attributes of a single event suffices since the entire data set is mocked)
+			sentEvents := ceCli.Sent()
+			reqr.Greater(len(sentEvents), 0, "Expected at least one CloudEvent to be sent")
+			assrt.Equal(sentEvents[0].Type(), "com.amazon.sqs.message")
+			assrt.Equal("arn:aws:sqs:us-fake-0:123456789012:MyQueue", sentEvents[0].Source(), "CloudEvent source should match queue ARN")
+			assrt.Contains(sentEvents[0].ID(), tMsgIDPrefix, "CloudEvent id should match SQS message ID")
+
+			// assertions on post-test queue state
+			assrt.Len(ceCli.Sent(), tc.numMsgs, "Received more events than expected")
+			assrt.Equal(tc.numMsgs, sqsCli.totalDeleted, "Not all processed messages were deleted")
+			assrt.Empty(sqsCli.inFlightMsgs, "Found unprocessed in-flight messages")
+
+			// assertions on API requests and their parameters
+			rcvMsgRequests := sqsCli.rcvMsgRecorder.requests
+			reqr.Greater(len(rcvMsgRequests), 0, "Expected at least one ReceiveMessage request")
+			reqr.NotNil(rcvMsgRequests[0].VisibilityTimeout, "VisibilityTimeout should be set in ReceiveMessage requests")
+			assrt.EqualValues(tVisibilityTimeout, *rcvMsgRequests[0].VisibilityTimeout)
 		})
 	}
 }
@@ -175,10 +197,14 @@ type standardMockSQSClient struct {
 	sqsiface.SQSAPI
 
 	sync.Mutex
-	availMsgs    []*sqs.Message
-	inFlightMsgs []*sqs.Message
+	availMsgs []*sqs.Message
 
+	// messages which have been received but not yet deleted are kept in this queue
+	inFlightMsgs []*sqs.Message
+	// counter for the number messages explicitly deleted by message deleters
 	totalDeleted int
+
+	rcvMsgRecorder receiveMessageRequestRecorder
 }
 
 func (*standardMockSQSClient) GetQueueUrl(*sqs.GetQueueUrlInput) (*sqs.GetQueueUrlOutput, error) { //nolint:golint,stylecheck
@@ -190,6 +216,8 @@ func (*standardMockSQSClient) GetQueueUrl(*sqs.GetQueueUrlInput) (*sqs.GetQueueU
 func (c *standardMockSQSClient) ReceiveMessageWithContext(_ context.Context,
 	in *sqs.ReceiveMessageInput, _ ...request.Option) (*sqs.ReceiveMessageOutput, error) {
 
+	c.rcvMsgRecorder.Record(in)
+
 	c.Lock()
 	defer c.Unlock()
 
@@ -311,3 +339,17 @@ func TestDeleteMessageBatchWithContext(t *testing.T) {
 	assert.EqualValues(t, expect, sqsClient.inFlightMsgs)
 	assert.Equal(t, len(in.Entries), sqsClient.totalDeleted)
 }
+
+// receiveMessageRequestRecorder records calls to ReceiveMessage.
+type receiveMessageRequestRecorder struct {
+	sync.Mutex
+	requests []*sqs.ReceiveMessageInput
+}
+
+// receiveMessageRequestRecorder records calls to ReceiveMessage.
+func (r *receiveMessageRequestRecorder) Record(req *sqs.ReceiveMessageInput) {
+	r.Lock()
+	defer r.Unlock()
+
+	r.requests = append(r.requests, req)
+}