Repository: apache/camel-k
Pull request #3322 (2022-06-01 08:35:29)

[TEST] Add autoscaling test
<!-- Description -->




<!--
Enter your extended release note in the below block. If the PR requires
additional action from users switching to the new release, include the string
"action required". If no release note is required, write "NONE". 

You can (optionally) mark this PR with labels "kind/bug" or "kind/feature" to make sure
the text is added to the right section of the release notes. 
-->

**Release Note**
```release-note
NONE
```



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 042111400..d9bfa41fb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,7 @@
 - Kamel run default service nodePort is invalid [\#3296](https://github.com/apache/camel-k/issues/3296)
 - Cannot find RestProducerFactory in Registry or as a Component to use" [\#3290](https://github.com/apache/camel-k/issues/3290)
 - Never stale a bug issue [\#3283](https://github.com/apache/camel-k/issues/3283)
+- Moving main to 1.10.0-SNAPSHOT [\#3280](https://github.com/apache/camel-k/issues/3280)
 - Release 1.9.2 [\#3275](https://github.com/apache/camel-k/issues/3275)
 - Add e2e test for kamel run --dev in a warmed-up environment [\#3211](https://github.com/apache/camel-k/issues/3211)
 - Access channel in different namespace [\#3009](https://github.com/apache/camel-k/issues/3009)
diff --git a/docs/modules/ROOT/pages/contributing/local-development.adoc b/docs/modules/ROOT/pages/contributing/local-development.adoc
index 547b2295b..f2774ea01 100644
--- a/docs/modules/ROOT/pages/contributing/local-development.adoc
+++ b/docs/modules/ROOT/pages/contributing/local-development.adoc
@@ -65,6 +65,21 @@ Camel K integrations are based on https://github.com/apache/camel-k-runtime[Came
 ./kamel install --skip-operator-setup -n operator-test --registry localhost:5000 --force --runtime-version $version
 ----
 
-The `$version` variable must be replaced with the version you are building. For example, `1.3.1-SNAPSHOT`. With these instructions, the operator will pick up and use the snapshot version you have released locally.
+The `$version` variable must be replaced with the version you are building. For example, `1.3.1-SNAPSHOT`. With these instructions, the operator will pick up and use the snapshot version you have released locally. In order to use the local maven repository, you will also need to edit your IntegrationPlatform as follow:
+----
+$ k edit ip -n operator-test
+
+...
+  spec:
+    build:
+      maven:
+        cliOptions:
+        - -V
+        localRepository: /home/user/.m2/repository
+        settings: {}
+...
+----
+pointing the `localRepository` where your local maven is storing the artifacts (it will look for the camel-k-runtime dependencies there).
+
 
 Alternatively, if no local registry is available, you can use another type of registry as explained in xref:installation/registry/registry.adoc[the Registry section].
diff --git a/examples/wiretap/README.md b/examples/wiretap/README.md
new file mode 100644
index 000000000..daed2a363
--- /dev/null
+++ b/examples/wiretap/README.md
@@ -0,0 +1,3 @@
+# Camel K WireTap example
+
+Find useful examples about how use a "WireTap" EIP in a Camel K integration.
diff --git a/examples/wiretap/WireTap.java b/examples/wiretap/WireTap.java
new file mode 100644
index 000000000..556b524d5
--- /dev/null
+++ b/examples/wiretap/WireTap.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// To run this integrations use:
+//
+// kamel run WireTap.java --dev
+//
+
+import org.apache.camel.builder.RouteBuilder;
+
+public class WireTap extends RouteBuilder {
+
+  @Override
+  public void configure() throws Exception {
+	  from("timer:tick")
+	    .wireTap("direct:tap")
+        .log("Hello from the timer!");
+
+      from("direct:tap")
+        .log("Hello from the tap!");
+  }
+}
diff --git a/go.mod b/go.mod
index 1fd5047e7..33bc7bbd0 100644
--- a/go.mod
+++ b/go.mod
@@ -45,6 +45,7 @@ require (
 	golang.org/x/oauth2 v0.0.0-20220411215720-9780585627b5
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211
+	golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11 // indirect
 	gopkg.in/inf.v0 v0.9.1
 	gopkg.in/yaml.v2 v2.4.0
 	k8s.io/api v0.22.5
diff --git a/pkg/cmd/run.go b/pkg/cmd/run.go
index a51154c5e..ae9e3384c 100644
--- a/pkg/cmd/run.go
+++ b/pkg/cmd/run.go
@@ -34,6 +34,7 @@ import (
 	"io"
 	"io/fs"
 	"io/ioutil"
+	"net/url"
 	"os"
 	"os/signal"
 	"path/filepath"
@@ -95,7 +96,7 @@ func newCmdRun(rootCmdOptions *RootCmdOptions) (*cobra.Command, *runCmdOptions)
 
 	cmd.Flags().String("name", "", "The integration name")
 	cmd.Flags().StringArrayP("connect", "c", nil, "A Service that the integration should bind to, specified as [[apigroup/]version:]kind:[namespace/]name")
-	cmd.Flags().StringArrayP("dependency", "d", nil, "A dependency that should be included, e.g., \"-d camel-mail\" for a Camel component, \"-d mvn:org.my:app:1.0\" for a Maven dependency or \"file://localPath[:targetPath]\" for local files (experimental)")
+	cmd.Flags().StringArrayP("dependency", "d", nil, "A dependency that should be included, e.g., \"-d camel-mail\" for a Camel component, \"-d mvn:org.my:app:1.0\" for a Maven dependency or \"file://localPath[?targetPath=<path>&registry=<registry URL>&skipChecksums=<true>&skipPOM=<true>]\" for local files (experimental)")
 	cmd.Flags().BoolP("wait", "w", false, "Wait for the integration to be running")
 	cmd.Flags().StringP("kit", "k", "", "The kit used to run the integration")
 	cmd.Flags().StringArrayP("property", "p", nil, "Add a runtime property or properties file (syntax: [my-key=my-value|file:/path/to/my-conf.properties])")
@@ -155,6 +156,7 @@ type runCmdOptions struct {
 	Labels          []string `mapstructure:"labels" yaml:",omitempty"`
 	Annotations     []string `mapstructure:"annotations" yaml:",omitempty"`
 	Sources         []string `mapstructure:"sources" yaml:",omitempty"`
+	RegistryOptions url.Values
 }
 
 func (o *runCmdOptions) preRunE(cmd *cobra.Command, args []string) error {
@@ -801,9 +803,44 @@ func resolvePodTemplate(ctx context.Context, cmd *cobra.Command, templateSrc str
 	return err
 }
 
+func parseFileURI(uri string) *url.URL {
+	file := new(url.URL)
+	file.Scheme = "file"
+	path := strings.TrimPrefix(uri, "file://")
+	i := strings.IndexByte(path, '?')
+	if i > 0 {
+		file.Path = path[:i]
+		file.RawQuery = path[i+1:]
+	} else {
+		file.Path = path
+	}
+	return file
+}
+
+func (o *runCmdOptions) getRegistry(platform *v1.IntegrationPlatform) string {
+	registry := o.RegistryOptions.Get("registry")
+	if registry != "" {
+		return registry
+	}
+	return platform.Status.Build.Registry.Address
+}
+
+func (o *runCmdOptions) skipChecksums() bool {
+	return o.RegistryOptions.Get("skipChecksums") == "true"
+}
+
+func (o *runCmdOptions) skipPom() bool {
+	return o.RegistryOptions.Get("skipPOM") == "true"
+}
+
+func (o *runCmdOptions) getTargetPath() string {
+	return o.RegistryOptions.Get("targetPath")
+}
+
 func (o *runCmdOptions) uploadFileOrDirectory(platform *v1.IntegrationPlatform, item string, integrationName string, cmd *cobra.Command, integration *v1.Integration) error {
-	path := strings.TrimPrefix(item, "file://")
-	localPath, targetPath := getPaths(path, runtimeos.GOOS, filepath.IsAbs(path))
+	uri := parseFileURI(item)
+	o.RegistryOptions = uri.Query()
+	localPath, targetPath := uri.Path, o.getTargetPath()
 	options := o.getSpectrumOptions(platform, cmd)
 	dirName, err := getDirName(localPath)
 	if err != nil {
@@ -849,23 +886,6 @@ func (o *runCmdOptions) uploadFileOrDirectory(platform *v1.IntegrationPlatform,
 	})
 }
 
-func getPaths(path string, os string, isAbs bool) (localPath string, targetPath string) {
-	localPath = path
-	targetPath = ""
-	parts := strings.Split(path, ":")
-	if len(parts) > 1 {
-		if os != "windows" || !isAbs {
-			localPath = parts[0]
-			targetPath = parts[1]
-		} else if isAbs && len(parts) == 3 {
-			// special case on Windows for absolute paths e.g C:\foo\bar\test.csv:remote/path
-			localPath = fmt.Sprintf("%s:%s", parts[0], parts[1])
-			targetPath = parts[2]
-		}
-	}
-	return localPath, targetPath
-}
-
 func getMountPath(targetPath string, dirName string, path string) (string, error) {
 	// if the target path is a file then use that as the exact mount path
 	if filepath.Ext(targetPath) != "" {
@@ -909,9 +929,13 @@ func (o *runCmdOptions) uploadPomFromJar(gav maven.Dependency, path string, plat
 			}
 		}
 		if pomExtracted {
-			gav.Type = "pom"
-			// Swallow error as this is not a mandatory step
-			o.uploadAsMavenArtifact(gav, pomPath, platform, ns, options, cmd)
+			if o.skipPom() {
+				o.PrintfVerboseOutf(cmd, "Skipping uploading extracted POM from %s \n", path)
+			} else {
+				gav.Type = "pom"
+				// Swallow error as this is not a mandatory step
+				o.uploadAsMavenArtifact(gav, pomPath, platform, ns, options, cmd)
+			}
 		}
 		return nil
 	})
@@ -978,7 +1002,7 @@ func (o *runCmdOptions) extractGav(src *zip.File, localPath string, cmd *cobra.C
 
 func (o *runCmdOptions) uploadAsMavenArtifact(dependency maven.Dependency, path string, platform *v1.IntegrationPlatform, ns string, options spectrum.Options, cmd *cobra.Command) error {
 	artifactHTTPPath := getArtifactHTTPPath(dependency, platform, ns)
-	options.Target = fmt.Sprintf("%s/%s:%s", platform.Status.Build.Registry.Address, artifactHTTPPath, dependency.Version)
+	options.Target = fmt.Sprintf("%s/%s:%s", o.getRegistry(platform), artifactHTTPPath, dependency.Version)
 	if runtimeos.GOOS == "windows" {
 		// workaround for https://github.com/container-tools/spectrum/issues/8
 		// work with relative paths instead
@@ -993,7 +1017,11 @@ func (o *runCmdOptions) uploadAsMavenArtifact(dependency maven.Dependency, path
 		return err
 	}
 	o.PrintfVerboseOutf(cmd, "Uploaded: %s to %s \n", path, options.Target)
-	return uploadChecksumFiles(path, options, platform, artifactHTTPPath, dependency)
+	if o.skipChecksums() {
+		o.PrintfVerboseOutf(cmd, "Skipping generating and uploading checksum files for %s \n", path)
+		return nil
+	}
+	return o.uploadChecksumFiles(path, options, platform, artifactHTTPPath, dependency)
 }
 
 // Deprecated: workaround for https://github.com/container-tools/spectrum/issues/8
@@ -1041,18 +1069,18 @@ func extractGavFromPom(path string, gav maven.Dependency) maven.Dependency {
 	return gav
 }
 
-func uploadChecksumFiles(path string, options spectrum.Options, platform *v1.IntegrationPlatform, artifactHTTPPath string, dependency maven.Dependency) error {
+func (o *runCmdOptions) uploadChecksumFiles(path string, options spectrum.Options, platform *v1.IntegrationPlatform, artifactHTTPPath string, dependency maven.Dependency) error {
 	return util.WithTempDir("camel-k", func(tmpDir string) error {
 		// #nosec G401
-		if err := uploadChecksumFile(md5.New(), tmpDir, "_md5", path, options, platform, artifactHTTPPath, dependency); err != nil {
+		if err := o.uploadChecksumFile(md5.New(), tmpDir, "_md5", path, options, platform, artifactHTTPPath, dependency); err != nil {
 			return err
 		}
 		// #nosec G401
-		return uploadChecksumFile(sha1.New(), tmpDir, "_sha1", path, options, platform, artifactHTTPPath, dependency)
+		return o.uploadChecksumFile(sha1.New(), tmpDir, "_sha1", path, options, platform, artifactHTTPPath, dependency)
 	})
 }
 
-func uploadChecksumFile(hash hash.Hash, tmpDir string, ext string, path string, options spectrum.Options, platform *v1.IntegrationPlatform, artifactHTTPPath string, dependency maven.Dependency) error {
+func (o *runCmdOptions) uploadChecksumFile(hash hash.Hash, tmpDir string, ext string, path string, options spectrum.Options, platform *v1.IntegrationPlatform, artifactHTTPPath string, dependency maven.Dependency) error {
 	file, err := os.Open(path)
 	if err != nil {
 		return err
@@ -1078,7 +1106,7 @@ func uploadChecksumFile(hash hash.Hash, tmpDir string, ext string, path string,
 	if err = writeChecksumToFile(filepath, hash); err != nil {
 		return err
 	}
-	options.Target = fmt.Sprintf("%s/%s%s:%s", platform.Status.Build.Registry.Address, artifactHTTPPath, ext, dependency.Version)
+	options.Target = fmt.Sprintf("%s/%s%s:%s", o.getRegistry(platform), artifactHTTPPath, ext, dependency.Version)
 	_, err = spectrum.Build(options, fmt.Sprintf("%s:.", filepath))
 	return err
 }
diff --git a/pkg/trait/gc.go b/pkg/trait/gc.go
index 1ff3204f5..a088367fb 100644
--- a/pkg/trait/gc.go
+++ b/pkg/trait/gc.go
@@ -19,6 +19,7 @@ package trait
 
 import (
 	"context"
+	"fmt"
 	"path/filepath"
 	"regexp"
 	"strconv"
@@ -26,6 +27,10 @@ import (
 	"sync"
 	"time"
 
+	"github.com/pkg/errors"
+	"golang.org/x/time/rate"
+
+	authorization "k8s.io/api/authorization/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -35,17 +40,20 @@ import (
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/discovery/cached/disk"
 	"k8s.io/client-go/discovery/cached/memory"
-
 	ctrl "sigs.k8s.io/controller-runtime/pkg/client"
 
 	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
+	"github.com/apache/camel-k/pkg/util"
 )
 
 var (
-	toFileName                  = regexp.MustCompile(`[^(\w/\.)]`)
-	diskCachedDiscoveryClient   discovery.CachedDiscoveryInterface
-	memoryCachedDiscoveryClient discovery.CachedDiscoveryInterface
-	discoveryClientLock         sync.Mutex
+	toFileName = regexp.MustCompile(`[^(\w/\.)]`)
+
+	lock                  sync.Mutex
+	rateLimiter           = rate.NewLimiter(rate.Every(time.Minute), 1)
+	collectableGVKs       = make(map[schema.GroupVersionKind]struct{})
+	memoryCachedDiscovery discovery.CachedDiscoveryInterface
+	diskCachedDiscovery   discovery.CachedDiscoveryInterface
 )
 
 type discoveryCacheType string
@@ -81,8 +89,7 @@ func (t *garbageCollectorTrait) Configure(e *Environment) (bool, error) {
 		t.DiscoveryCache = &s
 	}
 
-	return e.IntegrationInPhase(v1.IntegrationPhaseInitialization) || e.IntegrationInRunningPhases(),
-		nil
+	return e.IntegrationInPhase(v1.IntegrationPhaseInitialization) || e.IntegrationInRunningPhases(), nil
 }
 
 func (t *garbageCollectorTrait) Apply(e *Environment) error {
@@ -92,12 +99,9 @@ func (t *garbageCollectorTrait) Apply(e *Environment) error {
 		// Register a post action that deletes the existing resources that are labelled
 		// with the previous integration generations.
 		// TODO: this should be refined so that it's run when all the replicas for the newer generation
-		// are ready. This is to be added when the integration scale status is refined with ready replicas
+		// are ready.
 		e.PostActions = append(e.PostActions, func(env *Environment) error {
-			// The collection and deletion are performed asynchronously to avoid blocking
-			// the reconciliation loop.
-			go t.garbageCollectResources(env)
-			return nil
+			return t.garbageCollectResources(env)
 		})
 
 		fallthrough
@@ -121,43 +125,40 @@ func (t *garbageCollectorTrait) Apply(e *Environment) error {
 	return nil
 }
 
-func (t *garbageCollectorTrait) garbageCollectResources(e *Environment) {
+func (t *garbageCollectorTrait) garbageCollectResources(e *Environment) error {
+	deletableGVKs, err := t.getDeletableTypes(e)
+	if err != nil {
+		return errors.Wrap(err, "cannot discover GVK types")
+	}
+
 	integration, _ := labels.NewRequirement(v1.IntegrationLabel, selection.Equals, []string{e.Integration.Name})
 	generation, err := labels.NewRequirement("camel.apache.org/generation", selection.LessThan, []string{strconv.FormatInt(e.Integration.GetGeneration(), 10)})
 	if err != nil {
-		t.L.ForIntegration(e.Integration).Errorf(err, "cannot determine generation requirement")
-		return
+		return errors.Wrap(err, "cannot determine generation requirement")
 	}
 	selector := labels.NewSelector().
 		Add(*integration).
 		Add(*generation)
 
-	deletableGVKs, err := t.getDeletableTypes(e)
-	if err != nil {
-		t.L.ForIntegration(e.Integration).Errorf(err, "cannot discover GVK types")
-		return
-	}
-
-	t.deleteEachOf(deletableGVKs, e, selector)
+	return t.deleteEachOf(e.Ctx, deletableGVKs, e, selector)
 }
 
-func (t *garbageCollectorTrait) deleteEachOf(gvks map[schema.GroupVersionKind]struct{}, e *Environment, selector labels.Selector) {
-	for gvk := range gvks {
+func (t *garbageCollectorTrait) deleteEachOf(ctx context.Context, deletableGVKs map[schema.GroupVersionKind]struct{}, e *Environment, selector labels.Selector) error {
+	for GVK := range deletableGVKs {
 		resources := unstructured.UnstructuredList{
 			Object: map[string]interface{}{
-				"apiVersion": gvk.GroupVersion().String(),
-				"kind":       gvk.Kind,
+				"apiVersion": GVK.GroupVersion().String(),
+				"kind":       GVK.Kind,
 			},
 		}
 		options := []ctrl.ListOption{
 			ctrl.InNamespace(e.Integration.Namespace),
 			ctrl.MatchingLabelsSelector{Selector: selector},
 		}
-		if err := t.Client.List(context.TODO(), &resources, options...); err != nil {
-			if !k8serrors.IsNotFound(err) && !k8serrors.IsForbidden(err) {
-				t.L.ForIntegration(e.Integration).Errorf(err, "cannot list child resources: %v", gvk)
+		if err := t.Client.List(ctx, &resources, options...); err != nil {
+			if !k8serrors.IsNotFound(err) {
+				return errors.Wrap(err, "cannot list child resources")
 			}
-
 			continue
 		}
 
@@ -166,7 +167,7 @@ func (t *garbageCollectorTrait) deleteEachOf(gvks map[schema.GroupVersionKind]st
 			if !t.canBeDeleted(e, r) {
 				continue
 			}
-			err := t.Client.Delete(context.TODO(), &r, ctrl.PropagationPolicy(metav1.DeletePropagationBackground))
+			err := t.Client.Delete(ctx, &r, ctrl.PropagationPolicy(metav1.DeletePropagationBackground))
 			if err != nil {
 				// The resource may have already been deleted
 				if !k8serrors.IsNotFound(err) {
@@ -177,6 +178,8 @@ func (t *garbageCollectorTrait) deleteEachOf(gvks map[schema.GroupVersionKind]st
 			}
 		}
 	}
+
+	return nil
 }
 
 func (t *garbageCollectorTrait) canBeDeleted(e *Environment, u unstructured.Unstructured) bool {
@@ -190,64 +193,94 @@ func (t *garbageCollectorTrait) canBeDeleted(e *Environment, u unstructured.Unst
 }
 
 func (t *garbageCollectorTrait) getDeletableTypes(e *Environment) (map[schema.GroupVersionKind]struct{}, error) {
+	lock.Lock()
+	defer lock.Unlock()
+
+	// Rate limit to avoid Discovery and SelfSubjectRulesReview requests at every reconciliation.
+	if !rateLimiter.Allow() {
+		// Return the cached set of garbage collectable GVKs.
+		return collectableGVKs, nil
+	}
+
 	// We rely on the discovery API to retrieve all the resources GVK,
 	// that results in an unbounded set that can impact garbage collection latency when scaling up.
-	discoveryClient, err := t.discoveryClient(e)
+	discoveryClient, err := t.discoveryClient()
 	if err != nil {
 		return nil, err
 	}
 	resources, err := discoveryClient.ServerPreferredNamespacedResources()
 	// Swallow group discovery errors, e.g., Knative serving exposes
 	// an aggregated API for custom.metrics.k8s.io that requires special
-	// authentication scheme while discovering preferred resources
+	// authentication scheme while discovering preferred resources.
 	if err != nil && !discovery.IsGroupDiscoveryFailedError(err) {
 		return nil, err
 	}
 
 	// We only take types that support the "delete" verb,
 	// to prevents from performing queries that we know are going to return "MethodNotAllowed".
-	return groupVersionKinds(discovery.FilteredBy(discovery.SupportsAllVerbs{Verbs: []string{"delete"}}, resources)),
-		nil
-}
+	APIResourceLists := discovery.FilteredBy(discovery.SupportsAllVerbs{Verbs: []string{"delete"}}, resources)
+
+	// Retrieve the permissions granted to the operator service account.
+	// We assume the operator has only to garbage collect the resources it has created.
+	srr := &authorization.SelfSubjectRulesReview{
+		Spec: authorization.SelfSubjectRulesReviewSpec{
+			Namespace: e.Integration.Namespace,
+		},
+	}
+	res, err := e.Client.AuthorizationV1().SelfSubjectRulesReviews().Create(e.Ctx, srr, metav1.CreateOptions{})
+	if err != nil {
+		return nil, err
+	}
 
-func groupVersionKinds(rls []*metav1.APIResourceList) map[schema.GroupVersionKind]struct{} {
-	GVKs := map[schema.GroupVersionKind]struct{}{}
-	for _, rl := range rls {
-		for _, r := range rl.APIResources {
-			GVKs[schema.FromAPIVersionAndKind(rl.GroupVersion, r.Kind)] = struct{}{}
+	GVKs := make(map[schema.GroupVersionKind]struct{})
+	for _, APIResourceList := range APIResourceLists {
+		for _, resource := range APIResourceList.APIResources {
+		rule:
+			for _, rule := range res.Status.ResourceRules {
+				if !util.StringSliceContainsAnyOf(rule.Verbs, "delete", "*") {
+					continue
+				}
+				for _, group := range rule.APIGroups {
+					for _, name := range rule.Resources {
+						if (resource.Group == group || group == "*") && (resource.Name == name || name == "*") {
+							GVK := schema.FromAPIVersionAndKind(APIResourceList.GroupVersion, resource.Kind)
+							GVKs[GVK] = struct{}{}
+							break rule
+						}
+					}
+				}
+			}
 		}
 	}
-	return GVKs
-}
+	collectableGVKs = GVKs
 
-func (t *garbageCollectorTrait) discoveryClient(e *Environment) (discovery.DiscoveryInterface, error) {
-	discoveryClientLock.Lock()
-	defer discoveryClientLock.Unlock()
+	return collectableGVKs, nil
+}
 
+func (t *garbageCollectorTrait) discoveryClient() (discovery.DiscoveryInterface, error) {
 	switch *t.DiscoveryCache {
 	case diskDiscoveryCache:
-		if diskCachedDiscoveryClient != nil {
-			return diskCachedDiscoveryClient, nil
+		if diskCachedDiscovery != nil {
+			return diskCachedDiscovery, nil
 		}
 		config := t.Client.GetConfig()
 		httpCacheDir := filepath.Join(mustHomeDir(), ".kube", "http-cache")
 		diskCacheDir := filepath.Join(mustHomeDir(), ".kube", "cache", "discovery", toHostDir(config.Host))
 		var err error
-		diskCachedDiscoveryClient, err = disk.NewCachedDiscoveryClientForConfig(config, diskCacheDir, httpCacheDir, 10*time.Minute)
-		return diskCachedDiscoveryClient, err
+		diskCachedDiscovery, err = disk.NewCachedDiscoveryClientForConfig(config, diskCacheDir, httpCacheDir, 10*time.Minute)
+		return diskCachedDiscovery, err
 
 	case memoryDiscoveryCache:
-		if memoryCachedDiscoveryClient != nil {
-			return memoryCachedDiscoveryClient, nil
+		if memoryCachedDiscovery != nil {
+			return memoryCachedDiscovery, nil
 		}
-		memoryCachedDiscoveryClient = memory.NewMemCacheClient(t.Client.Discovery())
-		return memoryCachedDiscoveryClient, nil
+		memoryCachedDiscovery = memory.NewMemCacheClient(t.Client.Discovery())
+		return memoryCachedDiscovery, nil
 
 	case disabledDiscoveryCache, "":
 		return t.Client.Discovery(), nil
 
 	default:
-		t.L.ForIntegration(e.Integration).Infof("unsupported discovery cache type: %s", *t.DiscoveryCache)
-		return t.Client.Discovery(), nil
+		return nil, fmt.Errorf("unsupported discovery cache type: %s", *t.DiscoveryCache)
 	}
 }
diff --git a/pkg/util/source/inspector.go b/pkg/util/source/inspector.go
index 651af7344..be2b86ddc 100644
--- a/pkg/util/source/inspector.go
+++ b/pkg/util/source/inspector.go
@@ -40,9 +40,11 @@ var (
 	singleQuotedTo          = regexp.MustCompile(`\.to\s*\(\s*'([a-zA-Z0-9-]+:[^']+)'`)
 	singleQuotedToD         = regexp.MustCompile(`\.toD\s*\(\s*'([a-zA-Z0-9-]+:[^']+)'`)
 	singleQuotedToF         = regexp.MustCompile(`\.toF\s*\(\s*'([a-zA-Z0-9-]+:[^']+)'`)
+	singleQuotedWireTap     = regexp.MustCompile(`\.wireTap\s*\(\s*'([a-zA-Z0-9-]+:[^']+)'`)
 	doubleQuotedTo          = regexp.MustCompile(`\.to\s*\(\s*"([a-zA-Z0-9-]+:[^"]+)"`)
 	doubleQuotedToD         = regexp.MustCompile(`\.toD\s*\(\s*"([a-zA-Z0-9-]+:[^"]+)"`)
 	doubleQuotedToF         = regexp.MustCompile(`\.toF\s*\(\s*"([a-zA-Z0-9-]+:[^"]+)"`)
+	doubleQuotedWireTap     = regexp.MustCompile(`\.wireTap\s*\(\s*"([a-zA-Z0-9-]+:[^"]+)"`)
 	languageRegexp          = regexp.MustCompile(`language\s*\(\s*["|']([a-zA-Z0-9-]+[^"|']+)["|']\s*,.*\)`)
 	camelTypeRegexp         = regexp.MustCompile(`.*(org.apache.camel.*Component|DataFormat|Language)`)
 	jsonLibraryRegexp       = regexp.MustCompile(`.*JsonLibrary\.Jackson.*`)
diff --git a/pkg/util/source/inspector_groovy.go b/pkg/util/source/inspector_groovy.go
index 318107862..686814084 100644
--- a/pkg/util/source/inspector_groovy.go
+++ b/pkg/util/source/inspector_groovy.go
@@ -42,6 +42,8 @@ func (i GroovyInspector) Extract(source v1.SourceSpec, meta *Metadata) error {
 		doubleQuotedToD,
 		singleQuotedToF,
 		doubleQuotedToF,
+		singleQuotedWireTap,
+		doubleQuotedWireTap,
 	)
 
 	meta.FromURIs = append(meta.FromURIs, from...)
diff --git a/pkg/util/source/inspector_java_script.go b/pkg/util/source/inspector_java_script.go
index d8679e0c9..ebcbe37bb 100644
--- a/pkg/util/source/inspector_java_script.go
+++ b/pkg/util/source/inspector_java_script.go
@@ -42,6 +42,8 @@ func (i JavaScriptInspector) Extract(source v1.SourceSpec, meta *Metadata) error
 		doubleQuotedToD,
 		singleQuotedToF,
 		doubleQuotedToF,
+		singleQuotedWireTap,
+		doubleQuotedWireTap,
 	)
 
 	meta.FromURIs = append(meta.FromURIs, from...)
diff --git a/pkg/util/source/inspector_java_source.go b/pkg/util/source/inspector_java_source.go
index 80981e74b..00a9f391f 100644
--- a/pkg/util/source/inspector_java_source.go
+++ b/pkg/util/source/inspector_java_source.go
@@ -37,6 +37,7 @@ func (i JavaSourceInspector) Extract(source v1.SourceSpec, meta *Metadata) error
 		doubleQuotedTo,
 		doubleQuotedToD,
 		doubleQuotedToF,
+		doubleQuotedWireTap,
 	)
 
 	meta.FromURIs = append(meta.FromURIs, from...)
diff --git a/pkg/util/source/inspector_kotlin.go b/pkg/util/source/inspector_kotlin.go
index b2485c29b..753d0dbbf 100644
--- a/pkg/util/source/inspector_kotlin.go
+++ b/pkg/util/source/inspector_kotlin.go
@@ -37,6 +37,7 @@ func (i KotlinInspector) Extract(source v1.SourceSpec, meta *Metadata) error {
 		doubleQuotedTo,
 		doubleQuotedToD,
 		doubleQuotedToF,
+		doubleQuotedWireTap,
 	)
 
 	meta.FromURIs = append(meta.FromURIs, from...)
diff --git a/pkg/util/source/inspector_xml.go b/pkg/util/source/inspector_xml.go
index 18c48277c..51baaee24 100644
--- a/pkg/util/source/inspector_xml.go
+++ b/pkg/util/source/inspector_xml.go
@@ -62,7 +62,7 @@ func (i XMLInspector) Extract(source v1.SourceSpec, meta *Metadata) error {
 						meta.FromURIs = append(meta.FromURIs, a.Value)
 					}
 				}
-			case "to", "toD", "toF":
+			case "to", "toD", "toF", "wireTap":
 				for _, a := range se.Attr {
 					if a.Name.Local == "uri" {
 						meta.ToURIs = append(meta.ToURIs, a.Value)
diff --git a/pkg/util/source/inspector_yaml.go b/pkg/util/source/inspector_yaml.go
index 379bd100b..99ed8b876 100644
--- a/pkg/util/source/inspector_yaml.go
+++ b/pkg/util/source/inspector_yaml.go
@@ -161,7 +161,7 @@ func (i YAMLInspector) parseStep(key string, content interface{}, meta *Metadata
 		switch key {
 		case "from":
 			meta.FromURIs = append(meta.FromURIs, maybeURI)
-		case "to", "to-d", "toD":
+		case "to", "to-d", "toD", "wire-tap", "wireTap":
 			meta.ToURIs = append(meta.ToURIs, maybeURI)
 		}
 	}


Test patch:
diff --git a/e2e/knative/knative_test.go b/e2e/knative/knative_test.go
index fe1e6f976..7d7c780dc 100644
--- a/e2e/knative/knative_test.go
+++ b/e2e/knative/knative_test.go
@@ -124,6 +124,11 @@ func TestRunFlow(t *testing.T) {
 		Expect(Kamel("run", "-n", ns, "files/flow.yaml").Execute()).To(Succeed())
 		Eventually(IntegrationPodPhase(ns, "flow"), TestTimeoutLong).Should(Equal(v1.PodRunning))
 		Eventually(IntegrationConditionStatus(ns, "flow", camelv1.IntegrationConditionReady), TestTimeoutShort).Should(Equal(v1.ConditionTrue))
+		
+		t.Run("Scale to zero", func(t *testing.T) {
+			Eventually(IntegrationPod(ns, "flow"), TestTimeoutLong).Should(BeNil())
+		})
+		
 		Expect(Kamel("delete", "--all", "-n", ns).Execute()).To(Succeed())
 	})
 }
diff --git a/e2e/registry/registry_maven_wagon_test.go b/e2e/registry/registry_maven_wagon_test.go
index 76e2bbc99..4abebc239 100644
--- a/e2e/registry/registry_maven_wagon_test.go
+++ b/e2e/registry/registry_maven_wagon_test.go
@@ -50,7 +50,7 @@ func TestImageRegistryIsAMavenRepository(t *testing.T) {
 		t.Run("image registry is a maven repository", func(t *testing.T) {
 			// Create integration that should decrypt an encrypted message to "foobar" and log it
 			name := "foobar-decryption"
-			jar, err := filepath.Abs("files/sample-decryption-1.0.jar")
+			jar, err := filepath.Abs("files/sample-decryption-1.0.jar?skipPOM=true")
 			assert.Nil(t, err)
 			pom, err := filepath.Abs("files/sample-decryption-1.0.pom")
 			assert.Nil(t, err)
@@ -89,7 +89,7 @@ func TestImageRegistryIsAMavenRepository(t *testing.T) {
 			Expect(Kamel("run", "files/LaughingRoute.java",
 				"--name", name,
 				"-p", fmt.Sprintf("location=%s", customPath),
-				"-d", fmt.Sprintf("file://files/laugh.txt:%slaugh.txt", customPath),
+				"-d", fmt.Sprintf("file://files/laugh.txt?targetPath=%slaugh.txt", customPath),
 				"-n", ns,
 			).Execute()).To(Succeed())
 
@@ -104,7 +104,7 @@ func TestImageRegistryIsAMavenRepository(t *testing.T) {
 			Expect(Kamel("run", "files/LaughingRoute.java",
 				"--name", name,
 				"-p", "location=files/",
-				"-d", fmt.Sprintf("file://files/laughs/:files/"),
+				"-d", fmt.Sprintf("file://files/laughs/?targetPath=files/"),
 				"-n", ns,
 			).Execute()).To(Succeed())
 
diff --git a/pkg/cmd/run_test.go b/pkg/cmd/run_test.go
index 46c032bc0..be71a8425 100644
--- a/pkg/cmd/run_test.go
+++ b/pkg/cmd/run_test.go
@@ -677,69 +677,3 @@ func TestMissingTrait(t *testing.T) {
 	assert.Equal(t, "Error: bogus.fail=i-must-fail is not a valid trait property\n", output)
 	assert.NotNil(t, err)
 }
-
-func TestGetsPaths(t *testing.T) {
-	tests := []struct {
-		path       string
-		localPath  string
-		remotePath string
-		os         string
-		isAbs      bool
-	}{
-		{
-			path:       "C:\\USER\\HOME\\:remote/path",
-			localPath:  "C:\\USER\\HOME\\",
-			remotePath: "remote/path",
-			os:         "windows",
-			isAbs:      true,
-		},
-		{
-			path:       "src\\main\\resources:remote/path",
-			localPath:  "src\\main\\resources",
-			remotePath: "remote/path",
-			os:         "windows",
-		},
-		{
-			path:       "C:\\USER\\HOME\\",
-			localPath:  "C:\\USER\\HOME\\",
-			remotePath: "",
-			os:         "windows",
-			isAbs:      true,
-		},
-		{
-			path:       "src\\main\\resources",
-			localPath:  "src\\main\\resources",
-			remotePath: "",
-			os:         "windows",
-		},
-		{
-			path:       "/home/user/name/dir:/remote/path",
-			localPath:  "/home/user/name/dir",
-			remotePath: "/remote/path",
-			os:         "linux",
-			isAbs:      true,
-		}, {
-			path:       "/home/user/name/dir",
-			localPath:  "/home/user/name/dir",
-			remotePath: "",
-			os:         "linux",
-			isAbs:      true,
-		}, {
-			path:       "src/main/resources:remote/path",
-			localPath:  "src/main/resources",
-			remotePath: "remote/path",
-			os:         "linux",
-		}, {
-			path:       "src/main/resources",
-			localPath:  "src/main/resources",
-			remotePath: "",
-			os:         "linux",
-		},
-	}
-	for _, test := range tests {
-		localPath, targetPath := getPaths(test.path, test.os, test.isAbs)
-		assert.Equal(t, test.localPath, localPath)
-		assert.Equal(t, test.remotePath, targetPath)
-
-	}
-}
diff --git a/pkg/util/source/inspector_groovy_test.go b/pkg/util/source/inspector_groovy_test.go
index 7f091f3b8..d0d2a5e80 100644
--- a/pkg/util/source/inspector_groovy_test.go
+++ b/pkg/util/source/inspector_groovy_test.go
@@ -48,6 +48,11 @@ from("direct:start")
     .to('kamelet:foo/bar?baz=test')
 `
 
+const GroovyWireTapEipSingleQuote = `
+from("direct:start")
+    .wireTap('kamelet:foo/bar?baz=test')
+`
+
 func TestGroovyKamelet(t *testing.T) {
 	tc := []struct {
 		source   string
@@ -69,6 +74,10 @@ func TestGroovyKamelet(t *testing.T) {
 			source:   GroovyKameletEndpointSingleQuote,
 			kamelets: []string{"foo/bar"},
 		},
+		{
+			source:   GroovyWireTapEipSingleQuote,
+			kamelets: []string{"foo/bar"},
+		},
 	}
 
 	for i := range tc {
diff --git a/pkg/util/source/inspector_java_script_test.go b/pkg/util/source/inspector_java_script_test.go
index f9e1634a6..3ed09d928 100644
--- a/pkg/util/source/inspector_java_script_test.go
+++ b/pkg/util/source/inspector_java_script_test.go
@@ -37,6 +37,11 @@ from("direct:start")
     .to("kamelet:foo/bar?baz=test")
 `
 
+const JavaScriptWireTapEipSingleQuote = `
+from("direct:start")
+    .wireTap('kamelet:foo/bar?baz=test')
+`
+
 func TestJavaScriptKamelet(t *testing.T) {
 	tc := []struct {
 		source   string
@@ -50,6 +55,10 @@ func TestJavaScriptKamelet(t *testing.T) {
 			source:   JavaScriptKameletEndpoint,
 			kamelets: []string{"foo/bar"},
 		},
+		{
+			source:   JavaScriptWireTapEipSingleQuote,
+			kamelets: []string{"foo/bar"},
+		},
 	}
 
 	for i := range tc {
diff --git a/pkg/util/source/inspector_java_source_test.go b/pkg/util/source/inspector_java_source_test.go
index 353643130..af1a8e3af 100644
--- a/pkg/util/source/inspector_java_source_test.go
+++ b/pkg/util/source/inspector_java_source_test.go
@@ -37,6 +37,11 @@ from("direct:start")
     .to("kamelet:foo/bar?baz=test")
 `
 
+const JavaSourceWireTapEip = `
+from("direct:start")
+    .wireTap("kamelet:foo/bar?baz=test")
+`
+
 func TestJavaSourceKamelet(t *testing.T) {
 	tc := []struct {
 		source   string
@@ -50,6 +55,10 @@ func TestJavaSourceKamelet(t *testing.T) {
 			source:   JavaSourceKameletEndpoint,
 			kamelets: []string{"foo/bar"},
 		},
+		{
+			source:   JavaSourceWireTapEip,
+			kamelets: []string{"foo/bar"},
+		},
 	}
 
 	for i := range tc {
diff --git a/pkg/util/source/inspector_kotlin_test.go b/pkg/util/source/inspector_kotlin_test.go
index d94ee7fd9..1b08fe2b3 100644
--- a/pkg/util/source/inspector_kotlin_test.go
+++ b/pkg/util/source/inspector_kotlin_test.go
@@ -37,6 +37,11 @@ from("direct:start")
     .to("kamelet:foo/bar?baz=test")
 `
 
+const KotlinWireTapEip = `
+from("direct:start")
+    .wireTap("kamelet:foo/bar?baz=test")
+`
+
 func TestKotlinKamelet(t *testing.T) {
 	tc := []struct {
 		source   string
@@ -50,6 +55,10 @@ func TestKotlinKamelet(t *testing.T) {
 			source:   KotlinKameletEndpoint,
 			kamelets: []string{"foo/bar"},
 		},
+		{
+			source:   KotlinWireTapEip,
+			kamelets: []string{"foo/bar"},
+		},
 	}
 
 	for i := range tc {
diff --git a/pkg/util/source/inspector_xml_test.go b/pkg/util/source/inspector_xml_test.go
index 52445e820..8552a84b7 100644
--- a/pkg/util/source/inspector_xml_test.go
+++ b/pkg/util/source/inspector_xml_test.go
@@ -46,6 +46,15 @@ const XMLKameletEndpoint = `
 </camelContext>
 `
 
+const XMLWireTapEndpoint = `
+<camelContext xmlns="http://camel.apache.org/schema/spring">
+  <route>
+    <from uri="direct:start"/>
+    <wireTap uri="kamelet:foo/bar?baz=test"/>
+  </route>
+</camelContext>
+`
+
 func TestXMLKamelet(t *testing.T) {
 	tc := []struct {
 		source   string
@@ -59,6 +68,10 @@ func TestXMLKamelet(t *testing.T) {
 			source:   XMLKameletEndpoint,
 			kamelets: []string{"foo/bar"},
 		},
+		{
+			source:   XMLWireTapEndpoint,
+			kamelets: []string{"foo/bar"},
+		},
 	}
 
 	for i := range tc {
diff --git a/pkg/util/source/inspector_yaml_test.go b/pkg/util/source/inspector_yaml_test.go
index 88d77d5ba..1a85b1fd5 100644
--- a/pkg/util/source/inspector_yaml_test.go
+++ b/pkg/util/source/inspector_yaml_test.go
@@ -83,6 +83,26 @@ const YAMLInDepthChannel = `
                 uri: knative:endpoint/service
 `
 
+const YAMLWireTapKnativeEIP = `
+- from:
+    uri: knative:channel/mychannel
+    parameters:
+      period: "1000"
+    steps:
+      - wireTap:
+          uri: knative:channel/mychannel
+`
+
+const YAMLWireTapJmsEIP = `
+- from:
+    uri: knative:channel/mychannel
+    parameters:
+      period: "1000"
+    steps:
+      - wireTap:
+          uri: jms:queue:foo
+`
+
 func TestYAMLDependencies(t *testing.T) {
 	tests := []struct {
 		name                string
@@ -125,6 +145,22 @@ func TestYAMLDependencies(t *testing.T) {
 				`mvn:org.apache.camel.k:camel-k-knative-consumer`,
 			},
 		},
+		{
+			name:   "wire-tap-knative",
+			source: YAMLWireTapKnativeEIP,
+			dependencies: []string{
+				`mvn:org.apache.camel.k:camel-k-knative-producer`,
+				`mvn:org.apache.camel.k:camel-k-knative-consumer`,
+			},
+		},
+		{
+			name:   "wire-tap-jms",
+			source: YAMLWireTapJmsEIP,
+			dependencies: []string{
+				`mvn:org.apache.camel.k:camel-k-knative-consumer`,
+				`camel:jms`,
+			},
+		},
 	}
 	for i := range tests {
 		test := tests[i]