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 776cd022503e825dbd69715daaa18ecd1c6e710c Author: Dhvani Dave <[email protected]> AuthorDate: Thu Oct 22 22:03:16 2026 -0400 derive artemis management endpoint from integration properties (#6890) --- .../ROOT/pages/contributing/keda-scaler.adoc | 8 +++--- e2e/arkmq/arkmq_keda_test.go | 25 ++++++++++++++++- e2e/arkmq/files/keda-arkmq-to-log-auto.yaml | 4 +-- ...og-auto.yaml => keda-it-arkmq-to-log-auto.yaml} | 26 ++++++++---------- pkg/trait/keda.go | 23 +++++++++++++++- pkg/trait/keda/mapper.go | 15 ++++++++-- pkg/trait/keda/scalers/artemis.go | 9 +++++- pkg/trait/keda/scalers/artemis_test.go | 32 ++++++++++++++++++++-- pkg/trait/keda/scalers/kafka.go | 2 +- pkg/trait/keda/scalers/kafka_test.go | 8 +++--- 10 files changed, 118 insertions(+), 34 deletions(-) diff --git a/docs/modules/ROOT/pages/contributing/keda-scaler.adoc b/docs/modules/ROOT/pages/contributing/keda-scaler.adoc index dcce94205..92f040176 100644 --- a/docs/modules/ROOT/pages/contributing/keda-scaler.adoc +++ b/docs/modules/ROOT/pages/contributing/keda-scaler.adoc @@ -13,7 +13,7 @@ When the KEDA trait is enabled, `kedaTrait.Configure` runs automatic discovery u Discovery follows this flow: . `metadata.ExtractAll` inspects all Integration sources and extracts their Camel `from` URIs. -. Each URI is passed to `MapToKedaTrigger`. +. Each URI is passed to `MapToKedaTrigger`, together with the Integration properties. . `ParseComponentURI` separates the URI into its component name, path value, and query parameters. . The component name is used to look up a `ScaleMapper` in the registry. . If a mapper is registered, it converts the path and parameters into a KEDA scaler type and metadata, which become a `KedaTrigger`. @@ -31,12 +31,12 @@ The registry and interface are defined in `pkg/trait/keda/mapper.go`: ---- type ScaleMapper interface { Component() string - Map(pathValue string, params map[string]string) (kedaType string, metadata map[string]string) + Map(pathValue string, params map[string]string, props map[string]string) (kedaType string, metadata map[string]string) } ---- `Component` returns the Camel component name used as the registry key. -`Map` receives the path part of the Camel URI and its decoded query parameters, and returns the corresponding KEDA scaler type and metadata. +`Map` receives the path part of the Camel URI, its decoded query parameters, and the Integration `camel.component.<component>.*` properties with that prefix removed, and returns the corresponding KEDA scaler type and metadata. Each implementation registers itself by calling `keda.Register` from an `init` function. The blank import of `pkg/trait/keda/scalers` in `pkg/trait/keda.go` ensures that these registration functions run when the trait package is loaded. @@ -99,7 +99,7 @@ func (m *MyComponentScaler) Component() string { return "my-component" } -func (m *MyComponentScaler) Map(pathValue string, params map[string]string) (string, map[string]string) { +func (m *MyComponentScaler) Map(pathValue string, params map[string]string, props map[string]string) (string, map[string]string) { metadata := make(map[string]string) if pathValue != "" { diff --git a/e2e/arkmq/arkmq_keda_test.go b/e2e/arkmq/arkmq_keda_test.go index ec9a80d76..d782c393e 100644 --- a/e2e/arkmq/arkmq_keda_test.go +++ b/e2e/arkmq/arkmq_keda_test.go @@ -45,11 +45,34 @@ func TestArkMQKedaAutoDiscovery(t *testing.T) { // 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["managementEndpoint"]).To(Equal("my-broker-ss-0.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") } + +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")) + + 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 index f2c262042..328f5b399 100644 --- a/e2e/arkmq/files/keda-arkmq-to-log-auto.yaml +++ b/e2e/arkmq/files/keda-arkmq-to-log-auto.yaml @@ -17,7 +17,8 @@ # 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 +# - management endpoint is derived from the broker url +# - broker name is merged via autoMetadata apiVersion: camel.apache.org/v1 kind: Pipe @@ -39,5 +40,4 @@ spec: maxReplicaCount: 1 autoMetadata: artemis-queue: - managementEndpoint: my-broker-ss-0.my-broker-hdls-svc.arkmq.svc:8161 brokerName: amq-broker \ No newline at end of file diff --git a/e2e/arkmq/files/keda-arkmq-to-log-auto.yaml b/e2e/arkmq/files/keda-it-arkmq-to-log-auto.yaml similarity index 69% copy from e2e/arkmq/files/keda-arkmq-to-log-auto.yaml copy to e2e/arkmq/files/keda-it-arkmq-to-log-auto.yaml index f2c262042..003a18232 100644 --- a/e2e/arkmq/files/keda-arkmq-to-log-auto.yaml +++ b/e2e/arkmq/files/keda-it-arkmq-to-log-auto.yaml @@ -15,29 +15,25 @@ # 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 +kind: Integration metadata: - name: keda-arkmq-to-log-auto + name: keda-it-arkmq-to-log-auto namespace: arkmq spec: - source: - ref: - kind: ActiveMQArtemisAddress - apiVersion: broker.amq.io/v1beta1 - name: my-queue - sink: - uri: log:info + sources: + - content: | + from("amqp:queue:my-queue") + .log("${body}"); + name: routes.java traits: + camel: + properties: + - camel.component.amqp.broker-url=amqp://my-broker-hdls-svc.arkmq.svc:61616 keda: enabled: true minReplicaCount: 0 maxReplicaCount: 1 autoMetadata: artemis-queue: - managementEndpoint: my-broker-ss-0.my-broker-hdls-svc.arkmq.svc:8161 - brokerName: amq-broker \ No newline at end of file + brokerName: amq-broker diff --git a/pkg/trait/keda.go b/pkg/trait/keda.go index e34452b3f..27a699de4 100644 --- a/pkg/trait/keda.go +++ b/pkg/trait/keda.go @@ -28,6 +28,7 @@ import ( "github.com/apache/camel-k/v2/pkg/metadata" kedamapper "github.com/apache/camel-k/v2/pkg/trait/keda" _ "github.com/apache/camel-k/v2/pkg/trait/keda/scalers" + "github.com/apache/camel-k/v2/pkg/util/property" corev1 "k8s.io/api/core/v1" "k8s.io/utils/ptr" ) @@ -164,8 +165,9 @@ func (t *kedaTrait) autoDiscoverTriggers(e *Environment) error { return err } + props := integrationProperties(e) for _, fromURI := range meta.FromURIs { - trigger, err := kedamapper.MapToKedaTrigger(fromURI) + trigger, err := kedamapper.MapToKedaTrigger(fromURI, props) if err != nil { return err } @@ -183,3 +185,22 @@ func (t *kedaTrait) autoDiscoverTriggers(e *Environment) error { return nil } + +func integrationProperties(e *Environment) map[string]string { + props := make(map[string]string) + for _, prop := range e.collectConfigurationPairs("property") { + props[prop.Name] = prop.Value + } + if ct := e.Catalog.GetTrait(camelTraitID); ct != nil { + if camel, ok := ct.(*camelTrait); ok { + for _, prop := range camel.Properties { + k, v := property.SplitPropertyFileEntry(prop) + if k != "" { + props[k] = v + } + } + } + } + + return props +} diff --git a/pkg/trait/keda/mapper.go b/pkg/trait/keda/mapper.go index eca30cfe2..27b6952bd 100644 --- a/pkg/trait/keda/mapper.go +++ b/pkg/trait/keda/mapper.go @@ -18,13 +18,15 @@ limitations under the License. package keda import ( + "strings" + traitv1 "github.com/apache/camel-k/v2/pkg/apis/camel/v1/trait" ) // ScaleMapper defines the interface for mapping Camel component URIs to KEDA triggers. type ScaleMapper interface { Component() string - Map(pathValue string, params map[string]string) (kedaType string, metadata map[string]string) + Map(pathValue string, params map[string]string, props map[string]string) (kedaType string, metadata map[string]string) } var registry = map[string]ScaleMapper{} @@ -39,7 +41,7 @@ func GetMapper(component string) (ScaleMapper, bool) { return m, ok } -func MapToKedaTrigger(rawURI string) (*traitv1.KedaTrigger, error) { +func MapToKedaTrigger(rawURI string, props map[string]string) (*traitv1.KedaTrigger, error) { scheme, pathValue, params, err := ParseComponentURI(rawURI) if err != nil { return nil, err @@ -51,7 +53,14 @@ func MapToKedaTrigger(rawURI string) (*traitv1.KedaTrigger, error) { if !found { return nil, nil } - kedaType, metadata := mapper.Map(pathValue, params) + prefix := "camel.component." + scheme + "." + componentProps := map[string]string{} + for k, v := range props { + if strings.HasPrefix(k, prefix) { + componentProps[strings.TrimPrefix(k, prefix)] = v + } + } + kedaType, metadata := mapper.Map(pathValue, params, componentProps) return &traitv1.KedaTrigger{ Type: kedaType, diff --git a/pkg/trait/keda/scalers/artemis.go b/pkg/trait/keda/scalers/artemis.go index da588b41c..262179e53 100644 --- a/pkg/trait/keda/scalers/artemis.go +++ b/pkg/trait/keda/scalers/artemis.go @@ -18,6 +18,8 @@ limitations under the License. package scalers import ( + "net" + "net/url" "strings" keda "github.com/apache/camel-k/v2/pkg/trait/keda" @@ -34,7 +36,7 @@ func (a *ArtemisScaler) Component() string { return "amqp" } -func (a *ArtemisScaler) Map(pathValue string, params map[string]string) (string, map[string]string) { +func (a *ArtemisScaler) Map(pathValue string, params map[string]string, props map[string]string) (string, map[string]string) { metadata := make(map[string]string) queueName := strings.TrimPrefix(pathValue, "queue:") if queueName != "" && !strings.HasPrefix(queueName, "topic:") { @@ -42,6 +44,11 @@ func (a *ArtemisScaler) Map(pathValue string, params map[string]string) (string, // For anycast queues created by ArkMQ, the address name matches the queue name. metadata["brokerAddress"] = queueName } + if brokerURL, ok := props["broker-url"]; ok { + if u, err := url.Parse(brokerURL); err == nil && u.Hostname() != "" { + metadata["managementEndpoint"] = net.JoinHostPort(u.Hostname(), "8161") + } + } return "artemis-queue", metadata } diff --git a/pkg/trait/keda/scalers/artemis_test.go b/pkg/trait/keda/scalers/artemis_test.go index 5a296dd3e..c295eb466 100644 --- a/pkg/trait/keda/scalers/artemis_test.go +++ b/pkg/trait/keda/scalers/artemis_test.go @@ -39,6 +39,7 @@ func TestArtemisScalerMap(t *testing.T) { name string pathValue string params map[string]string + props map[string]string expectedType string expectedMeta map[string]string }{ @@ -69,6 +70,18 @@ func TestArtemisScalerMap(t *testing.T) { expectedType: "artemis-queue", expectedMeta: map[string]string{}, }, + { + name: "broker url sets management endpoint", + pathValue: "queue:orders", + params: map[string]string{}, + props: map[string]string{"broker-url": "amqp://my-broker-hdls-svc.arkmq.svc:61616"}, + expectedType: "artemis-queue", + expectedMeta: map[string]string{ + "queueName": "orders", + "brokerAddress": "orders", + "managementEndpoint": "my-broker-hdls-svc.arkmq.svc:8161", + }, + }, { name: "empty path", pathValue: "", @@ -80,7 +93,7 @@ func TestArtemisScalerMap(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - kedaType, metadata := scaler.Map(tt.pathValue, tt.params) + kedaType, metadata := scaler.Map(tt.pathValue, tt.params, tt.props) assert.Equal(t, tt.expectedType, kedaType) assert.Equal(t, tt.expectedMeta, metadata) }) @@ -88,7 +101,7 @@ func TestArtemisScalerMap(t *testing.T) { } func TestMapToKedaTriggerArtemis(t *testing.T) { - trigger, err := keda.MapToKedaTrigger("amqp:queue:my-queue") + trigger, err := keda.MapToKedaTrigger("amqp:queue:my-queue", nil) require.NoError(t, err) require.NotNil(t, trigger) assert.Equal(t, "artemis-queue", trigger.Type) @@ -97,3 +110,18 @@ func TestMapToKedaTriggerArtemis(t *testing.T) { "brokerAddress": "my-queue", }, trigger.Metadata) } + +func TestMapToKedaTriggerArtemisWithBrokerURL(t *testing.T) { + props := map[string]string{ + "camel.component.amqp.broker-url": "amqp://my-broker-hdls-svc.arkmq.svc:61616", + "camel.component.kafka.brokers": "broker:9092", + } + trigger, err := keda.MapToKedaTrigger("amqp:queue:my-queue", props) + require.NoError(t, err) + require.NotNil(t, trigger) + assert.Equal(t, map[string]string{ + "queueName": "my-queue", + "brokerAddress": "my-queue", + "managementEndpoint": "my-broker-hdls-svc.arkmq.svc:8161", + }, trigger.Metadata) +} diff --git a/pkg/trait/keda/scalers/kafka.go b/pkg/trait/keda/scalers/kafka.go index 873664a00..b2117bf17 100644 --- a/pkg/trait/keda/scalers/kafka.go +++ b/pkg/trait/keda/scalers/kafka.go @@ -29,7 +29,7 @@ func init() { func (k *KafkaScaler) Component() string { return "kafka" } -func (k *KafkaScaler) Map(pathValue string, params map[string]string) (string, map[string]string) { +func (k *KafkaScaler) Map(pathValue string, params map[string]string, props map[string]string) (string, map[string]string) { metadata := make(map[string]string) if pathValue != "" { metadata["topic"] = pathValue diff --git a/pkg/trait/keda/scalers/kafka_test.go b/pkg/trait/keda/scalers/kafka_test.go index dcfbec2cb..f74c9bf36 100644 --- a/pkg/trait/keda/scalers/kafka_test.go +++ b/pkg/trait/keda/scalers/kafka_test.go @@ -76,7 +76,7 @@ func TestKafkaScalerMap(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - kedaType, metadata := scaler.Map(tt.pathValue, tt.params) + kedaType, metadata := scaler.Map(tt.pathValue, tt.params, nil) assert.Equal(t, tt.expectedType, kedaType) assert.Equal(t, tt.expectedMeta, metadata) }) @@ -84,7 +84,7 @@ func TestKafkaScalerMap(t *testing.T) { } func TestMapToKedaTriggerKafka(t *testing.T) { - trigger, err := keda.MapToKedaTrigger("kafka:orders?brokers=broker:9092&groupId=grp") + trigger, err := keda.MapToKedaTrigger("kafka:orders?brokers=broker:9092&groupId=grp", nil) require.NoError(t, err) require.NotNil(t, trigger) assert.Equal(t, "kafka", trigger.Type) @@ -96,13 +96,13 @@ func TestMapToKedaTriggerKafka(t *testing.T) { } func TestMapToKedaTriggerUnsupported(t *testing.T) { - trigger, err := keda.MapToKedaTrigger("timer:tick") + trigger, err := keda.MapToKedaTrigger("timer:tick", nil) require.NoError(t, err) assert.Nil(t, trigger, "unsupported component should return nil trigger") } func TestMapToKedaTriggerEmpty(t *testing.T) { - trigger, err := keda.MapToKedaTrigger("") + trigger, err := keda.MapToKedaTrigger("", nil) require.NoError(t, err) assert.Nil(t, trigger) }
