Repository: triggermesh/aws-event-sources
Pull request #213 (2020-10-27 22:16:02)

Receive and delete SQS messages in batches
Process events in batches and increase the throughput from **~0.8 message/sec** to **~25 messagse/sec** (**~7.2 messages/sec** if we exclude the backoff  change in #214), considering a payload of 2 KiB.

We could improve this even more by also sending CloudEvents in batches, but neither the Go SDK not Knative dispatchers support that. As a workaround, I refactored the `sendSQSEvents` function to send each message in a separate goroutine.

There is still room for improvement, e.g. we could run multiple receiver goroutines and perform deletions in the background, but that's a topic for future experiments.

antoineco: Worth mentioning: SQS doesn't guarantee that `Receive` always returns a number of messages equal to `MaxNumberOfMessages`, even if the queue has this number of messages available. During my tests, processing exactly 100 messages each time, I noticed that we were sometimes processing batches of random sizes:

```sql
returned  10  events  # 10
returned  10  events  # 20
returned  8   events  # 28
returned  10  events  # 38
returned  10  events  # 48
returned  10  events  # 58
returned  2   events  # 60
returned  9   events  # 69
returned  10  events  # 79
returned  10  events  # 89
returned  2   events  # 91
returned  9   events  # 100
```

Patch:
diff --git a/pkg/adapter/awssqssource/adapter.go b/pkg/adapter/awssqssource/adapter.go
index fe7b5f7..1731459 100644
--- a/pkg/adapter/awssqssource/adapter.go
+++ b/pkg/adapter/awssqssource/adapter.go
@@ -19,9 +19,12 @@ package awssqssource
 import (
 	"context"
 	"fmt"
+	"strings"
+	"sync"
 	"time"
 
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 
@@ -38,6 +41,14 @@ import (
 	"github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
 )
 
+// Highest possible value for the MaxNumberOfMessages request parameter.
+// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html
+const maxReceiveMsgBatchSize = 10
+
+// Highest possible duration of a long polling request.
+// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-short-and-long-polling.html#sqs-long-polling
+const maxLongPollingWaitTimeSeconds = 20
+
 // envConfig is a set parameters sourced from the environment for the source's
 // adapter.
 type envConfig struct {
@@ -100,23 +111,25 @@ func (a *adapter) Start(ctx context.Context) error {
 	err = backoff.Run(ctx.Done(), func(ctx context.Context) (bool, error) {
 		messages, err := a.getMessages(queueURL)
 		if err != nil {
-			a.logger.Errorw("Failed to get messages from SQS queue", "error", err)
+			a.logger.Errorw("Failed to get messages from the SQS queue", zap.Error(err))
 			return false, nil
 		}
 
-		for _, message := range messages {
-			err = a.sendSQSEvent(message)
-			if err != nil {
-				a.logger.Errorw("Failed to send event", "error", err)
-				continue
-			}
+		if len(messages) == 0 {
+			return true, nil
+		}
 
-			// Delete message from queue if we pushed successfully
-			err = a.deleteMessage(queueURL, message.ReceiptHandle)
-			if err != nil {
-				a.logger.Errorw("Failed to delete message from SQS queue", "error", err)
-				continue
-			}
+		sent, err := a.sendSQSEvents(messages)
+		if err != nil {
+			// log the error but proceed with deleting the events
+			// that were successfully sent to the sink
+			a.logger.Errorw("Failed to send events to the sink", zap.Error(err))
+		}
+
+		err = a.deleteMessages(queueURL, sent)
+		if err != nil {
+			a.logger.Errorw("Failed to delete message from the SQS queue", zap.Error(err))
+			return false, nil
 		}
 
 		return true, nil
@@ -137,8 +150,10 @@ func (a *adapter) queueLookup(queueName string) (*sqs.GetQueueUrlOutput, error)
 // occurs that error will be returned.
 func (a *adapter) getMessages(queueURL string) ([]*sqs.Message, error) {
 	params := sqs.ReceiveMessageInput{
-		AttributeNames: aws.StringSlice([]string{sqs.QueueAttributeNameAll}),
-		QueueUrl:       &queueURL,
+		AttributeNames:      aws.StringSlice([]string{sqs.QueueAttributeNameAll}),
+		QueueUrl:            &queueURL,
+		MaxNumberOfMessages: aws.Int64(maxReceiveMsgBatchSize),
+		WaitTimeSeconds:     aws.Int64(maxLongPollingWaitTimeSeconds),
 	}
 	resp, err := a.sqsClient.ReceiveMessage(&params)
 	if err != nil {
@@ -147,39 +162,142 @@ func (a *adapter) getMessages(queueURL string) ([]*sqs.Message, error) {
 	return resp.Messages, nil
 }
 
-func (a *adapter) sendSQSEvent(msg *sqs.Message) error {
-	a.logger.Infof("Processing message with ID: %s", *msg.MessageId)
+// sendSQSEvents sends SQS messages to the event sink.
+func (a *adapter) sendSQSEvents(msgs []*sqs.Message) (sent []*sqs.Message, err error) {
+	// NOTE(antoineco): the CloudEvents SDK for Go doesn't support the
+	// batched content mode, although it is defined in the spec.
+	// https://github.com/cloudevents/spec/blob/v1.0/http-protocol-binding.md#33-batched-content-mode
+
+	concurrentSent := concurrentMsgSlice{
+		msgs: make([]*sqs.Message, 0, len(msgs)),
+	}
+
+	errCh := make(chan error, len(msgs))
+	defer close(errCh)
+
+	for _, msg := range msgs {
+		a.logger.Debugf("Processing messages ID %s", *msg.MessageId)
+
+		go func(msg *sqs.Message) {
+			err := sendSQSEvent(a.ceClient, msg, a.arn)
+			if err != nil {
+				errCh <- fmt.Errorf("message ID %s: %w", *msg.MessageId, err)
+				return
+			}
+
+			concurrentSent.append(msg)
+
+			// always write to errCh to notify that sendSQSEvent returned
+			errCh <- err
+		}(msg)
+	}
+
+	var errs []error
+
+	for i := 0; i < cap(errCh); i++ {
+		if err := <-errCh; err != nil {
+			errs = append(errs, err)
+		}
+	}
+
+	if len(errs) > 0 {
+		err = &errList{errs: errs}
+	}
+
+	return concurrentSent.msgs, err
+}
+
+// concurrentMsgSlice protects writes to a slice of Messages using a mutex.
+type concurrentMsgSlice struct {
+	sync.Mutex
+	msgs []*sqs.Message
+}
+
+// append appends a new Message to the Messages slice in a thread-safe manner.
+func (ms *concurrentMsgSlice) append(msg *sqs.Message) {
+	ms.Lock()
+	ms.msgs = append(ms.msgs, msg)
+	ms.Unlock()
+}
 
+// sendSQSEvent sends a single SQS message to the event sink.
+func sendSQSEvent(cli cloudevents.Client, msg *sqs.Message, queueARN arn.ARN) error {
 	// TODO: work on CE attributes contract
 	subject, exist := msg.Attributes[sqs.MessageSystemAttributeNameSenderId]
 	if !exist {
 		subject = msg.MessageId
 	}
-	event := cloudevents.NewEvent(cloudevents.VersionV1)
-	event.SetType(v1alpha1.AWSEventType(a.arn.Service, v1alpha1.AWSSQSGenericEventType))
+
+	event := cloudevents.NewEvent()
+	event.SetType(v1alpha1.AWSEventType(queueARN.Service, v1alpha1.AWSSQSGenericEventType))
 	event.SetSubject(*subject)
-	event.SetSource(a.arn.String())
+	event.SetSource(queueARN.String())
 	event.SetID(*msg.MessageId)
 	if err := event.SetData(cloudevents.ApplicationJSON, msg); err != nil {
 		return fmt.Errorf("failed to set event data: %w", err)
 	}
 
-	if result := a.ceClient.Send(context.Background(), event); !cloudevents.IsACK(result) {
+	if result := cli.Send(context.Background(), event); !cloudevents.IsACK(result) {
 		return result
 	}
 	return nil
 }
 
-// deleteMessage deletes a message from the SQS queue.
-func (a *adapter) deleteMessage(queueURL string, receiptHandle *string) error {
-	deleteParams := &sqs.DeleteMessageInput{
-		QueueUrl:      &queueURL,
-		ReceiptHandle: receiptHandle,
+// deleteMessages deletes messages from the SQS queue.
+func (a *adapter) deleteMessages(queueURL string, msgs []*sqs.Message) error {
+	deleteEntries := make([]*sqs.DeleteMessageBatchRequestEntry, len(msgs))
+	for i, msg := range msgs {
+		deleteEntries[i] = &sqs.DeleteMessageBatchRequestEntry{
+			Id:            msg.MessageId,
+			ReceiptHandle: msg.ReceiptHandle,
+		}
+	}
+
+	deleteParams := &sqs.DeleteMessageBatchInput{
+		QueueUrl: &queueURL,
+		Entries:  deleteEntries,
 	}
-	if _, err := a.sqsClient.DeleteMessage(deleteParams); err != nil {
+	if _, err := a.sqsClient.DeleteMessageBatch(deleteParams); err != nil {
 		return err
 	}
 
-	a.logger.Debug("Message deleted")
+	if a.logger.Desugar().Core().Enabled(zapcore.DebugLevel) {
+		msgIds := make([]string, len(deleteEntries))
+		for i, msg := range deleteEntries {
+			msgIds[i] = *msg.Id
+		}
+		a.logger.Debugf("Deleted message IDs %v", msgIds)
+	}
+
 	return nil
 }
+
+type errList struct {
+	errs []error
+}
+
+var _ error = (*errList)(nil)
+
+// Error implements the error interface.
+func (e *errList) Error() string {
+	if e == nil || len(e.errs) == 0 {
+		return ""
+	}
+
+	if len(e.errs) == 1 {
+		return e.errs[0].Error()
+	}
+
+	var b strings.Builder
+
+	b.WriteByte('[')
+	for i, err := range e.errs {
+		b.WriteString(err.Error())
+		if i != len(e.errs)-1 {
+			b.WriteString(", ")
+		}
+	}
+	b.WriteByte(']')
+
+	return b.String()
+}


Test patch:
diff --git a/pkg/adapter/awssqssource/adapter_test.go b/pkg/adapter/awssqssource/adapter_test.go
index 77c2515..57a32ef 100644
--- a/pkg/adapter/awssqssource/adapter_test.go
+++ b/pkg/adapter/awssqssource/adapter_test.go
@@ -18,9 +18,13 @@ package awssqssource
 
 import (
 	"errors"
+	"sort"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/sqs"
@@ -32,61 +36,68 @@ import (
 
 type mockedReceiveMsgs struct {
 	sqsiface.SQSAPI
-	Resp sqs.ReceiveMessageOutput
+	resp *sqs.ReceiveMessageOutput
 	err  error
 }
 
 type mockedDeleteMsgs struct {
 	sqsiface.SQSAPI
-	Resp sqs.DeleteMessageOutput
+	resp *sqs.DeleteMessageBatchOutput
 	err  error
 }
 
 type mockedGetQueueURL struct {
 	sqsiface.SQSAPI
-	Resp sqs.GetQueueUrlOutput
+	resp *sqs.GetQueueUrlOutput
 	err  error
 }
 
 func (m mockedReceiveMsgs) ReceiveMessage(in *sqs.ReceiveMessageInput) (*sqs.ReceiveMessageOutput, error) {
-	return &m.Resp, m.err
+	return m.resp, m.err
 }
 
-func (m mockedDeleteMsgs) DeleteMessage(in *sqs.DeleteMessageInput) (*sqs.DeleteMessageOutput, error) {
-	return &m.Resp, m.err
+func (m mockedDeleteMsgs) DeleteMessageBatch(in *sqs.DeleteMessageBatchInput) (*sqs.DeleteMessageBatchOutput, error) {
+	return m.resp, m.err
 }
 
 func (m mockedGetQueueURL) GetQueueUrl(*sqs.GetQueueUrlInput) (*sqs.GetQueueUrlOutput, error) { //nolint:golint,stylecheck
-	return &m.Resp, m.err
+	return m.resp, m.err
 }
 
 func TestQueueLookup(t *testing.T) {
 	cases := []struct {
-		Resp     sqs.GetQueueUrlOutput
+		resp     *sqs.GetQueueUrlOutput
 		err      error
-		Expected *string
+		expected *string
 	}{
 		{ // Case 1, expect parsed responses
-			Resp:     sqs.GetQueueUrlOutput{QueueUrl: aws.String("testQueueURL")},
+			resp: &sqs.GetQueueUrlOutput{
+				QueueUrl: aws.String("testQueueURL"),
+			},
 			err:      nil,
-			Expected: aws.String("testQueueURL"),
+			expected: aws.String("testQueueURL"),
 		},
 		{ // Case 2, expect error
-			Resp:     sqs.GetQueueUrlOutput{QueueUrl: aws.String("")},
+			resp: &sqs.GetQueueUrlOutput{
+				QueueUrl: aws.String(""),
+			},
 			err:      errors.New("fake getQueueUrl error"),
-			Expected: aws.String(""),
+			expected: aws.String(""),
 		},
 	}
 
 	for _, c := range cases {
 		a := &adapter{
-			logger:    loggingtesting.TestLogger(t),
-			sqsClient: mockedGetQueueURL{Resp: c.Resp, err: c.err},
+			logger: loggingtesting.TestLogger(t),
+			sqsClient: mockedGetQueueURL{
+				resp: c.resp,
+				err:  c.err,
+			},
 		}
 
-		url, err := a.queueLookup("testQueue")
+		url, err := a.queueLookup("")
 		assert.Equal(t, c.err, err)
-		assert.Equal(t, c.Expected, url.QueueUrl)
+		assert.Equal(t, c.expected, url.QueueUrl)
 	}
 }
 
@@ -94,47 +105,49 @@ func TestGetMessages(t *testing.T) {
 	const queueURL = "mockURL"
 
 	cases := []struct {
-		Resp     sqs.ReceiveMessageOutput
+		resp     *sqs.ReceiveMessageOutput
 		err      error
-		Expected []sqs.Message
+		expected []*sqs.Message
 	}{
 		{ // Case 1, expect parsed responses
-			Resp: sqs.ReceiveMessageOutput{
-				Messages: []*sqs.Message{
-					{},
-				},
-			},
-			err: nil,
-			Expected: []sqs.Message{
-				{},
+			resp: &sqs.ReceiveMessageOutput{
+				Messages: make([]*sqs.Message, 2),
 			},
+			err:      nil,
+			expected: make([]*sqs.Message, 2),
 		},
-		{ // Case 2, not messages returned
-			Resp:     sqs.ReceiveMessageOutput{},
+		{ // Case 2, no message returned
+			resp:     &sqs.ReceiveMessageOutput{},
 			err:      errors.New("no message found"),
-			Expected: []sqs.Message{},
+			expected: []*sqs.Message{},
 		},
 	}
 
 	for _, c := range cases {
 		a := &adapter{
-			logger:    loggingtesting.TestLogger(t),
-			sqsClient: mockedReceiveMsgs{Resp: c.Resp, err: c.err},
+			logger: loggingtesting.TestLogger(t),
+			sqsClient: mockedReceiveMsgs{
+				resp: c.resp,
+				err:  c.err},
 		}
 
 		msgs, err := a.getMessages(queueURL)
 		assert.Equal(t, c.err, err)
-		assert.Equal(t, len(c.Expected), len(msgs))
+		assert.Equal(t, len(c.expected), len(msgs))
 	}
 }
 
-func TestPushMessage(t *testing.T) {
-	msg := sqs.Message{
-		MessageId: aws.String("foo"),
-		Body:      aws.String("bar"),
-		Attributes: map[string]*string{
-			"SentTimestamp": aws.String("1549540781"),
-		}}
+func TestSendEvents(t *testing.T) {
+	msgs := []*sqs.Message{
+		{
+			MessageId: aws.String("0001"),
+			Body:      aws.String("msg1"),
+		},
+		{
+			MessageId: aws.String("0002"),
+			Body:      aws.String("msg2"),
+		},
+	}
 
 	ceClient := adaptertest.NewTestClient()
 
@@ -143,23 +156,40 @@ func TestPushMessage(t *testing.T) {
 		ceClient: ceClient,
 	}
 
-	err := a.sendSQSEvent(&msg)
+	sent, err := a.sendSQSEvents(msgs)
+	ceClientSentEvents := ceClient.Sent()
 	assert.NoError(t, err)
+	assert.Len(t, sent, len(msgs), "The function didn't return the expected number of messages")
+	require.Len(t, ceClientSentEvents, len(msgs), "The client didn't send the expected number of messages")
 
-	gotEvents := ceClient.Sent()
-	assert.Len(t, gotEvents, 1, "Expected 1 event, got %d", len(gotEvents))
+	sort.Sort(eventsByID(ceClientSentEvents))
 
-	var gotData sqs.Message
-	err = gotEvents[0].DataAs(&gotData)
-	assert.NoError(t, err)
-	assert.EqualValues(t, msg, gotData, "Expected event %q, got %q", msg, gotData)
+	for i, e := range ceClientSentEvents {
+		sentMsg := &sqs.Message{}
+		err := e.DataAs(sentMsg)
+		assert.NoError(t, err)
+		assert.EqualValues(t, msgs[i], sentMsg, "%d: sent payload differs from original message", i)
+	}
 }
 
+type eventsByID []cloudevents.Event
+
+func (ce eventsByID) Len() int           { return len(ce) }
+func (ce eventsByID) Less(i, j int) bool { return ce[i].ID() < ce[j].ID() }
+func (ce eventsByID) Swap(i, j int)      { ce[i], ce[j] = ce[j], ce[i] }
+
 func TestDeleteMessage(t *testing.T) {
 	const queueURL = "mockURL"
 
-	msg := sqs.Message{
-		ReceiptHandle: aws.String("foo"),
+	msgs := []*sqs.Message{
+		{
+			MessageId:     aws.String("0001"),
+			ReceiptHandle: aws.String("0001"),
+		},
+		{
+			MessageId:     aws.String("0002"),
+			ReceiptHandle: aws.String("0002"),
+		},
 	}
 
 	a := &adapter{
@@ -167,18 +197,55 @@ func TestDeleteMessage(t *testing.T) {
 	}
 
 	a.sqsClient = mockedDeleteMsgs{
-		Resp: sqs.DeleteMessageOutput{},
+		resp: &sqs.DeleteMessageBatchOutput{},
 		err:  nil,
 	}
 
-	err := a.deleteMessage(queueURL, msg.ReceiptHandle)
+	err := a.deleteMessages(queueURL, msgs)
 	assert.NoError(t, err)
 
 	a.sqsClient = mockedDeleteMsgs{
-		Resp: sqs.DeleteMessageOutput{},
+		resp: &sqs.DeleteMessageBatchOutput{},
 		err:  errors.New("fake deleteMessage error"),
 	}
 
-	err = a.deleteMessage(queueURL, msg.ReceiptHandle)
+	err = a.deleteMessages(queueURL, msgs)
 	assert.Error(t, err)
 }
+
+func TestErrList(t *testing.T) {
+	testCases := []struct {
+		name   string
+		errors []error
+		expect string
+	}{
+		{
+			name:   "no error",
+			errors: nil,
+			expect: "",
+		},
+		{
+			name: "one error",
+			errors: []error{
+				errors.New("err1"),
+			},
+			expect: "err1",
+		},
+		{
+			name: "multiple errors",
+			errors: []error{
+				errors.New("err1"),
+				errors.New("err2"),
+				errors.New("err3"),
+			},
+			expect: "[err1, err2, err3]",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			errs := &errList{errs: tc.errors}
+			assert.EqualError(t, errs, tc.expect)
+		})
+	}
+}