Repository: knative-sandbox/kn-plugin-func
Pull request #711 (2021-12-08 20:20:51)

feat: custom default HTTP transport

# Changes


- :gift: use in cluster dialer by default:
  This sets http.DefaultTransport to our specialized RoundTripper.
  The RoundTripper first tries dial in standard way.
  If dial operation fails due to hostname resolution error the in cluster dialer will be used.

- :broom: dialer pod auto-exit:
  The pod should exit (be Completed) when func exits (even in case of SIGKILL).



knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/kn-plugin-func/pull/711#" title="Author self-approved">matejvasek</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Fkn-plugin-func).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/kn-plugin-func/blob/main/OWNERS)~~ [matejvasek]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
matejvasek: @lkingland not this automatically enables things like:
`func emit --sink http://dialer-test-service.dialer-test-ns.svc`
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/711?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#711](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/711?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (0b5d3f8) into [main](https://codecov.io/gh/knative-sandbox/kn-plugin-func/commit/f63af0d34e97a549df51c47f12e22a7558504278?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (f63af0d) will **not change** coverage.
> The diff coverage is `n/a`.

> :exclamation: Current head 0b5d3f8 differs from pull request most recent head 4969a9d. Consider uploading reports for the commit 4969a9d to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/711/graphs/tree.svg?width=650&height=150&src=pr&token=da1B9bMsQ9&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/711?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@           Coverage Diff           @@
##             main     #711   +/-   ##
=======================================
  Coverage   39.79%   39.79%           
=======================================
  Files          42       42           
  Lines        3968     3968           
=======================================
  Hits         1579     1579           
  Misses       2168     2168           
  Partials      221      221           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/711?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/711?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [f63af0d...4969a9d](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/711?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

matejvasek: e2e are failing but that's not fault of this PR, the `main` is broken.

Patch:
diff --git a/cmd/build.go b/cmd/build.go
index 8ba3c46b..f4c3687b 100644
--- a/cmd/build.go
+++ b/cmd/build.go
@@ -28,9 +28,8 @@ func newBuildClient(cfg buildConfig) (*fn.Client, error) {
 	pusherOption := fn.WithPusher(nil)
 	if cfg.Push {
 		credentialsProvider := docker.NewCredentialsProvider(
-			newCredentialsCallback(),
-			docker.CheckAuth,
-			newChooseHelperCallback(),
+			docker.WithPromptForCredentials(newPromptForCredentials()),
+			docker.WithPromptForCredentialStore(newPromptForCredentialStore()),
 		)
 		pusher, err := docker.NewPusher(
 			docker.WithCredentialsProvider(credentialsProvider),
diff --git a/cmd/deploy.go b/cmd/deploy.go
index 8ff33687..b7cc84e0 100644
--- a/cmd/deploy.go
+++ b/cmd/deploy.go
@@ -27,9 +27,8 @@ func newDeployClient(cfg deployConfig) (*fn.Client, error) {
 	builder := buildpacks.NewBuilder()
 
 	credentialsProvider := docker.NewCredentialsProvider(
-		newCredentialsCallback(),
-		docker.CheckAuth,
-		newChooseHelperCallback())
+		docker.WithPromptForCredentials(newPromptForCredentials()),
+		docker.WithPromptForCredentialStore(newPromptForCredentialStore()))
 	pusher, err := docker.NewPusher(
 		docker.WithCredentialsProvider(credentialsProvider),
 		docker.WithProgressListener(listener))
@@ -195,7 +194,7 @@ func runDeploy(cmd *cobra.Command, _ []string, clientFn deployClientFn) (err err
 	// (for example kubectl usually uses ~/.kube/config)
 }
 
-func newCredentialsCallback() func(registry string) (docker.Credentials, error) {
+func newPromptForCredentials() func(registry string) (docker.Credentials, error) {
 	firstTime := true
 	return func(registry string) (docker.Credentials, error) {
 		var result docker.Credentials
@@ -229,7 +228,7 @@ func newCredentialsCallback() func(registry string) (docker.Credentials, error)
 	}
 }
 
-func newChooseHelperCallback() docker.ChooseCredentialHelperCallback {
+func newPromptForCredentialStore() docker.ChooseCredentialHelperCallback {
 	return func(availableHelpers []string) (string, error) {
 		if len(availableHelpers) < 1 {
 			fmt.Fprintf(os.Stderr, `Credentials will not be saved.
diff --git a/cmd/func/main.go b/cmd/func/main.go
index bc79adf8..89782a2f 100644
--- a/cmd/func/main.go
+++ b/cmd/func/main.go
@@ -2,10 +2,13 @@ package main
 
 import (
 	"context"
+	"net/http"
 	"os"
 	"os/signal"
 	"syscall"
 
+	funcHttp "knative.dev/kn-plugin-func/http"
+
 	"knative.dev/kn-plugin-func/cmd"
 )
 
@@ -14,6 +17,11 @@ import (
 var date, vers, hash string
 
 func main() {
+
+	rt := funcHttp.NewRoundTripper()
+	http.DefaultTransport = rt
+	defer rt.Close()
+
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
diff --git a/docker/credentials_helper.go b/docker/credentials_helper.go
index cb715eb0..37944b04 100644
--- a/docker/credentials_helper.go
+++ b/docker/credentials_helper.go
@@ -12,7 +12,6 @@ import (
 	"runtime"
 	"strings"
 
-	"github.com/containers/image/v5/types"
 	"github.com/docker/docker-credential-helpers/client"
 	"github.com/docker/docker-credential-helpers/credentials"
 )
@@ -65,15 +64,15 @@ func setCredentialHelperToConfig(confFilePath, helper string) error {
 	return nil
 }
 
-func getCredentialsByCredentialHelper(confFilePath, registry string) (types.DockerAuthConfig, error) {
-	result := types.DockerAuthConfig{}
+func getCredentialsByCredentialHelper(confFilePath, registry string) (Credentials, error) {
+	result := Credentials{}
 
 	helper, err := getCredentialHelperFromConfig(confFilePath)
 	if err != nil && !os.IsNotExist(err) {
-		return types.DockerAuthConfig{}, fmt.Errorf("failed to get helper from config: %w", err)
+		return result, fmt.Errorf("failed to get helper from config: %w", err)
 	}
 	if helper == "" {
-		return types.DockerAuthConfig{}, errCredentialsNotFound
+		return result, errCredentialsNotFound
 	}
 
 	helperName := fmt.Sprintf("docker-credential-%s", helper)
diff --git a/docker/pusher.go b/docker/pusher.go
index b892d2f1..ea5b5f1f 100644
--- a/docker/pusher.go
+++ b/docker/pusher.go
@@ -14,6 +14,8 @@ import (
 	"regexp"
 	"strings"
 
+	"github.com/containers/image/v5/pkg/docker/config"
+
 	"github.com/google/go-containerregistry/pkg/authn"
 	"github.com/google/go-containerregistry/pkg/name"
 	"github.com/google/go-containerregistry/pkg/v1/remote/transport"
@@ -22,7 +24,6 @@ import (
 
 	fn "knative.dev/kn-plugin-func"
 
-	"github.com/containers/image/v5/pkg/docker/config"
 	containersTypes "github.com/containers/image/v5/types"
 	"github.com/docker/docker/api/types"
 )
@@ -42,9 +43,9 @@ var ErrUnauthorized = errors.New("bad credentials")
 
 // VerifyCredentialsCallback checks if credentials are accepted by the registry.
 // If credentials are incorrect this callback shall return ErrUnauthorized.
-type VerifyCredentialsCallback func(ctx context.Context, username, password, registry string) error
+type VerifyCredentialsCallback func(ctx context.Context, registry string, credentials Credentials) error
 
-func CheckAuth(ctx context.Context, username, password, registry string) error {
+func CheckAuth(ctx context.Context, registry string, credentials Credentials) error {
 	serverAddress := registry
 	if !strings.HasPrefix(serverAddress, "https://") && !strings.HasPrefix(serverAddress, "http://") {
 		serverAddress = "https://" + serverAddress
@@ -53,8 +54,8 @@ func CheckAuth(ctx context.Context, username, password, registry string) error {
 	url := fmt.Sprintf("%s/v2", serverAddress)
 
 	authenticator := &authn.Basic{
-		Username: username,
-		Password: password,
+		Username: credentials.Username,
+		Password: credentials.Password,
 	}
 
 	reg, err := name.NewRegistry(registry)
@@ -92,22 +93,74 @@ func CheckAuth(ctx context.Context, username, password, registry string) error {
 
 type ChooseCredentialHelperCallback func(available []string) (string, error)
 
-// NewCredentialsProvider returns new CredentialsProvider that tires to get credentials from `docker` and `func` config files.
+type credentialProviderConfig struct {
+	promptForCredentials        CredentialsCallback
+	verifyCredentials           VerifyCredentialsCallback
+	promptForCredentialStore    ChooseCredentialHelperCallback
+	additionalCredentialLoaders []CredentialsCallback
+}
+
+type CredentialProviderOptions func(opts *credentialProviderConfig)
+
+// WithPromptForCredentials sets custom callback that is supposed to
+// interactively ask for credentials in case the credentials cannot be found in configuration files.
+// The callback may be called multiple times in case incorrect credentials were returned before.
+func WithPromptForCredentials(cbk CredentialsCallback) CredentialProviderOptions {
+	return func(opts *credentialProviderConfig) {
+		opts.promptForCredentials = cbk
+	}
+}
+
+// WithVerifyCredentials sets custom callback for credentials validation.
+func WithVerifyCredentials(cbk VerifyCredentialsCallback) CredentialProviderOptions {
+	return func(opts *credentialProviderConfig) {
+		opts.verifyCredentials = cbk
+	}
+}
+
+// WithPromptForCredentialStore sets custom callback that is supposed to
+// interactively ask user which credentials store/helper is used to store credentials obtained
+// from user.
+func WithPromptForCredentialStore(cbk ChooseCredentialHelperCallback) CredentialProviderOptions {
+	return func(opts *credentialProviderConfig) {
+		opts.promptForCredentialStore = cbk
+	}
+}
+
+// WithAdditionalCredentialLoaders adds custom callbacks for credential retrieval.
+// The callbacks are supposed to be non-interactive as opposed to WithPromptForCredentials.
+//
+// This might be useful when credentials are shared with some other service.
+//
+// Example: OpenShift builtin registry shares credentials with the cluster (k8s) credentials.
+func WithAdditionalCredentialLoaders(loaders ...CredentialsCallback) CredentialProviderOptions {
+	return func(opts *credentialProviderConfig) {
+		opts.additionalCredentialLoaders = append(opts.additionalCredentialLoaders, loaders...)
+	}
+}
+
+// NewCredentialsProvider returns new CredentialsProvider that tires to get credentials from docker/func config files.
 //
 // In case getting credentials from the config files fails
-// the caller provided callback will be invoked to obtain credentials.
-// The callback may be called multiple times in case it returned credentials that are not correct (see verifyCredentials).
+// the caller provided callback (see WithPromptForCredentials) will be invoked to obtain credentials.
+// The callback may be called multiple times in case the returned credentials
+// are not correct (see WithVerifyCredentials).
 //
-// When the callback succeeds the credentials will be saved by using helper defined in the `func` config.
-// If the helper is not defined in the config the chooseCredentialHelper parameter will be used to pick one.
+// When the callback succeeds the credentials will be saved by using helper defined in the func config.
+// If the helper is not defined in the config file
+// it may be picked by provided callback (see WithPromptForCredentialStore).
 // The picked value will be saved in the func config.
 //
-// To verify that credentials are correct the verifyCredentials parameter is used.
-// If verifyCredentials is not set then CheckAuth will be used as a fallback.
-func NewCredentialsProvider(
-	getCredentials CredentialsCallback,
-	verifyCredentials VerifyCredentialsCallback,
-	chooseCredentialHelper ChooseCredentialHelperCallback) CredentialsProvider {
+// To verify that credentials are correct callback will be used (see WithVerifyCredentials).
+// If the callback is not set then CheckAuth will be used as a fallback.
+func NewCredentialsProvider(opts ...CredentialProviderOptions) CredentialsProvider {
+	var conf credentialProviderConfig
+
+	for _, o := range opts {
+		o(&conf)
+	}
+
+	askUser, verifyCredentials, chooseCredentialHelper := conf.promptForCredentials, conf.verifyCredentials, conf.promptForCredentialStore
 
 	if verifyCredentials == nil {
 		verifyCredentials = CheckAuth
@@ -131,31 +184,40 @@ func NewCredentialsProvider(
 	}
 	dockerConfigPath := filepath.Join(home, ".docker", "config.json")
 
+	var authLoaders = []CredentialsCallback{
+		func(registry string) (Credentials, error) {
+			creds, err := config.GetCredentials(sys, registry)
+			if err != nil {
+				return Credentials{}, err
+			}
+			return Credentials{
+				Username: creds.Username,
+				Password: creds.Password,
+			}, nil
+		},
+		func(registry string) (Credentials, error) {
+			return getCredentialsByCredentialHelper(authFilePath, registry)
+		},
+		func(registry string) (Credentials, error) {
+			return getCredentialsByCredentialHelper(dockerConfigPath, registry)
+		},
+	}
+
+	authLoaders = append(conf.additionalCredentialLoaders, authLoaders...)
+
 	return func(ctx context.Context, registry string) (Credentials, error) {
 		result := Credentials{}
 
-		for _, load := range []func() (containersTypes.DockerAuthConfig, error){
-			func() (containersTypes.DockerAuthConfig, error) {
-				return config.GetCredentials(sys, registry)
-			},
-			func() (containersTypes.DockerAuthConfig, error) {
-				return getCredentialsByCredentialHelper(authFilePath, registry)
-			},
-			func() (containersTypes.DockerAuthConfig, error) {
-				return getCredentialsByCredentialHelper(dockerConfigPath, registry)
-			},
-		} {
-			var credentials containersTypes.DockerAuthConfig
-			credentials, err = load()
+		for _, load := range authLoaders {
+
+			result, err = load(registry)
 
 			if err != nil && !errors.Is(err, errCredentialsNotFound) {
 				return Credentials{}, err
 			}
 
-			if credentials != (containersTypes.DockerAuthConfig{}) {
-				result.Username, result.Password = credentials.Username, credentials.Password
-
-				err = verifyCredentials(ctx, result.Username, result.Password, registry)
+			if result != (Credentials{}) {
+				err = verifyCredentials(ctx, registry, result)
 				if err == nil {
 					return result, nil
 				} else {
@@ -167,12 +229,12 @@ func NewCredentialsProvider(
 		}
 
 		for {
-			result, err = getCredentials(registry)
+			result, err = askUser(registry)
 			if err != nil {
 				return Credentials{}, err
 			}
 
-			err = verifyCredentials(ctx, result.Username, result.Password, registry)
+			err = verifyCredentials(ctx, registry, result)
 			if err == nil {
 				err = setCredentialsByCredentialHelper(authFilePath, registry, result.Username, result.Password)
 				if err != nil {
diff --git a/http/transport.go b/http/transport.go
new file mode 100644
index 00000000..7ac5acad
--- /dev/null
+++ b/http/transport.go
@@ -0,0 +1,92 @@
+package http
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"sync"
+	"time"
+
+	"knative.dev/kn-plugin-func/k8s"
+)
+
+type RoundTripCloser interface {
+	http.RoundTripper
+	io.Closer
+}
+
+// NewRoundTripper returns new closable RoundTripper that first tries to dial connection in standard way,
+// if the dial operation fails due to hostname resolution the RoundTripper tries to dial from in cluster pod.
+//
+// This is useful for accessing cluster internal services (pushing a CloudEvent into Knative broker).
+func NewRoundTripper() RoundTripCloser {
+	d := &dialer{
+		netDialer: net.Dialer{
+			Timeout:   30 * time.Second,
+			KeepAlive: 30 * time.Second,
+		},
+	}
+	return &roundTripCloser{
+		Transport: http.Transport{
+			Proxy:                 http.ProxyFromEnvironment,
+			DialContext:           d.DialContext,
+			ForceAttemptHTTP2:     true,
+			MaxIdleConns:          100,
+			IdleConnTimeout:       90 * time.Second,
+			TLSHandshakeTimeout:   10 * time.Second,
+			ExpectContinueTimeout: 1 * time.Second,
+		},
+		d: d,
+	}
+}
+
+type roundTripCloser struct {
+	http.Transport
+	d *dialer
+}
+
+func (r *roundTripCloser) Close() error {
+	return r.d.Close()
+}
+
+type dialer struct {
+	o               sync.Once
+	netDialer       net.Dialer
+	inClusterDialer k8s.ContextDialer
+}
+
+func (d *dialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
+	conn, err := d.netDialer.DialContext(ctx, network, address)
+	if err == nil {
+		return conn, nil
+	}
+
+	var dnsErr *net.DNSError
+	if !(errors.As(err, &dnsErr) && dnsErr.IsNotFound) {
+		return nil, err
+	}
+	err = nil
+
+	d.o.Do(func() {
+		d.inClusterDialer, err = k8s.NewInClusterDialer(ctx)
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	if d.inClusterDialer == nil {
+		return nil, errors.New("failed to init in cluster dialer")
+	}
+
+	return d.inClusterDialer.DialContext(ctx, network, address)
+}
+
+func (d *dialer) Close() error {
+	if d.inClusterDialer != nil {
+		return d.inClusterDialer.Close()
+	}
+	return nil
+}
diff --git a/k8s/dialer.go b/k8s/dialer.go
index 9af287bc..263d1bf6 100644
--- a/k8s/dialer.go
+++ b/k8s/dialer.go
@@ -49,7 +49,9 @@ type ContextDialer interface {
 //         Transport: transport,
 //     }
 func NewInClusterDialer(ctx context.Context) (ContextDialer, error) {
-	c := &contextDialer{}
+	c := &contextDialer{
+		detachChan: make(chan struct{}),
+	}
 	err := c.startDialerPod(ctx)
 	if err != nil {
 		return nil, err
@@ -58,10 +60,11 @@ func NewInClusterDialer(ctx context.Context) (ContextDialer, error) {
 }
 
 type contextDialer struct {
-	coreV1    v1.CoreV1Interface
-	restConf  *restclient.Config
-	podName   string
-	namespace string
+	coreV1     v1.CoreV1Interface
+	restConf   *restclient.Config
+	podName    string
+	namespace  string
+	detachChan chan struct{}
 }
 
 func (c *contextDialer) DialContext(ctx context.Context, network string, addr string) (net.Conn, error) {
@@ -87,6 +90,9 @@ func (c *contextDialer) DialContext(ctx context.Context, network string, addr st
 }
 
 func (c *contextDialer) Close() error {
+	// closing the channel will cause stdin of the attached container to return EOF
+	// as a result the pod exits -- it transits to Completed state
+	close(c.detachChan)
 	ctx, cancel := context.WithTimeout(context.Background(), time.Minute*1)
 	defer cancel()
 	delOpts := metaV1.DeleteOptions{}
@@ -136,10 +142,11 @@ func (c *contextDialer) startDialerPod(ctx context.Context) (err error) {
 		Spec: coreV1.PodSpec{
 			Containers: []coreV1.Container{
 				{
-					Name:    c.podName,
-					Image:   socatImage,
-					Stdin:   true,
-					Command: []string{"sleep", "infinity"},
+					Name:      c.podName,
+					Image:     socatImage,
+					Stdin:     true,
+					StdinOnce: true,
+					Args:      []string{"-u", "-", "OPEN:/dev/null,append"},
 				},
 			},
 			DNSPolicy:     coreV1.DNSClusterFirst,
@@ -164,10 +171,24 @@ func (c *contextDialer) startDialerPod(ctx context.Context) (err error) {
 	}
 
 	if err != nil {
-		err = fmt.Errorf("failed to start dialer container: %w", err)
+		return fmt.Errorf("failed to start dialer container: %w", err)
 	}
 
-	return err
+	// attaching to the stdin to automatically Complete the pod on exit
+	go func() {
+		_ = c.attach(emptyBlockingReader(c.detachChan), io.Discard, io.Discard)
+	}()
+
+	return nil
+}
+
+// reader that returns no data and blocks until
+// the channel is closed or data are sent to the channel
+type emptyBlockingReader chan struct{}
+
+func (e emptyBlockingReader) Read(p []byte) (n int, err error) {
+	<-e
+	return 0, io.EOF
 }
 
 func (c *contextDialer) exec(hostPort string, in io.Reader, out, errOut io.Writer) error {
@@ -200,6 +221,35 @@ func (c *contextDialer) exec(hostPort string, in io.Reader, out, errOut io.Write
 	})
 }
 
+func (c *contextDialer) attach(in io.Reader, out, errOut io.Writer) error {
+
+	restClient := c.coreV1.RESTClient()
+	req := restClient.Post().
+		Resource("pods").
+		Name(c.podName).
+		Namespace(c.namespace).
+		SubResource("attach")
+	req.VersionedParams(&coreV1.PodAttachOptions{
+		Container: c.podName,
+		Stdin:     true,
+		Stdout:    true,
+		Stderr:    true,
+		TTY:       false,
+	}, scheme.ParameterCodec)
+
+	executor, err := remotecommand.NewSPDYExecutor(c.restConf, "POST", req.URL())
+	if err != nil {
+		return err
+	}
+
+	return executor.Stream(remotecommand.StreamOptions{
+		Stdin:  in,
+		Stdout: out,
+		Stderr: errOut,
+		Tty:    false,
+	})
+}
+
 func (c *contextDialer) podReady(ctx context.Context) (errChan <-chan error) {
 	d := make(chan error)
 	errChan = d


Test patch:
diff --git a/docker/pusher_test.go b/docker/pusher_test.go
index d504cc21..ed8a11da 100644
--- a/docker/pusher_test.go
+++ b/docker/pusher_test.go
@@ -107,10 +107,11 @@ func TestNewCredentialsProvider(t *testing.T) {
 	}
 
 	type args struct {
-		credentialsCallback CredentialsCallback
-		verifyCredentials   VerifyCredentialsCallback
-		registry            string
-		setUpEnv            setUpEnv
+		promptUser        CredentialsCallback
+		verifyCredentials VerifyCredentialsCallback
+		additionalLoaders []CredentialsCallback
+		registry          string
+		setUpEnv          setUpEnv
 	}
 	tests := []struct {
 		name string
@@ -120,47 +121,47 @@ func TestNewCredentialsProvider(t *testing.T) {
 		{
 			name: "test user callback correct password on first try",
 			args: args{
-				credentialsCallback: correctPwdCallback,
-				verifyCredentials:   correctVerifyCbk,
-				registry:            "docker.io",
+				promptUser:        correctPwdCallback,
+				verifyCredentials: correctVerifyCbk,
+				registry:          "docker.io",
 			},
 			want: Credentials{Username: dockerIoUser, Password: dockerIoUserPwd},
 		},
 		{
 			name: "test user callback correct password on second try",
 			args: args{
-				credentialsCallback: pwdCbkFirstWrongThenCorrect(t),
-				verifyCredentials:   correctVerifyCbk,
-				registry:            "docker.io",
+				promptUser:        pwdCbkFirstWrongThenCorrect(t),
+				verifyCredentials: correctVerifyCbk,
+				registry:          "docker.io",
 			},
 			want: Credentials{Username: dockerIoUser, Password: dockerIoUserPwd},
 		},
 		{
 			name: "get quay-io credentials with func config populated",
 			args: args{
-				credentialsCallback: pwdCbkThatShallNotBeCalled(t),
-				verifyCredentials:   correctVerifyCbk,
-				registry:            "quay.io",
-				setUpEnv:            withPopulatedFuncAuthConfig,
+				promptUser:        pwdCbkThatShallNotBeCalled(t),
+				verifyCredentials: correctVerifyCbk,
+				registry:          "quay.io",
+				setUpEnv:          withPopulatedFuncAuthConfig,
 			},
 			want: Credentials{Username: quayIoUser, Password: quayIoUserPwd},
 		},
 		{
 			name: "get docker-io credentials with func config populated",
 			args: args{
-				credentialsCallback: pwdCbkThatShallNotBeCalled(t),
-				verifyCredentials:   correctVerifyCbk,
-				registry:            "docker.io",
-				setUpEnv:            withPopulatedFuncAuthConfig,
+				promptUser:        pwdCbkThatShallNotBeCalled(t),
+				verifyCredentials: correctVerifyCbk,
+				registry:          "docker.io",
+				setUpEnv:          withPopulatedFuncAuthConfig,
 			},
 			want: Credentials{Username: dockerIoUser, Password: dockerIoUserPwd},
 		},
 		{
 			name: "get quay-io credentials with docker config populated",
 			args: args{
-				credentialsCallback: pwdCbkThatShallNotBeCalled(t),
-				verifyCredentials:   correctVerifyCbk,
-				registry:            "quay.io",
+				promptUser:        pwdCbkThatShallNotBeCalled(t),
+				verifyCredentials: correctVerifyCbk,
+				registry:          "quay.io",
 				setUpEnv: all(
 					withPopulatedDockerAuthConfig,
 					setUpMockHelper("docker-credential-mock", helperWithQuayIO)),
@@ -170,10 +171,20 @@ func TestNewCredentialsProvider(t *testing.T) {
 		{
 			name: "get docker-io credentials with docker config populated",
 			args: args{
-				credentialsCallback: pwdCbkThatShallNotBeCalled(t),
-				verifyCredentials:   correctVerifyCbk,
-				registry:            "docker.io",
-				setUpEnv:            withPopulatedDockerAuthConfig,
+				promptUser:        pwdCbkThatShallNotBeCalled(t),
+				verifyCredentials: correctVerifyCbk,
+				registry:          "docker.io",
+				setUpEnv:          withPopulatedDockerAuthConfig,
+			},
+			want: Credentials{Username: dockerIoUser, Password: dockerIoUserPwd},
+		},
+		{
+			name: "get docker-io credentials from custom loader",
+			args: args{
+				promptUser:        pwdCbkThatShallNotBeCalled(t),
+				verifyCredentials: correctVerifyCbk,
+				registry:          "docker.io",
+				additionalLoaders: []CredentialsCallback{correctPwdCallback},
 			},
 			want: Credentials{Username: dockerIoUser, Password: dockerIoUserPwd},
 		},
@@ -186,7 +197,10 @@ func TestNewCredentialsProvider(t *testing.T) {
 				defer tt.args.setUpEnv(t)()
 			}
 
-			credentialsProvider := NewCredentialsProvider(tt.args.credentialsCallback, tt.args.verifyCredentials, nil)
+			credentialsProvider := NewCredentialsProvider(
+				WithPromptForCredentials(tt.args.promptUser),
+				WithVerifyCredentials(tt.args.verifyCredentials),
+				WithAdditionalCredentialLoaders(tt.args.additionalLoaders...))
 			got, err := credentialsProvider(context.Background(), tt.args.registry)
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
@@ -228,7 +242,10 @@ func TestCredentialsProviderSavingFromUserInput(t *testing.T) {
 		return "", errors.New("this callback shall not be invoked")
 	}
 
-	credentialsProvider := NewCredentialsProvider(pwdCbk, correctVerifyCbk, chooseNoStore)
+	credentialsProvider := NewCredentialsProvider(
+		WithPromptForCredentials(pwdCbk),
+		WithVerifyCredentials(correctVerifyCbk),
+		WithPromptForCredentialStore(chooseNoStore))
 	_, err := credentialsProvider(context.Background(), "docker.io")
 	if err != nil {
 		t.Errorf("unexpected error: %v", err)
@@ -244,7 +261,10 @@ func TestCredentialsProviderSavingFromUserInput(t *testing.T) {
 	if credsInStore != 0 {
 		t.Errorf("expected to have zero credentials in store, but has: %d", credsInStore)
 	}
-	credentialsProvider = NewCredentialsProvider(pwdCbk, correctVerifyCbk, chooseMockStore)
+	credentialsProvider = NewCredentialsProvider(
+		WithPromptForCredentials(pwdCbk),
+		WithVerifyCredentials(correctVerifyCbk),
+		WithPromptForCredentialStore(chooseMockStore))
 	_, err = credentialsProvider(context.Background(), "docker.io")
 	if err != nil {
 		t.Errorf("unexpected error: %v", err)
@@ -263,9 +283,10 @@ func TestCredentialsProviderSavingFromUserInput(t *testing.T) {
 	if len(l) != 1 {
 		t.Errorf("expected to have exactly one credentials in store, but has: %d", credsInStore)
 	}
-	credentialsProvider = NewCredentialsProvider(pwdCbkThatShallNotBeCalled(t),
-		correctVerifyCbk,
-		shallNotBeInvoked)
+	credentialsProvider = NewCredentialsProvider(
+		WithPromptForCredentials(pwdCbkThatShallNotBeCalled(t)),
+		WithVerifyCredentials(correctVerifyCbk),
+		WithPromptForCredentialStore(shallNotBeInvoked))
 	_, err = credentialsProvider(context.Background(), "docker.io")
 	if err != nil {
 		t.Errorf("unexpected error: %v", err)
@@ -381,7 +402,8 @@ func correctPwdCallback(registry string) (Credentials, error) {
 	return Credentials{}, errors.New("this cbk don't know the pwd")
 }
 
-func correctVerifyCbk(ctx context.Context, username, password, registry string) error {
+func correctVerifyCbk(ctx context.Context, registry string, creds Credentials) error {
+	username, password := creds.Username, creds.Password
 	if username == dockerIoUser && password == dockerIoUserPwd && registry == "docker.io" {
 		return nil
 	}
@@ -738,7 +760,11 @@ func TestCheckAuth(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if err := CheckAuth(tt.args.ctx, tt.args.username, tt.args.password, tt.args.registry); (err != nil) != tt.wantErr {
+			creds := Credentials{
+				Username: tt.args.username,
+				Password: tt.args.password,
+			}
+			if err := CheckAuth(tt.args.ctx, tt.args.registry, creds); (err != nil) != tt.wantErr {
 				t.Errorf("CheckAuth() error = %v, wantErr %v", err, tt.wantErr)
 			}
 		})
diff --git a/k8s/dialer_test.go b/k8s/dialer_test.go
index 17e30923..c98d10b8 100644
--- a/k8s/dialer_test.go
+++ b/k8s/dialer_test.go
@@ -13,12 +13,14 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"knative.dev/kn-plugin-func/k8s"
 
 	coreV1 "k8s.io/api/core/v1"
 	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/util/rand"
 )
 
 func TestDialInClusterService(t *testing.T) {
@@ -35,7 +37,7 @@ func TestDialInClusterService(t *testing.T) {
 
 	testingNS := &coreV1.Namespace{
 		ObjectMeta: metaV1.ObjectMeta{
-			Name: "dialer-test-ns",
+			Name: "dialer-test-ns-" + rand.String(5),
 		},
 		Spec: coreV1.NamespaceSpec{},
 	}
@@ -44,7 +46,9 @@ func TestDialInClusterService(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer cliSet.CoreV1().Namespaces().Delete(ctx, testingNS.Name, deleteOpts)
+	t.Cleanup(func() {
+		cliSet.CoreV1().Namespaces().Delete(ctx, testingNS.Name, deleteOpts)
+	})
 	t.Log("created namespace: ", testingNS.Name)
 
 	nginxPod := &coreV1.Pod{
@@ -69,7 +73,9 @@ func TestDialInClusterService(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer cliSet.CoreV1().Pods(testingNS.Name).Delete(ctx, nginxPod.Name, deleteOpts)
+	t.Cleanup(func() {
+		cliSet.CoreV1().Pods(testingNS.Name).Delete(ctx, nginxPod.Name, deleteOpts)
+	})
 	t.Log("created pod: ", nginxPod.Name)
 
 	nginxService := &coreV1.Service{
@@ -95,14 +101,21 @@ func TestDialInClusterService(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer cliSet.CoreV1().Services(testingNS.Name).Delete(ctx, nginxService.Name, deleteOpts)
+	t.Cleanup(func() {
+		cliSet.CoreV1().Services(testingNS.Name).Delete(ctx, nginxService.Name, deleteOpts)
+	})
 	t.Log("created svc: ", nginxService.Name)
 
+	// wait for service to start
+	time.Sleep(time.Second * 10)
+
 	dialer, err := k8s.NewInClusterDialer(ctx)
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer dialer.Close()
+	t.Cleanup(func() {
+		dialer.Close()
+	})
 
 	transport := &http.Transport{
 		DialContext: dialer.DialContext,
@@ -154,7 +167,9 @@ func TestDialUnreachable(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer dialer.Close()
+	t.Cleanup(func() {
+		dialer.Close()
+	})
 
 	transport := &http.Transport{
 		DialContext: dialer.DialContext,