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 f26e98cdca6f0c32c11107d12be7055669e3b5c7 Author: Anshul Jagota <[email protected]> AuthorDate: Tue Oct 27 21:22:25 2026 +0530 refactor(test): address review comments on KEDA test workflow --- e2e/keda/{setup => files}/kafka-ephemeral.yaml | 0 e2e/keda/files/keda-it-kafkatopic-to-log-auto.yaml | 22 ++++++++- .../keda-kafka-auto-discovery-with-manual.yaml | 24 +++++++++- e2e/keda/files/keda-kafka-auto-metadata.yaml | 26 ++++++++++- e2e/keda/files/timer-to-kafka.yaml | 19 +++++++- e2e/keda/keda_kafka_test.go | 54 ++++++++++------------ e2e/keda/setup/setup.sh | 8 +--- 7 files changed, 111 insertions(+), 42 deletions(-) diff --git a/e2e/keda/setup/kafka-ephemeral.yaml b/e2e/keda/files/kafka-ephemeral.yaml similarity index 100% rename from e2e/keda/setup/kafka-ephemeral.yaml rename to e2e/keda/files/kafka-ephemeral.yaml diff --git a/e2e/keda/files/keda-it-kafkatopic-to-log-auto.yaml b/e2e/keda/files/keda-it-kafkatopic-to-log-auto.yaml index 2eb2ee53f..ff98774ce 100644 --- a/e2e/keda/files/keda-it-kafkatopic-to-log-auto.yaml +++ b/e2e/keda/files/keda-it-kafkatopic-to-log-auto.yaml @@ -1,3 +1,20 @@ +# --------------------------------------------------------------------------- +# 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: @@ -7,7 +24,7 @@ spec: sources: - content: | from("kafka:my-topic?brokers=my-cluster-kafka-bootstrap.keda.svc:9092&groupId=auto-group") - .log("${body}"); + .to("log:info"); name: routes.java traits: keda: @@ -15,4 +32,5 @@ spec: minReplicaCount: 0 maxReplicaCount: 1 cooldownPeriod: 20 - pollingInterval: 10 \ No newline at end of file + pollingInterval: 10 + # No triggers specified - auto-discovery should create kafka scaler diff --git a/e2e/keda/files/keda-kafka-auto-discovery-with-manual.yaml b/e2e/keda/files/keda-kafka-auto-discovery-with-manual.yaml index e15909c64..0f286dd53 100644 --- a/e2e/keda/files/keda-kafka-auto-discovery-with-manual.yaml +++ b/e2e/keda/files/keda-kafka-auto-discovery-with-manual.yaml @@ -1,3 +1,24 @@ +# --------------------------------------------------------------------------- +# 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: @@ -14,10 +35,11 @@ spec: enabled: true minReplicaCount: 0 maxReplicaCount: 2 + # Manual cron trigger - kafka should still be auto-discovered triggers: - type: cron metadata: timezone: "UTC" start: "0 * * * *" end: "30 * * * *" - desiredReplicas: "1" \ No newline at end of file + desiredReplicas: "1" diff --git a/e2e/keda/files/keda-kafka-auto-metadata.yaml b/e2e/keda/files/keda-kafka-auto-metadata.yaml index 82e19bef1..4ff96ae44 100644 --- a/e2e/keda/files/keda-kafka-auto-metadata.yaml +++ b/e2e/keda/files/keda-kafka-auto-metadata.yaml @@ -1,3 +1,23 @@ +# --------------------------------------------------------------------------- +# 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: @@ -14,6 +34,8 @@ spec: enabled: true minReplicaCount: 0 maxReplicaCount: 2 + # Extra metadata to merge into auto-discovered kafka trigger autoMetadata: - lagThreshold: "100" - activationLagThreshold: "10" \ No newline at end of file + kafka: + lagThreshold: "100" + activationLagThreshold: "10" diff --git a/e2e/keda/files/timer-to-kafka.yaml b/e2e/keda/files/timer-to-kafka.yaml index a6b7760d9..b41c46018 100644 --- a/e2e/keda/files/timer-to-kafka.yaml +++ b/e2e/keda/files/timer-to-kafka.yaml @@ -1,3 +1,20 @@ +# --------------------------------------------------------------------------- +# 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: @@ -7,4 +24,4 @@ spec: source: uri: timer:foo sink: - uri: kafka:my-topic?brokers=my-cluster-kafka-bootstrap.keda.svc:9092 \ No newline at end of file + uri: kafka:my-topic?brokers=my-cluster-kafka-bootstrap.keda.svc:9092 diff --git a/e2e/keda/keda_kafka_test.go b/e2e/keda/keda_kafka_test.go index c5b474bd6..db8245adc 100644 --- a/e2e/keda/keda_kafka_test.go +++ b/e2e/keda/keda_kafka_test.go @@ -25,7 +25,7 @@ import ( "testing" . "github.com/onsi/gomega" - "github.com/onsi/gomega/gstruct" + . "github.com/onsi/gomega/gstruct" corev1 "k8s.io/api/core/v1" . "github.com/apache/camel-k/v2/e2e/support" @@ -34,36 +34,33 @@ import ( func TestKafkaKedaAutoscale(t *testing.T) { WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { + ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/kafka-ephemeral.yaml", "-n", ns)) + ExpectExecSucceed(t, g, Kubectl("rollout", "status", "deployment", "my-cluster-kafka", "-n", ns, "--timeout=180s")) + t.Run("Serverless Kafka Autoscale", func(t *testing.T) { consumerName := "keda-kafka-auto-discovery" producerName := "timer-to-kafka" - // Ensure cleanup on failure - defer func() { - _ = Kubectl("delete", "pipe", producerName, "-n", ns).Run() - _ = Kubectl("delete", "it", consumerName, "-n", ns).Run() - }() - - ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-it-kafkatopic-to-log-auto.yaml")) + ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-it-kafkatopic-to-log-auto.yaml", "-n", ns)) // 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))) + Should(PointTo(BeNumerically("==", 1))) // Scale down to 0 when idle g.Eventually(IntegrationStatusReplicas(t, ctx, ns, consumerName), TestTimeoutMedium). - Should(gstruct.PointTo(BeNumerically("==", 0))) + Should(PointTo(BeNumerically("==", 0))) // Start producer to generate traffic - ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/timer-to-kafka.yaml")) + ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/timer-to-kafka.yaml", "-n", ns)) g.Eventually(PipeConditionStatus(t, ctx, ns, producerName, v1.PipeConditionReady), TestTimeoutMedium). Should(Equal(corev1.ConditionTrue)) // Consumer scales up to 1 g.Eventually(IntegrationStatusReplicas(t, ctx, ns, consumerName), TestTimeoutShort). - Should(gstruct.PointTo(BeNumerically("==", 1))) + Should(PointTo(BeNumerically("==", 1))) // Verify messages are being received and processed g.Eventually(IntegrationLogs(t, ctx, ns, consumerName), TestTimeoutShort). @@ -74,8 +71,13 @@ func TestKafkaKedaAutoscale(t *testing.T) { // Scale back to 0 g.Eventually(IntegrationStatusReplicas(t, ctx, ns, consumerName), TestTimeoutMedium). - Should(gstruct.PointTo(BeNumerically("==", 0))) + Should(PointTo(BeNumerically("==", 0))) + + ExpectExecSucceed(t, g, Kubectl("delete", "pipe", "--all", "-n", ns)) + ExpectExecSucceed(t, g, Kubectl("delete", "it", "--all", "-n", ns)) }) + + ExpectExecSucceed(t, g, Kubectl("delete", "-f", "files/kafka-ephemeral.yaml", "-n", ns)) }, "keda") } @@ -84,11 +86,7 @@ func TestKafkaKedaAutoDiscovery(t *testing.T) { t.Run("Auto-discovery Kafka", func(t *testing.T) { integrationName := "keda-kafka-auto-discovery" - defer func() { - _ = Kubectl("delete", "it", integrationName, "-n", ns).Run() - }() - - ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-it-kafkatopic-to-log-auto.yaml")) + ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-it-kafkatopic-to-log-auto.yaml", "-n", ns)) // Wait for ScaledObject g.Eventually(ScaledObject(t, ctx, ns, integrationName), TestTimeoutMedium). @@ -102,6 +100,8 @@ func TestKafkaKedaAutoDiscovery(t *testing.T) { 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.keda.svc:9092")) g.Expect(scaledObj.Spec.Triggers[0].Metadata["consumerGroup"]).To(Equal("auto-group")) + + ExpectExecSucceed(t, g, Kubectl("delete", "it", integrationName, "-n", ns)) }) }, "keda") } @@ -111,11 +111,7 @@ func TestKafkaKedaAutoDiscoveryWithManualTrigger(t *testing.T) { t.Run("Auto-discovery with manual trigger", func(t *testing.T) { integrationName := "keda-kafka-auto-discovery-with-manual" - defer func() { - _ = Kubectl("delete", "it", integrationName, "-n", ns).Run() - }() - - ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-kafka-auto-discovery-with-manual.yaml")) + ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-kafka-auto-discovery-with-manual.yaml", "-n", ns)) g.Eventually(ScaledObject(t, ctx, ns, integrationName), TestTimeoutMedium). ShouldNot(BeNil()) @@ -135,6 +131,8 @@ func TestKafkaKedaAutoDiscoveryWithManualTrigger(t *testing.T) { } 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)) }) }, "keda") } @@ -144,11 +142,7 @@ func TestKafkaKedaAutoMetadata(t *testing.T) { t.Run("Auto-discovery with autoMetadata merge", func(t *testing.T) { integrationName := "keda-kafka-auto-metadata" - defer func() { - _ = Kubectl("delete", "it", integrationName, "-n", ns).Run() - }() - - ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-kafka-auto-metadata.yaml")) + ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-kafka-auto-metadata.yaml", "-n", ns)) g.Eventually(ScaledObject(t, ctx, ns, integrationName), TestTimeoutMedium). ShouldNot(BeNil()) @@ -162,6 +156,8 @@ func TestKafkaKedaAutoMetadata(t *testing.T) { g.Expect(scaledObj.Spec.Triggers[0].Metadata["consumerGroup"]).To(Equal("metadata-group")) 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)) }) }, "keda") -} \ No newline at end of file +} diff --git a/e2e/keda/setup/setup.sh b/e2e/keda/setup/setup.sh index 5062250ee..636b779cd 100644 --- a/e2e/keda/setup/setup.sh +++ b/e2e/keda/setup/setup.sh @@ -17,13 +17,7 @@ # limitations under the License. # --------------------------------------------------------------------------- -SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" - -#### 1. Setup KEDA Operator +#### Setup KEDA Operator # Installs KEDA operator into '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 - -#### 2. Setup Plain Ephemeral Kafka (No Operator required) -kubectl apply -f $SCRIPT_DIR/kafka-ephemeral.yaml -n keda -kubectl rollout status deployment my-cluster-kafka -n keda --timeout=180s \ No newline at end of file
