This is an automated email from the ASF dual-hosted git repository.
squakez pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-k.git
The following commit(s) were added to refs/heads/main by this push:
new 732085555 feat(trait/keda): add AWS SQS scaler mapping (#6914)
732085555 is described below
commit 73208555537b937c8a9bf733bbe7986ee115fd9f
Author: sarankumar <[email protected]>
AuthorDate: Wed Nov 11 21:27:40 2026 +0530
feat(trait/keda): add AWS SQS scaler mapping (#6914)
* feat(trait/keda): add AWS SQS scaler mapping
Map the aws2-sqs consumer to the KEDA aws-sqs-queue scaler. queueURL
comes from the queueUrl parameter, a queue ARN, or the queue name with
region and queueOwnerAWSAccountId. awsRegion comes from region or the
ARN, and awsEndpoint from uriEndpointOverride when overrideEndpoint is
enabled.
Fixes #6908
* test(trait/keda): cover aws2-sqs URI to trigger mapping
Add MapToKedaTrigger tests for full aws2-sqs endpoint URIs, including
static credentials and an ElasticMQ endpoint override. When no full
queue URL can be built, use the queue name as queueURL, which KEDA
accepts when it is unambiguous.
* test(e2e/keda): add AWS SQS ElasticMQ autoscale coverage
Align aws2-sqs ScaleMapper with the props-aware Map signature from
main, and add KEDA e2e fixtures/tests using ElasticMQ (publish,
consume, scale to 0) plus auto-discovery metadata checks.
Signed-off-by: sarankumar1325
<[email protected]>
* test(e2e/keda): use a random namespace for aws2-sqs tests
Switch to WithNewTestNamespace so the framework can delete resources.
Substitute %%% in ElasticMQ endpoint URLs so KEDA (in another namespace)
can reach the queue via elasticmq.<ns>.svc.
Signed-off-by: sarankumar1325
<[email protected]>
* test(e2e/keda): reuse shared cloneAndReplaceNamespace helper
Drop the duplicate helper now that pulsar e2e already defines it.
Signed-off-by: sarankumar1325
<[email protected]>
* test(e2e/keda): use the full ElasticMQ queue URL in the SQS trigger
KEDA's aws-sqs-queue scaler sends queueURL verbatim as the QueueUrl of
GetQueueAttributes. ElasticMQ rejects the plain name "my-queue" there
("Invalid queue url, the path should be /<accountId>/<queueName>"), so
the scaler errored on every poll, KEDA kept the idle consumer at one
replica, and TestAws2SqsKedaAutoscale timed out waiting for zero.
Signed-off-by: sarankumar1325
<[email protected]>
---------
Signed-off-by: sarankumar1325
<[email protected]>
---
docs/modules/traits/pages/keda.adoc | 1 +
e2e/keda/files/elasticmq-ephemeral.yaml | 93 +++++++++++
.../files/keda-it-aws2-sqs-auto-discovery.yaml | 36 ++++
e2e/keda/files/keda-it-aws2-sqs-to-log-auto.yaml | 49 ++++++
e2e/keda/files/timer-to-aws2-sqs.yaml | 30 ++++
e2e/keda/keda_aws2_sqs_test.go | 106 ++++++++++++
pkg/trait/keda/scalers/aws2_sqs.go | 78 +++++++++
pkg/trait/keda/scalers/aws2_sqs_test.go | 182 +++++++++++++++++++++
8 files changed, 575 insertions(+)
diff --git a/docs/modules/traits/pages/keda.adoc
b/docs/modules/traits/pages/keda.adoc
index cd282d900..cc83648ca 100644
--- a/docs/modules/traits/pages/keda.adoc
+++ b/docs/modules/traits/pages/keda.adoc
@@ -76,6 +76,7 @@ When `keda.auto` is enabled, Camel K inspects the consumer
endpoints in an Integ
Automatic discovery currently supports the following Camel components:
* `kafka`: creates an Apache Kafka trigger from the topic, broker, and
consumer-group values in the endpoint URI. See the
https://keda.sh/docs/latest/scalers/apache-kafka/[KEDA Apache Kafka scaler]
documentation for the available trigger metadata.
+* `aws2-sqs`: creates an AWS SQS Queue trigger. `queueURL` comes from the
`queueUrl` endpoint parameter, or is built from a queue ARN, or from the queue
name with the `region` and `queueOwnerAWSAccountId` parameters. Otherwise the
plain queue name is used, which KEDA accepts when it is unambiguous.
`awsRegion` comes from `region` or the ARN, and `awsEndpoint` from
`uriEndpointOverride` when `overrideEndpoint=true`. Set other metadata, such as
`queueLength`, with `keda.autoMetadata`, and [...]
* `pulsar`: creates an Apache Pulsar trigger from the topic in the endpoint
URI and the `subscriptionName` option (Camel's default `subs` when it is not
set). When the service URL (`serviceUrl` option or
`camel.component.pulsar.service-url` property) uses the `pulsar://` scheme,
`adminURL` is set to `http://<host>:8080`; for TLS or a custom admin port, set
it with `keda.autoMetadata`. See the
https://keda.sh/docs/latest/scalers/pulsar/[KEDA Apache Pulsar scaler]
documentation for the ava [...]
Unsupported Camel components do not produce an automatically discovered
trigger. You can still configure their KEDA triggers explicitly with
`keda.triggers`.
diff --git a/e2e/keda/files/elasticmq-ephemeral.yaml
b/e2e/keda/files/elasticmq-ephemeral.yaml
new file mode 100644
index 000000000..df11a5f97
--- /dev/null
+++ b/e2e/keda/files/elasticmq-ephemeral.yaml
@@ -0,0 +1,93 @@
+# ---------------------------------------------------------------------------
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+# ---------------------------------------------------------------------------
+
+apiVersion: v1
+kind: Secret
+metadata:
+ name: aws-sqs-credentials-secret
+type: Opaque
+stringData:
+ AWS_ACCESS_KEY_ID: test
+ AWS_SECRET_ACCESS_KEY: test
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+ name: elasticmq-config
+data:
+ elasticmq.conf: |
+ include classpath("application.conf")
+
+ queues {
+ my-queue {
+ defaultVisibilityTimeout = 10 seconds
+ delay = 0 seconds
+ receiveMessageWait = 0 seconds
+ }
+ }
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+ name: elasticmq
+spec:
+ replicas: 1
+ selector:
+ matchLabels:
+ app: elasticmq
+ template:
+ metadata:
+ labels:
+ app: elasticmq
+ spec:
+ containers:
+ - name: elasticmq
+ image: softwaremill/elasticmq-native:latest
+ ports:
+ - containerPort: 9324
+ name: sqs
+ - containerPort: 9325
+ name: ui
+ volumeMounts:
+ - name: config
+ mountPath: /opt/elasticmq.conf
+ subPath: elasticmq.conf
+ resources:
+ requests:
+ memory: "128Mi"
+ cpu: "100m"
+ limits:
+ memory: "256Mi"
+ volumes:
+ - name: config
+ configMap:
+ name: elasticmq-config
+---
+apiVersion: v1
+kind: Service
+metadata:
+ name: elasticmq
+spec:
+ selector:
+ app: elasticmq
+ ports:
+ - name: sqs
+ port: 9324
+ targetPort: 9324
+ - name: ui
+ port: 9325
+ targetPort: 9325
diff --git a/e2e/keda/files/keda-it-aws2-sqs-auto-discovery.yaml
b/e2e/keda/files/keda-it-aws2-sqs-auto-discovery.yaml
new file mode 100644
index 000000000..15470bdff
--- /dev/null
+++ b/e2e/keda/files/keda-it-aws2-sqs-auto-discovery.yaml
@@ -0,0 +1,36 @@
+# ---------------------------------------------------------------------------
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+# ---------------------------------------------------------------------------
+
+apiVersion: camel.apache.org/v1
+kind: Integration
+metadata:
+ name: keda-aws2-sqs-auto-discovery
+spec:
+ flows:
+ - route:
+ from:
+ steps:
+ - to: log:info
+ uri:
aws2-sqs:my-queue?region=us-east-1&accessKey=test&secretKey=test&overrideEndpoint=true&uriEndpointOverride=http://elasticmq.%%%.svc:9324
+ traits:
+ keda:
+ enabled: true
+ minReplicaCount: 0
+ maxReplicaCount: 1
+ cooldownPeriod: 20
+ pollingInterval: 10
+ # No triggers specified - auto-discovery should create aws-sqs-queue
scaler
diff --git a/e2e/keda/files/keda-it-aws2-sqs-to-log-auto.yaml
b/e2e/keda/files/keda-it-aws2-sqs-to-log-auto.yaml
new file mode 100644
index 000000000..71eba41ff
--- /dev/null
+++ b/e2e/keda/files/keda-it-aws2-sqs-to-log-auto.yaml
@@ -0,0 +1,49 @@
+# ---------------------------------------------------------------------------
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+# ---------------------------------------------------------------------------
+
+apiVersion: camel.apache.org/v1
+kind: Integration
+metadata:
+ name: keda-it-aws2-sqs-to-log-auto
+spec:
+ flows:
+ - route:
+ from:
+ steps:
+ - to: log:info
+ uri:
aws2-sqs:my-queue?region=us-east-1&accessKey=test&secretKey=test&overrideEndpoint=true&uriEndpointOverride=http://elasticmq.%%%.svc:9324&autoCreateQueue=true
+ traits:
+ keda:
+ enabled: true
+ minReplicaCount: 0
+ maxReplicaCount: 1
+ cooldownPeriod: 20
+ pollingInterval: 10
+ # Manual trigger with secrets so KEDA can poll ElasticMQ. KEDA sends
queueURL as the QueueUrl
+ # of GetQueueAttributes, and ElasticMQ rejects a plain queue name there,
so use the full URL.
+ triggers:
+ - type: aws-sqs-queue
+ metadata:
+ queueURL: http://elasticmq.%%%.svc:9324/000000000000/my-queue
+ awsRegion: us-east-1
+ awsEndpoint: http://elasticmq.%%%.svc:9324
+ queueLength: "1"
+ secrets:
+ - name: aws-sqs-credentials-secret
+ mapping:
+ AWS_ACCESS_KEY_ID: awsAccessKeyID
+ AWS_SECRET_ACCESS_KEY: awsSecretAccessKey
diff --git a/e2e/keda/files/timer-to-aws2-sqs.yaml
b/e2e/keda/files/timer-to-aws2-sqs.yaml
new file mode 100644
index 000000000..785b6d8b2
--- /dev/null
+++ b/e2e/keda/files/timer-to-aws2-sqs.yaml
@@ -0,0 +1,30 @@
+# ---------------------------------------------------------------------------
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+# ---------------------------------------------------------------------------
+
+apiVersion: camel.apache.org/v1
+kind: Integration
+metadata:
+ name: timer-to-aws2-sqs
+spec:
+ flows:
+ - route:
+ from:
+ uri: timer:foo?period=1000
+ steps:
+ - setBody:
+ constant: "Body is null"
+ - to:
aws2-sqs:my-queue?region=us-east-1&accessKey=test&secretKey=test&overrideEndpoint=true&uriEndpointOverride=http://elasticmq.%%%.svc:9324&autoCreateQueue=true
diff --git a/e2e/keda/keda_aws2_sqs_test.go b/e2e/keda/keda_aws2_sqs_test.go
new file mode 100644
index 000000000..1d1d0bab4
--- /dev/null
+++ b/e2e/keda/keda_aws2_sqs_test.go
@@ -0,0 +1,106 @@
+//go:build integration
+// +build integration
+
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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 keda
+
+import (
+ "context"
+ "testing"
+
+ . "github.com/onsi/gomega"
+ . "github.com/onsi/gomega/gstruct"
+ corev1 "k8s.io/api/core/v1"
+
+ . "github.com/apache/camel-k/v2/e2e/support"
+ v1 "github.com/apache/camel-k/v2/pkg/apis/camel/v1"
+)
+
+func TestAws2SqsKedaAutoscale(t *testing.T) {
+ WithNewTestNamespace(t, func(ctx context.Context, g *WithT, ns string) {
+ ExpectExecSucceed(t, g, Kubectl("apply", "-f",
"files/elasticmq-ephemeral.yaml", "-n", ns))
+ ExpectExecSucceed(t, g, Kubectl("rollout", "status",
"deployment", "elasticmq", "-n", ns, "--timeout=180s"))
+
+ t.Run("Serverless AWS SQS Autoscale", func(t *testing.T) {
+ consumerName := "keda-it-aws2-sqs-to-log-auto"
+ producerName := "timer-to-aws2-sqs"
+
+ ExpectExecSucceed(t, g, Kubectl("apply", "-f",
cloneAndReplaceNamespace(t, "files/keda-it-aws2-sqs-to-log-auto.yaml", ns),
"-n", ns))
+
+ // Start consumer
+ g.Eventually(IntegrationConditionStatus(t, ctx, ns,
consumerName, v1.IntegrationConditionReady), TestTimeoutLong).
+ Should(Equal(corev1.ConditionTrue))
+ g.Eventually(IntegrationStatusReplicas(t, ctx, ns,
consumerName), TestTimeoutShort).
+ Should(PointTo(BeNumerically("==", 1)))
+
+ // Scale down to 0 when idle
+ g.Eventually(IntegrationStatusReplicas(t, ctx, ns,
consumerName), TestTimeoutMedium).
+ Should(PointTo(BeNumerically("==", 0)))
+
+ // Start producer to generate traffic
+ ExpectExecSucceed(t, g, Kubectl("apply", "-f",
cloneAndReplaceNamespace(t, "files/timer-to-aws2-sqs.yaml", ns), "-n", ns))
+ g.Eventually(IntegrationConditionStatus(t, ctx, ns,
producerName, v1.IntegrationConditionReady), TestTimeoutLong).
+ Should(Equal(corev1.ConditionTrue))
+
+ // Consumer scales up to 1
+ g.Eventually(IntegrationStatusReplicas(t, ctx, ns,
consumerName), TestTimeoutShort).
+ Should(PointTo(BeNumerically("==", 1)))
+
+ // Verify messages are being received and processed
+ g.Eventually(IntegrationLogs(t, ctx, ns, consumerName),
TestTimeoutShort).
+ Should(ContainSubstring("Body is null"))
+
+ // Stop producer
+ ExpectExecSucceed(t, g, Kubectl("delete", "it",
producerName, "-n", ns))
+
+ // Scale back to 0
+ g.Eventually(IntegrationStatusReplicas(t, ctx, ns,
consumerName), TestTimeoutMedium).
+ Should(PointTo(BeNumerically("==", 0)))
+
+ ExpectExecSucceed(t, g, Kubectl("delete", "it",
"--all", "-n", ns))
+ })
+
+ ExpectExecSucceed(t, g, Kubectl("delete", "-f",
"files/elasticmq-ephemeral.yaml", "-n", ns))
+ })
+}
+
+func TestAws2SqsKedaAutoDiscovery(t *testing.T) {
+ WithNewTestNamespace(t, func(ctx context.Context, g *WithT, ns string) {
+ t.Run("Auto-discovery AWS SQS", func(t *testing.T) {
+ integrationName := "keda-aws2-sqs-auto-discovery"
+
+ ExpectExecSucceed(t, g, Kubectl("apply", "-f",
cloneAndReplaceNamespace(t, "files/keda-it-aws2-sqs-auto-discovery.yaml", ns),
"-n", ns))
+
+ // Wait for ScaledObject
+ g.Eventually(ScaledObject(t, ctx, ns, integrationName),
TestTimeoutMedium).
+ ShouldNot(BeNil())
+
+ // Verify auto-discovered trigger from aws2-sqs URI
mapping
+ scaledObj := ScaledObject(t, ctx, ns, integrationName)()
+ g.Expect(scaledObj).NotTo(BeNil())
+ g.Expect(scaledObj.Spec.Triggers).To(HaveLen(1))
+
g.Expect(scaledObj.Spec.Triggers[0].Type).To(Equal("aws-sqs-queue"))
+
g.Expect(scaledObj.Spec.Triggers[0].Metadata["queueURL"]).To(Equal("my-queue"))
+
g.Expect(scaledObj.Spec.Triggers[0].Metadata["awsRegion"]).To(Equal("us-east-1"))
+
g.Expect(scaledObj.Spec.Triggers[0].Metadata["awsEndpoint"]).To(Equal("http://elasticmq.";
+ ns + ".svc:9324"))
+
+ ExpectExecSucceed(t, g, Kubectl("delete", "it",
integrationName, "-n", ns))
+ })
+ })
+}
diff --git a/pkg/trait/keda/scalers/aws2_sqs.go
b/pkg/trait/keda/scalers/aws2_sqs.go
new file mode 100644
index 000000000..8e3233688
--- /dev/null
+++ b/pkg/trait/keda/scalers/aws2_sqs.go
@@ -0,0 +1,78 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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 scalers
+
+import (
+ "fmt"
+ "strings"
+
+ keda "github.com/apache/camel-k/v2/pkg/trait/keda"
+)
+
+// AWS2SQSScaler maps the Camel aws2-sqs consumer to the KEDA aws-sqs-queue
scaler.
+type AWS2SQSScaler struct{}
+
+func init() {
+ keda.Register(&AWS2SQSScaler{})
+}
+
+func (s *AWS2SQSScaler) Component() string {
+ return "aws2-sqs"
+}
+
+// Map derives the KEDA queueURL and awsRegion from the endpoint URI.
+// An explicit queueUrl parameter wins. Otherwise the URL is built from an
+// SQS queue ARN, or from a queue name together with the region and
+// queueOwnerAWSAccountId parameters. When no full URL can be built, the
+// plain queue name is used, which KEDA accepts when it is unambiguous.
+func (s *AWS2SQSScaler) Map(pathValue string, params map[string]string, props
map[string]string) (string, map[string]string) {
+ metadata := make(map[string]string)
+
+ region := params["region"]
+ pathValue = strings.TrimPrefix(pathValue, "//")
+ queueName := pathValue
+ account := params["queueOwnerAWSAccountId"]
+
+ // arn:aws:sqs:<region>:<account>:<queue>
+ if parts := strings.Split(pathValue, ":"); len(parts) == 6 && parts[0]
== "arn" && parts[2] == "sqs" {
+ if region == "" {
+ region = parts[3]
+ }
+ account = parts[4]
+ queueName = parts[5]
+ }
+
+ if region != "" {
+ metadata["awsRegion"] = region
+ }
+
+ switch {
+ case params["queueUrl"] != "":
+ metadata["queueURL"] = params["queueUrl"]
+ case queueName != "" && account != "" && region != "":
+ metadata["queueURL"] =
fmt.Sprintf("https://sqs.%s.amazonaws.com/%s/%s";, region, account, queueName)
+ case queueName != "":
+ metadata["queueURL"] = queueName
+ }
+
+ if params["overrideEndpoint"] == "true" &&
params["uriEndpointOverride"] != "" {
+ metadata["awsEndpoint"] = params["uriEndpointOverride"]
+ }
+
+ return "aws-sqs-queue", metadata
+}
diff --git a/pkg/trait/keda/scalers/aws2_sqs_test.go
b/pkg/trait/keda/scalers/aws2_sqs_test.go
new file mode 100644
index 000000000..e31802dab
--- /dev/null
+++ b/pkg/trait/keda/scalers/aws2_sqs_test.go
@@ -0,0 +1,182 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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 scalers
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ keda "github.com/apache/camel-k/v2/pkg/trait/keda"
+)
+
+func TestAWS2SQSScalerRegistered(t *testing.T) {
+ mapper, found := keda.GetMapper("aws2-sqs")
+ require.True(t, found, "aws2-sqs scaler should be registered")
+ assert.Equal(t, "aws2-sqs", mapper.Component())
+}
+
+func TestAWS2SQSScalerMap(t *testing.T) {
+ scaler := &AWS2SQSScaler{}
+
+ tests := []struct {
+ name string
+ pathValue string
+ params map[string]string
+ expectedMeta map[string]string
+ }{
+ {
+ name: "explicit queue URL",
+ pathValue: "orders",
+ params: map[string]string{
+ "region": "eu-west-1",
+ "queueUrl":
"https://sqs.eu-west-1.amazonaws.com/123456789012/orders";,
+ },
+ expectedMeta: map[string]string{
+ "awsRegion": "eu-west-1",
+ "queueURL":
"https://sqs.eu-west-1.amazonaws.com/123456789012/orders";,
+ },
+ },
+ {
+ name: "queue name with owner account and region",
+ pathValue: "orders",
+ params: map[string]string{"region": "us-east-1",
"queueOwnerAWSAccountId": "123456789012"},
+ expectedMeta: map[string]string{
+ "awsRegion": "us-east-1",
+ "queueURL":
"https://sqs.us-east-1.amazonaws.com/123456789012/orders";,
+ },
+ },
+ {
+ name: "queue ARN",
+ pathValue:
"arn:aws:sqs:ap-south-1:123456789012:orders.fifo",
+ params: map[string]string{},
+ expectedMeta: map[string]string{
+ "awsRegion": "ap-south-1",
+ "queueURL":
"https://sqs.ap-south-1.amazonaws.com/123456789012/orders.fifo";,
+ },
+ },
+ {
+ name: "double-slash URI form",
+ pathValue: "//orders",
+ params: map[string]string{"region": "us-east-1",
"queueOwnerAWSAccountId": "123456789012"},
+ expectedMeta: map[string]string{
+ "awsRegion": "us-east-1",
+ "queueURL":
"https://sqs.us-east-1.amazonaws.com/123456789012/orders";,
+ },
+ },
+ {
+ name: "queue name and region only",
+ pathValue: "orders",
+ params: map[string]string{"region": "us-east-1"},
+ expectedMeta: map[string]string{"awsRegion":
"us-east-1", "queueURL": "orders"},
+ },
+ {
+ name: "endpoint override",
+ pathValue: "orders",
+ params: map[string]string{
+ "region": "us-east-1",
+ "queueOwnerAWSAccountId": "000000000000",
+ "overrideEndpoint": "true",
+ "uriEndpointOverride":
"http://localstack:4566";,
+ },
+ expectedMeta: map[string]string{
+ "awsRegion": "us-east-1",
+ "queueURL":
"https://sqs.us-east-1.amazonaws.com/000000000000/orders";,
+ "awsEndpoint": "http://localstack:4566";,
+ },
+ },
+ {
+ name: "endpoint override not enabled",
+ pathValue: "orders",
+ params: map[string]string{
+ "region": "us-east-1",
+ "uriEndpointOverride": "http://localstack:4566";,
+ },
+ expectedMeta: map[string]string{"awsRegion":
"us-east-1", "queueURL": "orders"},
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ kedaType, metadata := scaler.Map(tt.pathValue,
tt.params, nil)
+ assert.Equal(t, "aws-sqs-queue", kedaType)
+ assert.Equal(t, tt.expectedMeta, metadata)
+ })
+ }
+}
+
+func TestMapToKedaTriggerAWS2SQS(t *testing.T) {
+ tests := []struct {
+ name string
+ uri string
+ expectedMeta map[string]string
+ }{
+ {
+ name: "queue name with static credentials",
+ uri:
"aws2-sqs:my-queue?region=us-east-1&accessKey=test&secretKey=test",
+ expectedMeta: map[string]string{
+ "awsRegion": "us-east-1",
+ "queueURL": "my-queue",
+ },
+ },
+ {
+ name: "endpoint override",
+ uri:
"aws2-sqs:my-queue?region=us-east-1&accessKey=test&secretKey=test&overrideEndpoint=true&uriEndpointOverride=http://elasticmq:9324";,
+ expectedMeta: map[string]string{
+ "awsRegion": "us-east-1",
+ "queueURL": "my-queue",
+ "awsEndpoint": "http://elasticmq:9324";,
+ },
+ },
+ {
+ name: "queue owner account",
+ uri:
"aws2-sqs:my-queue?region=eu-west-1&queueOwnerAWSAccountId=123456789012",
+ expectedMeta: map[string]string{
+ "awsRegion": "eu-west-1",
+ "queueURL":
"https://sqs.eu-west-1.amazonaws.com/123456789012/my-queue";,
+ },
+ },
+ {
+ name: "queue ARN",
+ uri:
"aws2-sqs:arn:aws:sqs:us-east-1:123456789012:my-queue?accessKey=test&secretKey=test",
+ expectedMeta: map[string]string{
+ "awsRegion": "us-east-1",
+ "queueURL":
"https://sqs.us-east-1.amazonaws.com/123456789012/my-queue";,
+ },
+ },
+ {
+ name: "double-slash form",
+ uri: "aws2-sqs://my-queue?region=us-east-1",
+ expectedMeta: map[string]string{
+ "awsRegion": "us-east-1",
+ "queueURL": "my-queue",
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ trigger, err := keda.MapToKedaTrigger(tt.uri, nil)
+ require.NoError(t, err)
+ require.NotNil(t, trigger)
+ assert.Equal(t, "aws-sqs-queue", trigger.Type)
+ assert.Equal(t, tt.expectedMeta, trigger.Metadata)
+ })
+ }
+}
