Add SQS Custom endpoint support (#1632)

* feat: Add support for custom SQS endpoints Signed-off-by: itamarom <[email protected]>
argoproj · Feb 15, 2022 · 00e2ae8 · 00e2ae8
1 parent 4721cf3
commit 00e2ae8
Show file tree

Hide file tree

Showing 10 changed files with 483 additions and 389 deletions.
diff --git a/USERS.md b/USERS.md
@@ -21,6 +21,7 @@ Organizations below are **officially** using Argo Events. Please send a PR with
 1. [Helio](https://helio.exchange)
 1. [InsideBoard](https://www.insideboard.com)
 1. [Intuit](https://www.intuit.com/)
+1. [Loam](https://www.getloam.com/)
 1. [Mobimeo GmbH](https://mobimeo.com/en/home/)
 1. [OneCause](https://www.onecause.com/)
 1. [Produvar](https://www.produvar.com/)

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
@@ -2212,6 +2212,10 @@
           "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"
         },
+        "endpoint": {
+          "description": "Endpoint configures connection to a specific SQS endpoint instead of Amazons servers",
+          "type": "string"
+        },
         "filter": {
           "$ref": "#/definitions/io.argoproj.eventsource.v1alpha1.EventSourceFilter",
           "description": "Filter"

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
@@ -75,7 +75,13 @@ func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byt
 		return errors.Wrapf(err, "failed to create aws session for %s", el.GetEventName())
 	}
 
-	sqsClient := sqslib.New(awsSession)
+	var sqsClient *sqslib.SQS
+
+	if sqsEventSource.Endpoint == "" {
+		sqsClient = sqslib.New(awsSession)
+	} else {
+		sqsClient = sqslib.New(awsSession, &aws.Config{Endpoint: &sqsEventSource.Endpoint, Region: &sqsEventSource.Region})
+	}
 
 	log.Info("fetching queue url...")
 	getQueueURLInput := &sqslib.GetQueueUrlInput{

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
@@ -645,6 +645,9 @@ type SQSEventSource struct {
 	// Filter
 	// +optional
 	Filter *EventSourceFilter `json:"filter,omitempty" protobuf:"bytes,11,opt,name=filter"`
+	// Endpoint configures connection to a specific SQS endpoint instead of Amazons servers
+	// +optional
+	Endpoint string `json:"endpoint" protobuf:"bytes,12,opt,name=endpoint"`
 }
 
 // PubSubEventSource refers to event-source for GCP PubSub related events.