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 8160ae4fd Fix #6912: Add Spring RabbitMQ KEDA scaler mapping (#6913)
8160ae4fd is described below
commit 8160ae4fda823ebafa91b034e9d14cc4ab2f15eb
Author: Bharat Kaurav <[email protected]>
AuthorDate: Fri Nov 13 19:55:56 2026 +0530
Fix #6912: Add Spring RabbitMQ KEDA scaler mapping (#6913)
* Fix #6912: Add Spring RabbitMQ KEDA scaler mapping
* Fix #6912: Add e2e test for Spring RabbitMQ KEDA scaler
Adapt the scaler to the Map signature that takes integration
properties, and add a keda workflow test that runs an ephemeral
RabbitMQ broker, checks the auto-discovered trigger, and verifies
scale 0 -> 1 -> 0 under producer traffic.
* Fix #6912: Run Spring RabbitMQ KEDA e2e test in a random namespace
* Fix #6912: Discover the RabbitMQ host from the connection factory bean
Read the camel.beans properties of the connection factory referenced by
the spring-rabbitmq connectionFactory option and set the KEDA host from
its uri, addresses or host/port/virtualHost. Credentials are not copied
into the trigger.
Also drop the e2e cloneAndReplaceNamespace copy now that main has one,
and the replicas==1 check right after Ready, which raced with KEDA
scaling the idle consumer to 0.
* Fix #6912: Prefer a host endpoint parameter over the discovered broker
address
* Fix #6912: Drop host discovery from the connection factory bean
Bean-based discovery is fragile and overlaps with #6942. Keep queueName
from queues and an explicit host endpoint parameter; otherwise host,
mode and value come from keda.autoMetadata.
---
docs/modules/traits/pages/keda.adoc | 1 +
.../files/keda-it-spring-rabbitmq-to-log-auto.yaml | 55 ++++++++++
e2e/keda/files/rabbitmq-ephemeral.yaml | 67 ++++++++++++
e2e/keda/files/timer-to-spring-rabbitmq.yaml | 43 ++++++++
e2e/keda/keda_spring_rabbitmq_test.go | 87 +++++++++++++++
pkg/trait/keda/scalers/spring_rabbitmq.go | 51 +++++++++
pkg/trait/keda/scalers/spring_rabbitmq_test.go | 118 +++++++++++++++++++++
7 files changed, 422 insertions(+)
diff --git a/docs/modules/traits/pages/keda.adoc
b/docs/modules/traits/pages/keda.adoc
index da08c06c3..ba1e2628f 100644
--- a/docs/modules/traits/pages/keda.adoc
+++ b/docs/modules/traits/pages/keda.adoc
@@ -79,5 +79,6 @@ 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 [...]
+* `spring-rabbitmq`: creates a RabbitMQ trigger whose `queueName` is the
`queues` value of the endpoint URI, when it names a single queue. The broker
address is not part of the URI: a `host` endpoint parameter, when present, is
used as the trigger `host` as is; otherwise set `host`, together with `mode`
and `value`, with `keda.autoMetadata`. See the
https://keda.sh/docs/latest/scalers/rabbitmq-queue/[KEDA RabbitMQ Queue scaler]
documentation for the available trigger metadata.
Unsupported Camel components do not produce an automatically discovered
trigger. You can still configure their KEDA triggers explicitly with
`keda.triggers`.
\ No newline at end of file
diff --git a/e2e/keda/files/keda-it-spring-rabbitmq-to-log-auto.yaml
b/e2e/keda/files/keda-it-spring-rabbitmq-to-log-auto.yaml
new file mode 100644
index 000000000..0cbf83df2
--- /dev/null
+++ b/e2e/keda/files/keda-it-spring-rabbitmq-to-log-auto.yaml
@@ -0,0 +1,55 @@
+# ---------------------------------------------------------------------------
+# 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.
+# ---------------------------------------------------------------------------
+
+
+# Spring RabbitMQ trigger is auto-discovered from the consumer URI (queueName
comes from `queues`).
+# The broker address is not part of the URI, so host, mode and value are
merged via autoMetadata.
+
+apiVersion: camel.apache.org/v1
+kind: Integration
+metadata:
+ name: keda-it-spring-rabbitmq-to-log-auto
+spec:
+ flows:
+ - beans:
+ - name: rabbitConnectionFactory
+ type: org.springframework.amqp.rabbit.connection.CachingConnectionFactory
+ properties:
+ host: my-rabbitmq.%%%.svc
+ port: 5672
+ username: admin
+ password: admin
+ - route:
+ from:
+ uri: spring-rabbitmq:my-exchange?queues=my-queue&routingKey=my-key
+ steps:
+ - to: log:info
+ traits:
+ camel:
+ properties:
+ -
camel.component.spring-rabbitmq.connection-factory=#rabbitConnectionFactory
+ keda:
+ enabled: true
+ minReplicaCount: 0
+ maxReplicaCount: 1
+ cooldownPeriod: 20
+ pollingInterval: 10
+ autoMetadata:
+ rabbitmq:
+ host: amqp://admin:admin@my-rabbitmq.%%%.svc:5672/
+ mode: QueueLength
+ value: "5"
diff --git a/e2e/keda/files/rabbitmq-ephemeral.yaml
b/e2e/keda/files/rabbitmq-ephemeral.yaml
new file mode 100644
index 000000000..a4cd3b8b8
--- /dev/null
+++ b/e2e/keda/files/rabbitmq-ephemeral.yaml
@@ -0,0 +1,67 @@
+# ---------------------------------------------------------------------------
+# 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: apps/v1
+kind: Deployment
+metadata:
+ name: my-rabbitmq
+spec:
+ replicas: 1
+ selector:
+ matchLabels:
+ app: my-rabbitmq
+ template:
+ metadata:
+ labels:
+ app: my-rabbitmq
+ spec:
+ containers:
+ - name: rabbitmq
+ image: rabbitmq:4-alpine
+ env:
+ # The default guest user can only connect from localhost
+ - name: RABBITMQ_DEFAULT_USER
+ value: "admin"
+ - name: RABBITMQ_DEFAULT_PASS
+ value: "admin"
+ ports:
+ - containerPort: 5672
+ name: amqp
+ readinessProbe:
+ exec:
+ command: ["rabbitmq-diagnostics", "-q",
"check_port_connectivity"]
+ initialDelaySeconds: 10
+ periodSeconds: 5
+ resources:
+ requests:
+ memory: "256Mi"
+ cpu: "200m"
+ limits:
+ memory: "512Mi"
+---
+apiVersion: v1
+kind: Service
+metadata:
+ name: my-rabbitmq
+spec:
+ selector:
+ app: my-rabbitmq
+ ports:
+ - name: amqp
+ port: 5672
+ targetPort: 5672
diff --git a/e2e/keda/files/timer-to-spring-rabbitmq.yaml
b/e2e/keda/files/timer-to-spring-rabbitmq.yaml
new file mode 100644
index 000000000..c47c30c78
--- /dev/null
+++ b/e2e/keda/files/timer-to-spring-rabbitmq.yaml
@@ -0,0 +1,43 @@
+# ---------------------------------------------------------------------------
+# 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-spring-rabbitmq
+spec:
+ flows:
+ - beans:
+ - name: rabbitConnectionFactory
+ type: org.springframework.amqp.rabbit.connection.CachingConnectionFactory
+ properties:
+ host: my-rabbitmq.%%%.svc
+ port: 5672
+ username: admin
+ password: admin
+ - route:
+ from:
+ uri: timer:foo?period=1000
+ steps:
+ - setBody:
+ constant: "Hello RabbitMQ"
+ - to: spring-rabbitmq:my-exchange?routingKey=my-key
+ traits:
+ camel:
+ properties:
+ -
camel.component.spring-rabbitmq.connection-factory=#rabbitConnectionFactory
diff --git a/e2e/keda/keda_spring_rabbitmq_test.go
b/e2e/keda/keda_spring_rabbitmq_test.go
new file mode 100644
index 000000000..569da5ab4
--- /dev/null
+++ b/e2e/keda/keda_spring_rabbitmq_test.go
@@ -0,0 +1,87 @@
+//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 TestSpringRabbitMQKedaAutoscale(t *testing.T) {
+ WithNewTestNamespace(t, func(ctx context.Context, g *WithT, ns string) {
+ ExpectExecSucceed(t, g, Kubectl("apply", "-f",
"files/rabbitmq-ephemeral.yaml", "-n", ns))
+ ExpectExecSucceed(t, g, Kubectl("rollout", "status",
"deployment", "my-rabbitmq", "-n", ns, "--timeout=180s"))
+
+ t.Run("Serverless Spring RabbitMQ Autoscale", func(t
*testing.T) {
+ consumerName := "keda-it-spring-rabbitmq-to-log-auto"
+ producerName := "timer-to-spring-rabbitmq"
+
+ consumerFile := cloneAndReplaceNamespace(t,
"files/keda-it-spring-rabbitmq-to-log-auto.yaml", ns)
+ ExpectExecSucceed(t, g, Kubectl("apply", "-f",
consumerFile, "-n", ns))
+
+ // Trigger is auto-discovered from the consumer URI
+ g.Eventually(ScaledObject(t, ctx, ns, consumerName),
TestTimeoutMedium).
+ ShouldNot(BeNil())
+ scaledObj := ScaledObject(t, ctx, ns, consumerName)()
+ g.Expect(scaledObj).NotTo(BeNil())
+ g.Expect(scaledObj.Spec.Triggers).To(HaveLen(1))
+
g.Expect(scaledObj.Spec.Triggers[0].Type).To(Equal("rabbitmq"))
+
g.Expect(scaledObj.Spec.Triggers[0].Metadata["queueName"]).To(Equal("my-queue"))
+
g.Expect(scaledObj.Spec.Triggers[0].Metadata["mode"]).To(Equal("QueueLength"))
+
+ // Start consumer; it declares the exchange, queue and
binding
+ g.Eventually(IntegrationConditionStatus(t, ctx, ns,
consumerName, v1.IntegrationConditionReady), TestTimeoutLong).
+ Should(Equal(corev1.ConditionTrue))
+
+ // Scale down to 0 when idle
+ g.Eventually(IntegrationStatusReplicas(t, ctx, ns,
consumerName), TestTimeoutMedium).
+ Should(PointTo(BeNumerically("==", 0)))
+
+ // Start producer to generate traffic
+ producerFile := cloneAndReplaceNamespace(t,
"files/timer-to-spring-rabbitmq.yaml", ns)
+ ExpectExecSucceed(t, g, Kubectl("apply", "-f",
producerFile, "-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), TestTimeoutMedium).
+ Should(PointTo(BeNumerically("==", 1)))
+
+ // Verify messages are being received and processed
+ g.Eventually(IntegrationLogs(t, ctx, ns, consumerName),
TestTimeoutMedium).
+ Should(ContainSubstring("Hello RabbitMQ"))
+
+ // 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)))
+ })
+ })
+}
diff --git a/pkg/trait/keda/scalers/spring_rabbitmq.go
b/pkg/trait/keda/scalers/spring_rabbitmq.go
new file mode 100644
index 000000000..902df21a7
--- /dev/null
+++ b/pkg/trait/keda/scalers/spring_rabbitmq.go
@@ -0,0 +1,51 @@
+/*
+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 (
+ "strings"
+
+ keda "github.com/apache/camel-k/v2/pkg/trait/keda"
+)
+
+// SpringRabbitMQScaler maps Camel Spring RabbitMQ consumer endpoints to the
KEDA rabbitmq scaler.
+type SpringRabbitMQScaler struct{}
+
+func init() {
+ keda.Register(&SpringRabbitMQScaler{})
+}
+
+func (s *SpringRabbitMQScaler) Component() string {
+ return "spring-rabbitmq"
+}
+
+// Map takes the queue from the `queues` parameter. The URI path is the
exchange name, which the
+// scaler does not use. The broker address is taken from a `host` parameter
when the user sets one;
+// otherwise it has to be provided with keda.autoMetadata.
+func (s *SpringRabbitMQScaler) Map(pathValue string, params map[string]string,
props map[string]string) (string, map[string]string) {
+ metadata := make(map[string]string)
+ // A KEDA rabbitmq trigger watches a single queue, so a list of queues
is left for the user to choose.
+ if v, ok := params["queues"]; ok && v != "" && !strings.Contains(v,
",") {
+ metadata["queueName"] = v
+ }
+ if v := params["host"]; v != "" {
+ metadata["host"] = v
+ }
+
+ return "rabbitmq", metadata
+}
diff --git a/pkg/trait/keda/scalers/spring_rabbitmq_test.go
b/pkg/trait/keda/scalers/spring_rabbitmq_test.go
new file mode 100644
index 000000000..90159d0ec
--- /dev/null
+++ b/pkg/trait/keda/scalers/spring_rabbitmq_test.go
@@ -0,0 +1,118 @@
+/*
+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 TestSpringRabbitMQScalerRegistered(t *testing.T) {
+ mapper, found := keda.GetMapper("spring-rabbitmq")
+ require.True(t, found, "spring-rabbitmq scaler should be registered")
+ assert.Equal(t, "spring-rabbitmq", mapper.Component())
+}
+
+func TestSpringRabbitMQScalerMap(t *testing.T) {
+ scaler := &SpringRabbitMQScaler{}
+
+ tests := []struct {
+ name string
+ pathValue string
+ params map[string]string
+ props map[string]string
+ expectedType string
+ expectedMeta map[string]string
+ }{
+ {
+ name: "single queue",
+ pathValue: "orders-exchange",
+ params: map[string]string{"queues": "orders",
"routingKey": "new", "autoDeclare": "true"},
+ expectedType: "rabbitmq",
+ expectedMeta: map[string]string{
+ "queueName": "orders",
+ },
+ },
+ {
+ name: "no queue",
+ pathValue: "orders-exchange",
+ params: map[string]string{},
+ expectedType: "rabbitmq",
+ expectedMeta: map[string]string{},
+ },
+ {
+ name: "empty queue",
+ pathValue: "default",
+ params: map[string]string{"queues": ""},
+ expectedType: "rabbitmq",
+ expectedMeta: map[string]string{},
+ },
+ {
+ name: "several queues",
+ pathValue: "orders-exchange",
+ params: map[string]string{"queues":
"orders,invoices"},
+ expectedType: "rabbitmq",
+ expectedMeta: map[string]string{},
+ },
+ {
+ name: "host parameter",
+ pathValue: "orders-exchange",
+ params: map[string]string{"queues": "orders",
"host": "amqp://user:pass@my-rabbitmq:5672/vh"},
+ expectedType: "rabbitmq",
+ expectedMeta: map[string]string{
+ "queueName": "orders",
+ "host":
"amqp://user:pass@my-rabbitmq:5672/vh",
+ },
+ },
+ {
+ name: "connection factory bean is not used for
host",
+ pathValue: "orders-exchange",
+ params: map[string]string{"queues": "orders"},
+ props: map[string]string{
+
"camel.component.spring-rabbitmq.connection-factory": "#cf",
+ "camel.beans.cf.host":
"my-rabbitmq",
+ },
+ expectedType: "rabbitmq",
+ expectedMeta: map[string]string{
+ "queueName": "orders",
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ kedaType, metadata := scaler.Map(tt.pathValue,
tt.params, tt.props)
+ assert.Equal(t, tt.expectedType, kedaType)
+ assert.Equal(t, tt.expectedMeta, metadata)
+ })
+ }
+}
+
+func TestMapToKedaTriggerSpringRabbitMQ(t *testing.T) {
+ trigger, err :=
keda.MapToKedaTrigger("spring-rabbitmq:orders-exchange?queues=orders&routingKey=new",
nil)
+ require.NoError(t, err)
+ require.NotNil(t, trigger)
+ assert.Equal(t, "rabbitmq", trigger.Type)
+ assert.Equal(t, map[string]string{
+ "queueName": "orders",
+ }, trigger.Metadata)
+}
