feat: Added dlq option to aws-sqs eventsource spec (#1403)

* feat: Added dlq option to aws-sqs eventsource spec Signed-off-by: Daniel Soifer <[email protected]> * codegen Signed-off-by: Daniel Soifer <[email protected]>
argoproj · Nov 2, 2021 · 4a1bf22 · 4a1bf22
1 parent 73bafd4
commit 4a1bf22
Show file tree

Hide file tree

Showing 9 changed files with 453 additions and 365 deletions.
diff --git a/api/event-source.html b/api/event-source.html
diff --git a/api/event-source.md b/api/event-source.md
diff --git a/api/jsonschema/schema.json b/api/jsonschema/schema.json
@@ -1968,6 +1968,10 @@
           "$ref": "#/definitions/io.k8s.api.core.v1.SecretKeySelector",
           "description": "AccessKey refers K8s secret containing aws access key"
         },
+        "dlq": {
+          "description": "DLQ specifies if a dead-letter queue is configured for messages that can't be processed successfully. If set to true, messages with invalid payload won't be acknowledged to allow to forward them farther to the dead-letter queue. The default value is false.",
+          "type": "boolean"
+        },
         "jsonBody": {
           "description": "JSONBody specifies that all event body payload coming from this source will be JSON",
           "type": "boolean"

diff --git a/api/openapi-spec/swagger.json b/api/openapi-spec/swagger.json
diff --git a/eventsources/sources/awssqs/start.go b/eventsources/sources/awssqs/start.go
@@ -142,7 +142,10 @@ func (el *EventListener) processMessage(ctx context.Context, message *sqslib.Mes
 	if err != nil {
 		log.Errorw("failed to marshal event data, will process next message...", zap.Error(err))
 		el.Metrics.EventProcessingFailed(el.GetEventSourceName(), el.GetEventName())
-		ack()
+		// Don't ack if a DLQ is configured to allow to forward the message to the DLQ
+		if !el.SQSEventSource.DLQ {
+			ack()
+		}
 		return
 	}
 	if err = dispatch(eventBytes); err != nil {

diff --git a/pkg/apis/eventsource/v1alpha1/generated.pb.go b/pkg/apis/eventsource/v1alpha1/generated.pb.go
diff --git a/pkg/apis/eventsource/v1alpha1/generated.proto b/pkg/apis/eventsource/v1alpha1/generated.proto
diff --git a/pkg/apis/eventsource/v1alpha1/openapi_generated.go b/pkg/apis/eventsource/v1alpha1/openapi_generated.go
diff --git a/pkg/apis/eventsource/v1alpha1/types.go b/pkg/apis/eventsource/v1alpha1/types.go
@@ -604,6 +604,11 @@ type SQSEventSource struct {
 	// Metadata holds the user defined metadata which will passed along the event payload.
 	// +optional
 	Metadata map[string]string `json:"metadata,omitempty" protobuf:"bytes,9,rep,name=metadata"`
+	// DLQ specifies if a dead-letter queue is configured for messages that can't be processed successfully.
+	// If set to true, messages with invalid payload won't be acknowledged to allow to forward them farther to the dead-letter queue.
+	// The default value is false.
+	// +optional
+	DLQ bool `json:"dlq,omitempty" protobuf:"varint,10,opt,name=dlq"`
 }
 
 // PubSubEventSource refers to event-source for GCP PubSub related events.