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 6fb71384a92069cff33dd3b7b4deb54c1238f486 Author: UchihaIthachi <[email protected]> AuthorDate: Thu Nov 5 23:29:54 2026 +0530 feat(trait/keda): add Apache Pulsar scaler mapping Signed-off-by: UchihaIthachi <[email protected]> --- docs/modules/traits/pages/keda.adoc | 3 +- e2e/keda/files/keda-it-pulsar-to-log-auto.yaml | 38 ++++++++ e2e/keda/files/pulsar-ephemeral.yaml | 71 ++++++++++++++ e2e/keda/files/timer-to-pulsar.yaml | 34 +++++++ e2e/keda/keda_pulsar_test.go | 86 +++++++++++++++++ pkg/trait/keda/scalers/pulsar.go | 72 +++++++++++++++ pkg/trait/keda/scalers/pulsar_test.go | 123 +++++++++++++++++++++++++ 7 files changed, 426 insertions(+), 1 deletion(-) diff --git a/docs/modules/traits/pages/keda.adoc b/docs/modules/traits/pages/keda.adoc index 13fddb354..cd282d900 100644 --- a/docs/modules/traits/pages/keda.adoc +++ b/docs/modules/traits/pages/keda.adoc @@ -73,8 +73,9 @@ NOTE: the variable names are "snake case" if you're using in `kamel` CLI, for ex When `keda.auto` is enabled, Camel K inspects the consumer endpoints in an Integration and creates KEDA triggers for components with a registered scaler mapping. A scaler is the KEDA adapter that reads metrics from an external system and uses them to determine when a workload should scale. See the https://keda.sh/docs/latest/concepts/scaling-deployments/[KEDA scaler concepts] documentation for more information. -Automatic discovery currently supports the following Camel component: +Automatic discovery currently supports the following Camel components: * `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. +* `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`. diff --git a/e2e/keda/files/keda-it-pulsar-to-log-auto.yaml b/e2e/keda/files/keda-it-pulsar-to-log-auto.yaml new file mode 100644 index 000000000..01b3c97a8 --- /dev/null +++ b/e2e/keda/files/keda-it-pulsar-to-log-auto.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: keda-it-pulsar-to-log-auto +spec: + flows: + - route: + from: + uri: pulsar:persistent://public/default/my-topic?subscriptionName=my-subscription + steps: + - to: log:info + traits: + camel: + properties: + - camel.component.pulsar.service-url=pulsar://pulsar.keda.svc:6650 + keda: + enabled: true + minReplicaCount: 0 + maxReplicaCount: 1 + cooldownPeriod: 20 + pollingInterval: 10 diff --git a/e2e/keda/files/pulsar-ephemeral.yaml b/e2e/keda/files/pulsar-ephemeral.yaml new file mode 100644 index 000000000..4f7d5e112 --- /dev/null +++ b/e2e/keda/files/pulsar-ephemeral.yaml @@ -0,0 +1,71 @@ +# --------------------------------------------------------------------------- +# 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: pulsar +spec: + replicas: 1 + selector: + matchLabels: + app: pulsar + template: + metadata: + labels: + app: pulsar + spec: + containers: + - name: pulsar + image: apachepulsar/pulsar:4.0.4 + # Standalone broker without the stream storage and functions worker, to keep it small. + command: ["bin/pulsar", "standalone", "-nss", "-nfw"] + env: + - name: PULSAR_MEM + value: "-Xms512m -Xmx512m -XX:MaxDirectMemorySize=256m" + ports: + - containerPort: 6650 + name: broker + - containerPort: 8080 + name: admin + readinessProbe: + httpGet: + path: /admin/v2/clusters + port: 8080 + initialDelaySeconds: 20 + periodSeconds: 5 + resources: + requests: + memory: "1Gi" + cpu: "250m" + limits: + memory: "1536Mi" +--- +apiVersion: v1 +kind: Service +metadata: + name: pulsar +spec: + selector: + app: pulsar + ports: + - name: broker + port: 6650 + targetPort: 6650 + - name: admin + port: 8080 + targetPort: 8080 diff --git a/e2e/keda/files/timer-to-pulsar.yaml b/e2e/keda/files/timer-to-pulsar.yaml new file mode 100644 index 000000000..52262bfc1 --- /dev/null +++ b/e2e/keda/files/timer-to-pulsar.yaml @@ -0,0 +1,34 @@ +# --------------------------------------------------------------------------- +# 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-pulsar +spec: + flows: + - route: + from: + uri: timer:foo?period=1000 + steps: + - setBody: + constant: "Hello Pulsar" + - to: pulsar:persistent://public/default/my-topic + traits: + camel: + properties: + - camel.component.pulsar.service-url=pulsar://pulsar.keda.svc:6650 diff --git a/e2e/keda/keda_pulsar_test.go b/e2e/keda/keda_pulsar_test.go new file mode 100644 index 000000000..b20e33a62 --- /dev/null +++ b/e2e/keda/keda_pulsar_test.go @@ -0,0 +1,86 @@ +//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 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 TestPulsarKedaAutoscale(t *testing.T) { + WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { + ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/pulsar-ephemeral.yaml", "-n", ns)) + ExpectExecSucceed(t, g, Kubectl("rollout", "status", "deployment", "pulsar", "-n", ns, "--timeout=300s")) + + t.Run("Pulsar auto-discovered trigger", func(t *testing.T) { + consumerName := "keda-it-pulsar-to-log-auto" + producerName := "timer-to-pulsar" + + ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-it-pulsar-to-log-auto.yaml", "-n", ns)) + + // The ScaledObject uses the trigger discovered from the Pulsar endpoint URI + g.Eventually(ScaledObject(t, ctx, ns, consumerName), TestTimeoutMedium).ShouldNot(BeNil()) + scaledObj := ScaledObject(t, ctx, ns, consumerName)() + g.Expect(scaledObj.Spec.Triggers).To(HaveLen(1)) + g.Expect(scaledObj.Spec.Triggers[0].Type).To(Equal("pulsar")) + g.Expect(scaledObj.Spec.Triggers[0].Metadata["topic"]).To(Equal("persistent://public/default/my-topic")) + g.Expect(scaledObj.Spec.Triggers[0].Metadata["subscription"]).To(Equal("my-subscription")) + g.Expect(scaledObj.Spec.Triggers[0].Metadata["adminURL"]).To(Equal("http://pulsar.keda.svc:8080";)) + + // Start consumer; it creates the subscription + g.Eventually(IntegrationConditionStatus(t, ctx, ns, consumerName, v1.IntegrationConditionReady), TestTimeoutLong). + Should(Equal(corev1.ConditionTrue)) + + // Scale down to 0 when idle + g.Eventually(IntegrationStatusReplicas(t, ctx, ns, consumerName), TestTimeoutMedium). + Should(PointTo(BeNumerically("==", 0))) + + // Start producer to build a backlog + ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/timer-to-pulsar.yaml", "-n", ns)) + g.Eventually(IntegrationConditionStatus(t, ctx, ns, producerName, v1.IntegrationConditionReady), TestTimeoutLong). + Should(Equal(corev1.ConditionTrue)) + + // Consumer scales up to 1 and processes the messages + g.Eventually(IntegrationStatusReplicas(t, ctx, ns, consumerName), TestTimeoutMedium). + Should(PointTo(BeNumerically("==", 1))) + g.Eventually(IntegrationLogs(t, ctx, ns, consumerName), TestTimeoutMedium). + Should(ContainSubstring("Hello Pulsar")) + + // Stop producer and scale back to 0 + ExpectExecSucceed(t, g, Kubectl("delete", "it", producerName, "-n", ns)) + 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/pulsar-ephemeral.yaml", "-n", ns)) + }, "keda") +} diff --git a/pkg/trait/keda/scalers/pulsar.go b/pkg/trait/keda/scalers/pulsar.go new file mode 100644 index 000000000..92fe20f74 --- /dev/null +++ b/pkg/trait/keda/scalers/pulsar.go @@ -0,0 +1,72 @@ +/* +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" + + keda "github.com/apache/camel-k/v2/pkg/trait/keda" +) + +const ( + // pulsarDefaultSubscription is the default value of the Camel Pulsar subscriptionName option. + pulsarDefaultSubscription = "subs" + // pulsarAdminPort is the default port of the Pulsar admin REST API. + pulsarAdminPort = "8080" +) + +// PulsarScaler maps Camel Pulsar consumer endpoints to the KEDA pulsar scaler. +type PulsarScaler struct{} + +func init() { + keda.Register(&PulsarScaler{}) +} + +func (p *PulsarScaler) Component() string { + return "pulsar" +} + +func (p *PulsarScaler) Map(pathValue string, params map[string]string, props map[string]string) (string, map[string]string) { + metadata := make(map[string]string) + // The Camel path is already in the format KEDA expects, e.g. persistent://public/default/my-topic. + if pathValue != "" { + metadata["topic"] = pathValue + } + + subscription := params["subscriptionName"] + if subscription == "" { + subscription = props["camel.component.pulsar.subscription-name"] + } + if subscription == "" { + subscription = pulsarDefaultSubscription + } + metadata["subscription"] = subscription + + // The admin URL is not part of the Camel configuration. Derive it from a plain pulsar:// service URL + // on the default admin port; for other setups (TLS, custom ports) it can be set with keda.autoMetadata. + serviceURL := params["serviceUrl"] + if serviceURL == "" { + serviceURL = props["camel.component.pulsar.service-url"] + } + if u, err := url.Parse(serviceURL); err == nil && u.Scheme == "pulsar" && u.Hostname() != "" { + metadata["adminURL"] = "http://"; + net.JoinHostPort(u.Hostname(), pulsarAdminPort) + } + + return "pulsar", metadata +} diff --git a/pkg/trait/keda/scalers/pulsar_test.go b/pkg/trait/keda/scalers/pulsar_test.go new file mode 100644 index 000000000..26f7790bc --- /dev/null +++ b/pkg/trait/keda/scalers/pulsar_test.go @@ -0,0 +1,123 @@ +/* +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 TestPulsarScalerRegistered(t *testing.T) { + mapper, found := keda.GetMapper("pulsar") + require.True(t, found, "pulsar scaler should be registered") + assert.Equal(t, "pulsar", mapper.Component()) +} + +func TestPulsarScalerMap(t *testing.T) { + scaler := &PulsarScaler{} + + tests := []struct { + name string + pathValue string + params map[string]string + props map[string]string + expectedMeta map[string]string + }{ + { + name: "subscription and service URL from the endpoint", + pathValue: "persistent://public/default/orders", + params: map[string]string{"subscriptionName": "orders-sub", "serviceUrl": "pulsar://pulsar.ns.svc:6650"}, + expectedMeta: map[string]string{ + "topic": "persistent://public/default/orders", + "subscription": "orders-sub", + "adminURL": "http://pulsar.ns.svc:8080";, + }, + }, + { + name: "subscription and service URL from component properties", + pathValue: "persistent://public/default/orders", + params: map[string]string{}, + props: map[string]string{ + "camel.component.pulsar.subscription-name": "props-sub", + "camel.component.pulsar.service-url": "pulsar://broker:6650", + }, + expectedMeta: map[string]string{ + "topic": "persistent://public/default/orders", + "subscription": "props-sub", + "adminURL": "http://broker:8080";, + }, + }, + { + name: "endpoint values take precedence over properties", + pathValue: "persistent://public/default/orders", + params: map[string]string{"subscriptionName": "uri-sub", "serviceUrl": "pulsar://uri-host:6650"}, + props: map[string]string{ + "camel.component.pulsar.subscription-name": "props-sub", + "camel.component.pulsar.service-url": "pulsar://props-host:6650", + }, + expectedMeta: map[string]string{ + "topic": "persistent://public/default/orders", + "subscription": "uri-sub", + "adminURL": "http://uri-host:8080";, + }, + }, + { + name: "Camel default subscription and no service URL", + pathValue: "persistent://public/default/orders", + params: map[string]string{}, + expectedMeta: map[string]string{ + "topic": "persistent://public/default/orders", + "subscription": "subs", + }, + }, + { + name: "TLS service URL leaves adminURL to autoMetadata", + pathValue: "persistent://public/default/orders", + params: map[string]string{"serviceUrl": "pulsar+ssl://broker:6651"}, + expectedMeta: map[string]string{ + "topic": "persistent://public/default/orders", + "subscription": "subs", + }, + }, + } + + 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, "pulsar", kedaType) + assert.Equal(t, tt.expectedMeta, metadata) + }) + } +} + +func TestMapToKedaTriggerPulsar(t *testing.T) { + trigger, err := keda.MapToKedaTrigger( + "pulsar:persistent://public/default/orders?subscriptionName=orders-sub&serviceUrl=pulsar://broker:6650", nil) + require.NoError(t, err) + require.NotNil(t, trigger) + assert.Equal(t, "pulsar", trigger.Type) + assert.Equal(t, map[string]string{ + "topic": "persistent://public/default/orders", + "subscription": "orders-sub", + "adminURL": "http://broker:8080";, + }, trigger.Metadata) +}
