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 266afaf9d80f35e98f68091aa7d9d1b8f42fd482 Author: Anshul Jagota <[email protected]> AuthorDate: Wed Nov 4 19:10:43 2026 +0530 refactor(test): remove duplicated KEDA tests from kafka and arkmq, onboard artemis to keda workflow --- e2e/arkmq/arkmq_keda_test.go | 115 --------------- e2e/arkmq/files/keda-arkmq-to-log-auto.yaml | 53 ------- e2e/arkmq/files/keda-it-arkmq-to-log-auto.yaml | 52 ------- e2e/arkmq/setup/setup.sh | 5 - .../keda-kafka-auto-discovery-with-manual.yaml | 45 ------ e2e/kafka/files/keda-kafka-auto-metadata.yaml | 41 ------ e2e/kafka/files/keda-kafkatopic-to-log.yaml | 45 ------ e2e/kafka/kafka_autoscale_keda_test.go | 155 --------------------- e2e/kafka/setup/setup.sh | 5 - e2e/keda/files/artemis-ephemeral.yaml | 65 +++++++++ .../files/keda-it-artemis-to-log-auto.yaml} | 23 ++- .../files/timer-to-artemis.yaml} | 27 ++-- e2e/keda/keda_artemis_test.go | 107 ++++++++++++++ e2e/keda/keda_kafka_test.go | 2 +- 14 files changed, 197 insertions(+), 543 deletions(-) diff --git a/e2e/arkmq/arkmq_keda_test.go b/e2e/arkmq/arkmq_keda_test.go deleted file mode 100644 index b47143aa8..000000000 --- a/e2e/arkmq/arkmq_keda_test.go +++ /dev/null @@ -1,115 +0,0 @@ -//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/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 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")) - g.Expect(scaledObj.Spec.Triggers[0].Metadata["managementEndpoint"]).To(Equal("my-broker-hdls-svc.arkmq.svc:8161")) - // Merged from autoMetadata - g.Expect(scaledObj.Spec.Triggers[0].Metadata["brokerName"]).To(Equal("amq-broker")) - - g.Eventually(IntegrationConditionStatus(t, ctx, ns, integrationName, v1.IntegrationConditionReady), TestTimeoutMedium). - Should(Equal(corev1.ConditionTrue)) - g.Eventually(IntegrationStatusReplicas(t, ctx, ns, integrationName), TestTimeoutShort). - Should(gstruct.PointTo(BeNumerically("==", 1))) - g.Eventually(IntegrationStatusReplicas(t, ctx, ns, integrationName), TestTimeoutShort). - Should(gstruct.PointTo(BeNumerically("==", 0))) - producerName := "timer-to-arkmq" - ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/timer-to-arkmq.yaml")) - g.Eventually(IntegrationConditionStatus(t, ctx, ns, producerName, v1.IntegrationConditionReady), TestTimeoutMedium). - Should(Equal(corev1.ConditionTrue)) - g.Eventually(IntegrationStatusReplicas(t, ctx, ns, integrationName), TestTimeoutShort). - Should(gstruct.PointTo(BeNumerically("==", 1))) - g.Eventually(IntegrationLogs(t, ctx, ns, integrationName)).Should(ContainSubstring("Body is null")) - ExpectExecSucceed(t, g, Kubectl("delete", "pipe", producerName, "-n", ns)) - g.Eventually(IntegrationStatusReplicas(t, ctx, ns, integrationName), TestTimeoutMedium). - Should(gstruct.PointTo(BeNumerically("==", 0))) - - ExpectExecSucceed(t, g, Kubectl("delete", "pipe", integrationName, "-n", ns)) - }) - }, "arkmq") -} - -func TestArkMQKedaAutoDiscoveryIntegration(t *testing.T) { - WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { - t.Run("Auto-discovery ArkMQ queue from Integration", func(t *testing.T) { - ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-it-arkmq-to-log-auto.yaml")) - integrationName := "keda-it-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")) - g.Expect(scaledObj.Spec.Triggers[0].Metadata["queueName"]).To(Equal("my-queue")) - g.Expect(scaledObj.Spec.Triggers[0].Metadata["brokerAddress"]).To(Equal("my-queue")) - 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")) - - g.Eventually(IntegrationConditionStatus(t, ctx, ns, integrationName, v1.IntegrationConditionReady), TestTimeoutMedium). - Should(Equal(corev1.ConditionTrue)) - g.Eventually(IntegrationStatusReplicas(t, ctx, ns, integrationName), TestTimeoutShort). - Should(gstruct.PointTo(BeNumerically("==", 1))) - g.Eventually(IntegrationStatusReplicas(t, ctx, ns, integrationName), TestTimeoutShort). - Should(gstruct.PointTo(BeNumerically("==", 0))) - producerName := "timer-to-arkmq" - ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/timer-to-arkmq.yaml")) - g.Eventually(IntegrationConditionStatus(t, ctx, ns, producerName, v1.IntegrationConditionReady), TestTimeoutMedium). - Should(Equal(corev1.ConditionTrue)) - g.Eventually(IntegrationStatusReplicas(t, ctx, ns, integrationName), TestTimeoutShort). - Should(gstruct.PointTo(BeNumerically("==", 1))) - g.Eventually(IntegrationLogs(t, ctx, ns, integrationName)).Should(ContainSubstring("Body is null")) - ExpectExecSucceed(t, g, Kubectl("delete", "pipe", producerName, "-n", ns)) - g.Eventually(IntegrationStatusReplicas(t, ctx, ns, integrationName), TestTimeoutMedium). - Should(gstruct.PointTo(BeNumerically("==", 0))) - - ExpectExecSucceed(t, g, Kubectl("delete", "it", 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 deleted file mode 100644 index dcfeee167..000000000 --- a/e2e/arkmq/files/keda-arkmq-to-log-auto.yaml +++ /dev/null @@ -1,53 +0,0 @@ -# --------------------------------------------------------------------------- -# 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 a manual KEDA artemis-queue trigger for an ArkMQ queue: -# - trigger metadata is provided manually -# - credentials come from my-broker-credentials-secret via TriggerAuthentication - -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 - cooldownPeriod: 20 - pollingInterval: 10 - triggers: - - type: artemis-queue - metadata: - managementEndpoint: my-broker-hdls-svc.arkmq.svc:8161 - queueName: my-queue - brokerAddress: my-queue - brokerName: amq-broker - secrets: - - name: my-broker-credentials-secret - mapping: - AMQ_USER: username - AMQ_PASSWORD: password \ No newline at end of file diff --git a/e2e/arkmq/files/keda-it-arkmq-to-log-auto.yaml b/e2e/arkmq/files/keda-it-arkmq-to-log-auto.yaml deleted file mode 100644 index fca43ccc7..000000000 --- a/e2e/arkmq/files/keda-it-arkmq-to-log-auto.yaml +++ /dev/null @@ -1,52 +0,0 @@ -# --------------------------------------------------------------------------- -# 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-arkmq-to-log-auto - namespace: arkmq -spec: - flows: - - route: - from: - steps: - - to: log:info - uri: amqp:queue:my-queue - traits: - camel: - properties: - - camel.component.amqp.broker-url=amqp://my-broker-hdls-svc.arkmq.svc:61616 - - quarkus.qpid-jms.url=amqp://my-broker-hdls-svc.arkmq.svc:61616 - keda: - enabled: true - minReplicaCount: 0 - maxReplicaCount: 1 - cooldownPeriod: 20 - pollingInterval: 10 - triggers: - - type: artemis-queue - metadata: - managementEndpoint: my-broker-hdls-svc.arkmq.svc:8161 - queueName: my-queue - brokerAddress: my-queue - brokerName: amq-broker - secrets: - - name: my-broker-credentials-secret - mapping: - AMQ_USER: username - AMQ_PASSWORD: password diff --git a/e2e/arkmq/setup/setup.sh b/e2e/arkmq/setup/setup.sh index 16badef17..3d157823f 100755 --- a/e2e/arkmq/setup/setup.sh +++ b/e2e/arkmq/setup/setup.sh @@ -41,8 +41,3 @@ 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/e2e/kafka/files/keda-kafka-auto-discovery-with-manual.yaml b/e2e/kafka/files/keda-kafka-auto-discovery-with-manual.yaml deleted file mode 100644 index 02557f1b3..000000000 --- a/e2e/kafka/files/keda-kafka-auto-discovery-with-manual.yaml +++ /dev/null @@ -1,45 +0,0 @@ -# --------------------------------------------------------------------------- -# 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 integration tests KEDA auto-discovery with manual trigger (Case 2): -# - Manual cron trigger is specified -# - Kafka trigger should still be auto-discovered (different type) -# - Result: 2 triggers (cron + kafka) - -apiVersion: camel.apache.org/v1 -kind: Integration -metadata: - name: keda-kafka-auto-discovery-with-manual - namespace: kafka -spec: - sources: - - content: | - from("kafka:my-topic?brokers=my-cluster-kafka-bootstrap.kafka.svc:9092&groupId=mixed-group") - .log("${body}"); - name: routes.java - traits: - keda: - enabled: true - minReplicaCount: 0 - maxReplicaCount: 2 - # Manual cron trigger - kafka should still be auto-discovered - triggers: - - type: cron - metadata: - timezone: "Etc/UTC" - start: "0 * * * *" - end: "59 * * * *" - desiredReplicas: "1" diff --git a/e2e/kafka/files/keda-kafka-auto-metadata.yaml b/e2e/kafka/files/keda-kafka-auto-metadata.yaml deleted file mode 100644 index c554ef792..000000000 --- a/e2e/kafka/files/keda-kafka-auto-metadata.yaml +++ /dev/null @@ -1,41 +0,0 @@ -# --------------------------------------------------------------------------- -# 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 integration tests KEDA autoMetadata merge (Case 3): -# - Kafka trigger is auto-discovered from URI -# - Additional metadata (lagThreshold) is merged via autoMetadata - -apiVersion: camel.apache.org/v1 -kind: Integration -metadata: - name: keda-kafka-auto-metadata - namespace: kafka -spec: - sources: - - content: | - from("kafka:my-topic?brokers=my-cluster-kafka-bootstrap.kafka.svc:9092&groupId=metadata-group") - .log("${body}"); - name: routes.java - traits: - keda: - enabled: true - minReplicaCount: 0 - maxReplicaCount: 1 - # Extra metadata to merge into auto-discovered kafka trigger - autoMetadata: - kafka: - lagThreshold: "100" - activationLagThreshold: "10" diff --git a/e2e/kafka/files/keda-kafkatopic-to-log.yaml b/e2e/kafka/files/keda-kafkatopic-to-log.yaml deleted file mode 100644 index 5d5d0abb9..000000000 --- a/e2e/kafka/files/keda-kafkatopic-to-log.yaml +++ /dev/null @@ -1,45 +0,0 @@ -# --------------------------------------------------------------------------- -# 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: Pipe -metadata: - name: keda-kafkatopic-to-log - namespace: kafka -spec: - sink: - uri: log:info - source: - ref: - kind: KafkaTopic - apiVersion: kafka.strimzi.io/v1beta2 - name: my-topic - properties: - groupId: group-1 - traits: - keda: - enabled: true - minReplicaCount: 0 - maxReplicaCount: 1 - cooldownPeriod: 20 - pollingInterval: 10 - triggers: - - type: kafka - metadata: - bootstrapServers: my-cluster-kafka-bootstrap.kafka.svc:9092 - consumerGroup: group-1 - topic: my-topic - lagThreshold: "10" diff --git a/e2e/kafka/kafka_autoscale_keda_test.go b/e2e/kafka/kafka_autoscale_keda_test.go deleted file mode 100644 index 7c05e50fa..000000000 --- a/e2e/kafka/kafka_autoscale_keda_test.go +++ /dev/null @@ -1,155 +0,0 @@ -//go:build integration -// +build integration - -// To enable compilation of this file in Goland, go to "Settings -> Go -> Vendoring & Build Tags -> Custom Tags" and add "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 kafka - -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 TestKafkaKedaAutoscale(t *testing.T) { - WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { - // NOTE: all resources are local to kafka namespace - - // The scenario is the following: - // 1. Start a Kafka consumer and wait to autoscale to 0 as no traffic - // 2. Start a Kafka producer and verify the consumer scales up and consume accordingly - // 3. Stop the Kafka producer and verify the consumer scales to 0 accordingly - t.Run("Serverless Kafka", func(t *testing.T) { - ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-kafkatopic-to-log.yaml")) - consumerName := "keda-kafkatopic-to-log" - // Start consumer - g.Eventually(IntegrationConditionStatus(t, ctx, ns, consumerName, v1.IntegrationConditionReady), TestTimeoutMedium). - Should(Equal(corev1.ConditionTrue)) - g.Eventually(IntegrationStatusReplicas(t, ctx, ns, consumerName), TestTimeoutShort). - Should(gstruct.PointTo(BeNumerically("==", 1))) - // It should scale to 0 after some time as there is no traffic - g.Eventually(IntegrationStatusReplicas(t, ctx, ns, consumerName), TestTimeoutShort). - Should(gstruct.PointTo(BeNumerically("==", 0))) - // Start producer - producerName := "timer-to-kafkatopic" - ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/timer-to-kafkatopic.yaml")) - g.Eventually(IntegrationConditionStatus(t, ctx, ns, producerName, v1.IntegrationConditionReady), TestTimeoutMedium). - Should(Equal(corev1.ConditionTrue)) - // The consumer will scale up - g.Eventually(IntegrationStatusReplicas(t, ctx, ns, consumerName), TestTimeoutShort). - Should(gstruct.PointTo(BeNumerically("==", 1))) - // Verify we are consuming some record (the body is null as the timer is pushing nothing) - g.Eventually(IntegrationLogs(t, ctx, ns, consumerName)).Should(ContainSubstring("Body is null")) - // Stop the producer - ExpectExecSucceed(t, g, Kubectl("delete", "pipe", producerName, "-n", ns)) - // Consumer should scale back to 0 after some time as there is no longer traffic - g.Eventually(IntegrationStatusReplicas(t, ctx, ns, consumerName), TestTimeoutMedium). - Should(gstruct.PointTo(BeNumerically("==", 0))) - ExpectExecSucceed(t, g, Kubectl("delete", "pipe", "--all", "-n", ns)) - }) - }, "kafka") -} - -func TestKafkaKedaAutoDiscovery(t *testing.T) { - WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { - t.Run("Auto-discovery Kafka", func(t *testing.T) { - ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-it-kafkatopic-to-log-auto.yaml")) - integrationName := "keda-kafka-auto-discovery" - - // Wait for ScaledObject - g.Eventually(ScaledObject(t, ctx, ns, integrationName), TestTimeoutMedium). - ShouldNot(BeNil()) - - // Verify the auto-discovered - 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("kafka")) - g.Expect(scaledObj.Spec.Triggers[0].Metadata["topic"]).To(Equal("my-topic")) - g.Expect(scaledObj.Spec.Triggers[0].Metadata["bootstrapServers"]).To(Equal("my-cluster-kafka-bootstrap.kafka.svc:9092")) - g.Expect(scaledObj.Spec.Triggers[0].Metadata["consumerGroup"]).To(Equal("auto-group")) - - ExpectExecSucceed(t, g, Kubectl("delete", "it", integrationName, "-n", ns)) - }) - }, "kafka") -} - -func TestKafkaKedaAutoDiscoveryWithManualTrigger(t *testing.T) { - WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { - t.Run("Auto-discovery with manual trigger (Case 2)", func(t *testing.T) { - ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-kafka-auto-discovery-with-manual.yaml")) - integrationName := "keda-kafka-auto-discovery-with-manual" - - 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(2)) - - var kafkaTrigger, cronTrigger interface{} - for _, trigger := range scaledObj.Spec.Triggers { - if trigger.Type == "kafka" { - kafkaTrigger = trigger - } - if trigger.Type == "cron" { - cronTrigger = trigger - } - } - g.Expect(kafkaTrigger).NotTo(BeNil(), "kafka trigger should be auto-discovered") - g.Expect(cronTrigger).NotTo(BeNil(), "cron trigger should exist from manual config") - - ExpectExecSucceed(t, g, Kubectl("delete", "it", integrationName, "-n", ns)) - }) - }, "kafka") -} - -func TestKafkaKedaAutoMetadata(t *testing.T) { - WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { - t.Run("Auto-discovery with autoMetadata merge (Case 3)", func(t *testing.T) { - ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-kafka-auto-metadata.yaml")) - integrationName := "keda-kafka-auto-metadata" - - g.Eventually(ScaledObject(t, ctx, ns, integrationName), TestTimeoutMedium). - ShouldNot(BeNil()) - - // Verify the trigger has merged metadata - 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("kafka")) - // Auto-discovered - g.Expect(scaledObj.Spec.Triggers[0].Metadata["topic"]).To(Equal("my-topic")) - g.Expect(scaledObj.Spec.Triggers[0].Metadata["bootstrapServers"]).To(Equal("my-cluster-kafka-bootstrap.kafka.svc:9092")) - g.Expect(scaledObj.Spec.Triggers[0].Metadata["consumerGroup"]).To(Equal("metadata-group")) - // Merge - g.Expect(scaledObj.Spec.Triggers[0].Metadata["lagThreshold"]).To(Equal("100")) - g.Expect(scaledObj.Spec.Triggers[0].Metadata["activationLagThreshold"]).To(Equal("10")) - - ExpectExecSucceed(t, g, Kubectl("delete", "it", integrationName, "-n", ns)) - }) - }, "kafka") -} diff --git a/e2e/kafka/setup/setup.sh b/e2e/kafka/setup/setup.sh index 3e20a4a9f..51061dd50 100755 --- a/e2e/kafka/setup/setup.sh +++ b/e2e/kafka/setup/setup.sh @@ -32,11 +32,6 @@ kubectl wait --for=condition=established crd/kafkas.kafka.strimzi.io --timeout=6 kubectl wait --for=condition=established crd/kafkanodepools.kafka.strimzi.io --timeout=60s kubectl wait --for=condition=established crd/kafkatopics.kafka.strimzi.io --timeout=60s -#### 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 - #### Setup a Kafka cluster which we'll use for testing kubectl apply -f ./e2e/kafka/setup/kafka-ephemeral.yaml kubectl wait kafka/my-cluster --for=condition=Ready --timeout=300s -n kafka diff --git a/e2e/keda/files/artemis-ephemeral.yaml b/e2e/keda/files/artemis-ephemeral.yaml new file mode 100644 index 000000000..ac70bfb63 --- /dev/null +++ b/e2e/keda/files/artemis-ephemeral.yaml @@ -0,0 +1,65 @@ +# --------------------------------------------------------------------------- +# 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-broker +spec: + replicas: 1 + selector: + matchLabels: + app: my-broker + template: + metadata: + labels: + app: my-broker + spec: + containers: + - name: artemis + image: apache/artemis:latest-alpine + env: + - name: ANONYMOUS_LOGIN + value: "true" + - name: EXTRA_ARGS + value: "--http-host 0.0.0.0 --relax-jolokia --name amq-broker --queues my-queue --allow-anonymous" + ports: + - containerPort: 61616 + name: amqp + - containerPort: 8161 + name: console + resources: + requests: + memory: "256Mi" + cpu: "200m" + limits: + memory: "512Mi" +--- +apiVersion: v1 +kind: Service +metadata: + name: my-broker-hdls-svc +spec: + selector: + app: my-broker + ports: + - name: amqp + port: 61616 + targetPort: 61616 + - name: console + port: 8161 + targetPort: 8161 diff --git a/e2e/kafka/files/keda-it-kafkatopic-to-log.yaml b/e2e/keda/files/keda-it-artemis-to-log-auto.yaml similarity index 75% rename from e2e/kafka/files/keda-it-kafkatopic-to-log.yaml rename to e2e/keda/files/keda-it-artemis-to-log-auto.yaml index 49cf75966..58d03bceb 100644 --- a/e2e/kafka/files/keda-it-kafkatopic-to-log.yaml +++ b/e2e/keda/files/keda-it-artemis-to-log-auto.yaml @@ -14,30 +14,29 @@ # See the License for the specific language governing permissions and # limitations under the License. # --------------------------------------------------------------------------- + apiVersion: camel.apache.org/v1 kind: Integration metadata: - name: keda-kafkatopic-to-log - namespace: kafka + name: keda-it-artemis-to-log-auto spec: flows: - route: from: steps: - to: log:info - uri: kafka:my-topic?brokers=my-cluster-kafka-bootstrap.kafka.svc%3A9092&groupId=group-1 + uri: amqp:queue:my-queue traits: + camel: + properties: + - camel.component.amqp.broker-url=amqp://my-broker-hdls-svc.keda.svc:61616 + - quarkus.qpid-jms.url=amqp://my-broker-hdls-svc.keda.svc:61616 keda: enabled: true minReplicaCount: 0 maxReplicaCount: 1 - cooldownPeriod: 20 + cooldownPeriod: 20 pollingInterval: 10 - triggers: - - type: kafka - metadata: - bootstrapServers: my-cluster-kafka-bootstrap.kafka.svc:9092 - consumerGroup: group-1 - topic: my-topic - lagThreshold: "10" - + autoMetadata: + artemis-queue: + brokerName: amq-broker diff --git a/e2e/kafka/files/keda-it-kafkatopic-to-log-auto.yaml b/e2e/keda/files/timer-to-artemis.yaml similarity index 69% rename from e2e/kafka/files/keda-it-kafkatopic-to-log-auto.yaml rename to e2e/keda/files/timer-to-artemis.yaml index ed0cb165e..b0f2a1e0e 100644 --- a/e2e/kafka/files/keda-it-kafkatopic-to-log-auto.yaml +++ b/e2e/keda/files/timer-to-artemis.yaml @@ -18,19 +18,18 @@ apiVersion: camel.apache.org/v1 kind: Integration metadata: - name: keda-kafka-auto-discovery - namespace: kafka + name: timer-to-artemis spec: - sources: - - content: | - from("kafka:my-topic?brokers=my-cluster-kafka-bootstrap.kafka.svc:9092&groupId=auto-group") - .log("${body}"); - name: routes.java + flows: + - route: + from: + uri: timer:foo?period=1000 + steps: + - setBody: + constant: "Body is null" + - to: amqp:queue:my-queue traits: - keda: - enabled: true - minReplicaCount: 0 - maxReplicaCount: 1 - cooldownPeriod: 20 - pollingInterval: 10 - # No triggers specified - auto-discovery should create kafka scaler \ No newline at end of file + camel: + properties: + - camel.component.amqp.broker-url=amqp://my-broker-hdls-svc.keda.svc:61616 + - quarkus.qpid-jms.url=amqp://my-broker-hdls-svc.keda.svc:61616 diff --git a/e2e/keda/keda_artemis_test.go b/e2e/keda/keda_artemis_test.go new file mode 100644 index 000000000..7b3132469 --- /dev/null +++ b/e2e/keda/keda_artemis_test.go @@ -0,0 +1,107 @@ +//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 TestArtemisKedaAutoscale(t *testing.T) { + WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { + ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/artemis-ephemeral.yaml", "-n", ns)) + ExpectExecSucceed(t, g, Kubectl("rollout", "status", "deployment", "my-broker", "-n", ns, "--timeout=180s")) + + t.Run("Serverless Artemis Autoscale", func(t *testing.T) { + consumerName := "keda-it-artemis-to-log-auto" + producerName := "timer-to-artemis" + + ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-it-artemis-to-log-auto.yaml", "-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", "files/timer-to-artemis.yaml", "-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/artemis-ephemeral.yaml", "-n", ns)) + }, "keda") +} + +func TestArtemisKedaAutoDiscovery(t *testing.T) { + WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { + t.Run("Auto-discovery Artemis", func(t *testing.T) { + integrationName := "keda-it-artemis-to-log-auto" + + ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-it-artemis-to-log-auto.yaml", "-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("artemis-queue")) + g.Expect(scaledObj.Spec.Triggers[0].Metadata["queueName"]).To(Equal("my-queue")) + g.Expect(scaledObj.Spec.Triggers[0].Metadata["brokerAddress"]).To(Equal("my-queue")) + g.Expect(scaledObj.Spec.Triggers[0].Metadata["managementEndpoint"]).To(Equal("my-broker-hdls-svc.keda.svc:8161")) + g.Expect(scaledObj.Spec.Triggers[0].Metadata["brokerName"]).To(Equal("amq-broker")) + + ExpectExecSucceed(t, g, Kubectl("delete", "it", integrationName, "-n", ns)) + }) + }, "keda") +} diff --git a/e2e/keda/keda_kafka_test.go b/e2e/keda/keda_kafka_test.go index 1eb7b34eb..d12b34f6d 100644 --- a/e2e/keda/keda_kafka_test.go +++ b/e2e/keda/keda_kafka_test.go @@ -164,4 +164,4 @@ func TestKafkaKedaAutoMetadata(t *testing.T) { ExpectExecSucceed(t, g, Kubectl("delete", "it", integrationName, "-n", ns)) }) }, "keda") -} \ No newline at end of file +}
