diff --git a/cmd/delete_test.go b/cmd/delete_test.go index 1c6efebb6b..569db90d73 100644 --- a/cmd/delete_test.go +++ b/cmd/delete_test.go @@ -353,7 +353,15 @@ func TestDelete_ByProjectClearsDeployedMarker(t *testing.T) { Runtime: "go", Registry: TestRegistry, Deployer: keda.KedaDeployerName, // intent - how to deploy - Deploy: fn.DeploySpec{Namespace: "myns", Deployer: keda.KedaDeployerName}, + Deploy: fn.DeploySpec{ + Namespace: "myns", + Deployer: keda.KedaDeployerName, + Options: fn.Options{ + Scale: &fn.ScaleOptions{ + KEDA: &fn.KEDAScaleOptions{Triggers: []fn.KEDATrigger{{Type: "http"}}}, + }, + }, + }, } f, err := fn.New().Init(f) if err != nil { @@ -441,7 +449,16 @@ func TestDelete_ByNameLeavesLocalFunctionUntouched(t *testing.T) { // after removal keeps the INTENT deployer intact and functional. func TestDelete_ByProjectPreservesDeployerForRedeploy(t *testing.T) { root := FromTempDirectory(t) - if _, err := fn.New().Init(fn.Function{Runtime: "go", Root: root, Registry: TestRegistry}); err != nil { + f, err := fn.New().Init(fn.Function{Runtime: "go", Root: root, Registry: TestRegistry}) + if err != nil { + t.Fatal(err) + } + // keda requires at least one trigger to be declared explicitly. + f.Deployer = keda.KedaDeployerName + f.Deploy.Options.Scale = &fn.ScaleOptions{ + KEDA: &fn.KEDAScaleOptions{Triggers: []fn.KEDATrigger{{Type: "http"}}}, + } + if err := f.Write(); err != nil { t.Fatal(err) } diff --git a/cmd/deploy.go b/cmd/deploy.go index b8b8273efd..a11f682565 100644 --- a/cmd/deploy.go +++ b/cmd/deploy.go @@ -168,7 +168,7 @@ EXAMPLES cmd.Flags().StringP("builder", "b", cfg.Builder, fmt.Sprintf("Builder to use when creating the function's container. Currently supported builders are %s.", KnownBuilders())) cmd.Flags().String("deployer", cfg.Deployer, - fmt.Sprintf("Type of deployment to use: '%s' for Knative Service, '%s' for Kubernetes Deployment, or '%s' for Deployment with a KEDA HTTP scaler ($FUNC_DEPLOYER)", deployers.Knative, deployers.Kubernetes, deployers.Keda)) + fmt.Sprintf("Type of deployment to use: '%s' for Knative Service, '%s' for Kubernetes Deployment, or '%s' for Deployment scaled by KEDA (HTTP, Kafka, or cron triggers) ($FUNC_DEPLOYER)", deployers.Knative, deployers.Kubernetes, deployers.Keda)) cmd.Flags().StringP("registry", "r", cfg.Registry, "Container registry + registry namespace. (ex 'ghcr.io/myuser'). The full image name is automatically determined using this along with function name. ($FUNC_REGISTRY)") cmd.Flags().Bool("registry-insecure", cfg.RegistryInsecure, "Skip TLS certificate verification when communicating in HTTPS with the registry. The value is persisted over consecutive runs ($FUNC_REGISTRY_INSECURE)") diff --git a/cmd/deploy_test.go b/cmd/deploy_test.go index 906993f54e..dc8ce1e280 100644 --- a/cmd/deploy_test.go +++ b/cmd/deploy_test.go @@ -8,6 +8,7 @@ import ( "os" "path/filepath" "reflect" + "slices" "strings" "testing" "time" @@ -2718,6 +2719,19 @@ func TestDeploy_DeployerSwitch(t *testing.T) { // Namespace set == already deployed, which is what the guard gates on. Deploy: fn.DeploySpec{Namespace: "myns", Deployer: tt.deployedDep}, } + // keda requires at least one trigger to be declared explicitly, + // but only matters when keda ends up the effective deployer for + // this attempt (an explicit switch away from it does not). + effectiveDeployer := tt.requested + if effectiveDeployer == "" { + effectiveDeployer = tt.deployedDep + } + if effectiveDeployer == keda.KedaDeployerName { + f.Deployer = keda.KedaDeployerName + f.Deploy.Options.Scale = &fn.ScaleOptions{ + KEDA: &fn.KEDAScaleOptions{Triggers: []fn.KEDATrigger{{Type: "http"}}}, + } + } if _, err := fn.New().Init(f); err != nil { t.Fatal(err) } @@ -2944,9 +2958,20 @@ func TestDeploy_ExposeIgnoredByDeployerNote(t *testing.T) { // route cases need OpenShift gate open; none/empty do not care. cleanup := k8s.SetOpenShiftForTest(true, nil) defer cleanup() - if _, err := fn.New().Init(fn.Function{Runtime: "go", Root: root}); err != nil { + f, err := fn.New().Init(fn.Function{Runtime: "go", Root: root}) + if err != nil { t.Fatal(err) } + if slices.Contains(tt.args, "keda") { + // keda requires at least one trigger to be declared explicitly. + f.Deployer = keda.KedaDeployerName + f.Deploy.Options.Scale = &fn.ScaleOptions{ + KEDA: &fn.KEDAScaleOptions{Triggers: []fn.KEDATrigger{{Type: "http"}}}, + } + if err := f.Write(); err != nil { + t.Fatal(err) + } + } builder := mock.NewBuilder() cmd := NewDeployCmd(NewTestClient( @@ -2958,7 +2983,7 @@ func TestDeploy_ExposeIgnoredByDeployerNote(t *testing.T) { var stderr strings.Builder cmd.SetOut(&stderr) cmd.SetErr(&stderr) - err := cmd.Execute() + err = cmd.Execute() if err != nil { t.Fatalf("unexpected error: %v", err) diff --git a/docs/reference/func_deploy.md b/docs/reference/func_deploy.md index 6fdb998af7..d59e738c09 100644 --- a/docs/reference/func_deploy.md +++ b/docs/reference/func_deploy.md @@ -126,7 +126,7 @@ func deploy -b, --builder string Builder to use when creating the function's container. Currently supported builders are "host", "pack" and "s2i". (default "pack") --builder-image string Specify a custom builder image for use by the builder other than its default. ($FUNC_BUILDER_IMAGE) -c, --confirm Prompt to confirm options interactively ($FUNC_CONFIRM) - --deployer string Type of deployment to use: 'knative' for Knative Service, 'raw' for Kubernetes Deployment, or 'keda' for Deployment with a KEDA HTTP scaler ($FUNC_DEPLOYER) (default "knative") + --deployer string Type of deployment to use: 'knative' for Knative Service, 'raw' for Kubernetes Deployment, or 'keda' for Deployment scaled by KEDA (HTTP, Kafka, or cron triggers) ($FUNC_DEPLOYER) (default "knative") --domain string Domain to use for the function's route. Cluster must be configured with domain matching for the given domain (ignored if unrecognized) ($FUNC_DOMAIN) -e, --env stringArray Environment variable to set in the form NAME=VALUE. You may provide this flag multiple times for setting multiple environment variables. To unset, specify the environment variable name followed by a "-" (e.g., NAME-). --expose string External exposure mode: 'route' for an OpenShift Route (OpenShift clusters only), 'none' for cluster-local. Default: no exposure. Raw and keda deployers only. ($FUNC_EXPOSE) diff --git a/docs/reference/func_yaml.md b/docs/reference/func_yaml.md index 6c0eabba13..178f506def 100644 --- a/docs/reference/func_yaml.md +++ b/docs/reference/func_yaml.md @@ -35,6 +35,17 @@ build: s2i: example.com/user/my-s2i-node-builder ``` +### `deployer` + +The type of deployment to use when deploying the function. Possible values are: +- `knative` (default): deploys a Knative Service, scaled by Knative's KPA (Knative Pod Autoscaler). +- `raw`: deploys a plain Kubernetes Deployment with a static replica count. +- `keda`: deploys a plain Kubernetes Deployment scaled by [KEDA](https://keda.sh), based on triggers such as incoming HTTP traffic or Kafka consumer lag. See [`options.scale.keda`](#options) below. + +```yaml +deployer: keda +``` + ### `git` If using a `git` build strategy, this field is used to specify the git URL as well @@ -139,6 +150,18 @@ Options allows you to set specific configuration for the deployed function, allo - `metric`: Defines which metric type is watched by the Autoscaler. Could be `concurrency` (default) or `rps`. See related [Knative docs](https://knative.dev/docs/serving/autoscaling/autoscaling-metrics/). - `target`: Recommendation for when to scale up based on the concurrent number of incoming request. Defaults to `options.resources.limits.concurrency` when given. Can be float value greater than 0.01, default is 100. See related [Knative docs](https://knative.dev/docs/serving/autoscaling/concurrency/#soft-limit). - `utilization`: Percentage of concurrent requests utilization before scaling up. Can be float value between 1 and 100, default is 70. See related [Knative docs](https://knative.dev/docs/serving/autoscaling/concurrency/#target-utilization). + - `kpa`: Knative-specific autoscaling config, used only with `deployer: knative`. Alternative location for `metric`, `target` and `utilization` above, kept separate so KPA-specific settings don't get confused with KEDA's. + - `metric`, `target`, `utilization`: same meaning as above. + - `keda`: KEDA-specific scaling config, required when `deployer: keda`. + - `triggers`: a list of KEDA triggers. At least one is required. Each trigger has a `type` of `http`, `kafka`, or `cron`: + - `http`: scales based on incoming HTTP request rate. No additional fields. + - `kafka`: scales based on consumer group lag. Requires [`run.kafka`](#runkafka) to be configured. + - `lagThreshold`: average consumer lag per partition that triggers scaling up. Default is 10. + - `activationLagThreshold`: lag below which KEDA keeps replicas at 0 when `scale.min` is 0. Default is 0. + - `cron`: scales based on a time window. + - `timezone`: e.g. `Europe/Istanbul`. + - `start`, `end`: cron expressions defining the active window, e.g. `0 8 * * *`. + - `desiredReplicas`: number of replicas to scale to during the active window. - `resources` - `requests` - `cpu`: A CPU resource request for the container with deployed function. See related [Kubernetes docs](https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/#requests-and-limits). @@ -166,6 +189,72 @@ options: concurrency: 100 ``` +Example using `deployer: keda` with an HTTP trigger and a Kafka trigger: + +```yaml +deployer: keda +options: + scale: + min: 0 + max: 10 + keda: + triggers: + - type: http + - type: kafka + lagThreshold: 5 + activationLagThreshold: 0 +``` + +Example using `deployer: knative` with explicit KPA settings: + +```yaml +deployer: knative +options: + scale: + min: 1 + max: 10 + kpa: + metric: concurrency + target: 50 +``` + +### `run.kafka` + +When set, the function is deployed as a Kafka consumer: it reads CloudEvents from a Kafka +topic instead of (or, with the KEDA `http` trigger, in addition to) serving HTTP requests. +Requires `invoke: cloudevent` and the Go runtime. + +- `brokers`: comma-separated list of Kafka broker addresses. +- `topic`: the topic to consume. +- `consumerGroup`: the Kafka consumer group ID. +- `securityProtocol`: one of `PLAINTEXT`, `SSL`, `SASL_PLAINTEXT`, `SASL_SSL`. +- `tls`: TLS configuration, required for `SSL` and `SASL_SSL`. + - `caCert`: path to the CA certificate PEM file used to verify the broker certificate. Typically mounted via [`volumes`](#volumes). + - `clientCert`, `clientKey`: paths to the client certificate/key PEM files, for mutual TLS. + - `skipVerify`: skip broker certificate verification (development only). +- `sasl`: SASL configuration, required for `SASL_PLAINTEXT` and `SASL_SSL`. + - `mechanism`: one of `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`. + - `user`: SASL username. Supports `{{ secret:name:key }}` and `{{ configMap:name:key }}` syntax, or a plain value. + - `password`: SASL password. Supports `{{ secret:name:key }}` and `{{ configMap:name:key }}` syntax. + +```yaml +run: + kafka: + brokers: "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9093" + topic: "my-topic" + consumerGroup: "my-function-group" + securityProtocol: "SASL_SSL" + tls: + caCert: "/etc/kafka/ca/ca.crt" + sasl: + mechanism: "SCRAM-SHA-512" + user: "my-kafka-user" + password: "{{ secret:my-kafka-user:password }}" + volumes: + - secret: my-cluster-cluster-ca-cert + path: /etc/kafka/ca +``` + ### `runtime` The language runtime for your function. For example `python`. diff --git a/docs/testing-deployments/a-raw-no-kafka/deployment.yaml b/docs/testing-deployments/a-raw-no-kafka/deployment.yaml new file mode 100644 index 0000000000..6bc867fdac --- /dev/null +++ b/docs/testing-deployments/a-raw-no-kafka/deployment.yaml @@ -0,0 +1,105 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + annotations: + deployment.kubernetes.io/revision: "1" + function.knative.dev/deployer: raw + creationTimestamp: "2026-09-03T11:50:09Z" + generation: 1 + labels: + boson.dev/function: "true" + function.knative.dev/name: test-raw-no-kafka + function.knative.dev/runtime: go + name: test-raw-no-kafka + namespace: default + resourceVersion: "572" + uid: 19429fe7-f249-403e-ba0f-a8876f804a35 +spec: + progressDeadlineSeconds: 600 + replicas: 1 + revisionHistoryLimit: 10 + selector: + matchLabels: + boson.dev/function: "true" + function.knative.dev/name: test-raw-no-kafka + function.knative.dev/runtime: go + strategy: + rollingUpdate: + maxSurge: 25% + maxUnavailable: 25% + type: RollingUpdate + template: + metadata: + annotations: + function.knative.dev/deployer: raw + labels: + boson.dev/function: "true" + function.knative.dev/name: test-raw-no-kafka + function.knative.dev/runtime: go + spec: + containers: + - env: + - name: BUILT + value: 20260903T145008 + - name: ADDRESS + value: 0.0.0.0 + image: index.docker.io/aliok/test-raw-no-kafka@sha256:ec7977a4995f8a8335a3aee3cfd2c7fab714dfd14bb275b66fa523dcbbfd9ce6 + imagePullPolicy: IfNotPresent + livenessProbe: + failureThreshold: 3 + httpGet: + path: /health/liveness + port: 8080 + scheme: HTTP + periodSeconds: 10 + successThreshold: 1 + timeoutSeconds: 1 + name: user-container + ports: + - containerPort: 8080 + protocol: TCP + readinessProbe: + failureThreshold: 3 + httpGet: + path: /health/readiness + port: 8080 + scheme: HTTP + periodSeconds: 10 + successThreshold: 1 + timeoutSeconds: 1 + resources: {} + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: + - ALL + runAsNonRoot: true + seccompProfile: + type: RuntimeDefault + terminationMessagePath: /dev/termination-log + terminationMessagePolicy: File + dnsPolicy: ClusterFirst + restartPolicy: Always + schedulerName: default-scheduler + securityContext: {} + terminationGracePeriodSeconds: 30 +status: + availableReplicas: 1 + conditions: + - lastTransitionTime: "2026-09-03T11:50:14Z" + lastUpdateTime: "2026-09-03T11:50:14Z" + message: Deployment has minimum availability. + reason: MinimumReplicasAvailable + status: "True" + type: Available + - lastTransitionTime: "2026-09-03T11:50:09Z" + lastUpdateTime: "2026-09-03T11:50:14Z" + message: ReplicaSet "test-raw-no-kafka-67bd4cbb95" has successfully progressed. + reason: NewReplicaSetAvailable + status: "True" + type: Progressing + observedGeneration: 1 + readyReplicas: 1 + replicas: 1 + terminatingReplicas: 0 + updatedReplicas: 1 diff --git a/docs/testing-deployments/a-raw-no-kafka/func.yaml b/docs/testing-deployments/a-raw-no-kafka/func.yaml new file mode 100644 index 0000000000..4fee8dd2e7 --- /dev/null +++ b/docs/testing-deployments/a-raw-no-kafka/func.yaml @@ -0,0 +1,16 @@ +# $schema: https://raw.githubusercontent.com/knative/func/main/schema/func_yaml-schema.json +# yaml-language-server: $schema=https://raw.githubusercontent.com/knative/func/main/schema/func_yaml-schema.json +specVersion: 0.37.0 +name: test-raw-no-kafka +runtime: go +registry: docker.io/aliok +namespace: default +deployer: raw +created: 2026-09-03T14:45:21.811082+03:00 +invoke: cloudevent +build: + builder: pack +deploy: + namespace: default + image: index.docker.io/aliok/test-raw-no-kafka@sha256:ec7977a4995f8a8335a3aee3cfd2c7fab714dfd14bb275b66fa523dcbbfd9ce6 + deployer: raw diff --git a/docs/testing-deployments/a-raw-no-kafka/service.yaml b/docs/testing-deployments/a-raw-no-kafka/service.yaml new file mode 100644 index 0000000000..1c7f73842d --- /dev/null +++ b/docs/testing-deployments/a-raw-no-kafka/service.yaml @@ -0,0 +1,41 @@ +apiVersion: v1 +kind: Service +metadata: + annotations: + function.knative.dev/deployer: raw + creationTimestamp: "2026-09-03T11:50:09Z" + labels: + boson.dev/function: "true" + function.knative.dev/name: test-raw-no-kafka + function.knative.dev/runtime: go + name: test-raw-no-kafka + namespace: default + ownerReferences: + - apiVersion: apps/v1 + controller: true + kind: Deployment + name: test-raw-no-kafka + uid: 19429fe7-f249-403e-ba0f-a8876f804a35 + resourceVersion: "540" + uid: a8f0754a-1783-4668-904a-32d3b5e4f87b +spec: + clusterIP: 10.96.182.163 + clusterIPs: + - 10.96.182.163 + internalTrafficPolicy: Cluster + ipFamilies: + - IPv4 + ipFamilyPolicy: SingleStack + ports: + - name: http + port: 80 + protocol: TCP + targetPort: 8080 + selector: + boson.dev/function: "true" + function.knative.dev/name: test-raw-no-kafka + function.knative.dev/runtime: go + sessionAffinity: None + type: ClusterIP +status: + loadBalancer: {} diff --git a/docs/testing-deployments/a-raw-no-kafka/testing.md b/docs/testing-deployments/a-raw-no-kafka/testing.md new file mode 100644 index 0000000000..a79f5a66c6 --- /dev/null +++ b/docs/testing-deployments/a-raw-no-kafka/testing.md @@ -0,0 +1,94 @@ +# Scenario A: RAW, no Kafka + +Raw deployer, no Kafka. The simplest case: Deployment + Service, nothing else. + +## Prerequisites + +- kind, kubectl, Go 1.25+, Docker, jq + +If using Colima and you hit DNS issues (image pulls failing, etc.), restart +it with explicit DNS: + +```bash +colima stop +colima start --dns 8.8.8.8 +``` + +## 1. Build the CLI + +```bash +cd ~/go/src/knative.dev/func +go build -o /tmp/func-local ./cmd/func +``` + +## 2. Create cluster + +```bash +kind create cluster --name test-a +``` + +## 3. Create and configure the function + +```bash +mkdir /tmp/test-raw-no-kafka && cd /tmp/test-raw-no-kafka +/tmp/func-local create -l go -t cloudevents +``` + +Replace the contents of `func.yaml` (keep the `created` line from the generated file): + +```yaml +created: +specVersion: 0.37.0 +name: test-raw-no-kafka +runtime: go +registry: docker.io/aliok +deployer: raw +invoke: cloudevent +``` + +## 4. Deploy + +```bash +cd /tmp/test-raw-no-kafka +FUNC_REGISTRY=docker.io/aliok /tmp/func-local deploy --verbose +``` + +## 5. Verify + +```bash +# Should exist: Deployment, Service +kubectl get deployment test-raw-no-kafka +kubectl get svc test-raw-no-kafka + +# Should NOT exist: HTTPScaledObject, ScaledObject, TriggerAuthentication, bridge Service +kubectl get httpscaledobject test-raw-no-kafka 2>&1 || true +kubectl get scaledobject test-raw-no-kafka-kafka 2>&1 || true +kubectl get triggerauthentication test-raw-no-kafka-kafka-auth 2>&1 || true +kubectl get svc test-raw-no-kafka-interceptor-proxy 2>&1 || true + +# No Kafka env vars +kubectl get deployment test-raw-no-kafka -o json \ + | jq '.spec.template.spec.containers[0].env[]? | select(.name | startswith("KAFKA"))' +# Should print nothing +``` + +## 6. Collect resources + +```bash +kubectl get deployment test-raw-no-kafka -o yaml > /tmp/test-raw-no-kafka/deployment.yaml +kubectl get svc test-raw-no-kafka -o yaml > /tmp/test-raw-no-kafka/service.yaml +``` + +## 7. Delete function + +```bash +cd /tmp/test-raw-no-kafka +/tmp/func-local delete +``` + +## 8. Cleanup + +```bash +kind delete cluster --name test-a +rm -rf /tmp/test-raw-no-kafka +``` diff --git a/docs/testing-deployments/b-raw-with-kafka/deployment.yaml b/docs/testing-deployments/b-raw-with-kafka/deployment.yaml new file mode 100644 index 0000000000..9107c77f90 --- /dev/null +++ b/docs/testing-deployments/b-raw-with-kafka/deployment.yaml @@ -0,0 +1,134 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + annotations: + deployment.kubernetes.io/revision: "1" + function.knative.dev/deployer: raw + creationTimestamp: "2026-09-03T12:23:21Z" + generation: 1 + labels: + boson.dev/function: "true" + function.knative.dev/name: test-raw-kafka + function.knative.dev/runtime: go + name: test-raw-kafka + namespace: default + resourceVersion: "2164" + uid: bcac4018-97c6-45ed-93d3-7d63d2d0efde +spec: + progressDeadlineSeconds: 600 + replicas: 1 + revisionHistoryLimit: 10 + selector: + matchLabels: + boson.dev/function: "true" + function.knative.dev/name: test-raw-kafka + function.knative.dev/runtime: go + strategy: + rollingUpdate: + maxSurge: 25% + maxUnavailable: 25% + type: RollingUpdate + template: + metadata: + annotations: + function.knative.dev/deployer: raw + labels: + boson.dev/function: "true" + function.knative.dev/name: test-raw-kafka + function.knative.dev/runtime: go + spec: + containers: + - env: + - name: BUILT + value: 20260903T152320 + - name: ADDRESS + value: 0.0.0.0 + - name: FUNC_TRANSPORT + value: kafka + - name: KAFKA_BROKERS + value: my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9093 + - name: KAFKA_TOPIC + value: test-topic + - name: KAFKA_CONSUMER_GROUP + value: test-raw-kafka-group + - name: KAFKA_SECURITY_PROTOCOL + value: SASL_SSL + - name: KAFKA_TLS_CA_CERT + value: /etc/kafka/ca/ca.crt + - name: KAFKA_SASL_MECHANISM + value: SCRAM-SHA-512 + - name: KAFKA_SASL_USER + value: my-kafka-user + - name: KAFKA_SASL_PASSWORD + valueFrom: + secretKeyRef: + key: password + name: my-kafka-user + image: index.docker.io/aliok/test-raw-kafka@sha256:ec7977a4995f8a8335a3aee3cfd2c7fab714dfd14bb275b66fa523dcbbfd9ce6 + imagePullPolicy: IfNotPresent + livenessProbe: + failureThreshold: 3 + httpGet: + path: /health/liveness + port: 8080 + scheme: HTTP + periodSeconds: 10 + successThreshold: 1 + timeoutSeconds: 1 + name: user-container + ports: + - containerPort: 8080 + protocol: TCP + readinessProbe: + failureThreshold: 3 + httpGet: + path: /health/readiness + port: 8080 + scheme: HTTP + periodSeconds: 10 + successThreshold: 1 + timeoutSeconds: 1 + resources: {} + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: + - ALL + runAsNonRoot: true + seccompProfile: + type: RuntimeDefault + terminationMessagePath: /dev/termination-log + terminationMessagePolicy: File + volumeMounts: + - mountPath: /etc/kafka/ca + name: secret-my-cluster-cluster-ca-cert + dnsPolicy: ClusterFirst + restartPolicy: Always + schedulerName: default-scheduler + securityContext: {} + terminationGracePeriodSeconds: 30 + volumes: + - name: secret-my-cluster-cluster-ca-cert + secret: + defaultMode: 420 + secretName: my-cluster-cluster-ca-cert +status: + availableReplicas: 1 + conditions: + - lastTransitionTime: "2026-09-03T12:23:38Z" + lastUpdateTime: "2026-09-03T12:23:38Z" + message: Deployment has minimum availability. + reason: MinimumReplicasAvailable + status: "True" + type: Available + - lastTransitionTime: "2026-09-03T12:23:21Z" + lastUpdateTime: "2026-09-03T12:23:38Z" + message: ReplicaSet "test-raw-kafka-7cff8c6957" has successfully progressed. + reason: NewReplicaSetAvailable + status: "True" + type: Progressing + observedGeneration: 1 + readyReplicas: 1 + replicas: 1 + terminatingReplicas: 0 + updatedReplicas: 1 diff --git a/docs/testing-deployments/b-raw-with-kafka/func.yaml b/docs/testing-deployments/b-raw-with-kafka/func.yaml new file mode 100644 index 0000000000..db57641007 --- /dev/null +++ b/docs/testing-deployments/b-raw-with-kafka/func.yaml @@ -0,0 +1,34 @@ +# $schema: https://raw.githubusercontent.com/knative/func/main/schema/func_yaml-schema.json +# yaml-language-server: $schema=https://raw.githubusercontent.com/knative/func/main/schema/func_yaml-schema.json +specVersion: 0.37.0 +name: test-raw-kafka +runtime: go +registry: docker.io/aliok +namespace: default +deployer: raw +created: 2026-09-03T12:00:00Z +invoke: cloudevent +build: + builder: pack +run: + volumes: + - secret: my-cluster-cluster-ca-cert + path: /etc/kafka/ca + kafka: + brokers: my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9093 + topic: test-topic + consumerGroup: test-raw-kafka-group + securityProtocol: SASL_SSL + tls: + caCert: /etc/kafka/ca/ca.crt + sasl: + mechanism: SCRAM-SHA-512 + user: my-kafka-user + password: '{{ secret:my-kafka-user:password }}' +deploy: + namespace: default + image: index.docker.io/aliok/test-raw-kafka@sha256:ec7977a4995f8a8335a3aee3cfd2c7fab714dfd14bb275b66fa523dcbbfd9ce6 + options: + scale: + min: 1 + deployer: raw diff --git a/docs/testing-deployments/b-raw-with-kafka/service.yaml b/docs/testing-deployments/b-raw-with-kafka/service.yaml new file mode 100644 index 0000000000..4fda599641 --- /dev/null +++ b/docs/testing-deployments/b-raw-with-kafka/service.yaml @@ -0,0 +1,41 @@ +apiVersion: v1 +kind: Service +metadata: + annotations: + function.knative.dev/deployer: raw + creationTimestamp: "2026-09-03T12:23:21Z" + labels: + boson.dev/function: "true" + function.knative.dev/name: test-raw-kafka + function.knative.dev/runtime: go + name: test-raw-kafka + namespace: default + ownerReferences: + - apiVersion: apps/v1 + controller: true + kind: Deployment + name: test-raw-kafka + uid: bcac4018-97c6-45ed-93d3-7d63d2d0efde + resourceVersion: "2106" + uid: 0f14041f-9659-4b68-8677-649e36228d70 +spec: + clusterIP: 10.96.7.21 + clusterIPs: + - 10.96.7.21 + internalTrafficPolicy: Cluster + ipFamilies: + - IPv4 + ipFamilyPolicy: SingleStack + ports: + - name: http + port: 80 + protocol: TCP + targetPort: 8080 + selector: + boson.dev/function: "true" + function.knative.dev/name: test-raw-kafka + function.knative.dev/runtime: go + sessionAffinity: None + type: ClusterIP +status: + loadBalancer: {} diff --git a/docs/testing-deployments/b-raw-with-kafka/testing.md b/docs/testing-deployments/b-raw-with-kafka/testing.md new file mode 100644 index 0000000000..fafe35ec2e --- /dev/null +++ b/docs/testing-deployments/b-raw-with-kafka/testing.md @@ -0,0 +1,221 @@ +# Scenario B: RAW, with Kafka + +Raw deployer with Kafka (SASL_SSL). Deployment + Service with Kafka env vars +and CA cert volume. No KEDA resources. + +## Prerequisites + +- kind, kubectl, Go 1.25+, Docker, jq + +If using Colima and you hit DNS issues (image pulls failing, etc.), restart +it with explicit DNS: + +```bash +colima stop +colima start --dns 8.8.8.8 +``` + +## 1. Build the CLI + +```bash +cd ~/go/src/knative.dev/func +go build -o /tmp/func-local ./cmd/func +``` + +## 2. Create cluster + +```bash +kind create cluster --name test-b +``` + +## 3. Install Strimzi and create Kafka cluster + user + topic + +```bash +kubectl create namespace kafka +kubectl apply -f 'https://strimzi.io/install/latest?namespace=kafka' -n kafka +kubectl wait --for=condition=Ready pods --all -n kafka --timeout=120s + +kubectl apply -n kafka -f - <<'EOF' +apiVersion: kafka.strimzi.io/v1 +kind: KafkaNodePool +metadata: + name: dual-role + labels: + strimzi.io/cluster: my-cluster +spec: + replicas: 1 + roles: [controller, broker] + storage: + type: jbod + volumes: + - id: 0 + type: persistent-claim + size: 1Gi + deleteClaim: true +--- +apiVersion: kafka.strimzi.io/v1 +kind: Kafka +metadata: + name: my-cluster + annotations: + strimzi.io/node-pools: enabled + strimzi.io/kraft: enabled +spec: + kafka: + version: 4.2.0 + authorization: + type: simple + superUsers: [ANONYMOUS] + listeners: + - name: plain + port: 9092 + type: internal + tls: false + - name: tls + port: 9093 + type: internal + tls: true + authentication: + type: scram-sha-512 + config: + offsets.topic.replication.factor: 1 + transaction.state.log.replication.factor: 1 + transaction.state.log.min.isr: 1 + entityOperator: + topicOperator: {} + userOperator: {} +EOF + +kubectl wait kafka/my-cluster --for=condition=Ready --timeout=300s -n kafka + +kubectl apply -n kafka -f - <<'EOF' +apiVersion: kafka.strimzi.io/v1 +kind: KafkaUser +metadata: + name: my-kafka-user + labels: + strimzi.io/cluster: my-cluster +spec: + authentication: + type: scram-sha-512 + authorization: + type: simple + acls: + - resource: { type: topic, name: test-topic, patternType: literal } + operations: [Read, Describe] + host: "*" + - resource: { type: group, name: "*", patternType: literal } + operations: [Read] + host: "*" +--- +apiVersion: kafka.strimzi.io/v1 +kind: KafkaTopic +metadata: + name: test-topic + labels: + strimzi.io/cluster: my-cluster +spec: + partitions: 3 + replicas: 1 +EOF + +kubectl wait kafkauser/my-kafka-user --for=condition=Ready --timeout=60s -n kafka +``` + +## 4. Copy secrets to default namespace + +```bash +kubectl get secret my-cluster-cluster-ca-cert -n kafka -o json \ + | jq 'del(.metadata.namespace,.metadata.resourceVersion,.metadata.uid,.metadata.creationTimestamp,.metadata.ownerReferences)' \ + | kubectl apply -n default -f - + +kubectl get secret my-kafka-user -n kafka -o json \ + | jq 'del(.metadata.namespace,.metadata.resourceVersion,.metadata.uid,.metadata.creationTimestamp,.metadata.ownerReferences)' \ + | kubectl apply -n default -f - +``` + +## 5. Create and configure the function + +```bash +mkdir /tmp/test-raw-kafka && cd /tmp/test-raw-kafka +/tmp/func-local create -l go -t cloudevents +``` + +Replace the contents of `func.yaml` (keep the `created` line from the generated file): + +```yaml +created: +specVersion: 0.37.0 +name: test-raw-kafka +runtime: go +registry: docker.io/aliok +deployer: raw +invoke: cloudevent +deploy: + options: + scale: + min: 1 +run: + kafka: + brokers: "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9093" + topic: "test-topic" + consumerGroup: "test-raw-kafka-group" + securityProtocol: "SASL_SSL" + tls: + caCert: "/etc/kafka/ca/ca.crt" + sasl: + mechanism: "SCRAM-SHA-512" + user: "my-kafka-user" + password: "{{ secret:my-kafka-user:password }}" + volumes: + - secret: my-cluster-cluster-ca-cert + path: /etc/kafka/ca +``` + +## 6. Deploy + +```bash +cd /tmp/test-raw-kafka +FUNC_REGISTRY=docker.io/aliok /tmp/func-local deploy --verbose +``` + +## 7. Verify + +```bash +# Should exist: Deployment, Service +kubectl get deployment test-raw-kafka +kubectl get svc test-raw-kafka + +# Should NOT exist: no KEDA resources (raw deployer) +kubectl get httpscaledobject test-raw-kafka 2>&1 || true +kubectl get scaledobject test-raw-kafka-kafka 2>&1 || true + +# Kafka env vars should be present +kubectl get deployment test-raw-kafka -o json \ + | jq '.spec.template.spec.containers[0].env[] | select(.name | startswith("KAFKA") or . == "FUNC_TRANSPORT")' + +# Volume mount for CA cert +kubectl get deployment test-raw-kafka -o json \ + | jq '.spec.template.spec.containers[0].volumeMounts' +``` + +## 8. Collect resources + +```bash +kubectl get deployment test-raw-kafka -o yaml > /tmp/test-raw-kafka/deployment.yaml +kubectl get svc test-raw-kafka -o yaml > /tmp/test-raw-kafka/service.yaml +``` + +## 9. Delete function + +```bash +cd /tmp/test-raw-kafka +/tmp/func-local delete +``` + +## 10. Cleanup + +```bash +kind delete cluster --name test-b +rm -rf /tmp/test-raw-kafka +``` diff --git a/docs/testing-deployments/c-knative-no-kafka/func.yaml b/docs/testing-deployments/c-knative-no-kafka/func.yaml new file mode 100644 index 0000000000..fee9060a44 --- /dev/null +++ b/docs/testing-deployments/c-knative-no-kafka/func.yaml @@ -0,0 +1,24 @@ +# $schema: https://raw.githubusercontent.com/knative/func/main/schema/func_yaml-schema.json +# yaml-language-server: $schema=https://raw.githubusercontent.com/knative/func/main/schema/func_yaml-schema.json +specVersion: 0.37.0 +name: test-knative-no-kafka +runtime: go +registry: docker.io/aliok +namespace: default +deployer: knative +created: 2026-09-03T15:37:23.632061+03:00 +invoke: cloudevent +build: + builder: pack +deploy: + namespace: default + image: index.docker.io/aliok/test-knative-no-kafka@sha256:ec7977a4995f8a8335a3aee3cfd2c7fab714dfd14bb275b66fa523dcbbfd9ce6 + options: + scale: + min: 0 + max: 5 + kpa: + metric: concurrency + target: 100 + utilization: 70 + deployer: knative diff --git a/docs/testing-deployments/c-knative-no-kafka/ksvc.yaml b/docs/testing-deployments/c-knative-no-kafka/ksvc.yaml new file mode 100644 index 0000000000..94a7b38303 --- /dev/null +++ b/docs/testing-deployments/c-knative-no-kafka/ksvc.yaml @@ -0,0 +1,85 @@ +apiVersion: serving.knative.dev/v1 +kind: Service +metadata: + annotations: + function.knative.dev/deployer: knative + serving.knative.dev/creator: kubernetes-admin + serving.knative.dev/lastModifier: kubernetes-admin + creationTimestamp: "2026-09-03T12:38:46Z" + generation: 1 + labels: + boson.dev/function: "true" + function.knative.dev/name: test-knative-no-kafka + function.knative.dev/runtime: go + name: test-knative-no-kafka + namespace: default + resourceVersion: "1724" + uid: d421a51b-1edc-4007-a131-211bed172476 +spec: + template: + metadata: + annotations: + autoscaling.knative.dev/max-scale: "5" + autoscaling.knative.dev/metric: concurrency + autoscaling.knative.dev/min-scale: "0" + autoscaling.knative.dev/target: "100.000000" + autoscaling.knative.dev/target-utilization-percentage: "70.000000" + function.knative.dev/deployer: knative + labels: + boson.dev/function: "true" + function.knative.dev/name: test-knative-no-kafka + function.knative.dev/runtime: go + spec: + containerConcurrency: 0 + containers: + - env: + - name: BUILT + value: 20260903T153846 + - name: ADDRESS + value: 0.0.0.0 + image: index.docker.io/aliok/test-knative-no-kafka@sha256:ec7977a4995f8a8335a3aee3cfd2c7fab714dfd14bb275b66fa523dcbbfd9ce6 + livenessProbe: + httpGet: + path: /health/liveness + port: 8080 + name: user-container + readinessProbe: + httpGet: + path: /health/readiness + port: 8080 + successThreshold: 1 + resources: {} + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: + - ALL + runAsNonRoot: true + seccompProfile: + type: RuntimeDefault + enableServiceLinks: false + timeoutSeconds: 300 + traffic: + - latestRevision: true + percent: 100 +status: + address: + url: http://test-knative-no-kafka.default.svc.cluster.local + conditions: + - lastTransitionTime: "2026-09-03T12:38:55Z" + status: "True" + type: ConfigurationsReady + - lastTransitionTime: "2026-09-03T12:38:55Z" + status: "True" + type: Ready + - lastTransitionTime: "2026-09-03T12:38:55Z" + status: "True" + type: RoutesReady + latestCreatedRevisionName: test-knative-no-kafka-00001 + latestReadyRevisionName: test-knative-no-kafka-00001 + observedGeneration: 1 + traffic: + - latestRevision: true + percent: 100 + revisionName: test-knative-no-kafka-00001 + url: http://test-knative-no-kafka.default.svc.cluster.local diff --git a/docs/testing-deployments/c-knative-no-kafka/testing.md b/docs/testing-deployments/c-knative-no-kafka/testing.md new file mode 100644 index 0000000000..4015cb61fb --- /dev/null +++ b/docs/testing-deployments/c-knative-no-kafka/testing.md @@ -0,0 +1,117 @@ +# Scenario C: Knative, no Kafka + +Knative deployer with KPA scaling (using the new `kpa` sub-key). +Creates a Knative Service with autoscaling annotations. + +## Prerequisites + +- kind, kubectl, Go 1.25+, Docker, jq + +If using Colima and you hit DNS issues (image pulls failing, etc.), restart +it with explicit DNS: + +```bash +colima stop +colima start --dns 8.8.8.8 +``` + +## 1. Build the CLI + +```bash +cd ~/go/src/knative.dev/func +go build -o /tmp/func-local ./cmd/func +``` + +## 2. Create cluster and install Knative Serving + +```bash +kind create cluster --name test-c + +kubectl apply -f https://github.com/knative/serving/releases/download/knative-v1.21.2/serving-crds.yaml +kubectl apply -f https://github.com/knative/serving/releases/download/knative-v1.21.2/serving-core.yaml +kubectl apply -f https://github.com/knative/net-kourier/releases/download/knative-v1.21.1/kourier.yaml +kubectl patch configmap/config-network -n knative-serving \ + --type merge -p '{"data":{"ingress.class":"kourier.ingress.networking.knative.dev"}}' +kubectl wait deployment --all -n knative-serving --for=condition=Available --timeout=120s +``` + +## 3. Create and configure the function + +```bash +mkdir /tmp/test-knative-no-kafka && cd /tmp/test-knative-no-kafka +/tmp/func-local create -l go -t cloudevents +``` + +Replace the contents of `func.yaml` (keep the `created` line from the generated file): + +```yaml +created: +specVersion: 0.37.0 +name: test-knative-no-kafka +runtime: go +registry: docker.io/aliok +deployer: knative +invoke: cloudevent +deploy: + options: + scale: + min: 0 + max: 5 + kpa: + metric: concurrency + target: 100 + utilization: 70 +``` + +## 4. Deploy + +```bash +cd /tmp/test-knative-no-kafka +FUNC_REGISTRY=docker.io/aliok /tmp/func-local deploy --verbose +``` + +## 5. Verify + +```bash +# Should exist: Knative Service (ksvc) +kubectl get ksvc test-knative-no-kafka + +# Check KPA annotations +kubectl get ksvc test-knative-no-kafka -o json \ + | jq '.spec.template.metadata.annotations | { + "autoscaling.knative.dev/minScale", + "autoscaling.knative.dev/maxScale", + "autoscaling.knative.dev/metric", + "autoscaling.knative.dev/target", + "autoscaling.knative.dev/target-utilization-percentage" + }' + +# Should NOT exist: no KEDA resources +kubectl get httpscaledobject test-knative-no-kafka 2>&1 || true +kubectl get scaledobject test-knative-no-kafka-kafka 2>&1 || true + +# No Kafka env vars +kubectl get ksvc test-knative-no-kafka -o json \ + | jq '.spec.template.spec.containers[0].env[]? | select(.name | startswith("KAFKA"))' +# Should print nothing +``` + +## 6. Collect resources + +```bash +kubectl get ksvc test-knative-no-kafka -o yaml > /tmp/test-knative-no-kafka/ksvc.yaml +``` + +## 7. Delete function + +```bash +cd /tmp/test-knative-no-kafka +/tmp/func-local delete +``` + +## 8. Cleanup + +```bash +kind delete cluster --name test-c +rm -rf /tmp/test-knative-no-kafka +``` diff --git a/docs/testing-deployments/d-knative-with-kafka/func.yaml b/docs/testing-deployments/d-knative-with-kafka/func.yaml new file mode 100644 index 0000000000..d06b7400b9 --- /dev/null +++ b/docs/testing-deployments/d-knative-with-kafka/func.yaml @@ -0,0 +1,38 @@ +# $schema: https://raw.githubusercontent.com/knative/func/main/schema/func_yaml-schema.json +# yaml-language-server: $schema=https://raw.githubusercontent.com/knative/func/main/schema/func_yaml-schema.json +specVersion: 0.37.0 +name: test-knative-kafka +runtime: go +registry: docker.io/aliok +namespace: default +deployer: knative +created: 2026-09-03T16:13:13.783633+03:00 +invoke: cloudevent +build: + builder: pack +run: + volumes: + - secret: my-cluster-cluster-ca-cert + path: /etc/kafka/ca + kafka: + brokers: my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9093 + topic: test-topic + consumerGroup: test-knative-kafka-group + securityProtocol: SASL_SSL + tls: + caCert: /etc/kafka/ca/ca.crt + sasl: + mechanism: SCRAM-SHA-512 + user: my-kafka-user + password: '{{ secret:my-kafka-user:password }}' +deploy: + namespace: default + image: index.docker.io/aliok/test-knative-kafka@sha256:ec7977a4995f8a8335a3aee3cfd2c7fab714dfd14bb275b66fa523dcbbfd9ce6 + options: + scale: + min: 1 + max: 10 + kpa: + metric: concurrency + target: 50 + deployer: knative diff --git a/docs/testing-deployments/d-knative-with-kafka/ksvc.yaml b/docs/testing-deployments/d-knative-with-kafka/ksvc.yaml new file mode 100644 index 0000000000..378c6fb8e2 --- /dev/null +++ b/docs/testing-deployments/d-knative-with-kafka/ksvc.yaml @@ -0,0 +1,113 @@ +apiVersion: serving.knative.dev/v1 +kind: Service +metadata: + annotations: + function.knative.dev/deployer: knative + serving.knative.dev/creator: kubernetes-admin + serving.knative.dev/lastModifier: kubernetes-admin + creationTimestamp: "2026-09-03T13:14:55Z" + generation: 1 + labels: + boson.dev/function: "true" + function.knative.dev/name: test-knative-kafka + function.knative.dev/runtime: go + name: test-knative-kafka + namespace: default + resourceVersion: "3396" + uid: 568f2812-4b0c-4df6-90e9-e2379e31ea41 +spec: + template: + metadata: + annotations: + autoscaling.knative.dev/max-scale: "10" + autoscaling.knative.dev/metric: concurrency + autoscaling.knative.dev/min-scale: "1" + autoscaling.knative.dev/target: "50.000000" + function.knative.dev/deployer: knative + labels: + boson.dev/function: "true" + function.knative.dev/name: test-knative-kafka + function.knative.dev/runtime: go + spec: + containerConcurrency: 0 + containers: + - env: + - name: BUILT + value: 20260903T161455 + - name: ADDRESS + value: 0.0.0.0 + - name: FUNC_TRANSPORT + value: kafka + - name: KAFKA_BROKERS + value: my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9093 + - name: KAFKA_TOPIC + value: test-topic + - name: KAFKA_CONSUMER_GROUP + value: test-knative-kafka-group + - name: KAFKA_SECURITY_PROTOCOL + value: SASL_SSL + - name: KAFKA_TLS_CA_CERT + value: /etc/kafka/ca/ca.crt + - name: KAFKA_SASL_MECHANISM + value: SCRAM-SHA-512 + - name: KAFKA_SASL_USER + value: my-kafka-user + - name: KAFKA_SASL_PASSWORD + valueFrom: + secretKeyRef: + key: password + name: my-kafka-user + image: index.docker.io/aliok/test-knative-kafka@sha256:ec7977a4995f8a8335a3aee3cfd2c7fab714dfd14bb275b66fa523dcbbfd9ce6 + livenessProbe: + httpGet: + path: /health/liveness + port: 8080 + name: user-container + readinessProbe: + httpGet: + path: /health/readiness + port: 8080 + successThreshold: 1 + resources: {} + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: + - ALL + runAsNonRoot: true + seccompProfile: + type: RuntimeDefault + volumeMounts: + - mountPath: /etc/kafka/ca + name: secret-my-cluster-cluster-ca-cert + readOnly: true + enableServiceLinks: false + timeoutSeconds: 300 + volumes: + - name: secret-my-cluster-cluster-ca-cert + secret: + secretName: my-cluster-cluster-ca-cert + traffic: + - latestRevision: true + percent: 100 +status: + address: + url: http://test-knative-kafka.default.svc.cluster.local + conditions: + - lastTransitionTime: "2026-09-03T13:15:06Z" + status: "True" + type: ConfigurationsReady + - lastTransitionTime: "2026-09-03T13:15:06Z" + status: "True" + type: Ready + - lastTransitionTime: "2026-09-03T13:15:06Z" + status: "True" + type: RoutesReady + latestCreatedRevisionName: test-knative-kafka-00001 + latestReadyRevisionName: test-knative-kafka-00001 + observedGeneration: 1 + traffic: + - latestRevision: true + percent: 100 + revisionName: test-knative-kafka-00001 + url: http://test-knative-kafka.default.svc.cluster.local diff --git a/docs/testing-deployments/d-knative-with-kafka/testing.md b/docs/testing-deployments/d-knative-with-kafka/testing.md new file mode 100644 index 0000000000..4bab3e94d1 --- /dev/null +++ b/docs/testing-deployments/d-knative-with-kafka/testing.md @@ -0,0 +1,239 @@ +# Scenario D: Knative, with Kafka + +Knative deployer with Kafka (SASL_SSL) and KPA scaling. +Creates a Knative Service with Kafka env vars/volumes and autoscaling +annotations. No KEDA resources. + +## Prerequisites + +- kind, kubectl, Go 1.25+, Docker, jq + +If using Colima and you hit DNS issues (image pulls failing, etc.), restart +it with explicit DNS: + +```bash +colima stop +colima start --dns 8.8.8.8 +``` + +## 1. Build the CLI + +```bash +cd ~/go/src/knative.dev/func +go build -o /tmp/func-local ./cmd/func +``` + +## 2. Create cluster and install Knative Serving + +```bash +kind create cluster --name test-d + +kubectl apply -f https://github.com/knative/serving/releases/download/knative-v1.21.2/serving-crds.yaml +kubectl apply -f https://github.com/knative/serving/releases/download/knative-v1.21.2/serving-core.yaml +kubectl apply -f https://github.com/knative/net-kourier/releases/download/knative-v1.21.1/kourier.yaml +kubectl patch configmap/config-network -n knative-serving \ + --type merge -p '{"data":{"ingress.class":"kourier.ingress.networking.knative.dev"}}' +kubectl wait deployment --all -n knative-serving --for=condition=Available --timeout=120s +``` + +## 3. Install Strimzi and create Kafka cluster + user + topic + +```bash +kubectl create namespace kafka +kubectl apply -f 'https://strimzi.io/install/latest?namespace=kafka' -n kafka +kubectl wait --for=condition=Ready pods --all -n kafka --timeout=120s + +kubectl apply -n kafka -f - <<'EOF' +apiVersion: kafka.strimzi.io/v1 +kind: KafkaNodePool +metadata: + name: dual-role + labels: + strimzi.io/cluster: my-cluster +spec: + replicas: 1 + roles: [controller, broker] + storage: + type: jbod + volumes: + - id: 0 + type: persistent-claim + size: 1Gi + deleteClaim: true +--- +apiVersion: kafka.strimzi.io/v1 +kind: Kafka +metadata: + name: my-cluster + annotations: + strimzi.io/node-pools: enabled + strimzi.io/kraft: enabled +spec: + kafka: + version: 4.2.0 + authorization: + type: simple + superUsers: [ANONYMOUS] + listeners: + - name: plain + port: 9092 + type: internal + tls: false + - name: tls + port: 9093 + type: internal + tls: true + authentication: + type: scram-sha-512 + config: + offsets.topic.replication.factor: 1 + transaction.state.log.replication.factor: 1 + transaction.state.log.min.isr: 1 + entityOperator: + topicOperator: {} + userOperator: {} +EOF + +kubectl wait kafka/my-cluster --for=condition=Ready --timeout=300s -n kafka + +kubectl apply -n kafka -f - <<'EOF' +apiVersion: kafka.strimzi.io/v1 +kind: KafkaUser +metadata: + name: my-kafka-user + labels: + strimzi.io/cluster: my-cluster +spec: + authentication: + type: scram-sha-512 + authorization: + type: simple + acls: + - resource: { type: topic, name: test-topic, patternType: literal } + operations: [Read, Describe] + host: "*" + - resource: { type: group, name: "*", patternType: literal } + operations: [Read] + host: "*" +--- +apiVersion: kafka.strimzi.io/v1 +kind: KafkaTopic +metadata: + name: test-topic + labels: + strimzi.io/cluster: my-cluster +spec: + partitions: 3 + replicas: 1 +EOF + +kubectl wait kafkauser/my-kafka-user --for=condition=Ready --timeout=60s -n kafka +``` + +## 4. Copy secrets to default namespace + +```bash +kubectl get secret my-cluster-cluster-ca-cert -n kafka -o json \ + | jq 'del(.metadata.namespace,.metadata.resourceVersion,.metadata.uid,.metadata.creationTimestamp,.metadata.ownerReferences)' \ + | kubectl apply -n default -f - + +kubectl get secret my-kafka-user -n kafka -o json \ + | jq 'del(.metadata.namespace,.metadata.resourceVersion,.metadata.uid,.metadata.creationTimestamp,.metadata.ownerReferences)' \ + | kubectl apply -n default -f - +``` + +## 5. Create and configure the function + +```bash +mkdir /tmp/test-knative-kafka && cd /tmp/test-knative-kafka +/tmp/func-local create -l go -t cloudevents +``` + +Replace the contents of `func.yaml` (keep the `created` line from the generated file): + +```yaml +created: +specVersion: 0.37.0 +name: test-knative-kafka +runtime: go +registry: docker.io/aliok +deployer: knative +invoke: cloudevent +deploy: + options: + scale: + min: 1 + max: 10 + kpa: + metric: concurrency + target: 50 +run: + kafka: + brokers: "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9093" + topic: "test-topic" + consumerGroup: "test-knative-kafka-group" + securityProtocol: "SASL_SSL" + tls: + caCert: "/etc/kafka/ca/ca.crt" + sasl: + mechanism: "SCRAM-SHA-512" + user: "my-kafka-user" + password: "{{ secret:my-kafka-user:password }}" + volumes: + - secret: my-cluster-cluster-ca-cert + path: /etc/kafka/ca +``` + +## 6. Deploy + +```bash +cd /tmp/test-knative-kafka +FUNC_REGISTRY=docker.io/aliok /tmp/func-local deploy --verbose +``` + +## 7. Verify + +```bash +# Should exist: Knative Service +kubectl get ksvc test-knative-kafka + +# KPA annotations +kubectl get ksvc test-knative-kafka -o json \ + | jq '.spec.template.metadata.annotations | { + "autoscaling.knative.dev/minScale", + "autoscaling.knative.dev/maxScale", + "autoscaling.knative.dev/metric", + "autoscaling.knative.dev/target" + }' + +# Kafka env vars should be present +kubectl get ksvc test-knative-kafka -o json \ + | jq '.spec.template.spec.containers[0].env[] | select(.name | startswith("KAFKA") or . == "FUNC_TRANSPORT")' + +# Volume mount for CA cert +kubectl get ksvc test-knative-kafka -o json \ + | jq '.spec.template.spec.containers[0].volumeMounts' + +# Should NOT exist: no KEDA resources (Knative deployer) +kubectl get scaledobject test-knative-kafka-kafka 2>&1 || true +``` + +## 8. Collect resources + +```bash +kubectl get ksvc test-knative-kafka -o yaml > /tmp/test-knative-kafka/ksvc.yaml +``` + +## 9. Delete function + +```bash +cd /tmp/test-knative-kafka +/tmp/func-local delete +``` + +## 10. Cleanup + +```bash +kind delete cluster --name test-d +rm -rf /tmp/test-knative-kafka +``` diff --git a/docs/testing-deployments/e-keda-no-kafka/bridge-service.yaml b/docs/testing-deployments/e-keda-no-kafka/bridge-service.yaml new file mode 100644 index 0000000000..b417a41cf9 --- /dev/null +++ b/docs/testing-deployments/e-keda-no-kafka/bridge-service.yaml @@ -0,0 +1,20 @@ +apiVersion: v1 +kind: Service +metadata: + creationTimestamp: "2026-09-03T13:28:17Z" + name: test-keda-no-kafka-interceptor-bridge + namespace: default + ownerReferences: + - apiVersion: apps/v1 + controller: true + kind: Deployment + name: test-keda-no-kafka + uid: e479851c-a037-4376-b546-99168f2fa679 + resourceVersion: "1290" + uid: d1a5eea4-1cfd-4969-ad31-845be4aa4abe +spec: + externalName: keda-add-ons-http-interceptor-proxy.keda.svc.cluster.local + sessionAffinity: None + type: ExternalName +status: + loadBalancer: {} diff --git a/docs/testing-deployments/e-keda-no-kafka/deployment.yaml b/docs/testing-deployments/e-keda-no-kafka/deployment.yaml new file mode 100644 index 0000000000..04452b6f9f --- /dev/null +++ b/docs/testing-deployments/e-keda-no-kafka/deployment.yaml @@ -0,0 +1,105 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + annotations: + deployment.kubernetes.io/revision: "2" + function.knative.dev/deployer: keda + creationTimestamp: "2026-09-03T13:28:11Z" + generation: 2 + labels: + boson.dev/function: "true" + function.knative.dev/name: test-keda-no-kafka + function.knative.dev/runtime: go + name: test-keda-no-kafka + namespace: default + resourceVersion: "2669" + uid: e479851c-a037-4376-b546-99168f2fa679 +spec: + progressDeadlineSeconds: 600 + replicas: 1 + revisionHistoryLimit: 10 + selector: + matchLabels: + boson.dev/function: "true" + function.knative.dev/name: test-keda-no-kafka + function.knative.dev/runtime: go + strategy: + rollingUpdate: + maxSurge: 25% + maxUnavailable: 25% + type: RollingUpdate + template: + metadata: + annotations: + function.knative.dev/deployer: keda + labels: + boson.dev/function: "true" + function.knative.dev/name: test-keda-no-kafka + function.knative.dev/runtime: go + spec: + containers: + - env: + - name: BUILT + value: 20260903T163748 + - name: ADDRESS + value: 0.0.0.0 + image: index.docker.io/aliok/test-keda-no-kafka@sha256:ec7977a4995f8a8335a3aee3cfd2c7fab714dfd14bb275b66fa523dcbbfd9ce6 + imagePullPolicy: IfNotPresent + livenessProbe: + failureThreshold: 3 + httpGet: + path: /health/liveness + port: 8080 + scheme: HTTP + periodSeconds: 10 + successThreshold: 1 + timeoutSeconds: 1 + name: user-container + ports: + - containerPort: 8080 + protocol: TCP + readinessProbe: + failureThreshold: 3 + httpGet: + path: /health/readiness + port: 8080 + scheme: HTTP + periodSeconds: 10 + successThreshold: 1 + timeoutSeconds: 1 + resources: {} + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: + - ALL + runAsNonRoot: true + seccompProfile: + type: RuntimeDefault + terminationMessagePath: /dev/termination-log + terminationMessagePolicy: File + dnsPolicy: ClusterFirst + restartPolicy: Always + schedulerName: default-scheduler + securityContext: {} + terminationGracePeriodSeconds: 30 +status: + availableReplicas: 1 + conditions: + - lastTransitionTime: "2026-09-03T13:28:16Z" + lastUpdateTime: "2026-09-03T13:28:16Z" + message: Deployment has minimum availability. + reason: MinimumReplicasAvailable + status: "True" + type: Available + - lastTransitionTime: "2026-09-03T13:28:11Z" + lastUpdateTime: "2026-09-03T13:37:49Z" + message: ReplicaSet "test-keda-no-kafka-58bf6cc4f5" has successfully progressed. + reason: NewReplicaSetAvailable + status: "True" + type: Progressing + observedGeneration: 2 + readyReplicas: 1 + replicas: 1 + terminatingReplicas: 0 + updatedReplicas: 1 diff --git a/docs/testing-deployments/e-keda-no-kafka/func.yaml b/docs/testing-deployments/e-keda-no-kafka/func.yaml new file mode 100644 index 0000000000..f920c4ea93 --- /dev/null +++ b/docs/testing-deployments/e-keda-no-kafka/func.yaml @@ -0,0 +1,23 @@ +# $schema: https://raw.githubusercontent.com/knative/func/main/schema/func_yaml-schema.json +# yaml-language-server: $schema=https://raw.githubusercontent.com/knative/func/main/schema/func_yaml-schema.json +specVersion: 0.37.0 +name: test-keda-no-kafka +runtime: go +registry: docker.io/aliok +namespace: default +deployer: keda +created: 2026-09-03T16:26:53.083146+03:00 +invoke: cloudevent +build: + builder: pack +deploy: + namespace: default + image: index.docker.io/aliok/test-keda-no-kafka@sha256:ec7977a4995f8a8335a3aee3cfd2c7fab714dfd14bb275b66fa523dcbbfd9ce6 + options: + scale: + min: 1 + max: 10 + keda: + triggers: + - type: http + deployer: keda diff --git a/docs/testing-deployments/e-keda-no-kafka/httpscaledobject.yaml b/docs/testing-deployments/e-keda-no-kafka/httpscaledobject.yaml new file mode 100644 index 0000000000..6e00c79bc9 --- /dev/null +++ b/docs/testing-deployments/e-keda-no-kafka/httpscaledobject.yaml @@ -0,0 +1,52 @@ +apiVersion: http.keda.sh/v1alpha1 +kind: HTTPScaledObject +metadata: + annotations: + function.knative.dev/deployer: keda + creationTimestamp: "2026-09-03T13:28:17Z" + finalizers: + - httpscaledobject.http.keda.sh + generation: 1 + labels: + boson.dev/function: "true" + function.knative.dev/name: test-keda-no-kafka + function.knative.dev/runtime: go + name: test-keda-no-kafka + namespace: default + ownerReferences: + - apiVersion: apps/v1 + controller: true + kind: Deployment + name: test-keda-no-kafka + uid: e479851c-a037-4376-b546-99168f2fa679 + resourceVersion: "1294" + uid: b7ad5cf8-20c0-4cd9-86b6-c94b9bce6d18 +spec: + hosts: + - test-keda-no-kafka-interceptor-bridge.default.svc + - test-keda-no-kafka-interceptor-bridge + replicas: + max: 10 + min: 1 + scaleTargetRef: + apiVersion: apps/v1 + kind: Deployment + name: test-keda-no-kafka + port: 80 + service: test-keda-no-kafka + scaledownPeriod: 300 + scalingMetric: + requestRate: + granularity: 1s + targetValue: 100 + window: 1m0s +status: + conditions: + - lastTransitionTime: "2026-09-03T13:28:17Z" + message: ScaledObject created and configured + observedGeneration: 1 + reason: Reconciled + status: "True" + type: Ready + targetService: test-keda-no-kafka:80 + targetWorkload: apps/v1/Deployment/test-keda-no-kafka diff --git a/docs/testing-deployments/e-keda-no-kafka/service.yaml b/docs/testing-deployments/e-keda-no-kafka/service.yaml new file mode 100644 index 0000000000..480ceb88d7 --- /dev/null +++ b/docs/testing-deployments/e-keda-no-kafka/service.yaml @@ -0,0 +1,41 @@ +apiVersion: v1 +kind: Service +metadata: + annotations: + function.knative.dev/deployer: keda + creationTimestamp: "2026-09-03T13:28:11Z" + labels: + boson.dev/function: "true" + function.knative.dev/name: test-keda-no-kafka + function.knative.dev/runtime: go + name: test-keda-no-kafka + namespace: default + ownerReferences: + - apiVersion: apps/v1 + controller: true + kind: Deployment + name: test-keda-no-kafka + uid: e479851c-a037-4376-b546-99168f2fa679 + resourceVersion: "1251" + uid: 4678bb2b-ba71-45b8-a0c7-3d51f1f458c7 +spec: + clusterIP: 10.96.195.54 + clusterIPs: + - 10.96.195.54 + internalTrafficPolicy: Cluster + ipFamilies: + - IPv4 + ipFamilyPolicy: SingleStack + ports: + - name: http + port: 80 + protocol: TCP + targetPort: 8080 + selector: + boson.dev/function: "true" + function.knative.dev/name: test-keda-no-kafka + function.knative.dev/runtime: go + sessionAffinity: None + type: ClusterIP +status: + loadBalancer: {} diff --git a/docs/testing-deployments/e-keda-no-kafka/testing.md b/docs/testing-deployments/e-keda-no-kafka/testing.md new file mode 100644 index 0000000000..e51ebdec42 --- /dev/null +++ b/docs/testing-deployments/e-keda-no-kafka/testing.md @@ -0,0 +1,118 @@ +# Scenario E: KEDA, no Kafka + +KEDA deployer, HTTP-only (same as today's default behavior). +Creates Deployment + Service + bridge Service + HTTPScaledObject. +No Kafka resources. + +## Prerequisites + +- kind, kubectl, Go 1.25+, Docker, jq + +If using Colima and you hit DNS issues (image pulls failing, etc.), restart +it with explicit DNS: + +```bash +colima stop +colima start --dns 8.8.8.8 +``` + +## 1. Build the CLI + +```bash +cd ~/go/src/knative.dev/func +go build -o /tmp/func-local ./cmd/func +``` + +## 2. Create cluster and install KEDA + +```bash +kind create cluster --name test-e + +kubectl apply --server-side -f https://github.com/kedacore/keda/releases/download/v2.17.0/keda-2.17.0.yaml +kubectl apply --server-side -f https://github.com/kedacore/keda/releases/download/v2.17.0/keda-2.17.0-core.yaml +kubectl wait deployment --all -n keda --for=condition=Available --timeout=120s + +kubectl apply --server-side -f https://github.com/kedacore/http-add-on/releases/download/v0.12.0/keda-add-ons-http-0.12.0-crds.yaml +kubectl apply --server-side -f https://github.com/kedacore/http-add-on/releases/download/v0.12.0/keda-add-ons-http-0.12.0.yaml +kubectl wait deployment --all -n keda --for=condition=Available --timeout=120s +``` + +## 3. Create and configure the function + +```bash +mkdir /tmp/test-keda-no-kafka && cd /tmp/test-keda-no-kafka +/tmp/func-local create -l go -t cloudevents +``` + +Replace the contents of `func.yaml` (keep the `created` line from the generated file): + +```yaml +created: +specVersion: 0.37.0 +name: test-keda-no-kafka +runtime: go +registry: docker.io/aliok +deployer: keda +invoke: cloudevent +deploy: + options: + scale: + min: 1 + max: 10 + keda: + triggers: + - type: http +``` + +## 4. Deploy + +```bash +cd /tmp/test-keda-no-kafka +FUNC_REGISTRY=docker.io/aliok /tmp/func-local deploy --verbose +``` + +## 5. Verify + +```bash +# Should exist: Deployment, Service, bridge Service, HTTPScaledObject +kubectl get deployment test-keda-no-kafka +kubectl get svc test-keda-no-kafka +kubectl get svc test-keda-no-kafka-interceptor-proxy +kubectl get httpscaledobject test-keda-no-kafka + +# HTTPScaledObject details +kubectl get httpscaledobject test-keda-no-kafka -o json \ + | jq '{hosts: .spec.hosts, replicas: .spec.replicas, scaleTargetRef: .spec.scaleTargetRef}' + +# Should NOT exist: no Kafka ScaledObject or TriggerAuthentication +kubectl get scaledobject test-keda-no-kafka-kafka 2>&1 || true +kubectl get triggerauthentication test-keda-no-kafka-kafka-auth 2>&1 || true + +# No Kafka env vars +kubectl get deployment test-keda-no-kafka -o json \ + | jq '.spec.template.spec.containers[0].env[]? | select(.name | startswith("KAFKA"))' +# Should print nothing +``` + +## 6. Collect resources + +```bash +kubectl get deployment test-keda-no-kafka -o yaml > /tmp/test-keda-no-kafka/deployment.yaml +kubectl get svc test-keda-no-kafka -o yaml > /tmp/test-keda-no-kafka/service.yaml +kubectl get svc test-keda-no-kafka-interceptor-proxy -o yaml > /tmp/test-keda-no-kafka/bridge-service.yaml +kubectl get httpscaledobject test-keda-no-kafka -o yaml > /tmp/test-keda-no-kafka/httpscaledobject.yaml +``` + +## 7. Delete function + +```bash +cd /tmp/test-keda-no-kafka +/tmp/func-local delete +``` + +## 8. Cleanup + +```bash +kind delete cluster --name test-e +rm -rf /tmp/test-keda-no-kafka +``` diff --git a/docs/testing-deployments/f-keda-with-kafka/deployment.yaml b/docs/testing-deployments/f-keda-with-kafka/deployment.yaml new file mode 100644 index 0000000000..40db506182 --- /dev/null +++ b/docs/testing-deployments/f-keda-with-kafka/deployment.yaml @@ -0,0 +1,134 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + annotations: + deployment.kubernetes.io/revision: "1" + function.knative.dev/deployer: keda + creationTimestamp: "2026-09-03T15:04:48Z" + generation: 4 + labels: + boson.dev/function: "true" + function.knative.dev/name: test-keda-kafka + function.knative.dev/runtime: go + name: test-keda-kafka + namespace: default + resourceVersion: "12157" + uid: 12d64f85-4bda-4d4a-8568-f45eb47628db +spec: + progressDeadlineSeconds: 600 + replicas: 3 + revisionHistoryLimit: 10 + selector: + matchLabels: + boson.dev/function: "true" + function.knative.dev/name: test-keda-kafka + function.knative.dev/runtime: go + strategy: + rollingUpdate: + maxSurge: 25% + maxUnavailable: 25% + type: RollingUpdate + template: + metadata: + annotations: + function.knative.dev/deployer: keda + labels: + boson.dev/function: "true" + function.knative.dev/name: test-keda-kafka + function.knative.dev/runtime: go + spec: + containers: + - env: + - name: BUILT + value: 20260903T180448 + - name: ADDRESS + value: 0.0.0.0 + - name: FUNC_TRANSPORT + value: kafka + - name: KAFKA_BROKERS + value: my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9093 + - name: KAFKA_TOPIC + value: test-topic + - name: KAFKA_CONSUMER_GROUP + value: test-keda-kafka-group + - name: KAFKA_SECURITY_PROTOCOL + value: SASL_SSL + - name: KAFKA_TLS_CA_CERT + value: /etc/kafka/ca/ca.crt + - name: KAFKA_SASL_MECHANISM + value: SCRAM-SHA-512 + - name: KAFKA_SASL_USER + value: my-kafka-user + - name: KAFKA_SASL_PASSWORD + valueFrom: + secretKeyRef: + key: password + name: my-kafka-user + image: index.docker.io/aliok/test-keda-kafka@sha256:9a40b49d3f87baf3efe03bbdf6cc1d4b85276cc650e438231050985048ef400e + imagePullPolicy: IfNotPresent + livenessProbe: + failureThreshold: 3 + httpGet: + path: /health/liveness + port: 8080 + scheme: HTTP + periodSeconds: 10 + successThreshold: 1 + timeoutSeconds: 1 + name: user-container + ports: + - containerPort: 8080 + protocol: TCP + readinessProbe: + failureThreshold: 3 + httpGet: + path: /health/readiness + port: 8080 + scheme: HTTP + periodSeconds: 10 + successThreshold: 1 + timeoutSeconds: 1 + resources: {} + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: + - ALL + runAsNonRoot: true + seccompProfile: + type: RuntimeDefault + terminationMessagePath: /dev/termination-log + terminationMessagePolicy: File + volumeMounts: + - mountPath: /etc/kafka/ca + name: secret-my-cluster-cluster-ca-cert + dnsPolicy: ClusterFirst + restartPolicy: Always + schedulerName: default-scheduler + securityContext: {} + terminationGracePeriodSeconds: 30 + volumes: + - name: secret-my-cluster-cluster-ca-cert + secret: + defaultMode: 420 + secretName: my-cluster-cluster-ca-cert +status: + availableReplicas: 3 + conditions: + - lastTransitionTime: "2026-09-03T15:04:48Z" + lastUpdateTime: "2026-09-03T15:05:05Z" + message: ReplicaSet "test-keda-kafka-67f4cbb4fd" has successfully progressed. + reason: NewReplicaSetAvailable + status: "True" + type: Progressing + - lastTransitionTime: "2026-09-03T15:06:03Z" + lastUpdateTime: "2026-09-03T15:06:03Z" + message: Deployment has minimum availability. + reason: MinimumReplicasAvailable + status: "True" + type: Available + observedGeneration: 4 + readyReplicas: 3 + replicas: 3 + terminatingReplicas: 0 + updatedReplicas: 3 diff --git a/docs/testing-deployments/f-keda-with-kafka/func.yaml b/docs/testing-deployments/f-keda-with-kafka/func.yaml new file mode 100644 index 0000000000..d4242ce60e --- /dev/null +++ b/docs/testing-deployments/f-keda-with-kafka/func.yaml @@ -0,0 +1,40 @@ +# $schema: https://raw.githubusercontent.com/knative/func/main/schema/func_yaml-schema.json +# yaml-language-server: $schema=https://raw.githubusercontent.com/knative/func/main/schema/func_yaml-schema.json +specVersion: 0.37.0 +name: test-keda-kafka +runtime: go +registry: docker.io/aliok +namespace: default +deployer: keda +created: 2026-09-03T17:12:41.109216+03:00 +invoke: cloudevent +build: + builder: pack +run: + volumes: + - secret: my-cluster-cluster-ca-cert + path: /etc/kafka/ca + kafka: + brokers: my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9093 + topic: test-topic + consumerGroup: test-keda-kafka-group + securityProtocol: SASL_SSL + tls: + caCert: /etc/kafka/ca/ca.crt + sasl: + mechanism: SCRAM-SHA-512 + user: my-kafka-user + password: '{{ secret:my-kafka-user:password }}' +deploy: + namespace: default + image: index.docker.io/aliok/test-keda-kafka@sha256:9a40b49d3f87baf3efe03bbdf6cc1d4b85276cc650e438231050985048ef400e + options: + scale: + min: 0 + max: 10 + keda: + triggers: + - type: kafka + lagThreshold: 5 + activationLagThreshold: 0 + deployer: keda diff --git a/docs/testing-deployments/f-keda-with-kafka/scaledobject.yaml b/docs/testing-deployments/f-keda-with-kafka/scaledobject.yaml new file mode 100644 index 0000000000..630173242e --- /dev/null +++ b/docs/testing-deployments/f-keda-with-kafka/scaledobject.yaml @@ -0,0 +1,68 @@ +apiVersion: keda.sh/v1alpha1 +kind: ScaledObject +metadata: + creationTimestamp: "2026-09-03T15:05:06Z" + finalizers: + - finalizer.keda.sh + generation: 1 + labels: + scaledobject.keda.sh/name: test-keda-kafka-kafka + name: test-keda-kafka-kafka + namespace: default + ownerReferences: + - apiVersion: apps/v1 + blockOwnerDeletion: true + controller: true + kind: Deployment + name: test-keda-kafka + uid: 12d64f85-4bda-4d4a-8568-f45eb47628db + resourceVersion: "13494" + uid: e1f5fd03-431f-4f92-9b01-b24a6cb5e927 +spec: + cooldownPeriod: 300 + maxReplicaCount: 10 + minReplicaCount: 0 + scaleTargetRef: + kind: Deployment + name: test-keda-kafka + triggers: + - authenticationRef: + name: test-keda-kafka-kafka-auth + metadata: + activationLagThreshold: "0" + bootstrapServers: my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9093 + consumerGroup: test-keda-kafka-group + lagThreshold: "5" + sasl: scram_sha512 + tls: enable + topic: test-topic + type: kafka +status: + authenticationsTypes: test-keda-kafka-kafka-auth + conditions: + - message: ScaledObject is defined correctly and is ready for scaling + reason: ScaledObjectReady + status: "True" + type: Ready + - message: Scaling is performed because triggers are active + reason: ScalerActive + status: "True" + type: Active + - message: No fallbacks are active on this scaled object + reason: NoFallbackFound + status: "False" + type: Fallback + - status: Unknown + type: Paused + externalMetricNames: + - s0-kafka-test-topic + hpaName: keda-hpa-test-keda-kafka-kafka + lastActiveTime: "2026-09-03T15:13:36Z" + originalReplicaCount: 1 + scaleTargetGVKR: + group: apps + kind: Deployment + resource: deployments + version: v1 + scaleTargetKind: apps/v1.Deployment + triggersTypes: kafka diff --git a/docs/testing-deployments/f-keda-with-kafka/service.yaml b/docs/testing-deployments/f-keda-with-kafka/service.yaml new file mode 100644 index 0000000000..43bdf134b6 --- /dev/null +++ b/docs/testing-deployments/f-keda-with-kafka/service.yaml @@ -0,0 +1,41 @@ +apiVersion: v1 +kind: Service +metadata: + annotations: + function.knative.dev/deployer: keda + creationTimestamp: "2026-09-03T15:04:48Z" + labels: + boson.dev/function: "true" + function.knative.dev/name: test-keda-kafka + function.knative.dev/runtime: go + name: test-keda-kafka + namespace: default + ownerReferences: + - apiVersion: apps/v1 + controller: true + kind: Deployment + name: test-keda-kafka + uid: 12d64f85-4bda-4d4a-8568-f45eb47628db + resourceVersion: "11776" + uid: d19cf6f1-63cb-4f7a-ab8a-d312185b3b6c +spec: + clusterIP: 10.96.238.251 + clusterIPs: + - 10.96.238.251 + internalTrafficPolicy: Cluster + ipFamilies: + - IPv4 + ipFamilyPolicy: SingleStack + ports: + - name: http + port: 80 + protocol: TCP + targetPort: 8080 + selector: + boson.dev/function: "true" + function.knative.dev/name: test-keda-kafka + function.knative.dev/runtime: go + sessionAffinity: None + type: ClusterIP +status: + loadBalancer: {} diff --git a/docs/testing-deployments/f-keda-with-kafka/testing.md b/docs/testing-deployments/f-keda-with-kafka/testing.md new file mode 100644 index 0000000000..57087e7318 --- /dev/null +++ b/docs/testing-deployments/f-keda-with-kafka/testing.md @@ -0,0 +1,305 @@ +# Scenario F: KEDA, with Kafka (Kafka-only trigger, scaling test) + +KEDA deployer with Kafka (SASL_SSL) and explicit Kafka trigger. +Tests that the function scales up when messages are produced to the topic. + +Creates: Deployment, Service, **ScaledObject**, **TriggerAuthentication**. + +## Prerequisites + +- kind, kubectl, Go 1.25+, Docker, jq + +If using Colima and you hit DNS issues (image pulls failing, etc.), restart +it with explicit DNS: + +```bash +colima stop +colima start --dns 8.8.8.8 +``` + +## 1. Build the CLI + +```bash +cd ~/go/src/knative.dev/func +go build -o /tmp/func-local ./cmd/func +``` + +## 2. Create cluster and install KEDA + +```bash +kind create cluster --name test-f + +kubectl apply --server-side -f https://github.com/kedacore/keda/releases/download/v2.17.0/keda-2.17.0.yaml +kubectl apply --server-side -f https://github.com/kedacore/keda/releases/download/v2.17.0/keda-2.17.0-core.yaml +kubectl wait deployment --all -n keda --for=condition=Available --timeout=120s + +kubectl apply --server-side -f https://github.com/kedacore/http-add-on/releases/download/v0.12.0/keda-add-ons-http-0.12.0-crds.yaml +kubectl apply --server-side -f https://github.com/kedacore/http-add-on/releases/download/v0.12.0/keda-add-ons-http-0.12.0.yaml +kubectl wait deployment --all -n keda --for=condition=Available --timeout=120s +``` + +## 3. Install Strimzi and create Kafka cluster + user + topic + +```bash +kubectl create namespace kafka +kubectl apply -f 'https://strimzi.io/install/latest?namespace=kafka' -n kafka +kubectl wait --for=condition=Ready pods --all -n kafka --timeout=120s + +kubectl apply -n kafka -f - <<'EOF' +apiVersion: kafka.strimzi.io/v1 +kind: KafkaNodePool +metadata: + name: dual-role + labels: + strimzi.io/cluster: my-cluster +spec: + replicas: 1 + roles: [controller, broker] + storage: + type: jbod + volumes: + - id: 0 + type: persistent-claim + size: 1Gi + deleteClaim: true +--- +apiVersion: kafka.strimzi.io/v1 +kind: Kafka +metadata: + name: my-cluster + annotations: + strimzi.io/node-pools: enabled + strimzi.io/kraft: enabled +spec: + kafka: + version: 4.2.0 + authorization: + type: simple + superUsers: [ANONYMOUS] + listeners: + - name: plain + port: 9092 + type: internal + tls: false + - name: tls + port: 9093 + type: internal + tls: true + authentication: + type: scram-sha-512 + config: + offsets.topic.replication.factor: 1 + transaction.state.log.replication.factor: 1 + transaction.state.log.min.isr: 1 + entityOperator: + topicOperator: {} + userOperator: {} +EOF + +kubectl wait kafka/my-cluster --for=condition=Ready --timeout=300s -n kafka + +kubectl apply -n kafka -f - <<'EOF' +apiVersion: kafka.strimzi.io/v1 +kind: KafkaUser +metadata: + name: my-kafka-user + labels: + strimzi.io/cluster: my-cluster +spec: + authentication: + type: scram-sha-512 + authorization: + type: simple + acls: + - resource: { type: topic, name: test-topic, patternType: literal } + operations: [Read, Describe] + host: "*" + - resource: { type: group, name: "*", patternType: literal } + operations: [Read] + host: "*" +--- +apiVersion: kafka.strimzi.io/v1 +kind: KafkaTopic +metadata: + name: test-topic + labels: + strimzi.io/cluster: my-cluster +spec: + partitions: 3 + replicas: 1 +EOF + +kubectl wait kafkauser/my-kafka-user --for=condition=Ready --timeout=60s -n kafka +``` + +## 4. Copy secrets to default namespace + +```bash +kubectl get secret my-cluster-cluster-ca-cert -n kafka -o json \ + | jq 'del(.metadata.namespace,.metadata.resourceVersion,.metadata.uid,.metadata.creationTimestamp,.metadata.ownerReferences)' \ + | kubectl apply -n default -f - + +kubectl get secret my-kafka-user -n kafka -o json \ + | jq 'del(.metadata.namespace,.metadata.resourceVersion,.metadata.uid,.metadata.creationTimestamp,.metadata.ownerReferences)' \ + | kubectl apply -n default -f - +``` + +## 5. Create and configure the function + +```bash +mkdir /tmp/test-keda-kafka && cd /tmp/test-keda-kafka +/tmp/func-local create -l go -t cloudevents +``` + +Add a sleep to `function.go` so messages take time to process (otherwise the +consumer is too fast and KEDA never sees lag): + +```go +// In the Handle method, add after the fmt.Println lines: +time.Sleep(5 * time.Second) +``` + +And add `"time"` to the imports. + +Replace the contents of `func.yaml` (keep the `created` line from the generated file): + +```yaml +created: +specVersion: 0.37.0 +name: test-keda-kafka +runtime: go +registry: docker.io/aliok +deployer: keda +invoke: cloudevent +deploy: + options: + scale: + min: 0 + max: 10 + keda: + triggers: + - type: kafka + lagThreshold: 5 + activationLagThreshold: 0 +run: + kafka: + brokers: "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9093" + topic: "test-topic" + consumerGroup: "test-keda-kafka-group" + securityProtocol: "SASL_SSL" + tls: + caCert: "/etc/kafka/ca/ca.crt" + sasl: + mechanism: "SCRAM-SHA-512" + user: "my-kafka-user" + password: "{{ secret:my-kafka-user:password }}" + volumes: + - secret: my-cluster-cluster-ca-cert + path: /etc/kafka/ca +``` + +## 6. Deploy + +```bash +cd /tmp/test-keda-kafka +FUNC_REGISTRY=docker.io/aliok /tmp/func-local deploy --verbose +``` + +## 7. Verify + +```bash +# Should exist: Deployment, Service, ScaledObject, TriggerAuthentication +kubectl get deployment test-keda-kafka +kubectl get svc test-keda-kafka +kubectl get scaledobject test-keda-kafka-kafka +kubectl get triggerauthentication test-keda-kafka-kafka-auth + +# Should NOT exist: no HTTP resources (Kafka-only trigger) +kubectl get svc test-keda-kafka-interceptor-bridge 2>&1 || true +kubectl get httpscaledobject test-keda-kafka 2>&1 || true + +# ScaledObject details +kubectl get scaledobject test-keda-kafka-kafka -o json | jq '{ + scaleTargetRef: .spec.scaleTargetRef, + minReplicaCount: .spec.minReplicaCount, + maxReplicaCount: .spec.maxReplicaCount, + triggers: .spec.triggers +}' + +# TriggerAuthentication details +kubectl get triggerauthentication test-keda-kafka-kafka-auth -o json | jq '{ + secretTargetRef: .spec.secretTargetRef +}' + +# Kafka env vars should be present on the Deployment +kubectl get deployment test-keda-kafka -o json \ + | jq '.spec.template.spec.containers[0].env[] | select(.name | startswith("KAFKA") or . == "FUNC_TRANSPORT")' + +# Owner references should point to the Deployment +kubectl get scaledobject test-keda-kafka-kafka -o json | jq '.metadata.ownerReferences' +kubectl get triggerauthentication test-keda-kafka-kafka-auth -o json | jq '.metadata.ownerReferences' +``` + +## 8. Test Kafka scaling + +With `min: 0` and `lagThreshold: 5`, KEDA will scale from 0 to up to 10 +replicas based on consumer lag. The topic has 3 partitions, so max effective +replicas from Kafka alone is 3 (one consumer per partition). + +```bash +# Check current replicas — should be 0 (scaled to zero, no lag) +kubectl get deployment test-keda-kafka -o jsonpath='{.spec.replicas}' +echo + +# In a separate terminal, watch replicas: +# kubectl get deployment test-keda-kafka -w + +# Flood the topic with 10000 messages (uses the plain listener, no auth needed) +kubectl run kafka-producer -n kafka --rm -i --restart=Never \ + --image=quay.io/strimzi/kafka:latest-kafka-4.2.0 -- \ + bin/kafka-console-producer.sh \ + --bootstrap-server my-cluster-kafka-bootstrap:9092 \ + --topic test-topic \ + <<< "$(for i in $(seq 1 10000); do echo "message-$i"; done)" + +# Wait 30-60 seconds, then check replicas — should have scaled up +kubectl get deployment test-keda-kafka + +# Check consumer lag (the function's consumer group) +kubectl run kafka-lag -n kafka --rm -i --restart=Never \ + --image=quay.io/strimzi/kafka:latest-kafka-4.2.0 -- \ + bin/kafka-consumer-groups.sh \ + --bootstrap-server my-cluster-kafka-bootstrap:9092 \ + --describe --group test-keda-kafka-group +``` + +## 9. Collect resources + +```bash +kubectl get deployment test-keda-kafka -o yaml > /tmp/test-keda-kafka/deployment.yaml +kubectl get svc test-keda-kafka -o yaml > /tmp/test-keda-kafka/service.yaml +kubectl get scaledobject test-keda-kafka-kafka -o yaml > /tmp/test-keda-kafka/scaledobject.yaml +kubectl get triggerauthentication test-keda-kafka-kafka-auth -o yaml > /tmp/test-keda-kafka/triggerauthentication.yaml +``` + +## 10. Delete function + +```bash +cd /tmp/test-keda-kafka +/tmp/func-local delete +``` + +Verify cleanup: + +```bash +kubectl get scaledobject test-keda-kafka-kafka 2>&1 || true +kubectl get triggerauthentication test-keda-kafka-kafka-auth 2>&1 || true +# Both should return "not found" +``` + +## 11. Cleanup + +```bash +kind delete cluster --name test-f +rm -rf /tmp/test-keda-kafka +``` diff --git a/docs/testing-deployments/f-keda-with-kafka/triggerauthentication.yaml b/docs/testing-deployments/f-keda-with-kafka/triggerauthentication.yaml new file mode 100644 index 0000000000..3e2122103f --- /dev/null +++ b/docs/testing-deployments/f-keda-with-kafka/triggerauthentication.yaml @@ -0,0 +1,32 @@ +apiVersion: keda.sh/v1alpha1 +kind: TriggerAuthentication +metadata: + creationTimestamp: "2026-09-03T15:05:06Z" + finalizers: + - finalizer.keda.sh + generation: 1 + name: test-keda-kafka-kafka-auth + namespace: default + ownerReferences: + - apiVersion: apps/v1 + blockOwnerDeletion: true + controller: true + kind: Deployment + name: test-keda-kafka + uid: 12d64f85-4bda-4d4a-8568-f45eb47628db + resourceVersion: "11871" + uid: 8796adfc-3360-432e-ac05-2df90d480e8b +spec: + env: + - containerName: user-container + name: KAFKA_SASL_USER + parameter: username + secretTargetRef: + - key: password + name: my-kafka-user + parameter: password + - key: ca.crt + name: my-cluster-cluster-ca-cert + parameter: ca +status: + scaledobjects: test-keda-kafka-kafka diff --git a/docs/testing-deployments/g-migration/func.yaml b/docs/testing-deployments/g-migration/func.yaml new file mode 100644 index 0000000000..177035ad68 --- /dev/null +++ b/docs/testing-deployments/g-migration/func.yaml @@ -0,0 +1,27 @@ +# $schema: https://raw.githubusercontent.com/knative/func/main/schema/func_yaml-schema.json +# yaml-language-server: $schema=https://raw.githubusercontent.com/knative/func/main/schema/func_yaml-schema.json +specVersion: 0.37.0 +name: test-migration +runtime: go +registry: docker.io/aliok +namespace: default +deployer: knative +created: 2026-09-04T10:29:25.601427+03:00 +invoke: cloudevent +build: + builder: pack +deploy: + namespace: default + image: index.docker.io/aliok/test-migration@sha256:bcce986a9e2df7088268ef444ce64cf34d5b98b9d31cb68ed68bbedd7fb5120c + options: + scale: + min: 1 + max: 5 + metric: concurrency + target: 200 + utilization: 80 + kpa: + metric: concurrency + target: 200 + utilization: 80 + deployer: knative diff --git a/docs/testing-deployments/g-migration/ksvc.yaml b/docs/testing-deployments/g-migration/ksvc.yaml new file mode 100644 index 0000000000..826b21a96d --- /dev/null +++ b/docs/testing-deployments/g-migration/ksvc.yaml @@ -0,0 +1,85 @@ +apiVersion: serving.knative.dev/v1 +kind: Service +metadata: + annotations: + function.knative.dev/deployer: knative + serving.knative.dev/creator: kubernetes-admin + serving.knative.dev/lastModifier: kubernetes-admin + creationTimestamp: "2026-09-04T07:31:53Z" + generation: 1 + labels: + boson.dev/function: "true" + function.knative.dev/name: test-migration + function.knative.dev/runtime: go + name: test-migration + namespace: default + resourceVersion: "1684" + uid: 37033c72-75d8-4504-92e3-995efd918ace +spec: + template: + metadata: + annotations: + autoscaling.knative.dev/max-scale: "5" + autoscaling.knative.dev/metric: concurrency + autoscaling.knative.dev/min-scale: "1" + autoscaling.knative.dev/target: "200.000000" + autoscaling.knative.dev/target-utilization-percentage: "80.000000" + function.knative.dev/deployer: knative + labels: + boson.dev/function: "true" + function.knative.dev/name: test-migration + function.knative.dev/runtime: go + spec: + containerConcurrency: 0 + containers: + - env: + - name: BUILT + value: 20260904T103153 + - name: ADDRESS + value: 0.0.0.0 + image: index.docker.io/aliok/test-migration@sha256:bcce986a9e2df7088268ef444ce64cf34d5b98b9d31cb68ed68bbedd7fb5120c + livenessProbe: + httpGet: + path: /health/liveness + port: 8080 + name: user-container + readinessProbe: + httpGet: + path: /health/readiness + port: 8080 + successThreshold: 1 + resources: {} + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: + - ALL + runAsNonRoot: true + seccompProfile: + type: RuntimeDefault + enableServiceLinks: false + timeoutSeconds: 300 + traffic: + - latestRevision: true + percent: 100 +status: + address: + url: http://test-migration.default.svc.cluster.local + conditions: + - lastTransitionTime: "2026-09-04T07:32:05Z" + status: "True" + type: ConfigurationsReady + - lastTransitionTime: "2026-09-04T07:32:05Z" + status: "True" + type: Ready + - lastTransitionTime: "2026-09-04T07:32:05Z" + status: "True" + type: RoutesReady + latestCreatedRevisionName: test-migration-00001 + latestReadyRevisionName: test-migration-00001 + observedGeneration: 1 + traffic: + - latestRevision: true + percent: 100 + revisionName: test-migration-00001 + url: http://test-migration.default.svc.cluster.local diff --git a/docs/testing-deployments/g-migration/testing.md b/docs/testing-deployments/g-migration/testing.md new file mode 100644 index 0000000000..7e636cea93 --- /dev/null +++ b/docs/testing-deployments/g-migration/testing.md @@ -0,0 +1,110 @@ +# Scenario G: Migration (old flat scale fields to kpa sub-key) + +Tests that an old-format func.yaml (specVersion 0.36.0 with flat +metric/target/utilization) gets migrated to the `kpa` sub-key on deploy. + +## Prerequisites + +- kind, kubectl, Go 1.25+, Docker, jq + +If using Colima and you hit DNS issues (image pulls failing, etc.), restart +it with explicit DNS: + +```bash +colima stop +colima start --dns 8.8.8.8 +``` + +## 1. Build the CLI + +```bash +cd ~/go/src/knative.dev/func +go build -o /tmp/func-local ./cmd/func +``` + +## 2. Create cluster and install Knative Serving + +```bash +kind create cluster --name test-g + +kubectl apply -f https://github.com/knative/serving/releases/download/knative-v1.21.2/serving-crds.yaml +kubectl apply -f https://github.com/knative/serving/releases/download/knative-v1.21.2/serving-core.yaml +kubectl apply -f https://github.com/knative/net-kourier/releases/download/knative-v1.21.1/kourier.yaml +kubectl patch configmap/config-network -n knative-serving \ + --type merge -p '{"data":{"ingress.class":"kourier.ingress.networking.knative.dev"}}' +kubectl wait deployment --all -n knative-serving --for=condition=Available --timeout=120s +``` + +## 3. Create and configure the function + +```bash +mkdir /tmp/test-migration && cd /tmp/test-migration +/tmp/func-local create -l go -t cloudevents +``` + +Replace the contents of `func.yaml` (keep the `created` line from the generated file) — use OLD specVersion and flat fields: + +```yaml +created: +specVersion: 0.36.0 +name: test-migration +runtime: go +registry: docker.io/aliok +deployer: knative +invoke: cloudevent +deploy: + options: + scale: + min: 1 + max: 5 + metric: concurrency + target: 200 + utilization: 80 +``` + +## 4. Deploy + +```bash +cd /tmp/test-migration +FUNC_REGISTRY=docker.io/aliok /tmp/func-local deploy --verbose +``` + +## 5. Verify + +Check that func.yaml was migrated: + +```bash +cat /tmp/test-migration/func.yaml +# Should show: +# specVersion: 0.37.0 +# kpa: +# metric: concurrency +# target: 200 +# utilization: 80 +# (flat metric/target/utilization also still present for backwards compat) +``` + +Check that KPA annotations are correct on the Knative Service: + +```bash +kubectl get ksvc test-migration -o json \ + | jq '.spec.template.metadata.annotations | { + "autoscaling.knative.dev/metric", + "autoscaling.knative.dev/target", + "autoscaling.knative.dev/target-utilization-percentage" + }' +``` + +## 6. Delete function + +```bash +cd /tmp/test-migration +/tmp/func-local delete +``` + +## 7. Cleanup + +```bash +kind delete cluster --name test-g +rm -rf /tmp/test-migration +``` diff --git a/e2e/e2e_expose_test.go b/e2e/e2e_expose_test.go index aad8c367a7..83d7724f22 100644 --- a/e2e/e2e_expose_test.go +++ b/e2e/e2e_expose_test.go @@ -60,6 +60,30 @@ import ( // and s2i. Remote deploys keep pack: host cannot build in-cluster. // --------------------------------------------------------------------------- +// setKedaHTTPTrigger declares an explicit http KEDA trigger in the func.yaml +// at root. The keda deployer requires at least one trigger to be declared +// explicitly (it no longer infers one), so tests that only care about HTTP +// exposure/routing behavior need this to reach a deployable state. +func setKedaHTTPTrigger(t *testing.T, root string) { + t.Helper() + f, err := fn.NewFunction(root) + if err != nil { + t.Fatal(err) + } + // scale.keda requires deployer: keda, and this is written before the + // deploy command has a chance to set it from the --deployer flag. + f.Deployer = "keda" + if f.Deploy.Options.Scale == nil { + f.Deploy.Options.Scale = &fn.ScaleOptions{} + } + f.Deploy.Options.Scale.KEDA = &fn.KEDAScaleOptions{ + Triggers: []fn.KEDATrigger{{Type: "http"}}, + } + if err := f.Write(); err != nil { + t.Fatal(err) + } +} + // requiresOpenShift skips a test whose assertions need a real Route. func requiresOpenShift(t *testing.T) { t.Helper() @@ -194,11 +218,12 @@ func TestExpose_KedaRejectsLongName(t *testing.T) { if len(name) != 45 { t.Fatalf("test setup: expected a 45 character name, got %d", len(name)) } - fromCleanEnv(t, name) + root := fromCleanEnv(t, name) if err := newCmd(t, "init", "-l=go").Run(); err != nil { t.Fatal(err) } + setKedaHTTPTrigger(t, root) out, err := newCmdOutput(t, "deploy", "--builder=host", "--deployer=keda").CombinedOutput() if err == nil { @@ -379,6 +404,7 @@ func TestExpose_KedaRoute(t *testing.T) { if err := newCmd(t, "init", "-l=go").Run(); err != nil { t.Fatal(err) } + setKedaHTTPTrigger(t, root) if err := newCmd(t, "deploy", "--builder=host", "--deployer=keda", "--expose=route").Run(); err != nil { t.Fatal(err) } @@ -470,6 +496,7 @@ func TestExpose_KedaToggle(t *testing.T) { if err := newCmd(t, "init", "-l=go").Run(); err != nil { t.Fatal(err) } + setKedaHTTPTrigger(t, root) if err := newCmd(t, "deploy", "--builder=host", "--deployer=keda", "--expose=route").Run(); err != nil { t.Fatal(err) } @@ -535,6 +562,7 @@ func TestExpose_KedaDeleteCleansRoute(t *testing.T) { if err := newCmd(t, "init", "-l=go").Run(); err != nil { t.Fatal(err) } + setKedaHTTPTrigger(t, root) if err := newCmd(t, "deploy", "--builder=host", "--deployer=keda", "--expose=route").Run(); err != nil { t.Fatal(err) } @@ -587,6 +615,7 @@ func TestExpose_KedaRouteDomain(t *testing.T) { if err := newCmd(t, "init", "-l=go").Run(); err != nil { t.Fatal(err) } + setKedaHTTPTrigger(t, root) if err := newCmd(t, "deploy", "--builder=host", "--deployer=keda", "--expose=route", "--domain="+domain).Run(); err != nil { t.Fatal(err) } @@ -852,6 +881,7 @@ func TestExpose_RemoteKedaRoute(t *testing.T) { if err := newCmd(t, "init", "-l=go").Run(); err != nil { t.Fatal(err) } + setKedaHTTPTrigger(t, root) if err := newCmd(t, "deploy", "--remote", "--builder=pack", "--registry="+Registry, "--deployer=keda", "--expose=route").Run(); err != nil { t.Fatal(err) diff --git a/pkg/functions/function.go b/pkg/functions/function.go index e40cae3750..66fe103008 100644 --- a/pkg/functions/function.go +++ b/pkg/functions/function.go @@ -479,6 +479,7 @@ func (f Function) Validate() error { ValidateBuildEnvs(f.Build.BuildEnvs), ValidateEnvs(f.Run.Envs), validateOptions(f.Deploy.Options), + validateScaleDeployer(f.Deploy.Options.Scale, f.Deployer, f.Run.Kafka), ValidateLabels(f.Deploy.Labels), validateGit(f.Build.Git), validateKafka(f.Run.Kafka, f.Invoke, f.Runtime), @@ -505,6 +506,29 @@ func (f Function) Validate() error { return errors.New(b.String()) } +func validateScaleDeployer(scale *ScaleOptions, deployer string, kafka *KafkaConfig) (errors []string) { + if deployer == "keda" && (scale == nil || scale.KEDA == nil || len(scale.KEDA.Triggers) == 0) { + errors = append(errors, "deployer keda requires at least one trigger in scale.keda.triggers") + } + if scale == nil { + return + } + if scale.KEDA != nil && deployer != "keda" { + errors = append(errors, "options field \"scale.keda\" requires deployer: keda") + } + if scale.KPA != nil && deployer != "knative" && deployer != "" { + errors = append(errors, "options field \"scale.kpa\" requires deployer: knative") + } + if scale.KEDA != nil { + for i, t := range scale.KEDA.Triggers { + if t.Type == "kafka" && kafka == nil { + errors = append(errors, fmt.Sprintf("options field \"scale.keda.triggers[%d]\" has type kafka but run.kafka is not configured", i)) + } + } + } + return +} + var envPattern = regexp.MustCompile(`^{{\s*(\w+)\s*:(\w+)\s*}}$`) // Interpolate Env slice diff --git a/pkg/functions/function_migrations.go b/pkg/functions/function_migrations.go index 42995cc59c..4f0bec6f9e 100644 --- a/pkg/functions/function_migrations.go +++ b/pkg/functions/function_migrations.go @@ -99,6 +99,7 @@ var migrations = []migration{ {"0.34.0", migrateToSpecsStructure}, {"0.35.0", migrateFromInvokeStructure}, {"0.36.0", migratePersistentVolumeTypoFixup}, + {"0.37.0", migrateScaleKPA}, // New Migrations Here. } @@ -356,6 +357,40 @@ func migratePersistentVolumeTypoFixup(fn Function, m migration) (Function, error return fn, nil } +// migrateScaleKPA moves the flat metric/target/utilization fields under a kpa +// sub-key so that scaler-specific config is organized by type. +// The flat fields are kept alongside kpa for backwards compatibility with +// older CLI versions that don't know about the kpa sub-key. +func migrateScaleKPA(f Function, m migration) (Function, error) { + if f.Deploy.Options.Scale != nil { + hasKPAFields := f.Deploy.Options.Scale.Metric != nil || + f.Deploy.Options.Scale.Target != nil || + f.Deploy.Options.Scale.Utilization != nil + + if hasKPAFields && f.Deploy.Options.Scale.KPA == nil { + f.Deploy.Options.Scale.KPA = &KPAScaleOptions{ + Metric: f.Deploy.Options.Scale.Metric, + Target: f.Deploy.Options.Scale.Target, + Utilization: f.Deploy.Options.Scale.Utilization, + } + } + } + + if f.Deployer == "keda" { + if f.Deploy.Options.Scale == nil { + f.Deploy.Options.Scale = &ScaleOptions{} + } + if f.Deploy.Options.Scale.KEDA == nil || len(f.Deploy.Options.Scale.KEDA.Triggers) == 0 { + f.Deploy.Options.Scale.KEDA = &KEDAScaleOptions{ + Triggers: []KEDATrigger{{Type: "http"}}, + } + } + } + + f.SpecVersion = m.version + return f, nil +} + // The pertinent aspects of the Function's schema prior the 1.0.0 version migrations type migrateToSpecs_previousFunction struct { diff --git a/pkg/functions/function_migrations_unit_test.go b/pkg/functions/function_migrations_unit_test.go index b56abffaf0..9639fddbf7 100644 --- a/pkg/functions/function_migrations_unit_test.go +++ b/pkg/functions/function_migrations_unit_test.go @@ -316,3 +316,136 @@ func writeFunc(f Function, root string) error { } return os.WriteFile(root+"/func.yaml", bb, 0644) } + +func TestMigrateScaleKPA(t *testing.T) { + t.Run("flat fields move to kpa", func(t *testing.T) { + metric := "concurrency" + target := 100.0 + utilization := 70.0 + f := Function{ + SpecVersion: "0.36.0", + Deploy: DeploySpec{ + Options: Options{ + Scale: &ScaleOptions{ + Metric: &metric, + Target: &target, + Utilization: &utilization, + }, + }, + }, + } + + migrated, err := migrateScaleKPA(f, migration{version: "0.37.0"}) + if err != nil { + t.Fatal(err) + } + + if migrated.SpecVersion != "0.37.0" { + t.Errorf("specVersion = %q, want 0.37.0", migrated.SpecVersion) + } + if migrated.Deploy.Options.Scale.KPA == nil { + t.Fatal("expected kpa to be populated") + } + if *migrated.Deploy.Options.Scale.KPA.Metric != "concurrency" { + t.Errorf("kpa.metric = %q, want concurrency", *migrated.Deploy.Options.Scale.KPA.Metric) + } + if *migrated.Deploy.Options.Scale.KPA.Target != 100.0 { + t.Errorf("kpa.target = %f, want 100", *migrated.Deploy.Options.Scale.KPA.Target) + } + if *migrated.Deploy.Options.Scale.KPA.Utilization != 70.0 { + t.Errorf("kpa.utilization = %f, want 70", *migrated.Deploy.Options.Scale.KPA.Utilization) + } + // Flat fields are preserved for backwards compatibility + if migrated.Deploy.Options.Scale.Metric == nil { + t.Error("expected flat metric to be preserved") + } + }) + + t.Run("no-op when no scale fields", func(t *testing.T) { + f := Function{SpecVersion: "0.36.0"} + migrated, err := migrateScaleKPA(f, migration{version: "0.37.0"}) + if err != nil { + t.Fatal(err) + } + if migrated.SpecVersion != "0.37.0" { + t.Errorf("specVersion = %q, want 0.37.0", migrated.SpecVersion) + } + }) + + t.Run("no-op when kpa already set", func(t *testing.T) { + metric := "rps" + f := Function{ + SpecVersion: "0.36.0", + Deploy: DeploySpec{ + Options: Options{ + Scale: &ScaleOptions{ + KPA: &KPAScaleOptions{Metric: &metric}, + }, + }, + }, + } + migrated, err := migrateScaleKPA(f, migration{version: "0.37.0"}) + if err != nil { + t.Fatal(err) + } + if *migrated.Deploy.Options.Scale.KPA.Metric != "rps" { + t.Errorf("kpa.metric = %q, want rps (should not be overwritten)", *migrated.Deploy.Options.Scale.KPA.Metric) + } + }) + + t.Run("keda deployer gets http trigger", func(t *testing.T) { + f := Function{ + SpecVersion: "0.36.0", + Deployer: "keda", + } + migrated, err := migrateScaleKPA(f, migration{version: "0.37.0"}) + if err != nil { + t.Fatal(err) + } + if migrated.Deploy.Options.Scale == nil || migrated.Deploy.Options.Scale.KEDA == nil { + t.Fatal("expected scale.keda to be populated") + } + triggers := migrated.Deploy.Options.Scale.KEDA.Triggers + if len(triggers) != 1 || triggers[0].Type != "http" { + t.Errorf("expected [{http}], got %v", triggers) + } + }) + + t.Run("keda deployer with existing triggers unchanged", func(t *testing.T) { + f := Function{ + SpecVersion: "0.36.0", + Deployer: "keda", + Deploy: DeploySpec{ + Options: Options{ + Scale: &ScaleOptions{ + KEDA: &KEDAScaleOptions{ + Triggers: []KEDATrigger{{Type: "kafka"}}, + }, + }, + }, + }, + } + migrated, err := migrateScaleKPA(f, migration{version: "0.37.0"}) + if err != nil { + t.Fatal(err) + } + triggers := migrated.Deploy.Options.Scale.KEDA.Triggers + if len(triggers) != 1 || triggers[0].Type != "kafka" { + t.Errorf("expected [{kafka}], got %v", triggers) + } + }) + + t.Run("non-keda deployer no triggers added", func(t *testing.T) { + f := Function{ + SpecVersion: "0.36.0", + Deployer: "raw", + } + migrated, err := migrateScaleKPA(f, migration{version: "0.37.0"}) + if err != nil { + t.Fatal(err) + } + if migrated.Deploy.Options.Scale != nil { + t.Errorf("expected no scale options for raw deployer, got %v", migrated.Deploy.Options.Scale) + } + }) +} diff --git a/pkg/functions/function_options.go b/pkg/functions/function_options.go index 1af6a32b7e..dda6a51b46 100644 --- a/pkg/functions/function_options.go +++ b/pkg/functions/function_options.go @@ -12,8 +12,30 @@ type Options struct { } type ScaleOptions struct { - Min *int64 `yaml:"min,omitempty" jsonschema_extras:"minimum=0"` - Max *int64 `yaml:"max,omitempty" jsonschema_extras:"minimum=0"` + Min *int64 `yaml:"min,omitempty" jsonschema_extras:"minimum=0"` + Max *int64 `yaml:"max,omitempty" jsonschema_extras:"minimum=0"` + Metric *string `yaml:"metric,omitempty" jsonschema:"enum=concurrency,enum=rps"` + Target *float64 `yaml:"target,omitempty" jsonschema_extras:"minimum=0.01"` + Utilization *float64 `yaml:"utilization,omitempty" jsonschema:"minimum=1,maximum=100"` + KEDA *KEDAScaleOptions `yaml:"keda,omitempty"` + KPA *KPAScaleOptions `yaml:"kpa,omitempty"` +} + +type KEDAScaleOptions struct { + Triggers []KEDATrigger `yaml:"triggers,omitempty"` +} + +type KEDATrigger struct { + Type string `yaml:"type" jsonschema:"enum=http,enum=kafka,enum=cron"` + LagThreshold *int64 `yaml:"lagThreshold,omitempty" jsonschema_extras:"minimum=1"` + ActivationLagThreshold *int64 `yaml:"activationLagThreshold,omitempty" jsonschema_extras:"minimum=0"` + Timezone string `yaml:"timezone,omitempty"` + Start string `yaml:"start,omitempty"` + End string `yaml:"end,omitempty"` + DesiredReplicas *int64 `yaml:"desiredReplicas,omitempty" jsonschema_extras:"minimum=1"` +} + +type KPAScaleOptions struct { Metric *string `yaml:"metric,omitempty" jsonschema:"enum=concurrency,enum=rps"` Target *float64 `yaml:"target,omitempty" jsonschema_extras:"minimum=0.01"` Utilization *float64 `yaml:"utilization,omitempty" jsonschema:"minimum=1,maximum=100"` @@ -82,6 +104,18 @@ func validateOptions(options Options) (errors []string) { *options.Scale.Utilization)) } } + + if options.Scale.KEDA != nil && options.Scale.KPA != nil { + errors = append(errors, "options fields \"scale.keda\" and \"scale.kpa\" are mutually exclusive") + } + + if options.Scale.KEDA != nil { + errors = append(errors, validateKEDAScale(options.Scale.KEDA)...) + } + + if options.Scale.KPA != nil { + errors = append(errors, validateKPAScale(options.Scale.KPA)...) + } } // options.resource @@ -137,3 +171,63 @@ func validateOptions(options Options) (errors []string) { return } + +func validateKEDAScale(keda *KEDAScaleOptions) (errors []string) { + if len(keda.Triggers) == 0 { + errors = append(errors, "options field \"scale.keda.triggers\" must not be empty when scale.keda is set") + return + } + for i, t := range keda.Triggers { + switch t.Type { + case "http": + // no extra fields required + case "kafka": + if t.LagThreshold != nil && *t.LagThreshold < 1 { + errors = append(errors, fmt.Sprintf("options field \"scale.keda.triggers[%d].lagThreshold\" must be at least 1", i)) + } + if t.ActivationLagThreshold != nil && *t.ActivationLagThreshold < 0 { + errors = append(errors, fmt.Sprintf("options field \"scale.keda.triggers[%d].activationLagThreshold\" must not be negative", i)) + } + case "cron": + if t.Timezone == "" { + errors = append(errors, fmt.Sprintf("options field \"scale.keda.triggers[%d].timezone\" is required for cron triggers", i)) + } + if t.Start == "" { + errors = append(errors, fmt.Sprintf("options field \"scale.keda.triggers[%d].start\" is required for cron triggers", i)) + } + if t.End == "" { + errors = append(errors, fmt.Sprintf("options field \"scale.keda.triggers[%d].end\" is required for cron triggers", i)) + } + if t.DesiredReplicas == nil { + errors = append(errors, fmt.Sprintf("options field \"scale.keda.triggers[%d].desiredReplicas\" is required for cron triggers", i)) + } else if *t.DesiredReplicas < 1 { + errors = append(errors, fmt.Sprintf("options field \"scale.keda.triggers[%d].desiredReplicas\" must be at least 1", i)) + } + default: + errors = append(errors, fmt.Sprintf("options field \"scale.keda.triggers[%d].type\" has invalid value %q, allowed: http, kafka, cron", i, t.Type)) + } + } + return +} + +func validateKPAScale(kpa *KPAScaleOptions) (errors []string) { + if kpa.Metric != nil { + if *kpa.Metric != "concurrency" && *kpa.Metric != "rps" { + errors = append(errors, fmt.Sprintf("options field \"scale.kpa.metric\" has invalid value set: %s, allowed is only \"concurrency\" or \"rps\"", + *kpa.Metric)) + } + } + if kpa.Target != nil { + if *kpa.Target < 0.01 { + errors = append(errors, fmt.Sprintf("options field \"scale.kpa.target\" has value set to \"%f\", but it must not be less than 0.01", + *kpa.Target)) + } + } + if kpa.Utilization != nil { + if *kpa.Utilization < 1 || *kpa.Utilization > 100 { + errors = append(errors, fmt.Sprintf("options field \"scale.kpa.utilization\" has value set to \"%f\", but it must not be less than 1 or greater than 100", + *kpa.Utilization)) + } + } + return +} diff --git a/pkg/functions/function_options_unit_test.go b/pkg/functions/function_options_unit_test.go index 9b798b0362..54c84d2f48 100644 --- a/pkg/functions/function_options_unit_test.go +++ b/pkg/functions/function_options_unit_test.go @@ -310,6 +310,102 @@ func Test_validateOptions(t *testing.T) { }, 10, }, + { + "valid keda triggers", + Options{ + Scale: &ScaleOptions{ + KEDA: &KEDAScaleOptions{ + Triggers: []KEDATrigger{ + {Type: "http"}, + {Type: "kafka", LagThreshold: ptr.Int64(10)}, + }, + }, + }, + }, + 0, + }, + { + "empty keda triggers", + Options{ + Scale: &ScaleOptions{ + KEDA: &KEDAScaleOptions{}, + }, + }, + 1, + }, + { + "invalid keda trigger type", + Options{ + Scale: &ScaleOptions{ + KEDA: &KEDAScaleOptions{ + Triggers: []KEDATrigger{ + {Type: "invalid"}, + }, + }, + }, + }, + 1, + }, + { + "keda cron trigger missing fields", + Options{ + Scale: &ScaleOptions{ + KEDA: &KEDAScaleOptions{ + Triggers: []KEDATrigger{ + {Type: "cron"}, + }, + }, + }, + }, + 4, + }, + { + "valid keda cron trigger", + Options{ + Scale: &ScaleOptions{ + KEDA: &KEDAScaleOptions{ + Triggers: []KEDATrigger{ + {Type: "cron", Timezone: "UTC", Start: "0 8 * * *", End: "0 20 * * *", DesiredReplicas: ptr.Int64(3)}, + }, + }, + }, + }, + 0, + }, + { + "keda and kpa mutually exclusive", + Options{ + Scale: &ScaleOptions{ + KEDA: &KEDAScaleOptions{Triggers: []KEDATrigger{{Type: "http"}}}, + KPA: &KPAScaleOptions{Metric: ptr.String("concurrency")}, + }, + }, + 1, + }, + { + "valid kpa options", + Options{ + Scale: &ScaleOptions{ + KPA: &KPAScaleOptions{ + Metric: ptr.String("rps"), + Target: ptr.Float64(50), + Utilization: ptr.Float64(80), + }, + }, + }, + 0, + }, + { + "invalid kpa metric", + Options{ + Scale: &ScaleOptions{ + KPA: &KPAScaleOptions{ + Metric: ptr.String("bad"), + }, + }, + }, + 1, + }, } for _, tt := range tests { diff --git a/pkg/k8s/wait.go b/pkg/k8s/wait.go index 4d46736331..80af28781d 100644 --- a/pkg/k8s/wait.go +++ b/pkg/k8s/wait.go @@ -69,6 +69,10 @@ func checkIfDeploymentIsAvailable(ctx context.Context, clientset *kubernetes.Cli desiredReplicas := *deployment.Spec.Replicas + if desiredReplicas == 0 { + return true, nil + } + // Check if deployment is available for _, condition := range deployment.Status.Conditions { if condition.Type == appsv1.DeploymentAvailable && condition.Status == corev1.ConditionTrue { diff --git a/pkg/keda/deployer.go b/pkg/keda/deployer.go index 86d268d2eb..155314b33e 100644 --- a/pkg/keda/deployer.go +++ b/pkg/keda/deployer.go @@ -99,8 +99,14 @@ func (k *kedaDeployerDecorator) UpdateLabels(function fn.Function, labels map[st } func (d *Deployer) Deploy(ctx context.Context, f fn.Function) (fn.DeploymentResult, error) { - if err := validateBridgeName(f.Name); err != nil { - return fn.DeploymentResult{}, err + triggers := triggers(f) + wantHTTP := hasHTTPTrigger(triggers) + wantKafka := hasKafkaTrigger(triggers) + + if wantHTTP { + if err := validateBridgeName(f.Name); err != nil { + return fn.DeploymentResult{}, err + } } k8sClientset, err := k8s.NewKubernetesClientset() @@ -112,12 +118,13 @@ func (d *Deployer) Deploy(ctx context.Context, f fn.Function) (fn.DeploymentResu return fn.DeploymentResult{}, fmt.Errorf("failed to create dynamic client: %w", err) } - // Resolved once per deploy and threaded down - interceptorNS, exposeRefusal := interceptorNamespace(ctx, k8sClientset) - - // DNS label checks before we create anything on cluster - if err := d.validateExposure(f, exposeRefusal); err != nil { - return fn.DeploymentResult{}, err + var interceptorNS string + var exposeRefusal error + if wantHTTP { + interceptorNS, exposeRefusal = interceptorNamespace(ctx, k8sClientset) + if err := d.validateExposure(f, exposeRefusal); err != nil { + return fn.DeploymentResult{}, err + } } // execute raw deployment deployer @@ -126,7 +133,6 @@ func (d *Deployer) Deploy(ctx context.Context, f fn.Function) (fn.DeploymentResu return fn.DeploymentResult{}, fmt.Errorf("failed to deploy function via raw deployer: %w", err) } - // create additional required keda resources namespace := deployResult.Namespace deployment, err := k8sClientset.AppsV1().Deployments(namespace).Get(ctx, f.Name, metav1.GetOptions{}) @@ -139,39 +145,67 @@ func (d *Deployer) Deploy(ctx context.Context, f fn.Function) (fn.DeploymentResu return fn.DeploymentResult{}, fmt.Errorf("failed to get service %s/%s: %v", namespace, f.Name, err) } - ref := deployer.NewExposureRef(f.Name, namespace, interceptorNS) - if err := ensureInterceptorBridgeService(ctx, k8sClientset, ref, deployment); err != nil { - return fn.DeploymentResult{}, fmt.Errorf("failed to ensure proxy service exists: %w", err) - } - - labels, err := deployer.GenerateCommonLabels(f, d.decorator) - if err != nil { - return fn.DeploymentResult{}, fmt.Errorf("failed to generate common labels: %w", err) - } - annotations := deployer.GenerateCommonAnnotations(f, d.decorator, false, KedaDeployerName) - minScale, maxScale := replicaBounds(f) - target := deployTarget{ - clientset: k8sClientset, - dynClient: dynClient, - ref: ref, - deployment: deployment, - appService: appService, - labels: labels, - annotations: annotations, - minScale: minScale, - maxScale: maxScale, - } + + // HTTP trigger path: bridge Service + HTTPScaledObject var url string appliedExpose := "" - if d.exposer != nil && fn.ActiveExpose(f.Expose) { - if url, err = d.deployExposed(ctx, target); err != nil { - return fn.DeploymentResult{}, err + if wantHTTP { + ref := deployer.NewExposureRef(f.Name, namespace, interceptorNS) + if err := ensureInterceptorBridgeService(ctx, k8sClientset, ref, deployment); err != nil { + return fn.DeploymentResult{}, fmt.Errorf("failed to ensure proxy service exists: %w", err) + } + + labels, err := deployer.GenerateCommonLabels(f, d.decorator) + if err != nil { + return fn.DeploymentResult{}, fmt.Errorf("failed to generate common labels: %w", err) + } + annotations := deployer.GenerateCommonAnnotations(f, d.decorator, false, KedaDeployerName) + + target := deployTarget{ + clientset: k8sClientset, + dynClient: dynClient, + ref: ref, + deployment: deployment, + appService: appService, + labels: labels, + annotations: annotations, + minScale: minScale, + maxScale: maxScale, + } + + if d.exposer != nil && fn.ActiveExpose(f.Expose) { + if url, err = d.deployExposed(ctx, target); err != nil { + return fn.DeploymentResult{}, err + } + appliedExpose = f.Expose + } else { + if url, err = d.deployClusterLocal(ctx, target); err != nil { + return fn.DeploymentResult{}, err + } } - appliedExpose = f.Expose } else { - if url, err = d.deployClusterLocal(ctx, target); err != nil { - return fn.DeploymentResult{}, err + // No HTTP trigger — URL is the app service + url = fmt.Sprintf("http://%s.%s.svc:8080", f.Name, namespace) + } + + // Kafka trigger path: TriggerAuthentication + ScaledObject + if wantKafka && f.Run.Kafka != nil { + if needsTriggerAuth(f.Run.Kafka) { + ta := buildTriggerAuth(f, deployment, namespace) + if ta != nil { + if err := ensureTriggerAuth(ctx, dynClient, ta); err != nil { + return fn.DeploymentResult{}, fmt.Errorf("failed to ensure TriggerAuthentication: %w", err) + } + } + } + + kt := kafkaTrigger(triggers) + so := buildScaledObject(f, kt, deployment, namespace, minScale, maxScale) + if so != nil { + if err := ensureScaledObject(ctx, dynClient, so); err != nil { + return fn.DeploymentResult{}, fmt.Errorf("failed to ensure ScaledObject: %w", err) + } } } diff --git a/pkg/keda/kafka_scaling.go b/pkg/keda/kafka_scaling.go new file mode 100644 index 0000000000..b57ab54d09 --- /dev/null +++ b/pkg/keda/kafka_scaling.go @@ -0,0 +1,371 @@ +package keda + +import ( + "context" + "fmt" + "path/filepath" + "strings" + + v1 "k8s.io/api/apps/v1" + k8serrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/dynamic" + fn "knative.dev/func/pkg/functions" +) + +var ( + scaledObjectGVR = schema.GroupVersionResource{ + Group: "keda.sh", + Version: "v1alpha1", + Resource: "scaledobjects", + } + triggerAuthGVR = schema.GroupVersionResource{ + Group: "keda.sh", + Version: "v1alpha1", + Resource: "triggerauthentications", + } +) + +func scaledObjectName(funcName string) string { + return funcName + "-kafka" +} + +func triggerAuthName(funcName string) string { + return funcName + "-kafka-auth" +} + +// triggers returns the function's explicitly configured KEDA triggers, or a +// plain http trigger when none are configured at all. This only matters for +// callers of Deploy that bypass fn.Function.Validate (which requires +// deployer: keda to declare triggers explicitly) -- e.g. tests and other +// direct API consumers. It never infers a kafka trigger: that decision is +// never made silently, on any path. +func triggers(f fn.Function) []fn.KEDATrigger { + if f.Deploy.Options.Scale != nil && f.Deploy.Options.Scale.KEDA != nil { + return f.Deploy.Options.Scale.KEDA.Triggers + } + return []fn.KEDATrigger{{Type: "http"}} +} + +func hasHTTPTrigger(triggers []fn.KEDATrigger) bool { + for _, t := range triggers { + if t.Type == "http" { + return true + } + } + return false +} + +func hasKafkaTrigger(triggers []fn.KEDATrigger) bool { + for _, t := range triggers { + if t.Type == "kafka" { + return true + } + } + return false +} + +func kafkaTrigger(triggers []fn.KEDATrigger) fn.KEDATrigger { + for _, t := range triggers { + if t.Type == "kafka" { + return t + } + } + return fn.KEDATrigger{} +} + +// needsTriggerAuth returns true when the Kafka config uses SASL or TLS with +// secrets that must be referenced via a TriggerAuthentication. +func needsTriggerAuth(kafka *fn.KafkaConfig) bool { + if kafka == nil { + return false + } + if kafka.SASL != nil && kafka.SASL.Password != "" { + return true + } + if kafka.TLS != nil && kafka.TLS.CACert != "" { + return true + } + return false +} + +// parseSecretRef extracts the secret name and key from a {{ secret:name:key }} +// reference. Returns empty strings if the value is not a secret reference. +func parseSecretRef(value string) (secretName, secretKey string) { + if !strings.HasPrefix(value, "{{") { + return + } + trimmed := strings.Trim(value, "{} ") + parts := strings.Split(trimmed, ":") + if len(parts) == 3 && strings.TrimSpace(parts[0]) == "secret" { + return strings.TrimSpace(parts[1]), strings.TrimSpace(parts[2]) + } + return +} + +// findSecretForPath finds the volume secret name that backs a given file path. +// It matches by checking which volume's mount path is a parent of the cert path. +func findSecretForPath(certPath string, volumes []fn.Volume) (secretName, key string) { + for _, v := range volumes { + if v.Secret == nil || v.Path == nil { + continue + } + mountPath := *v.Path + if strings.HasPrefix(certPath, mountPath) { + rel, err := filepath.Rel(mountPath, certPath) + if err != nil { + continue + } + return *v.Secret, rel + } + } + return +} + +// buildTriggerAuth creates the unstructured TriggerAuthentication for Kafka SASL/TLS. +func buildTriggerAuth(f fn.Function, deployment *v1.Deployment, namespace string) *unstructured.Unstructured { + kafka := f.Run.Kafka + if kafka == nil { + return nil + } + + var secretRefs []interface{} + var envRefs []interface{} + + if kafka.SASL != nil && kafka.SASL.Password != "" { + secretName, secretKey := parseSecretRef(kafka.SASL.Password) + if secretName != "" { + secretRefs = append(secretRefs, map[string]interface{}{ + "parameter": "password", + "name": secretName, + "key": secretKey, + }) + } + + if kafka.SASL.User != "" { + userName, userKey := parseSecretRef(kafka.SASL.User) + if userName != "" { + secretRefs = append(secretRefs, map[string]interface{}{ + "parameter": "username", + "name": userName, + "key": userKey, + }) + } else { + envRefs = append(envRefs, map[string]interface{}{ + "parameter": "username", + "name": "KAFKA_SASL_USER", + "containerName": deployment.Spec.Template.Spec.Containers[0].Name, + }) + } + } + } + + if kafka.TLS != nil && kafka.TLS.CACert != "" { + caSecretName, caKey := findSecretForPath(kafka.TLS.CACert, f.Run.Volumes) + if caSecretName != "" { + secretRefs = append(secretRefs, map[string]interface{}{ + "parameter": "ca", + "name": caSecretName, + "key": caKey, + }) + } + } + + if len(secretRefs) == 0 && len(envRefs) == 0 { + return nil + } + + spec := map[string]interface{}{} + if len(secretRefs) > 0 { + spec["secretTargetRef"] = secretRefs + } + if len(envRefs) > 0 { + spec["env"] = envRefs + } + + ta := &unstructured.Unstructured{ + Object: map[string]interface{}{ + "apiVersion": "keda.sh/v1alpha1", + "kind": "TriggerAuthentication", + "metadata": map[string]interface{}{ + "name": triggerAuthName(f.Name), + "namespace": namespace, + "ownerReferences": []interface{}{ + map[string]interface{}{ + "apiVersion": "apps/v1", + "kind": "Deployment", + "name": deployment.Name, + "uid": string(deployment.UID), + "controller": true, + "blockOwnerDeletion": true, + }, + }, + }, + "spec": spec, + }, + } + + return ta +} + +// kedaSASLType maps func.yaml SASL mechanism names to KEDA trigger metadata values. +func kedaSASLType(mechanism string) string { + switch mechanism { + case "SCRAM-SHA-256": + return "scram_sha256" + case "SCRAM-SHA-512": + return "scram_sha512" + case "PLAIN": + return "plain" + default: + return "" + } +} + +// buildScaledObject creates the unstructured ScaledObject for Kafka consumer-lag scaling. +func buildScaledObject(f fn.Function, trigger fn.KEDATrigger, deployment *v1.Deployment, namespace string, minScale, maxScale int32) *unstructured.Unstructured { + kafka := f.Run.Kafka + if kafka == nil { + return nil + } + + lagThreshold := int64(10) + if trigger.LagThreshold != nil { + lagThreshold = *trigger.LagThreshold + } + + triggerMeta := map[string]interface{}{ + "bootstrapServers": kafka.Brokers, + "consumerGroup": kafka.ConsumerGroup, + "topic": kafka.Topic, + "lagThreshold": fmt.Sprintf("%d", lagThreshold), + } + + if trigger.ActivationLagThreshold != nil { + triggerMeta["activationLagThreshold"] = fmt.Sprintf("%d", *trigger.ActivationLagThreshold) + } + + if kafka.TLS != nil { + triggerMeta["tls"] = "enable" + } + + if kafka.SASL != nil && kafka.SASL.Mechanism != "" { + triggerMeta["sasl"] = kedaSASLType(kafka.SASL.Mechanism) + } + + triggerSpec := map[string]interface{}{ + "type": "kafka", + "metadata": triggerMeta, + } + + if needsTriggerAuth(kafka) { + triggerSpec["authenticationRef"] = map[string]interface{}{ + "name": triggerAuthName(f.Name), + } + } + + so := &unstructured.Unstructured{ + Object: map[string]interface{}{ + "apiVersion": "keda.sh/v1alpha1", + "kind": "ScaledObject", + "metadata": map[string]interface{}{ + "name": scaledObjectName(f.Name), + "namespace": namespace, + "ownerReferences": []interface{}{ + map[string]interface{}{ + "apiVersion": "apps/v1", + "kind": "Deployment", + "name": deployment.Name, + "uid": string(deployment.UID), + "controller": true, + "blockOwnerDeletion": true, + }, + }, + }, + "spec": map[string]interface{}{ + "scaleTargetRef": map[string]interface{}{ + "kind": "Deployment", + "name": deployment.Name, + }, + "minReplicaCount": int64(minScale), + "maxReplicaCount": int64(maxScale), + "cooldownPeriod": int64(300), + "triggers": []interface{}{ + triggerSpec, + }, + }, + }, + } + + return so +} + +// ensureScaledObject creates or updates a KEDA ScaledObject for Kafka scaling. +func ensureScaledObject(ctx context.Context, dynClient dynamic.Interface, so *unstructured.Unstructured) error { + ns := so.GetNamespace() + name := so.GetName() + client := dynClient.Resource(scaledObjectGVR).Namespace(ns) + + existing, err := client.Get(ctx, name, metav1.GetOptions{}) + if err != nil { + if k8serrors.IsNotFound(err) { + if _, err := client.Create(ctx, so, metav1.CreateOptions{}); err != nil { + return fmt.Errorf("failed to create ScaledObject %s/%s: %w", ns, name, err) + } + return nil + } + return fmt.Errorf("failed to get ScaledObject %s/%s: %w", ns, name, err) + } + + so.SetResourceVersion(existing.GetResourceVersion()) + if _, err := client.Update(ctx, so, metav1.UpdateOptions{}); err != nil { + return fmt.Errorf("failed to update ScaledObject %s/%s: %w", ns, name, err) + } + return nil +} + +// ensureTriggerAuth creates or updates a KEDA TriggerAuthentication. +func ensureTriggerAuth(ctx context.Context, dynClient dynamic.Interface, ta *unstructured.Unstructured) error { + ns := ta.GetNamespace() + name := ta.GetName() + client := dynClient.Resource(triggerAuthGVR).Namespace(ns) + + existing, err := client.Get(ctx, name, metav1.GetOptions{}) + if err != nil { + if k8serrors.IsNotFound(err) { + if _, err := client.Create(ctx, ta, metav1.CreateOptions{}); err != nil { + return fmt.Errorf("failed to create TriggerAuthentication %s/%s: %w", ns, name, err) + } + return nil + } + return fmt.Errorf("failed to get TriggerAuthentication %s/%s: %w", ns, name, err) + } + + ta.SetResourceVersion(existing.GetResourceVersion()) + if _, err := client.Update(ctx, ta, metav1.UpdateOptions{}); err != nil { + return fmt.Errorf("failed to update TriggerAuthentication %s/%s: %w", ns, name, err) + } + return nil +} + +// deleteScaledObject removes a ScaledObject if it exists. +func deleteScaledObject(ctx context.Context, dynClient dynamic.Interface, ns, name string) error { + client := dynClient.Resource(scaledObjectGVR).Namespace(ns) + err := client.Delete(ctx, name, metav1.DeleteOptions{}) + if err != nil && !k8serrors.IsNotFound(err) { + return fmt.Errorf("failed to delete ScaledObject %s/%s: %w", ns, name, err) + } + return nil +} + +// deleteTriggerAuth removes a TriggerAuthentication if it exists. +func deleteTriggerAuth(ctx context.Context, dynClient dynamic.Interface, ns, name string) error { + client := dynClient.Resource(triggerAuthGVR).Namespace(ns) + err := client.Delete(ctx, name, metav1.DeleteOptions{}) + if err != nil && !k8serrors.IsNotFound(err) { + return fmt.Errorf("failed to delete TriggerAuthentication %s/%s: %w", ns, name, err) + } + return nil +} diff --git a/pkg/keda/kafka_scaling_int_test.go b/pkg/keda/kafka_scaling_int_test.go new file mode 100644 index 0000000000..68b408c025 --- /dev/null +++ b/pkg/keda/kafka_scaling_int_test.go @@ -0,0 +1,228 @@ +//go:build integration + +package keda_test + +import ( + "context" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/util/rand" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/kubernetes" + + fn "knative.dev/func/pkg/functions" + "knative.dev/func/pkg/k8s" + "knative.dev/func/pkg/keda" + testingk8s "knative.dev/func/pkg/testing/k8s" +) + +var ( + scaledObjectGVR = schema.GroupVersionResource{ + Group: "keda.sh", + Version: "v1alpha1", + Resource: "scaledobjects", + } + triggerAuthGVR = schema.GroupVersionResource{ + Group: "keda.sh", + Version: "v1alpha1", + Resource: "triggerauthentications", + } +) + +// TestInt_KafkaScaling deploys a function with a Kafka-only KEDA trigger +// (no HTTP trigger, since a Deployment can only be owned by one ScaledObject +// and the http trigger's HTTPScaledObject creates its own -- see #4043) and +// verifies that the deployer creates a ScaledObject and TriggerAuthentication +// with the expected spec, and that both are cleaned up on removal. +// +// This does not require a reachable Kafka broker: KEDA admits and reconciles +// the ScaledObject as long as the trigger metadata is well-formed, regardless +// of whether it can actually reach the broker to read lag. +func TestInt_KafkaScaling(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), time.Minute*10) + t.Cleanup(cancel) + + name := "func-int-keda-kafka-" + rand.String(5) + ns := testingk8s.Namespace(t, ctx) + + cliSet, err := k8s.NewKubernetesClientset() + if err != nil { + t.Fatal(err) + } + + caSecretName := name + "-ca" + createSecretForTest(t, ctx, cliSet, ns, caSecretName, map[string][]byte{"ca.crt": []byte("placeholder-ca-cert")}) + + userSecretName := name + "-user" + createSecretForTest(t, ctx, cliSet, ns, userSecretName, map[string][]byte{"password": []byte("placeholder-password")}) + + minScale := int64(0) + maxScale := int64(10) + lagThreshold := int64(7) + + function := fn.Function{ + SpecVersion: "SNAPSHOT", + Root: "/non/existent", + Name: name, + Runtime: "blub", + Template: "cloudevents", + Created: time.Now(), + Deploy: fn.DeploySpec{ + // pinned prebuilt image: this test exercises the deployer's + // Kafka-scaling object creation, not the build/image flow + Image: "quay.io/mvasek/func-test-service@sha256:2eca4de00d7569c8791634bdbb0c4d5ec8fb061b001549314591e839dabd5269", + Namespace: ns, + Expose: "none", + Options: fn.Options{ + Scale: &fn.ScaleOptions{ + Min: &minScale, + Max: &maxScale, + KEDA: &fn.KEDAScaleOptions{ + Triggers: []fn.KEDATrigger{ + {Type: "kafka", LagThreshold: &lagThreshold}, + }, + }, + }, + }, + }, + Run: fn.RunSpec{ + Kafka: &fn.KafkaConfig{ + Brokers: "placeholder-kafka-bootstrap." + ns + ".svc:9093", + Topic: "test-topic", + ConsumerGroup: name + "-group", + SecurityProtocol: "SASL_SSL", + TLS: &fn.KafkaTLS{ + CACert: "/etc/kafka/ca/ca.crt", + }, + SASL: &fn.KafkaSASL{ + Mechanism: "SCRAM-SHA-512", + User: "test-kafka-user", + Password: "{{ secret:" + userSecretName + ":password }}", + }, + }, + Volumes: []fn.Volume{ + {Secret: ptrTo(caSecretName), Path: ptrTo("/etc/kafka/ca")}, + }, + }, + } + + deployer := keda.NewDeployer(keda.WithDeployerVerbose(false)) + remover := keda.NewRemover(false) + + _, err = deployer.Deploy(ctx, function) + if err != nil { + t.Fatalf("deploy failed: %v", err) + } + t.Cleanup(func() { + if err := remover.Remove(context.Background(), name, ns); err != nil { + t.Logf("error removing function: %v", err) + } + }) + + dynClient, err := k8s.NewDynamicClient() + if err != nil { + t.Fatal(err) + } + + so, err := dynClient.Resource(scaledObjectGVR).Namespace(ns).Get(ctx, name+"-kafka", metav1.GetOptions{}) + if err != nil { + t.Fatalf("expected ScaledObject to exist: %v", err) + } + + spec, _ := so.Object["spec"].(map[string]interface{}) + if spec["minReplicaCount"] != int64(0) { + t.Errorf("minReplicaCount = %v, want 0", spec["minReplicaCount"]) + } + if spec["maxReplicaCount"] != int64(10) { + t.Errorf("maxReplicaCount = %v, want 10", spec["maxReplicaCount"]) + } + triggersList, _ := spec["triggers"].([]interface{}) + if len(triggersList) != 1 { + t.Fatalf("expected 1 trigger, got %d", len(triggersList)) + } + trigger, _ := triggersList[0].(map[string]interface{}) + if trigger["type"] != "kafka" { + t.Errorf("trigger type = %v, want kafka", trigger["type"]) + } + meta, _ := trigger["metadata"].(map[string]interface{}) + if meta["lagThreshold"] != "7" { + t.Errorf("lagThreshold = %v, want 7", meta["lagThreshold"]) + } + if meta["consumerGroup"] != name+"-group" { + t.Errorf("consumerGroup = %v, want %s", meta["consumerGroup"], name+"-group") + } + if meta["sasl"] != "scram_sha512" { + t.Errorf("sasl = %v, want scram_sha512", meta["sasl"]) + } + if meta["tls"] != "enable" { + t.Errorf("tls = %v, want enable", meta["tls"]) + } + authRef, _ := trigger["authenticationRef"].(map[string]interface{}) + if authRef["name"] != name+"-kafka-auth" { + t.Errorf("authenticationRef.name = %v, want %s", authRef["name"], name+"-kafka-auth") + } + + ownerRefs, _ := so.Object["metadata"].(map[string]interface{})["ownerReferences"].([]interface{}) + if len(ownerRefs) != 1 { + t.Fatalf("expected 1 owner reference, got %d", len(ownerRefs)) + } + owner, _ := ownerRefs[0].(map[string]interface{}) + if owner["name"] != name || owner["kind"] != "Deployment" { + t.Errorf("unexpected owner reference: %v", owner) + } + + ta, err := dynClient.Resource(triggerAuthGVR).Namespace(ns).Get(ctx, name+"-kafka-auth", metav1.GetOptions{}) + if err != nil { + t.Fatalf("expected TriggerAuthentication to exist: %v", err) + } + + taSpec, _ := ta.Object["spec"].(map[string]interface{}) + secretRefs, _ := taSpec["secretTargetRef"].([]interface{}) + if len(secretRefs) != 2 { + t.Fatalf("expected 2 secretTargetRef entries (password, ca), got %d: %v", len(secretRefs), secretRefs) + } + envRefs, _ := taSpec["env"].([]interface{}) + if len(envRefs) != 1 { + t.Fatalf("expected 1 env entry (username), got %d: %v", len(envRefs), envRefs) + } + envRef, _ := envRefs[0].(map[string]interface{}) + if envRef["parameter"] != "username" || envRef["name"] != "KAFKA_SASL_USER" { + t.Errorf("unexpected env ref: %v", envRef) + } + + // Removal: TriggerAuthentication/ScaledObject carry KEDA's own finalizer, + // so deletion is asynchronous even after remover.Remove returns. + if err := remover.Remove(ctx, name, ns); err != nil { + t.Fatalf("remove failed: %v", err) + } + + err = wait.PollUntilContextTimeout(ctx, time.Second, time.Minute, true, func(ctx context.Context) (bool, error) { + _, soErr := dynClient.Resource(scaledObjectGVR).Namespace(ns).Get(ctx, name+"-kafka", metav1.GetOptions{}) + _, taErr := dynClient.Resource(triggerAuthGVR).Namespace(ns).Get(ctx, name+"-kafka-auth", metav1.GetOptions{}) + return apierrors.IsNotFound(soErr) && apierrors.IsNotFound(taErr), nil + }) + if err != nil { + t.Fatalf("expected ScaledObject and TriggerAuthentication to be removed: %v", err) + } +} + +func createSecretForTest(t *testing.T, ctx context.Context, cliSet *kubernetes.Clientset, ns, name string, data map[string][]byte) { + t.Helper() + _, err := cliSet.CoreV1().Secrets(ns).Create(ctx, &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Data: data, + Type: corev1.SecretTypeOpaque, + }, metav1.CreateOptions{}) + if err != nil { + t.Fatal(err) + } +} + +func ptrTo[T any](v T) *T { + return &v +} diff --git a/pkg/keda/kafka_scaling_test.go b/pkg/keda/kafka_scaling_test.go new file mode 100644 index 0000000000..ced2523278 --- /dev/null +++ b/pkg/keda/kafka_scaling_test.go @@ -0,0 +1,277 @@ +package keda + +import ( + "testing" + + v1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + fn "knative.dev/func/pkg/functions" +) + +func TestTriggers_NoScale(t *testing.T) { + f := fn.Function{Name: "test"} + got := triggers(f) + if len(got) != 1 || got[0].Type != "http" { + t.Errorf("expected [http] fallback, got %v", got) + } +} + +func TestTriggers_Explicit(t *testing.T) { + lag := int64(5) + f := fn.Function{ + Name: "test", + Deploy: fn.DeploySpec{ + Options: fn.Options{ + Scale: &fn.ScaleOptions{ + KEDA: &fn.KEDAScaleOptions{ + Triggers: []fn.KEDATrigger{ + {Type: "kafka", LagThreshold: &lag}, + }, + }, + }, + }, + }, + } + got := triggers(f) + if len(got) != 1 { + t.Fatalf("expected 1 trigger, got %d", len(got)) + } + if got[0].Type != "kafka" { + t.Errorf("expected kafka, got %s", got[0].Type) + } + if *got[0].LagThreshold != 5 { + t.Errorf("expected lagThreshold 5, got %d", *got[0].LagThreshold) + } +} + +func TestParseSecretRef(t *testing.T) { + tests := []struct { + input string + wantName string + wantKey string + }{ + {"{{ secret:my-secret:my-key }}", "my-secret", "my-key"}, + {"{{ secret:foo:bar }}", "foo", "bar"}, + {"plaintext-value", "", ""}, + {"{{ configMap:cm:key }}", "", ""}, + {"{{ invalid }}", "", ""}, + } + for _, tt := range tests { + name, key := parseSecretRef(tt.input) + if name != tt.wantName || key != tt.wantKey { + t.Errorf("parseSecretRef(%q) = (%q, %q), want (%q, %q)", tt.input, name, key, tt.wantName, tt.wantKey) + } + } +} + +func TestFindSecretForPath(t *testing.T) { + secret := "my-cluster-ca" + path := "/etc/kafka/ca" + volumes := []fn.Volume{ + {Secret: &secret, Path: &path}, + } + + name, key := findSecretForPath("/etc/kafka/ca/ca.crt", volumes) + if name != "my-cluster-ca" || key != "ca.crt" { + t.Errorf("got (%q, %q), want (my-cluster-ca, ca.crt)", name, key) + } + + name, _ = findSecretForPath("/other/path", volumes) + if name != "" { + t.Errorf("expected empty for non-matching path, got %q", name) + } +} + +func testDeployment() *v1.Deployment { + return &v1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-func", + Namespace: "default", + UID: types.UID("test-uid-123"), + }, + Spec: v1.DeploymentSpec{ + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "user-container"}}, + }, + }, + }, + } +} + +func TestBuildTriggerAuth(t *testing.T) { + f := fn.Function{ + Name: "test-func", + Run: fn.RunSpec{ + Kafka: &fn.KafkaConfig{ + Brokers: "broker:9093", + Topic: "topic", + ConsumerGroup: "group", + SecurityProtocol: "SASL_SSL", + TLS: &fn.KafkaTLS{ + CACert: "/etc/kafka/ca/ca.crt", + }, + SASL: &fn.KafkaSASL{ + Mechanism: "SCRAM-SHA-512", + User: "admin", + Password: "{{ secret:my-user:password }}", + }, + }, + Volumes: []fn.Volume{ + {Secret: strPtr("my-cluster-ca"), Path: strPtr("/etc/kafka/ca")}, + }, + }, + } + + ta := buildTriggerAuth(f, testDeployment(), "default") + if ta == nil { + t.Fatal("expected TriggerAuthentication, got nil") + } + + if ta.GetName() != "test-func-kafka-auth" { + t.Errorf("name = %q, want test-func-kafka-auth", ta.GetName()) + } + + spec, ok := ta.Object["spec"].(map[string]interface{}) + if !ok { + t.Fatal("missing spec") + } + refs, ok := spec["secretTargetRef"].([]interface{}) + if !ok { + t.Fatal("missing secretTargetRef") + } + if len(refs) != 2 { + t.Fatalf("expected 2 secretTargetRef entries, got %d", len(refs)) + } + + ref0 := refs[0].(map[string]interface{}) + if ref0["parameter"] != "password" || ref0["name"] != "my-user" || ref0["key"] != "password" { + t.Errorf("unexpected password ref: %v", ref0) + } + + ref1 := refs[1].(map[string]interface{}) + if ref1["parameter"] != "ca" || ref1["name"] != "my-cluster-ca" || ref1["key"] != "ca.crt" { + t.Errorf("unexpected ca ref: %v", ref1) + } + + envs, ok := spec["env"].([]interface{}) + if !ok { + t.Fatal("missing env") + } + if len(envs) != 1 { + t.Fatalf("expected 1 env entry, got %d", len(envs)) + } + env0 := envs[0].(map[string]interface{}) + if env0["parameter"] != "username" || env0["name"] != "KAFKA_SASL_USER" { + t.Errorf("unexpected env ref: %v", env0) + } +} + +func TestBuildScaledObject(t *testing.T) { + lag := int64(20) + f := fn.Function{ + Name: "test-func", + Run: fn.RunSpec{ + Kafka: &fn.KafkaConfig{ + Brokers: "broker:9093", + Topic: "my-topic", + ConsumerGroup: "my-group", + SecurityProtocol: "SASL_SSL", + TLS: &fn.KafkaTLS{CACert: "/etc/kafka/ca/ca.crt"}, + SASL: &fn.KafkaSASL{Mechanism: "SCRAM-SHA-512", Password: "{{ secret:s:k }}"}, + }, + }, + } + trigger := fn.KEDATrigger{Type: "kafka", LagThreshold: &lag} + + so := buildScaledObject(f, trigger, testDeployment(), "default", 0, 10) + if so == nil { + t.Fatal("expected ScaledObject, got nil") + } + + if so.GetName() != "test-func-kafka" { + t.Errorf("name = %q, want test-func-kafka", so.GetName()) + } + + spec := so.Object["spec"].(map[string]interface{}) + if spec["minReplicaCount"] != int64(0) { + t.Errorf("minReplicaCount = %v, want 0", spec["minReplicaCount"]) + } + if spec["maxReplicaCount"] != int64(10) { + t.Errorf("maxReplicaCount = %v, want 10", spec["maxReplicaCount"]) + } + + triggers := spec["triggers"].([]interface{}) + if len(triggers) != 1 { + t.Fatalf("expected 1 trigger, got %d", len(triggers)) + } + trigger0 := triggers[0].(map[string]interface{}) + meta := trigger0["metadata"].(map[string]interface{}) + if meta["bootstrapServers"] != "broker:9093" { + t.Errorf("bootstrapServers = %v", meta["bootstrapServers"]) + } + if meta["lagThreshold"] != "20" { + t.Errorf("lagThreshold = %v, want 20", meta["lagThreshold"]) + } + if meta["tls"] != "enable" { + t.Errorf("tls = %v, want enable", meta["tls"]) + } + if meta["sasl"] != "scram_sha512" { + t.Errorf("sasl = %v, want scram_sha512", meta["sasl"]) + } + + authRef := trigger0["authenticationRef"].(map[string]interface{}) + if authRef["name"] != "test-func-kafka-auth" { + t.Errorf("authenticationRef name = %v", authRef["name"]) + } +} + +func TestBuildScaledObject_DefaultLag(t *testing.T) { + f := fn.Function{ + Name: "test-func", + Run: fn.RunSpec{ + Kafka: &fn.KafkaConfig{ + Brokers: "broker:9092", + Topic: "t", + ConsumerGroup: "g", + }, + }, + } + trigger := fn.KEDATrigger{Type: "kafka"} + + so := buildScaledObject(f, trigger, testDeployment(), "default", 1, 5) + if so == nil { + t.Fatal("expected ScaledObject, got nil") + } + + spec := so.Object["spec"].(map[string]interface{}) + triggers := spec["triggers"].([]interface{}) + trigger0 := triggers[0].(map[string]interface{}) + meta := trigger0["metadata"].(map[string]interface{}) + if meta["lagThreshold"] != "10" { + t.Errorf("default lagThreshold = %v, want 10", meta["lagThreshold"]) + } + + // No TLS/SASL, so no authenticationRef + if _, ok := trigger0["authenticationRef"]; ok { + t.Error("expected no authenticationRef for plaintext Kafka") + } +} + +func TestKedaSASLType(t *testing.T) { + tests := map[string]string{ + "SCRAM-SHA-256": "scram_sha256", + "SCRAM-SHA-512": "scram_sha512", + "PLAIN": "plain", + "UNKNOWN": "", + } + for in, want := range tests { + if got := kedaSASLType(in); got != want { + t.Errorf("kedaSASLType(%q) = %q, want %q", in, got, want) + } + } +} + +func strPtr(s string) *string { return &s } diff --git a/pkg/keda/remover.go b/pkg/keda/remover.go index 21e5955a84..1214a34f17 100644 --- a/pkg/keda/remover.go +++ b/pkg/keda/remover.go @@ -50,6 +50,11 @@ func (remover *Remover) Remove(ctx context.Context, name, ns string) error { return fn.ErrNotHandled } + dynClient, err := k8s.NewDynamicClient() + if err != nil { + return fmt.Errorf("could not setup dynamic client: %w", err) + } + // Remove the recorded Route before deleting anything: keda's Route has no // owner reference (it would have to cross namespaces), so nothing collects // it, and its record - these Service annotations - is deleted with the @@ -57,10 +62,6 @@ func (remover *Remover) Remove(ctx context.Context, name, ns string) error { // A Route left unrecorded by a crash is not searched for; the next // exposed redeploy finds it by its function labels. if recordedNS := svc.Annotations[k8s.RouteNamespaceAnnotation]; recordedNS != "" { - dynClient, err := k8s.NewDynamicClient() - if err != nil { - return fmt.Errorf("could not setup dynamic client: %w", err) - } if err := ocproute.New(KedaDeployerName).Unexpose(ctx, dynClient, deployer.NewExposureRef(name, ns, recordedNS)); err != nil { return fmt.Errorf("could not remove the Route exposing function %q in namespace %q; "+ "nothing was deleted and the function is still running, if you fix this you can run delete again: %w", @@ -68,6 +69,13 @@ func (remover *Remover) Remove(ctx context.Context, name, ns string) error { } } + // Clean up Kafka scaling resources before deleting the Deployment. + // These have ownerReferences so they'd be garbage-collected, but + // explicit deletion avoids races with a slow GC. + // Ignore not-found: these resources may not exist (HTTP-only deploy). + _ = deleteScaledObject(ctx, dynClient, ns, scaledObjectName(name)) + _ = deleteTriggerAuth(ctx, dynClient, ns, triggerAuthName(name)) + deploymentClient := clientset.AppsV1().Deployments(ns) // Delete only the Deployment; owner references take the rest with it. diff --git a/pkg/knative/deployer.go b/pkg/knative/deployer.go index 4e719d6b05..fe0980ef90 100644 --- a/pkg/knative/deployer.go +++ b/pkg/knative/deployer.go @@ -598,20 +598,36 @@ func setServiceOptions(template *servingv1.RevisionTemplateSpec, options fn.Opti toRemove = append(toRemove, autoscaling.MaxScaleAnnotationKey) } - if options.Scale.Metric != nil { - toUpdate[autoscaling.MetricAnnotationKey] = *options.Scale.Metric + // KPA fields: prefer kpa sub-key, fall back to flat fields + metric := options.Scale.Metric + target := options.Scale.Target + utilization := options.Scale.Utilization + if options.Scale.KPA != nil { + if options.Scale.KPA.Metric != nil { + metric = options.Scale.KPA.Metric + } + if options.Scale.KPA.Target != nil { + target = options.Scale.KPA.Target + } + if options.Scale.KPA.Utilization != nil { + utilization = options.Scale.KPA.Utilization + } + } + + if metric != nil { + toUpdate[autoscaling.MetricAnnotationKey] = *metric } else { toRemove = append(toRemove, autoscaling.MetricAnnotationKey) } - if options.Scale.Target != nil { - toUpdate[autoscaling.TargetAnnotationKey] = fmt.Sprintf("%f", *options.Scale.Target) + if target != nil { + toUpdate[autoscaling.TargetAnnotationKey] = fmt.Sprintf("%f", *target) } else { toRemove = append(toRemove, autoscaling.TargetAnnotationKey) } - if options.Scale.Utilization != nil { - toUpdate[autoscaling.TargetUtilizationPercentageKey] = fmt.Sprintf("%f", *options.Scale.Utilization) + if utilization != nil { + toUpdate[autoscaling.TargetUtilizationPercentageKey] = fmt.Sprintf("%f", *utilization) } else { toRemove = append(toRemove, autoscaling.TargetUtilizationPercentageKey) } diff --git a/schema/func_yaml-schema.json b/schema/func_yaml-schema.json index ee142243d3..3b12e2c68d 100644 --- a/schema/func_yaml-schema.json +++ b/schema/func_yaml-schema.json @@ -290,6 +290,79 @@ "type": "object", "description": "HealthEndpoints specify the liveness and readiness endpoints for a Runtime" }, + "KEDAScaleOptions": { + "properties": { + "triggers": { + "items": { + "$schema": "http://json-schema.org/draft-04/schema#", + "$ref": "#/definitions/KEDATrigger" + }, + "type": "array" + } + }, + "additionalProperties": false, + "type": "object" + }, + "KEDATrigger": { + "required": [ + "type" + ], + "properties": { + "type": { + "enum": [ + "http", + "kafka", + "cron" + ], + "type": "string" + }, + "lagThreshold": { + "type": "integer", + "minimum": 1 + }, + "activationLagThreshold": { + "type": "integer", + "minimum": 0 + }, + "timezone": { + "type": "string" + }, + "start": { + "type": "string" + }, + "end": { + "type": "string" + }, + "desiredReplicas": { + "type": "integer", + "minimum": 1 + } + }, + "additionalProperties": false, + "type": "object" + }, + "KPAScaleOptions": { + "properties": { + "metric": { + "enum": [ + "concurrency", + "rps" + ], + "type": "string" + }, + "target": { + "type": "number", + "minimum": 0 + }, + "utilization": { + "maximum": 100, + "minimum": 1, + "type": "number" + } + }, + "additionalProperties": false, + "type": "object" + }, "KafkaConfig": { "required": [ "brokers", @@ -566,6 +639,14 @@ "maximum": 100, "minimum": 1, "type": "number" + }, + "keda": { + "$schema": "http://json-schema.org/draft-04/schema#", + "$ref": "#/definitions/KEDAScaleOptions" + }, + "kpa": { + "$schema": "http://json-schema.org/draft-04/schema#", + "$ref": "#/definitions/KPAScaleOptions" } }, "additionalProperties": false,