Use correct describer in pipelines provider (#3600)

* Use correct describer in pipelines provider

* Update integration test to catch different describers

* Update permissions for integration tests

* fix: use in-cluster dialer for raw/keda pipeline tests

The pipeline integration test was failing for raw and keda
deployers because their services are only reachable via
cluster-internal DNS (*.default.svc). The test used a plain
http.Get which can't resolve these names from the runner.

Use an in-cluster dialer (socat pod) for raw/keda, matching
the pattern already used in deployer integration tests.

Signed-off-by: Matej Vašek <matejvasek@gmail.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

---------

Signed-off-by: Matej Vašek <matejvasek@gmail.com>
Co-authored-by: Matej Vašek <matejvasek@gmail.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: 3 people

hack/cluster.sh

@@ -529,7 +529,12 @@ tekton() {
   $KUBECTL wait pod --for=condition=Ready --timeout=180s -n tekton-pipelines -l "app=tekton-pipelines-webhook"
   sleep 10
 
+  # Grant permissions for Knative resources (services, routes, etc.)
   $KUBECTL create clusterrolebinding "${namespace}:knative-serving-namespaced-admin" --clusterrole=knative-serving-namespaced-admin --serviceaccount="${namespace}:default"
+  # Grant permissions for standard Kubernetes resources (deployments, services, etc.) needed by k8s deployer
+  $KUBECTL create clusterrolebinding "${namespace}:admin" --clusterrole=admin --serviceaccount="${namespace}:default"
+  # Grant permissions for HTTPScaledObject Keda resources needed by keda deployer
+  $KUBECTL create clusterrolebinding "${namespace}:keda-add-ons-http-operator" --clusterrole=keda-add-ons-http-operator --serviceaccount="${namespace}:default"
 
   echo "${green}✅ Tekton${reset}"
 }

pkg/functions/client.go

@@ -188,6 +188,7 @@ type Instance struct {
 	Subscriptions []Subscription    `json:"subscriptions" yaml:"subscriptions"`
 	Labels        map[string]string `json:"labels" yaml:"labels" xml:"-"`
 	Middleware    Middleware        `json:"middleware,omitempty" yaml:"middleware,omitempty"`
+	Generation    int64             `json:"generation,omitempty" yaml:"generation,omitempty"`
 }
 
 // Subscriptions currently active to event sources

pkg/k8s/describer.go

@@ -85,6 +85,7 @@ func (d *Describer) Describe(ctx context.Context, name, namespace string) (fn.In
 		Middleware: fn.Middleware{
 			Version: middlewareVersion,
 		},
+		Generation: deployment.Generation,
 	}
 
 	return description, nil

pkg/keda/describer.go

@@ -104,6 +104,7 @@ func (d *Describer) Describe(ctx context.Context, name, namespace string) (fn.In
 		Middleware: fn.Middleware{
 			Version: middlewareVersion,
 		},
+		Generation: deployment.Generation,
 	}
 
 	return description, nil

pkg/knative/describer.go

@@ -82,12 +82,13 @@ func (d *Describer) Describe(ctx context.Context, name, namespace string) (fn.In
 	}
 
 	description := fn.Instance{
-		Name:      name,
-		Namespace: namespace,
-		Deployer:  KnativeDeployerName,
-		Route:     primaryRouteURL,
-		Routes:    routeURLs,
-		Labels:    service.Labels,
+		Name:       name,
+		Namespace:  namespace,
+		Deployer:   KnativeDeployerName,
+		Route:      primaryRouteURL,
+		Routes:     routeURLs,
+		Labels:     service.Labels,
+		Generation: service.Generation,
 	}
 
 	// get used image (including the sha)

pkg/pipelines/tekton/pipelines_int_test.go

@@ -25,6 +25,7 @@ import (
 	"knative.dev/func/pkg/docker"
 	fn "knative.dev/func/pkg/functions"
 	"knative.dev/func/pkg/k8s"
+	"knative.dev/func/pkg/keda"
 	"knative.dev/func/pkg/knative"
 	"knative.dev/func/pkg/oci"
 	"knative.dev/func/pkg/pipelines/tekton"
@@ -44,26 +45,34 @@ const (
 	TestNamespace = "default"
 )
 
-func newRemoteTestClient(verbose bool, opts ...fn.Option) *fn.Client {
+func newRemoteTestClient(verbose bool, deployer string, opts ...fn.Option) *fn.Client {
 	baseOpts := []fn.Option{
 		fn.WithBuilder(buildpacks.NewBuilder(buildpacks.WithVerbose(verbose))),
 		fn.WithPusher(docker.NewPusher(docker.WithCredentialsProvider(testCP))),
-		fn.WithDeployer(knative.NewDeployer(knative.WithDeployerVerbose(verbose))),
-		fn.WithDescribers(knative.NewDescriber(verbose), k8s.NewDescriber(verbose)),
-		fn.WithListers(knative.NewLister(verbose), k8s.NewLister(verbose)),
-		fn.WithRemovers(knative.NewRemover(verbose), k8s.NewRemover(verbose)),
+		fn.WithDescribers(knative.NewDescriber(verbose), k8s.NewDescriber(verbose), keda.NewDescriber(verbose)),
+		fn.WithListers(knative.NewLister(verbose), k8s.NewLister(verbose), keda.NewLister(verbose)),
+		fn.WithRemovers(knative.NewRemover(verbose), k8s.NewRemover(verbose), keda.NewRemover(verbose)),
 		fn.WithPipelinesProvider(tekton.NewPipelinesProvider(tekton.WithCredentialsProvider(testCP), tekton.WithVerbose(verbose))),
 	}
+
+	switch deployer {
+	case k8s.KubernetesDeployerName:
+		baseOpts = append(baseOpts, fn.WithDeployer(k8s.NewDeployer(k8s.WithDeployerVerbose(verbose))))
+	case keda.KedaDeployerName:
+		baseOpts = append(baseOpts, fn.WithDeployer(keda.NewDeployer(keda.WithDeployerVerbose(verbose))))
+	case knative.KnativeDeployerName:
+		baseOpts = append(baseOpts, fn.WithDeployer(knative.NewDeployer(knative.WithDeployerVerbose(verbose))))
+	}
+
 	return fn.New(append(baseOpts, opts...)...)
 }
 
 // assertFunctionEchoes returns without error when the function of the given
 // name echoes a parameter sent via a Get request.
-func assertFunctionEchoes(url string) (err error) {
+func assertFunctionEchoes(httpClient *http.Client, url string) (err error) {
 	token := time.Now().Format("20060102150405.000000000")
 
-	// res, err := http.Get("http://testremote-default.default.localtest.me?token=" + token)
-	res, err := http.Get(url + "?token=" + token)
+	res, err := httpClient.Get(url + "?token=" + token)
 	if err != nil {
 		return
 	}
@@ -82,6 +91,28 @@ func assertFunctionEchoes(url string) (err error) {
 	return
 }
 
+// httpClientForDeployer returns an HTTP client appropriate for the deployer type.
+// Raw and keda deployers expose cluster-internal services, so an in-cluster
+// dialer is needed. Knative services are externally accessible via ingress.
+func httpClientForDeployer(t *testing.T, ctx context.Context, deployer string) *http.Client {
+	switch deployer {
+	case k8s.KubernetesDeployerName, keda.KedaDeployerName:
+		dialer, err := k8s.NewInClusterDialer(ctx, k8s.GetClientConfig())
+		if err != nil {
+			t.Fatalf("failed to create in-cluster dialer: %v", err)
+		}
+		t.Cleanup(func() { _ = dialer.Close() })
+		return &http.Client{
+			Transport: &http.Transport{
+				DialContext: dialer.DialContext,
+			},
+			Timeout: time.Minute,
+		}
+	default:
+		return http.DefaultClient
+	}
+}
+
 func tektonTestsEnabled(t *testing.T) (enabled bool) {
 	enabled, _ = strconv.ParseBool(os.Getenv("FUNC_INT_TEKTON_ENABLED"))
 	if !enabled {
@@ -114,41 +145,51 @@ func TestInt_Remote_Default(t *testing.T) {
 		t.Skip()
 	}
 	skipOnUnsupportedArch(t)
-	_ = fromCleanEnvironment(t)
-	var (
-		err         error
-		url         string
-		verbose     = false
-		ctx, cancel = signal.NotifyContext(t.Context(), os.Interrupt)
-		client      = newRemoteTestClient(verbose,
-			fn.WithRepository("https://github.com/functions-dev/templates"))
-	)
-	defer cancel()
-
-	f := fn.Function{
-		Name:      "testremote-default",
-		Runtime:   "go",
-		Template:  "echo",
-		Registry:  TestRegistry,
-		Namespace: TestNamespace,
-		Build: fn.BuildSpec{
-			Builder: "pack", // TODO: test "s2i".  Currently it causes a 'no space left on device' error in GH actions.
-		},
-	}
 
-	if f, err = client.Init(f); err != nil {
-		t.Fatal(err)
-	}
-
-	if url, f, err = client.RunPipeline(ctx, f); err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		_ = client.Remove(ctx, "", "", f, true)
-	}()
-
-	if err := assertFunctionEchoes(url); err != nil {
-		t.Fatal(err)
+	for _, d := range []string{knative.KnativeDeployerName, k8s.KubernetesDeployerName, keda.KedaDeployerName} {
+		t.Run(d, func(t *testing.T) {
+			_ = fromCleanEnvironment(t)
+			var (
+				err         error
+				url         string
+				verbose     = false
+				ctx, cancel = signal.NotifyContext(t.Context(), os.Interrupt)
+				client      = newRemoteTestClient(verbose, d,
+					fn.WithRepository("https://github.com/functions-dev/templates"))
+			)
+			defer cancel()
+
+			f := fn.Function{
+				Name:      "testremote-default",
+				Runtime:   "go",
+				Template:  "echo",
+				Registry:  TestRegistry,
+				Namespace: TestNamespace,
+				Build: fn.BuildSpec{
+					Builder: "pack", // TODO: test "s2i".  Currently it causes a 'no space left on device' error in GH actions.
+				},
+				Deploy: fn.DeploySpec{
+					Deployer: d,
+				},
+			}
+
+			if f, err = client.Init(f); err != nil {
+				t.Fatal(err)
+			}
+
+			if url, f, err = client.RunPipeline(ctx, f); err != nil {
+				t.Fatal(err)
+			}
+			defer func() {
+				_ = client.Remove(ctx, "", "", f, true)
+			}()
+
+			httpClient := httpClientForDeployer(t, ctx, d)
+
+			if err := assertFunctionEchoes(httpClient, url); err != nil {
+				t.Fatal(err)
+			}
+		})
 	}
 }
 

pkg/pipelines/tekton/pipelines_provider.go

@@ -31,6 +31,7 @@ import (
 	fn "knative.dev/func/pkg/functions"
 	"knative.dev/func/pkg/k8s"
 	fnlabels "knative.dev/func/pkg/k8s/labels"
+	"knative.dev/func/pkg/keda"
 	"knative.dev/func/pkg/knative"
 	"knative.dev/func/pkg/oci"
 	"knative.dev/pkg/apis"
@@ -236,27 +237,33 @@ func (pp *PipelinesProvider) Run(ctx context.Context, f fn.Function) (string, fn
 		return "", f, fmt.Errorf("function pipeline run has failed with message: \n\n%s", message)
 	}
 
-	kClient, err := knative.NewServingClient(namespace)
-	if err != nil {
-		return "", f, fmt.Errorf("problem in retrieving status of deployed function: %v", err)
+	var describer fn.Describer
+	switch f.Deploy.Deployer {
+	case k8s.KubernetesDeployerName:
+		describer = k8s.NewDescriber(false)
+	case keda.KedaDeployerName:
+		describer = keda.NewDescriber(false)
+	default:
+		// default to knative
+		describer = knative.NewDescriber(false)
 	}
 
-	ksvc, err := kClient.GetService(ctx, f.Name)
+	obj, err := describer.Describe(ctx, f.Name, f.Namespace)
 	if err != nil {
 		return "", f, fmt.Errorf("problem in retrieving status of deployed function: %v", err)
 	}
 
-	if ksvc.Generation == 1 {
-		fmt.Fprintf(os.Stderr, "✅ Function deployed in namespace %q and exposed at URL: \n   %s\n", ksvc.Namespace, ksvc.Status.URL.String())
+	if obj.Generation == 1 {
+		fmt.Fprintf(os.Stderr, "✅ Function deployed in namespace %q and exposed at URL: \n   %s\n", obj.Namespace, obj.Route)
 	} else {
-		fmt.Fprintf(os.Stderr, "✅ Function updated in namespace %q and exposed at URL: \n   %s\n", ksvc.Namespace, ksvc.Status.URL.String())
+		fmt.Fprintf(os.Stderr, "✅ Function updated in namespace %q and exposed at URL: \n   %s\n", obj.Namespace, obj.Route)
 	}
 
-	if ksvc.Namespace != namespace {
-		fmt.Fprintf(os.Stderr, "Warning: Final ksvc namespace %q does not match expected %q", ksvc.Namespace, namespace)
+	if obj.Namespace != namespace {
+		fmt.Fprintf(os.Stderr, "Warning: Final function namespace %q does not match expected %q", obj.Namespace, namespace)
 	}
 
-	return ksvc.Status.URL.String(), f, nil
+	return obj.Route, f, nil
 }
 
 // Creates tar stream with the function sources as they were in "./source" directory.