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 99e886ff4 feat(trait/keda): add ActiveMQ scaler auto-discovery (#6904)
99e886ff4 is described below
commit 99e886ff40fa4a61c3f5a0af23453b056d8e700c
Author: Anshul Jagota <[email protected]>
AuthorDate: Fri Nov 13 19:53:13 2026 +0530
feat(trait/keda): add ActiveMQ scaler auto-discovery (#6904)
* feat(trait/keda): add ActiveMQ scaler auto-discovery
* test(e2e/keda): add ActiveMQ scaler e2e test and update mapper interface
* test(e2e/keda): use random namespace with cloneAndReplaceNamespace for
ActiveMQ tests
* fix(trait/keda): use strings.CutPrefix to satisfy modernize linter
* fix(e2e/keda): remove duplicate cloneAndReplaceNamespace from
keda_activemq_test
* test(e2e/keda): eliminate scale-to-zero cold start race and add ActiveMQ
readiness probe
* test(e2e/keda): pin ActiveMQ image to 5.19.2 for Camel Quarkus
compatibility
* test(e2e/keda): disable pooled connections for ActiveMQ integrations
---
docs/modules/traits/pages/keda.adoc | 3 +-
e2e/keda/files/activemq-ephemeral.yaml | 83 +++++++++++++
e2e/keda/files/keda-it-activemq-to-log-auto.yaml | 53 ++++++++
e2e/keda/files/timer-to-activemq.yaml | 38 ++++++
e2e/keda/keda_activemq_test.go | 95 ++++++++++++++
pkg/trait/keda/scalers/activemq.go | 79 ++++++++++++
pkg/trait/keda/scalers/activemq_test.go | 150 +++++++++++++++++++++++
7 files changed, 500 insertions(+), 1 deletion(-)
diff --git a/docs/modules/traits/pages/keda.adoc
b/docs/modules/traits/pages/keda.adoc
index cc83648ca..da08c06c3 100644
--- a/docs/modules/traits/pages/keda.adoc
+++ b/docs/modules/traits/pages/keda.adoc
@@ -75,8 +75,9 @@ When `keda.auto` is enabled, Camel K inspects the consumer
endpoints in an Integ
Automatic discovery currently supports the following Camel components:
+* `activemq`: creates an Apache ActiveMQ trigger from destination and broker
values in the endpoint URI. See the
https://keda.sh/docs/latest/scalers/activemq/[KEDA Apache ActiveMQ scaler]
documentation for the available trigger metadata.
* `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`.
+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/activemq-ephemeral.yaml
b/e2e/keda/files/activemq-ephemeral.yaml
new file mode 100644
index 000000000..0f8e08b30
--- /dev/null
+++ b/e2e/keda/files/activemq-ephemeral.yaml
@@ -0,0 +1,83 @@
+# ---------------------------------------------------------------------------
+# 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: my-activemq-credentials-secret
+type: Opaque
+stringData:
+ ACTIVEMQ_USER: admin
+ ACTIVEMQ_PASSWORD: admin
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+ name: my-activemq
+spec:
+ replicas: 1
+ selector:
+ matchLabels:
+ app: my-activemq
+ template:
+ metadata:
+ labels:
+ app: my-activemq
+ spec:
+ containers:
+ - name: activemq
+ image: apache/activemq-classic:5.19.2
+ env:
+ - name: ACTIVEMQ_OPTS
+ value: "-Djetty.host=0.0.0.0"
+ - name: ACTIVEMQ_USER
+ value: "admin"
+ - name: ACTIVEMQ_PASSWORD
+ value: "admin"
+ ports:
+ - containerPort: 61616
+ name: openwire
+ - containerPort: 8161
+ name: console
+ resources:
+ requests:
+ memory: "256Mi"
+ cpu: "200m"
+ limits:
+ memory: "512Mi"
+ readinessProbe:
+ tcpSocket:
+ port: 61616
+ initialDelaySeconds: 10
+ periodSeconds: 5
+ timeoutSeconds: 2
+---
+apiVersion: v1
+kind: Service
+metadata:
+ name: my-activemq-hdls-svc
+spec:
+ selector:
+ app: my-activemq
+ ports:
+ - name: openwire
+ port: 61616
+ targetPort: 61616
+ - name: console
+ port: 8161
+ targetPort: 8161
diff --git a/e2e/keda/files/keda-it-activemq-to-log-auto.yaml
b/e2e/keda/files/keda-it-activemq-to-log-auto.yaml
new file mode 100644
index 000000000..0c1355d8a
--- /dev/null
+++ b/e2e/keda/files/keda-it-activemq-to-log-auto.yaml
@@ -0,0 +1,53 @@
+# ---------------------------------------------------------------------------
+# 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-activemq-to-log-auto
+spec:
+ flows:
+ - route:
+ from:
+ steps:
+ - to: log:info
+ uri: activemq:queue:my-queue
+ traits:
+ camel:
+ properties:
+ -
camel.component.activemq.broker-url=tcp://my-activemq-hdls-svc.%%%.svc:61616
+ - camel.component.activemq.username=admin
+ - camel.component.activemq.password=admin
+ - camel.component.activemq.use-pooled-connection=false
+ keda:
+ enabled: true
+ minReplicaCount: 0
+ maxReplicaCount: 1
+ cooldownPeriod: 20
+ pollingInterval: 10
+ triggers:
+ - type: activemq
+ metadata:
+ managementEndpoint: my-activemq-hdls-svc.%%%.svc:8161
+ destinationName: my-queue
+ brokerName: localhost
+ secrets:
+ - name: my-activemq-credentials-secret
+ mapping:
+ ACTIVEMQ_USER: username
+ ACTIVEMQ_PASSWORD: password
diff --git a/e2e/keda/files/timer-to-activemq.yaml
b/e2e/keda/files/timer-to-activemq.yaml
new file mode 100644
index 000000000..4e1256149
--- /dev/null
+++ b/e2e/keda/files/timer-to-activemq.yaml
@@ -0,0 +1,38 @@
+# ---------------------------------------------------------------------------
+# 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-activemq
+spec:
+ flows:
+ - route:
+ from:
+ uri: timer:foo?period=1000
+ steps:
+ - setBody:
+ constant: "Body is null"
+ - to: activemq:queue:my-queue
+ traits:
+ camel:
+ properties:
+ -
camel.component.activemq.broker-url=tcp://my-activemq-hdls-svc.%%%.svc:61616
+ - camel.component.activemq.username=admin
+ - camel.component.activemq.password=admin
+ - camel.component.activemq.use-pooled-connection=false
diff --git a/e2e/keda/keda_activemq_test.go b/e2e/keda/keda_activemq_test.go
new file mode 100644
index 000000000..202705217
--- /dev/null
+++ b/e2e/keda/keda_activemq_test.go
@@ -0,0 +1,95 @@
+//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"
+ "fmt"
+ "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 TestActiveMQKedaAutoscale(t *testing.T) {
+ WithNewTestNamespace(t, func(ctx context.Context, g *WithT, ns string) {
+ ExpectExecSucceed(t, g, Kubectl("apply", "-f",
"files/activemq-ephemeral.yaml", "-n", ns))
+ ExpectExecSucceed(t, g, Kubectl("rollout", "status",
"deployment", "my-activemq", "-n", ns, "--timeout=180s"))
+
+ t.Run("Serverless ActiveMQ Autoscale", func(t *testing.T) {
+ consumerName := "keda-it-activemq-to-log-auto"
+ producerName := "timer-to-activemq"
+
+ // Start producer to generate traffic and populate the
queue
+ producerFile := cloneAndReplaceNamespace(t,
"files/timer-to-activemq.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))
+
+ // Start consumer; queue is non-empty so KEDA keeps
replicas active
+ consumerFile := cloneAndReplaceNamespace(t,
"files/keda-it-activemq-to-log-auto.yaml", ns)
+ ExpectExecSucceed(t, g, Kubectl("apply", "-f",
consumerFile, "-n", ns))
+ g.Eventually(IntegrationConditionStatus(t, ctx, ns,
consumerName, v1.IntegrationConditionReady), TestTimeoutLong).
+ Should(Equal(corev1.ConditionTrue))
+ 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("Body is null"))
+
+ // Stop producer to allow consumer to drain the queue
+ ExpectExecSucceed(t, g, Kubectl("delete", "it",
producerName, "-n", ns))
+
+ // Once idle, KEDA scales the consumer down to 0
+ g.Eventually(IntegrationStatusReplicas(t, ctx, ns,
consumerName), TestTimeoutMedium).
+ Should(PointTo(BeNumerically("==", 0)))
+ })
+ })
+}
+
+func TestActiveMQKedaAutoDiscovery(t *testing.T) {
+ WithNewTestNamespace(t, func(ctx context.Context, g *WithT, ns string) {
+ t.Run("Auto-discovery ActiveMQ", func(t *testing.T) {
+ integrationName := "keda-it-activemq-to-log-auto"
+
+ consumerFile := cloneAndReplaceNamespace(t,
"files/keda-it-activemq-to-log-auto.yaml", ns)
+ ExpectExecSucceed(t, g, Kubectl("apply", "-f",
consumerFile, "-n", ns))
+
+ // Wait for ScaledObject
+ g.Eventually(ScaledObject(t, ctx, ns, integrationName),
TestTimeoutMedium).
+ ShouldNot(BeNil())
+
+ // Verify auto-discovered trigger
+ 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("activemq"))
+
g.Expect(scaledObj.Spec.Triggers[0].Metadata["destinationName"]).To(Equal("my-queue"))
+
g.Expect(scaledObj.Spec.Triggers[0].Metadata["managementEndpoint"]).To(Equal(fmt.Sprintf("my-activemq-hdls-svc.%s.svc:8161",
ns)))
+
g.Expect(scaledObj.Spec.Triggers[0].Metadata["brokerName"]).To(Equal("localhost"))
+ })
+ })
+}
diff --git a/pkg/trait/keda/scalers/activemq.go
b/pkg/trait/keda/scalers/activemq.go
new file mode 100644
index 000000000..9aa95af82
--- /dev/null
+++ b/pkg/trait/keda/scalers/activemq.go
@@ -0,0 +1,79 @@
+/*
+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 (
+ "net"
+ "net/url"
+ "strings"
+
+ keda "github.com/apache/camel-k/v2/pkg/trait/keda"
+)
+
+type ActiveMQScaler struct{}
+
+func init() {
+ keda.Register(&ActiveMQScaler{})
+}
+
+func (a *ActiveMQScaler) Component() string {
+ return "activemq"
+}
+
+func (a *ActiveMQScaler) Map(pathValue string, params map[string]string, props
map[string]string) (string, map[string]string) {
+ metadata := make(map[string]string)
+
+ destination := pathValue
+ if d, ok := strings.CutPrefix(destination, "queue:"); ok {
+ destination = d
+ } else if d, ok := strings.CutPrefix(destination, "topic:"); ok {
+ destination = d
+ }
+
+ if destination != "" {
+ metadata["destinationName"] = destination
+ }
+ if brokerURL, ok := props["camel.component.activemq.broker-url"]; ok {
+ if u, err := url.Parse(brokerURL); err == nil && u.Hostname()
!= "" {
+ metadata["managementEndpoint"] =
net.JoinHostPort(u.Hostname(), "8161")
+ }
+ }
+ if v, ok := params["destinationName"]; ok {
+ metadata["destinationName"] = v
+ }
+ if v, ok := params["managementEndpoint"]; ok {
+ metadata["managementEndpoint"] = v
+ }
+ if v, ok := params["brokerName"]; ok {
+ metadata["brokerName"] = v
+ }
+ if v, ok := params["targetQueueSize"]; ok {
+ metadata["targetQueueSize"] = v
+ }
+ if v, ok := params["activationTargetQueueSize"]; ok {
+ metadata["activationTargetQueueSize"] = v
+ }
+ if v, ok := params["corsHeader"]; ok {
+ metadata["corsHeader"] = v
+ }
+ if v, ok := params["restAPITemplate"]; ok {
+ metadata["restAPITemplate"] = v
+ }
+
+ return "activemq", metadata
+}
diff --git a/pkg/trait/keda/scalers/activemq_test.go
b/pkg/trait/keda/scalers/activemq_test.go
new file mode 100644
index 000000000..7a5d96b12
--- /dev/null
+++ b/pkg/trait/keda/scalers/activemq_test.go
@@ -0,0 +1,150 @@
+/*
+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 TestActiveMQScalerRegistered(t *testing.T) {
+ mapper, found := keda.GetMapper("activemq")
+ require.True(t, found, "activemq scaler should be registered")
+ assert.Equal(t, "activemq", mapper.Component())
+}
+
+func TestActiveMQScalerMap(t *testing.T) {
+ scaler := &ActiveMQScaler{}
+
+ tests := []struct {
+ name string
+ pathValue string
+ params map[string]string
+ props map[string]string
+ expectedType string
+ expectedMeta map[string]string
+ }{
+ {
+ name: "full activemq params with queue prefix",
+ pathValue: "queue:orders",
+ params: map[string]string{
+ "managementEndpoint": "activemq:8161",
+ "brokerName": "my-broker",
+ "targetQueueSize": "50",
+ "activationTargetQueueSize": "5",
+ "corsHeader": "*",
+ "restAPITemplate":
"http://{endpoint}/api/jolokia";,
+ },
+ props: map[string]string{},
+ expectedType: "activemq",
+ expectedMeta: map[string]string{
+ "destinationName": "orders",
+ "managementEndpoint": "activemq:8161",
+ "brokerName": "my-broker",
+ "targetQueueSize": "50",
+ "activationTargetQueueSize": "5",
+ "corsHeader": "*",
+ "restAPITemplate":
"http://{endpoint}/api/jolokia";,
+ },
+ },
+ {
+ name: "only destination name without prefix",
+ pathValue: "my-queue",
+ params: map[string]string{},
+ props: map[string]string{},
+ expectedType: "activemq",
+ expectedMeta: map[string]string{
+ "destinationName": "my-queue",
+ },
+ },
+ {
+ name: "destination with topic prefix",
+ pathValue: "topic:notifications",
+ params: map[string]string{"managementEndpoint":
"activemq.default:8161"},
+ props: map[string]string{},
+ expectedType: "activemq",
+ expectedMeta: map[string]string{
+ "destinationName": "notifications",
+ "managementEndpoint": "activemq.default:8161",
+ },
+ },
+ {
+ name: "destinationName overridden by param",
+ pathValue: "queue:dummy",
+ params: map[string]string{
+ "destinationName": "actual-queue",
+ },
+ props: map[string]string{},
+ expectedType: "activemq",
+ expectedMeta: map[string]string{
+ "destinationName": "actual-queue",
+ },
+ },
+ {
+ name: "broker-url sets management endpoint",
+ pathValue: "queue:orders",
+ params: map[string]string{},
+ props: map[string]string{
+ "camel.component.activemq.broker-url":
"tcp://activemq-broker.default.svc:61616",
+ },
+ expectedType: "activemq",
+ expectedMeta: map[string]string{
+ "destinationName": "orders",
+ "managementEndpoint":
"activemq-broker.default.svc:8161",
+ },
+ },
+ }
+
+ 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 TestMapToKedaTriggerActiveMQ(t *testing.T) {
+ trigger, err :=
keda.MapToKedaTrigger("activemq:queue:orders?managementEndpoint=activemq:8161&brokerName=my-broker",
nil)
+ require.NoError(t, err)
+ require.NotNil(t, trigger)
+ assert.Equal(t, "activemq", trigger.Type)
+ assert.Equal(t, map[string]string{
+ "destinationName": "orders",
+ "managementEndpoint": "activemq:8161",
+ "brokerName": "my-broker",
+ }, trigger.Metadata)
+}
+
+func TestMapToKedaTriggerActiveMQWithBrokerURL(t *testing.T) {
+ props := map[string]string{
+ "camel.component.activemq.broker-url":
"tcp://my-activemq-hdls-svc.activemq.svc:61616",
+ }
+ trigger, err := keda.MapToKedaTrigger("activemq:queue:my-queue", props)
+ require.NoError(t, err)
+ require.NotNil(t, trigger)
+ assert.Equal(t, "activemq", trigger.Type)
+ assert.Equal(t, map[string]string{
+ "destinationName": "my-queue",
+ "managementEndpoint": "my-activemq-hdls-svc.activemq.svc:8161",
+ }, trigger.Metadata)
+}
