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 dc84bbcb2cf1b347f2771a501c5cf6952b0ef1a8 Author: Anshul Jagota <[email protected]> AuthorDate: Tue Oct 27 17:06:46 2026 +0530 refactor(test): introduce dedicated KEDA testing workflow and e2e suite --- .github/workflows/keda.yml | 85 +++++++++++ e2e/keda/files/keda-it-kafkatopic-to-log-auto.yaml | 18 +++ .../keda-kafka-auto-discovery-with-manual.yaml | 23 +++ e2e/keda/files/keda-kafka-auto-metadata.yaml | 19 +++ e2e/keda/files/timer-to-kafka.yaml | 10 ++ e2e/keda/keda_kafka_test.go | 167 +++++++++++++++++++++ e2e/keda/setup/kafka-ephemeral.yaml | 88 +++++++++++ e2e/keda/setup/setup.sh | 29 ++++ script/Makefile | 8 + 9 files changed, 447 insertions(+) diff --git a/.github/workflows/keda.yml b/.github/workflows/keda.yml new file mode 100644 index 000000000..5229927d0 --- /dev/null +++ b/.github/workflows/keda.yml @@ -0,0 +1,85 @@ +# --------------------------------------------------------------------------- +# 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. +# --------------------------------------------------------------------------- + +name: keda + +env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + +on: + pull_request: + branches: + - main + - "release-*" + paths-ignore: + - 'docs/**' + - 'java/**' + - 'proposals/**' + - '**.adoc' + - '**.md' + - 'KEYS' + - 'LICENSE' + - 'NOTICE' + push: + branches: + - main + - "release-*" + paths-ignore: + - 'docs/**' + - 'java/**' + - 'proposals/**' + - '**.adoc' + - '**.md' + - 'KEYS' + - 'LICENSE' + - 'NOTICE' + workflow_dispatch: + +concurrency: + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }} + cancel-in-progress: true + +jobs: + keda-test: + runs-on: ubuntu-latest + steps: + + - name: Checkout code + uses: actions/checkout@v7 + with: + persist-credentials: false + submodules: recursive + + - name: Infra setting + uses: ./.github/actions/infra-setting + + - name: Setup KEDA and Services + shell: bash + run: | + ./e2e/keda/setup/setup.sh + + - name: Install operator + shell: bash + run: | + kubectl create ns camel + kubectl apply -k install/overlays/all-namespaces/ --server-side --force-conflicts + kubectl wait --for=condition=available deployment/camel-k-operator -n camel --timeout=60s + + - name: Run KEDA test + shell: bash + run: | + DO_TEST_PREBUILD=false GOTESTFMT="-json 2>&1 | gotestfmt" make test-keda \ No newline at end of file diff --git a/e2e/keda/files/keda-it-kafkatopic-to-log-auto.yaml b/e2e/keda/files/keda-it-kafkatopic-to-log-auto.yaml new file mode 100644 index 000000000..2eb2ee53f --- /dev/null +++ b/e2e/keda/files/keda-it-kafkatopic-to-log-auto.yaml @@ -0,0 +1,18 @@ +apiVersion: camel.apache.org/v1 +kind: Integration +metadata: + name: keda-kafka-auto-discovery + namespace: keda +spec: + sources: + - content: | + from("kafka:my-topic?brokers=my-cluster-kafka-bootstrap.keda.svc:9092&groupId=auto-group") + .log("${body}"); + name: routes.java + traits: + keda: + enabled: true + minReplicaCount: 0 + maxReplicaCount: 1 + cooldownPeriod: 20 + pollingInterval: 10 \ No newline at end of file diff --git a/e2e/keda/files/keda-kafka-auto-discovery-with-manual.yaml b/e2e/keda/files/keda-kafka-auto-discovery-with-manual.yaml new file mode 100644 index 000000000..e15909c64 --- /dev/null +++ b/e2e/keda/files/keda-kafka-auto-discovery-with-manual.yaml @@ -0,0 +1,23 @@ +apiVersion: camel.apache.org/v1 +kind: Integration +metadata: + name: keda-kafka-auto-discovery-with-manual + namespace: keda +spec: + sources: + - content: | + from("kafka:my-topic?brokers=my-cluster-kafka-bootstrap.keda.svc:9092&groupId=manual-group") + .log("${body}"); + name: routes.java + traits: + keda: + enabled: true + minReplicaCount: 0 + maxReplicaCount: 2 + triggers: + - type: cron + metadata: + timezone: "UTC" + start: "0 * * * *" + end: "30 * * * *" + desiredReplicas: "1" \ No newline at end of file diff --git a/e2e/keda/files/keda-kafka-auto-metadata.yaml b/e2e/keda/files/keda-kafka-auto-metadata.yaml new file mode 100644 index 000000000..82e19bef1 --- /dev/null +++ b/e2e/keda/files/keda-kafka-auto-metadata.yaml @@ -0,0 +1,19 @@ +apiVersion: camel.apache.org/v1 +kind: Integration +metadata: + name: keda-kafka-auto-metadata + namespace: keda +spec: + sources: + - content: | + from("kafka:my-topic?brokers=my-cluster-kafka-bootstrap.keda.svc:9092&groupId=metadata-group") + .log("${body}"); + name: routes.java + traits: + keda: + enabled: true + minReplicaCount: 0 + maxReplicaCount: 2 + autoMetadata: + lagThreshold: "100" + activationLagThreshold: "10" \ No newline at end of file diff --git a/e2e/keda/files/timer-to-kafka.yaml b/e2e/keda/files/timer-to-kafka.yaml new file mode 100644 index 000000000..a6b7760d9 --- /dev/null +++ b/e2e/keda/files/timer-to-kafka.yaml @@ -0,0 +1,10 @@ +apiVersion: camel.apache.org/v1 +kind: Pipe +metadata: + name: timer-to-kafka + namespace: keda +spec: + source: + uri: timer:foo + sink: + uri: kafka:my-topic?brokers=my-cluster-kafka-bootstrap.keda.svc:9092 \ No newline at end of file diff --git a/e2e/keda/keda_kafka_test.go b/e2e/keda/keda_kafka_test.go new file mode 100644 index 000000000..c5b474bd6 --- /dev/null +++ b/e2e/keda/keda_kafka_test.go @@ -0,0 +1,167 @@ +//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 TestKafkaKedaAutoscale(t *testing.T) { + WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { + 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")) + + // 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))) + + // Scale down to 0 when idle + g.Eventually(IntegrationStatusReplicas(t, ctx, ns, consumerName), TestTimeoutMedium). + Should(gstruct.PointTo(BeNumerically("==", 0))) + + // Start producer to generate traffic + ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/timer-to-kafka.yaml")) + 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))) + + // 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", "pipe", producerName, "-n", ns)) + + // Scale back to 0 + g.Eventually(IntegrationStatusReplicas(t, ctx, ns, consumerName), TestTimeoutMedium). + Should(gstruct.PointTo(BeNumerically("==", 0))) + }) + }, "keda") +} + +func TestKafkaKedaAutoDiscovery(t *testing.T) { + WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { + 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")) + + // 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("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.keda.svc:9092")) + g.Expect(scaledObj.Spec.Triggers[0].Metadata["consumerGroup"]).To(Equal("auto-group")) + }) + }, "keda") +} + +func TestKafkaKedaAutoDiscoveryWithManualTrigger(t *testing.T) { + WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { + 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")) + + 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") + }) + }, "keda") +} + +func TestKafkaKedaAutoMetadata(t *testing.T) { + WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { + 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")) + + 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("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.keda.svc:9092")) + 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")) + }) + }, "keda") +} \ No newline at end of file diff --git a/e2e/keda/setup/kafka-ephemeral.yaml b/e2e/keda/setup/kafka-ephemeral.yaml new file mode 100644 index 000000000..c9afba58a --- /dev/null +++ b/e2e/keda/setup/kafka-ephemeral.yaml @@ -0,0 +1,88 @@ +# --------------------------------------------------------------------------- +# 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-cluster-kafka + namespace: keda + labels: + app: kafka +spec: + replicas: 1 + selector: + matchLabels: + app: kafka + template: + metadata: + labels: + app: kafka + spec: + containers: + - name: kafka + image: apache/kafka:3.8.0 + ports: + - name: plaintext + containerPort: 9092 + env: + - name: KAFKA_NODE_ID + value: "1" + - name: KAFKA_PROCESS_ROLES + value: "broker,controller" + - name: KAFKA_LISTENERS + value: "PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093" + - name: KAFKA_ADVERTISED_LISTENERS + value: "PLAINTEXT://my-cluster-kafka-bootstrap.keda.svc:9092" + - name: KAFKA_CONTROLLER_LISTENER_NAMES + value: "CONTROLLER" + - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP + value: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT" + - name: KAFKA_CONTROLLER_QUORUM_VOTERS + value: "1@localhost:9093" + - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR + value: "1" + - name: KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR + value: "1" + - name: KAFKA_TRANSACTION_STATE_LOG_MIN_ISR + value: "1" + - name: KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS + value: "0" + - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE + value: "true" + - name: CLUSTER_ID + value: "4L622nShTUiBenVznCXsla" + readinessProbe: + tcpSocket: + port: 9092 + initialDelaySeconds: 5 + periodSeconds: 2 + resources: + requests: + memory: "256Mi" + cpu: "250m" +--- +apiVersion: v1 +kind: Service +metadata: + name: my-cluster-kafka-bootstrap + namespace: keda +spec: + selector: + app: kafka + ports: + - name: plaintext + port: 9092 + targetPort: 9092 \ No newline at end of file diff --git a/e2e/keda/setup/setup.sh b/e2e/keda/setup/setup.sh new file mode 100644 index 000000000..5062250ee --- /dev/null +++ b/e2e/keda/setup/setup.sh @@ -0,0 +1,29 @@ +#!/bin/bash + +# --------------------------------------------------------------------------- +# 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. +# --------------------------------------------------------------------------- + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +#### 1. 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 diff --git a/script/Makefile b/script/Makefile index 818fab730..2deb97485 100644 --- a/script/Makefile +++ b/script/Makefile @@ -346,6 +346,14 @@ test-kafka: gotestfmt-install go test -timeout 20m -v ./e2e/kafka -tags=integration $(GOTESTFMT) || ((FAILED++)); \ exit $${FAILED} +# +# KEDA tests for automatic scaler discovery and autoscaling +# +test-keda: gotestfmt-install + FAILED=0; \ + go test -timeout 20m -v ./e2e/keda -tags=integration $(GOTESTFMT) || ((FAILED++)); \ + exit $${FAILED} + # # ArkMQ tests that require the presence of ArkMQ operator configuration #
