Add KEDA Kafka consumer-lag scaling support

cmd/delete_test.go

@@ -353,7 +353,13 @@ 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",
+			ActiveDeployer: keda.KedaDeployerName,
+		},
+		Scale: &fn.ScaleOptions{
+			KEDA: &fn.KEDAScaleOptions{Triggers: []fn.KEDATrigger{{Type: "http"}}},
+		},
 	}
 	f, err := fn.New().Init(f)
 	if err != nil {
@@ -382,8 +388,8 @@ func TestDelete_ByProjectClearsDeployedMarker(t *testing.T) {
 	if loaded.Deploy.Namespace != "" {
 		t.Fatalf("expected Deploy.Namespace cleared after a successful undeploy, got %q", loaded.Deploy.Namespace)
 	}
-	if loaded.Deploy.Deployer != "" {
-		t.Fatalf("expected Deploy.Deployer cleared after a successful undeploy, got %q", loaded.Deploy.Deployer)
+	if loaded.Deploy.ActiveDeployer != "" {
+		t.Fatalf("expected Deploy.Deployer cleared after a successful undeploy, got %q", loaded.Deploy.ActiveDeployer)
 	}
 	if loaded.Deployer != keda.KedaDeployerName {
 		t.Fatalf("expected the intended Deployer preserved as a remembered choice, got %q", loaded.Deployer)
@@ -401,7 +407,7 @@ func TestDelete_ByNameLeavesLocalFunctionUntouched(t *testing.T) {
 		Runtime:  "go",
 		Registry: TestRegistry,
 		Name:     "localfn",
-		Deploy:   fn.DeploySpec{Namespace: "myns", Deployer: keda.KedaDeployerName},
+		Deploy:   fn.DeploySpec{Namespace: "myns", ActiveDeployer: keda.KedaDeployerName},
 	}
 	f, err := fn.New().Init(f)
 	if err != nil {
@@ -441,7 +447,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.Scale = &fn.ScaleOptions{
+		KEDA: &fn.KEDAScaleOptions{Triggers: []fn.KEDATrigger{{Type: "http"}}},
+	}
+	if err := f.Write(); err != nil {
 		t.Fatal(err)
 	}
 
@@ -476,8 +491,8 @@ func TestDelete_ByProjectPreservesDeployerForRedeploy(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if loaded.Deploy.Deployer != keda.KedaDeployerName {
-		t.Fatalf("expected the flag-less redeploy to reuse the persisted %q deployer, got %q", keda.KedaDeployerName, loaded.Deploy.Deployer)
+	if loaded.Deploy.ActiveDeployer != keda.KedaDeployerName {
+		t.Fatalf("expected the flag-less redeploy to reuse the persisted %q deployer, got %q", keda.KedaDeployerName, loaded.Deploy.ActiveDeployer)
 	}
 }
 
@@ -490,7 +505,7 @@ func TestDelete_ByProjectThenRedeployWithDifferentDeployerNotBlocked(t *testing.
 		Root:     root,
 		Runtime:  "go",
 		Registry: TestRegistry,
-		Deploy:   fn.DeploySpec{Namespace: "myns", Deployer: keda.KedaDeployerName},
+		Deploy:   fn.DeploySpec{Namespace: "myns", ActiveDeployer: keda.KedaDeployerName},
 	}
 	f, err := fn.New().Init(f)
 	if err != nil {

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 or Kafka 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)")
@@ -306,8 +306,8 @@ func runDeploy(cmd *cobra.Command, newClient ClientFactory) (err error) {
 
 	// Back-compat: a function deployed before the deployer was recorded has a
 	// namespace but no deployer, which historically could only mean knative.
-	if f.Deploy.Namespace != "" && f.Deploy.Deployer == "" {
-		f.Deploy.Deployer = deployers.Knative
+	if f.Deploy.Namespace != "" && f.Deploy.ActiveDeployer == "" {
+		f.Deploy.ActiveDeployer = deployers.Knative
 	}
 
 	if f, err = cfg.Configure(f); err != nil { // Updates f with deploy cfg
@@ -368,7 +368,7 @@ func runDeploy(cmd *cobra.Command, newClient ClientFactory) (err error) {
 		var url string
 		// Invoke a remote build/push/deploy pipeline
 		// Returned is the function with fields like Registry, f.Deploy.Image &
-		// f.Deploy.Namespace, f.Deploy.Expose populated.
+		// f.Deploy.Namespace, f.Deploy.ActiveExpose populated.
 		if url, f, err = client.RunPipeline(cmd.Context(), f); err != nil {
 			return wrapDeploymentError(err)
 		}
@@ -377,7 +377,7 @@ func runDeploy(cmd *cobra.Command, newClient ClientFactory) (err error) {
 		// After a remote pipeline: intent was active, describer saw no
 		// applied expose. Typical cause is a func-util image that predates
 		// this field. Knative is excluded; it never applies expose.
-		if fn.ExposureRecordMissing(f.Expose, f.Deploy.Expose, f.Deploy.Deployer) {
+		if fn.ExposureRecordMissing(f.Expose, f.Deploy.ActiveExpose, f.Deploy.ActiveDeployer) {
 			fmt.Fprintf(cmd.OutOrStderr(), "Warning: expose %q was requested but the cluster's "+
 				"func-util image applied no external exposure; the function is running cluster-local\n", f.Expose)
 		}

cmd/deploy_test.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
+	"slices"
 	"strings"
 	"testing"
 	"time"
@@ -386,7 +387,7 @@ func TestDeploy_Envs(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	// Mock deployer records f.Deploy.Deployer so later redeploys are not
+	// Mock deployer records f.Deploy.ActiveDeployer so later redeploys are not
 	// treated as the pre-field legacy (knative) case.
 	clientFn := NewTestClient(fn.WithDeployer(mock.NewDeployer()))
 
@@ -1161,7 +1162,7 @@ func TestDeploy_NamespaceRedeployWarning(t *testing.T) {
 	f := fn.Function{
 		Runtime: "go",
 		Root:    root,
-		Deploy:  fn.DeploySpec{Namespace: "funcns", Deployer: deployers.Default},
+		Deploy:  fn.DeploySpec{Namespace: "funcns", ActiveDeployer: deployers.Default},
 	}
 	f, err := fn.New().Init(f)
 	if err != nil {
@@ -1211,8 +1212,8 @@ func TestDeploy_NamespaceUpdateWarning(t *testing.T) {
 		Runtime: "go",
 		Root:    root,
 		Deploy: fn.DeploySpec{
-			Namespace: "myns",
-			Deployer:  deployers.Default,
+			Namespace:      "myns",
+			ActiveDeployer: deployers.Default,
 		},
 	}
 	f, err := fn.New().Init(f)
@@ -1363,7 +1364,7 @@ func TestDeploy_NamespaceChangePreservesExternalRegistry(t *testing.T) {
 		Runtime:  "go",
 		Root:     root,
 		Registry: "docker.io/user",
-		Deploy:   fn.DeploySpec{Namespace: "ns1", Deployer: deployers.Default},
+		Deploy:   fn.DeploySpec{Namespace: "ns1", ActiveDeployer: deployers.Default},
 	}
 	f, err := fn.New().Init(f)
 	if err != nil {
@@ -1398,7 +1399,7 @@ func TestDeploy_NamespaceChangeUpdatesInternalRegistry(t *testing.T) {
 		Runtime:  "go",
 		Root:     root,
 		Registry: "image-registry.openshift-image-registry.svc:5000/ns1",
-		Deploy:   fn.DeploySpec{Namespace: "ns1", Deployer: deployers.Default},
+		Deploy:   fn.DeploySpec{Namespace: "ns1", ActiveDeployer: deployers.Default},
 	}
 	f, err := fn.New().Init(f)
 	if err != nil {
@@ -2599,8 +2600,8 @@ func TestDeploy_DeployerPersists(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		if loaded.Deploy.Deployer != deployers.Default {
-			t.Fatalf("expected persisted deployer %q, got %q", deployers.Default, loaded.Deploy.Deployer)
+		if loaded.Deploy.ActiveDeployer != deployers.Default {
+			t.Fatalf("expected persisted deployer %q, got %q", deployers.Default, loaded.Deploy.ActiveDeployer)
 		}
 	})
 
@@ -2629,8 +2630,8 @@ func TestDeploy_DeployerPersists(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		if loaded.Deploy.Deployer != other {
-			t.Fatalf("expected persisted deployer %q, got %q", other, loaded.Deploy.Deployer)
+		if loaded.Deploy.ActiveDeployer != other {
+			t.Fatalf("expected persisted deployer %q, got %q", other, loaded.Deploy.ActiveDeployer)
 		}
 
 		// no --deployer flag: the flag defaults to the persisted value
@@ -2648,8 +2649,8 @@ func TestDeploy_DeployerPersists(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		if loaded.Deploy.Deployer != other {
-			t.Fatalf("expected deployer to remain %q after a flag-less redeploy, got %q", other, loaded.Deploy.Deployer)
+		if loaded.Deploy.ActiveDeployer != other {
+			t.Fatalf("expected deployer to remain %q after a flag-less redeploy, got %q", other, loaded.Deploy.ActiveDeployer)
 		}
 	})
 }
@@ -2682,8 +2683,8 @@ func TestDeploy_DeployerGlobalConfig(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if loaded.Deploy.Deployer != k8s.KubernetesDeployerName {
-		t.Fatalf("expected the global config's deployer %q to seed the flagless deploy, got %q", k8s.KubernetesDeployerName, loaded.Deploy.Deployer)
+	if loaded.Deploy.ActiveDeployer != k8s.KubernetesDeployerName {
+		t.Fatalf("expected the global config's deployer %q to seed the flagless deploy, got %q", k8s.KubernetesDeployerName, loaded.Deploy.ActiveDeployer)
 	}
 }
 
@@ -2716,7 +2717,20 @@ func TestDeploy_DeployerSwitch(t *testing.T) {
 				Root:     root,
 				Registry: TestRegistry,
 				// Namespace set == already deployed, which is what the guard gates on.
-				Deploy: fn.DeploySpec{Namespace: "myns", Deployer: tt.deployedDep},
+				Deploy: fn.DeploySpec{Namespace: "myns", ActiveDeployer: 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.Scale = &fn.ScaleOptions{
+					KEDA: &fn.KEDAScaleOptions{Triggers: []fn.KEDATrigger{{Type: "http"}}},
+				}
 			}
 			if _, err := fn.New().Init(f); err != nil {
 				t.Fatal(err)
@@ -2824,8 +2838,8 @@ func TestDeploy_ExposeEmptyVsUnset(t *testing.T) {
 		if f.Expose != "" {
 			t.Errorf("expected intent expose empty, got %q", f.Expose)
 		}
-		if f.Deploy.Expose != "" {
-			t.Errorf("expected status expose empty, got %q", f.Deploy.Expose)
+		if f.Deploy.ActiveExpose != "" {
+			t.Errorf("expected status expose empty, got %q", f.Deploy.ActiveExpose)
 		}
 	})
 
@@ -2837,8 +2851,8 @@ func TestDeploy_ExposeEmptyVsUnset(t *testing.T) {
 			t.Fatalf("expected intent expose 'none', got %q", f.Expose)
 		}
 		// status is observed applied mode; "none"/empty both mean cluster-local
-		if f := loadFn(t, root); f.Deploy.Expose != "" {
-			t.Fatalf("expected status expose empty for cluster-local, got %q", f.Deploy.Expose)
+		if f := loadFn(t, root); f.Deploy.ActiveExpose != "" {
+			t.Fatalf("expected status expose empty for cluster-local, got %q", f.Deploy.ActiveExpose)
 		}
 
 		// redeploy without the flag should keep intent via flag default
@@ -2897,8 +2911,8 @@ func TestDeploy_ExposeRoutePersists(t *testing.T) {
 	if f.Expose != "route" {
 		t.Fatalf("expected intent expose 'route', got %q", f.Expose)
 	}
-	if f.Deploy.Expose != "route" {
-		t.Fatalf("expected status expose 'route', got %q", f.Deploy.Expose)
+	if f.Deploy.ActiveExpose != "route" {
+		t.Fatalf("expected status expose 'route', got %q", f.Deploy.ActiveExpose)
 	}
 }
 
@@ -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.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)
@@ -3016,7 +3041,7 @@ func TestDeploy_RemoteExposeRecordsObservation(t *testing.T) {
 			pipeliner.RunFn = func(f fn.Function) (string, fn.Function, error) {
 				// add exposure tracking to the base RunFn
 				url, f, err := base(f)
-				f.Deploy.Expose = tt.observed
+				f.Deploy.ActiveExpose = tt.observed
 				return url, f, err
 			}
 
@@ -3038,8 +3063,8 @@ func TestDeploy_RemoteExposeRecordsObservation(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			if f.Deploy.Expose != tt.wantRecord {
-				t.Errorf("Deploy.Expose = %q, want %q", f.Deploy.Expose, tt.wantRecord)
+			if f.Deploy.ActiveExpose != tt.wantRecord {
+				t.Errorf("Deploy.Expose = %q, want %q", f.Deploy.ActiveExpose, tt.wantRecord)
 			}
 			warned := strings.Contains(out.String(), "applied no external exposure")
 			if warned != tt.wantWarning {

cmd/func-util/main.go

@@ -159,7 +159,7 @@ func deploy(ctx context.Context) error {
 	// honors --deployer, which travels in func.yaml as intent.
 	deployer := f.Deployer
 	if deployer == "" {
-		deployer = f.Deploy.Deployer
+		deployer = f.Deploy.ActiveDeployer
 	}
 	if deployer == "" {
 		deployer = knative.KnativeDeployerName

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 or Kafka 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)