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
commit dd3e13f981d17f8a18aa7d48ab65245a1e425d32 Author: Dhvani Dave <[email protected]> AuthorDate: Sun Oct 18 04:37:31 2026 -0400 feat(keda): add ArtemisQueue scaler mapping (#6890) --- e2e/arkmq/arkmq_keda_test.go | 55 ++++++++++++++++ e2e/arkmq/files/keda-arkmq-to-log-auto.yaml | 43 +++++++++++++ e2e/arkmq/setup/setup.sh | 5 ++ pkg/trait/keda/scalers/artemis.go | 47 ++++++++++++++ pkg/trait/keda/scalers/artemis_test.go | 99 +++++++++++++++++++++++++++++ 5 files changed, 249 insertions(+) diff --git a/e2e/arkmq/arkmq_keda_test.go b/e2e/arkmq/arkmq_keda_test.go new file mode 100644 index 000000000..210a39ba6 --- /dev/null +++ b/e2e/arkmq/arkmq_keda_test.go @@ -0,0 +1,55 @@ +//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 arkmq + +import ( + "context" + "testing" + + . "github.com/onsi/gomega" + + . "github.com/apache/camel-k/v2/e2e/support" +) + +func TestArkMQKedaAutoDiscovery(t *testing.T) { + WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { + t.Run("Auto-discovery ArkMQ queue", func(t *testing.T) { + ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-arkmq-to-log-auto.yaml")) + integrationName := "keda-arkmq-to-log-auto" + + g.Eventually(ScaledObject(t, ctx, ns, integrationName), TestTimeoutMedium). + ShouldNot(BeNil()) + + 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("artemis-queue")) + // Auto-discovered from the amqp:queue URI + g.Expect(scaledObj.Spec.Triggers[0].Metadata["queueName"]).To(Equal("my-queue")) + g.Expect(scaledObj.Spec.Triggers[0].Metadata["brokerAddress"]).To(Equal("my-queue")) + // Merged from autoMetadata + g.Expect(scaledObj.Spec.Triggers[0].Metadata["managementEndpoint"]).To(Equal("my-broker-hdls-svc.arkmq.svc:8161")) + g.Expect(scaledObj.Spec.Triggers[0].Metadata["brokerName"]).To(Equal("amq-broker")) + + ExpectExecSucceed(t, g, Kubectl("delete", "pipe", integrationName, "-n", ns)) + }) + }, "arkmq") +} diff --git a/e2e/arkmq/files/keda-arkmq-to-log-auto.yaml b/e2e/arkmq/files/keda-arkmq-to-log-auto.yaml new file mode 100644 index 000000000..d63858e67 --- /dev/null +++ b/e2e/arkmq/files/keda-arkmq-to-log-auto.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. +# --------------------------------------------------------------------------- + +# This pipe tests KEDA auto-discovery for an ArkMQ queue: +# - artemis-queue trigger is auto-discovered from the amqp:queue URI +# - management endpoint and broker name are merged via autoMetadata + +apiVersion: camel.apache.org/v1 +kind: Pipe +metadata: + name: keda-arkmq-to-log-auto + namespace: arkmq +spec: + source: + ref: + kind: ActiveMQArtemisAddress + apiVersion: broker.amq.io/v1beta1 + name: my-queue + sink: + uri: log:info + traits: + keda: + enabled: true + minReplicaCount: 0 + maxReplicaCount: 1 + autoMetadata: + artemis-queue: + managementEndpoint: my-broker-hdls-svc.arkmq.svc:8161 + brokerName: amq-broker \ No newline at end of file diff --git a/e2e/arkmq/setup/setup.sh b/e2e/arkmq/setup/setup.sh index 3d157823f..16badef17 100755 --- a/e2e/arkmq/setup/setup.sh +++ b/e2e/arkmq/setup/setup.sh @@ -41,3 +41,8 @@ kubectl wait activemqartemis/my-broker --for=condition=Ready --timeout=300s -n a #### Setup an ActiveMQ Artemis queue address kubectl apply -f $SCRIPT_DIR/queue.yaml -n arkmq + +#### Setup Keda operator +# it creates by default a keda namespace +kubectl apply --server-side -f https://github.com/kedacore/keda/releases/download/v2.17.2/keda-2.17.2.yaml +kubectl rollout status deployment keda-operator -n keda --timeout=180s \ No newline at end of file diff --git a/pkg/trait/keda/scalers/artemis.go b/pkg/trait/keda/scalers/artemis.go new file mode 100644 index 000000000..da588b41c --- /dev/null +++ b/pkg/trait/keda/scalers/artemis.go @@ -0,0 +1,47 @@ +/* +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" +) + +// ArtemisScaler maps Camel AMQP queue endpoints to the KEDA artemis-queue scaler. +type ArtemisScaler struct{} + +func init() { + keda.Register(&ArtemisScaler{}) +} + +func (a *ArtemisScaler) Component() string { + return "amqp" +} + +func (a *ArtemisScaler) Map(pathValue string, params map[string]string) (string, map[string]string) { + metadata := make(map[string]string) + queueName := strings.TrimPrefix(pathValue, "queue:") + if queueName != "" && !strings.HasPrefix(queueName, "topic:") { + metadata["queueName"] = queueName + // For anycast queues created by ArkMQ, the address name matches the queue name. + metadata["brokerAddress"] = queueName + } + + return "artemis-queue", metadata +} diff --git a/pkg/trait/keda/scalers/artemis_test.go b/pkg/trait/keda/scalers/artemis_test.go new file mode 100644 index 000000000..5a296dd3e --- /dev/null +++ b/pkg/trait/keda/scalers/artemis_test.go @@ -0,0 +1,99 @@ +/* +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 TestArtemisScalerRegistered(t *testing.T) { + mapper, found := keda.GetMapper("amqp") + require.True(t, found, "artemis scaler should be registered for amqp") + assert.Equal(t, "amqp", mapper.Component()) +} + +func TestArtemisScalerMap(t *testing.T) { + scaler := &ArtemisScaler{} + + tests := []struct { + name string + pathValue string + params map[string]string + expectedType string + expectedMeta map[string]string + }{ + { + name: "queue prefix", + pathValue: "queue:orders", + params: map[string]string{}, + expectedType: "artemis-queue", + expectedMeta: map[string]string{ + "queueName": "orders", + "brokerAddress": "orders", + }, + }, + { + name: "no prefix", + pathValue: "orders", + params: map[string]string{}, + expectedType: "artemis-queue", + expectedMeta: map[string]string{ + "queueName": "orders", + "brokerAddress": "orders", + }, + }, + { + name: "topic is not mapped", + pathValue: "topic:events", + params: map[string]string{}, + expectedType: "artemis-queue", + expectedMeta: map[string]string{}, + }, + { + name: "empty path", + pathValue: "", + params: map[string]string{}, + expectedType: "artemis-queue", + expectedMeta: map[string]string{}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + kedaType, metadata := scaler.Map(tt.pathValue, tt.params) + assert.Equal(t, tt.expectedType, kedaType) + assert.Equal(t, tt.expectedMeta, metadata) + }) + } +} + +func TestMapToKedaTriggerArtemis(t *testing.T) { + trigger, err := keda.MapToKedaTrigger("amqp:queue:my-queue") + require.NoError(t, err) + require.NotNil(t, trigger) + assert.Equal(t, "artemis-queue", trigger.Type) + assert.Equal(t, map[string]string{ + "queueName": "my-queue", + "brokerAddress": "my-queue", + }, trigger.Metadata) +}
