[WIP] Scaling with KEDA Kafka scaler

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)
 	}
 

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)")

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)

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)

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`.

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

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

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: {}