You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by as...@apache.org on 2021/05/11 16:41:13 UTC

[camel-k] 03/04: chore(e2e): Polish upgrade e2e tests

This is an automated email from the ASF dual-hosted git repository.

astefanutti pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-k.git

commit 839608c85ca6a4b8baf246a86e0b2cd7313ce66c
Author: Antonin Stefanutti <an...@stefanutti.fr>
AuthorDate: Tue May 11 14:41:17 2021 +0200

    chore(e2e): Polish upgrade e2e tests
---
 e2e/support/test_support.go                        |  61 ----------
 ...pgrade_operator_test.go => cli_upgrade_test.go} |  30 ++---
 e2e/upgrade/olm_upgrade_test.go                    | 123 +++++++++++----------
 e2e/upgrade/util.go                                |  92 +++++++++++++++
 4 files changed, 171 insertions(+), 135 deletions(-)

diff --git a/e2e/support/test_support.go b/e2e/support/test_support.go
index 6e2632f..a65ca10 100644
--- a/e2e/support/test_support.go
+++ b/e2e/support/test_support.go
@@ -27,7 +27,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	v1alpha12 "github.com/operator-framework/api/pkg/operators/v1alpha1"
 	"io"
 	"io/ioutil"
 	"os"
@@ -35,7 +34,6 @@ import (
 	"strings"
 	"testing"
 	"time"
-	"unsafe"
 
 	"github.com/google/uuid"
 	"github.com/onsi/gomega"
@@ -673,65 +671,6 @@ func KnativeService(ns string, name string) func() *servingv1.Service {
 	}
 }
 
-func CKClusterServiceVersion(conditions func(v1alpha12.ClusterServiceVersion) bool ,ns string) func() *v1alpha12.ClusterServiceVersion {
-	return func() *v1alpha12.ClusterServiceVersion {
-		lst := v1alpha12.ClusterServiceVersionList{}
-		if err := TestClient().List(TestContext, &lst, ctrl.InNamespace(ns)); err != nil {
-			panic(err)
-		}
-		for _, s := range lst.Items {
-			if strings.Contains(s.Name, "camel-k") && conditions(s){
-				return &s
-			}
-		}
-		return nil
-	}
-}
-
-func CKClusterServiceVersionPhase(conditions func(v1alpha12.ClusterServiceVersion) bool, ns string) func() v1alpha12.ClusterServiceVersionPhase {
-	return func() v1alpha12.ClusterServiceVersionPhase {
-		if csv := CKClusterServiceVersion(conditions, ns)(); csv != nil && unsafe.Sizeof(csv.Status) > 0 {
-			return csv.Status.Phase
-		}
-		return ""
-	}
-}
-
-func CatalogSource(name string, ns string) func() *v1alpha12.CatalogSource {
-	return func() *v1alpha12.CatalogSource {
-		answer := v1alpha12.CatalogSource{
-			TypeMeta: metav1.TypeMeta{
-				Kind:       "CatalogSource",
-				APIVersion: v1alpha1.SchemeGroupVersion.String(),
-			},
-			ObjectMeta: metav1.ObjectMeta{
-				Namespace: ns,
-				Name:      name,
-			},
-		}
-		key := ctrl.ObjectKey{
-			Namespace: ns,
-			Name:      name,
-		}
-		if err := TestClient().Get(TestContext, key, &answer); err != nil && k8serrors.IsNotFound(err) {
-			return nil
-		} else if err != nil {
-			log.Errorf(err, "Error while retrieving CatalogSource %s", name)
-			return nil
-		}
-		return &answer
-	}
-}
-
-func CatalogSourcePhase(name string, ns string) func() string {
-	return func() string {
-		if source := CatalogSource(name, ns)(); source != nil && source.Status.GRPCConnectionState != nil {
-			return CatalogSource(name, ns)().Status.GRPCConnectionState.LastObservedState
-		}
-		return ""
-	}
-}
-
 func Deployment(ns string, name string) func() *appsv1.Deployment {
 	return func() *appsv1.Deployment {
 		answer := appsv1.Deployment{
diff --git a/e2e/upgrade/upgrade_operator_test.go b/e2e/upgrade/cli_upgrade_test.go
similarity index 79%
rename from e2e/upgrade/upgrade_operator_test.go
rename to e2e/upgrade/cli_upgrade_test.go
index 2ce5775..095943e 100644
--- a/e2e/upgrade/upgrade_operator_test.go
+++ b/e2e/upgrade/cli_upgrade_test.go
@@ -28,10 +28,10 @@ import (
 
 	. "github.com/onsi/gomega"
 
-	v1 "k8s.io/api/core/v1"
+	corev1 "k8s.io/api/core/v1"
 
 	. "github.com/apache/camel-k/e2e/support"
-	camelv1 "github.com/apache/camel-k/pkg/apis/camel/v1"
+	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
 	"github.com/apache/camel-k/pkg/util/defaults"
 )
 
@@ -49,14 +49,14 @@ func TestOperatorUpgrade(t *testing.T) {
 		kamel, ok := os.LookupEnv("RELEASED_KAMEL_BIN")
 		Expect(ok).To(BeTrue())
 
-		//set KAMEL_BIN only for this test - don't override the ENV variable for all tests
+		// Set KAMEL_BIN only for this test - don't override the ENV variable for all tests
 		Expect(os.Setenv("KAMEL_BIN", kamel)).To(Succeed())
 
 		Expect(Kamel("install", "--olm=false", "--cluster-setup", "--force").Execute()).To(Succeed())
 		Expect(Kamel("install", "--olm=false", "-n", ns).Execute()).To(Succeed())
 
 		// Check the operator pod is running
-		Eventually(OperatorPodPhase(ns), TestTimeoutMedium).Should(Equal(v1.PodRunning))
+		Eventually(OperatorPodPhase(ns), TestTimeoutMedium).Should(Equal(corev1.PodRunning))
 
 		// Refresh the test client to account for the newly installed CRDs
 		SyncClient()
@@ -67,11 +67,11 @@ func TestOperatorUpgrade(t *testing.T) {
 		// Run the Integration
 		name := "yaml"
 		Expect(Kamel("run", "-n", ns, "files/yaml.yaml").Execute()).To(Succeed())
-		Eventually(IntegrationPodPhase(ns, name), TestTimeoutMedium).Should(Equal(v1.PodRunning))
-		Eventually(IntegrationCondition(ns, name, camelv1.IntegrationConditionReady), TestTimeoutShort).Should(Equal(v1.ConditionTrue))
+		Eventually(IntegrationPodPhase(ns, name), TestTimeoutMedium).Should(Equal(corev1.PodRunning))
+		Eventually(IntegrationCondition(ns, name, v1.IntegrationConditionReady), TestTimeoutShort).Should(Equal(corev1.ConditionTrue))
 		// Check the Integration version
-		Eventually(IntegrationVersion(ns, "yaml")).Should(Equal(version))
-		kit := IntegrationKit(ns, "yaml")()
+		Eventually(IntegrationVersion(ns, name)).Should(Equal(version))
+		kit := IntegrationKit(ns, name)()
 
 		// Clear the KAMEL_BIN environment variable so that the current version is used from now on
 		Expect(os.Setenv("KAMEL_BIN", "")).To(Succeed())
@@ -84,27 +84,27 @@ func TestOperatorUpgrade(t *testing.T) {
 		// Check the operator image is the current built one
 		Eventually(OperatorImage(ns)).Should(Equal(image))
 		// Check the operator pod is running
-		Eventually(OperatorPodPhase(ns), TestTimeoutMedium).Should(Equal(v1.PodRunning))
+		Eventually(OperatorPodPhase(ns), TestTimeoutMedium).Should(Equal(corev1.PodRunning))
 		// Check the IntegrationPlatform has been reconciled
 		Eventually(PlatformVersion(ns), TestTimeoutMedium).Should(Equal(defaults.Version))
 
 		// Check the Integration hasn't been upgraded
-		Consistently(IntegrationVersion(ns, "yaml"), 3*time.Second).Should(Equal(version))
+		Consistently(IntegrationVersion(ns, name), 3*time.Second).Should(Equal(version))
 
 		// Force the Integration upgrade
-		Expect(Kamel("rebuild", "yaml", "-n", ns).Execute()).To(Succeed())
+		Expect(Kamel("rebuild", name, "-n", ns).Execute()).To(Succeed())
 
 		// Check the Integration version change
-		Eventually(IntegrationVersion(ns, "yaml")).Should(Equal(defaults.Version))
+		Eventually(IntegrationVersion(ns, name)).Should(Equal(defaults.Version))
 		// Check the previous kit is not garbage collected
 		Eventually(KitsWithVersion(ns, version)).Should(Equal(1))
 		// Check a new kit is created with the current version
 		Eventually(KitsWithVersion(ns, defaults.Version)).Should(Equal(1))
 		// Check the Integration uses the new kit
-		Eventually(IntegrationKit(ns, "yaml"), TestTimeoutMedium).ShouldNot(Equal(kit))
+		Eventually(IntegrationKit(ns, name), TestTimeoutMedium).ShouldNot(Equal(kit))
 		// Check the Integration runs correctly
-		Eventually(IntegrationPodPhase(ns, "yaml"), TestTimeoutMedium).Should(Equal(v1.PodRunning))
-		Eventually(IntegrationCondition(ns, name, camelv1.IntegrationConditionReady), TestTimeoutShort).Should(Equal(v1.ConditionTrue))
+		Eventually(IntegrationPodPhase(ns, name), TestTimeoutMedium).Should(Equal(corev1.PodRunning))
+		Eventually(IntegrationCondition(ns, name, v1.IntegrationConditionReady), TestTimeoutShort).Should(Equal(corev1.ConditionTrue))
 
 		// Clean up
 		Expect(Kamel("delete", "--all", "-n", ns).Execute()).To(Succeed())
diff --git a/e2e/upgrade/olm_upgrade_test.go b/e2e/upgrade/olm_upgrade_test.go
index 2913493..21f6237 100644
--- a/e2e/upgrade/olm_upgrade_test.go
+++ b/e2e/upgrade/olm_upgrade_test.go
@@ -23,19 +23,24 @@ package common
 
 import (
 	"fmt"
-	. "github.com/apache/camel-k/e2e/support"
-	. "github.com/onsi/gomega"
-	"github.com/operator-framework/api/pkg/lib/version"
-	"github.com/operator-framework/api/pkg/operators/v1alpha1"
-	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"os"
-	ctrl "sigs.k8s.io/controller-runtime/pkg/client"
 	"testing"
+
+	. "github.com/onsi/gomega"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	ctrlutil "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+
+	"github.com/operator-framework/api/pkg/lib/version"
+	olm "github.com/operator-framework/api/pkg/operators/v1alpha1"
+
+	. "github.com/apache/camel-k/e2e/support"
+	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
 )
 
-const CATALOG_SOURCE_NAME = "test-camel-k-source"
+const catalogSourceName = "test-camel-k-source"
 
 func TestOLMAutomaticUpgrade(t *testing.T) {
 	prevIIB := os.Getenv("CAMEL_K_PREV_IIB")
@@ -43,23 +48,23 @@ func TestOLMAutomaticUpgrade(t *testing.T) {
 	kamel := os.Getenv("RELEASED_KAMEL_BIN")
 
 	if prevIIB == "" || newIIB == "" {
-		t.Skip("OLM Upgrade test needs CAMEL_K_PREV_IIB and CAMEL_K_PREV_IIB ENV variables")
+		t.Skip("OLM Upgrade test requires the CAMEL_K_PREV_IIB and CAMEL_K_NEW_IIB environment variables")
 	}
 
 	WithNewTestNamespace(t, func(ns string) {
-		Expect(createCatalogSource(CATALOG_SOURCE_NAME, prevIIB, ns)).To(Succeed())
-		Eventually(CatalogSourcePhase(CATALOG_SOURCE_NAME, ns), TestTimeoutMedium).Should(Equal("READY"))
+		Expect(createOrUpdateCatalogSource(ns, catalogSourceName, prevIIB)).To(Succeed())
+		Eventually(catalogSourcePhase(ns, catalogSourceName), TestTimeoutMedium).Should(Equal("READY"))
 
-		//set KAMEL_BIN only for this test - don't override the ENV variable for all tests
+		// Set KAMEL_BIN only for this test - don't override the ENV variable for all tests
 		Expect(os.Setenv("KAMEL_BIN", kamel)).To(Succeed())
 
-		Expect(Kamel("install", "-n", ns, "--olm=true", "--olm-source", CATALOG_SOURCE_NAME, "--olm-source-namespace", ns).Execute()).To(Succeed())
+		Expect(Kamel("install", "-n", ns, "--olm=true", "--olm-source", catalogSourceName, "--olm-source-namespace", ns).Execute()).To(Succeed())
 
-		//find the only one Camel-K CSV
-		noAdditionalConditions := func(csv v1alpha1.ClusterServiceVersion) bool {
+		// Find the only one Camel-K CSV
+		noAdditionalConditions := func(csv olm.ClusterServiceVersion) bool {
 			return true
 		}
-		Eventually(CKClusterServiceVersionPhase(noAdditionalConditions, ns), TestTimeoutMedium).Should(Equal(v1alpha1.CSVPhaseSucceeded))
+		Eventually(clusterServiceVersionPhase(noAdditionalConditions, ns), TestTimeoutMedium).Should(Equal(olm.CSVPhaseSucceeded))
 
 		// Refresh the test client to account for the newly installed CRDs
 		SyncClient()
@@ -72,92 +77,92 @@ func TestOLMAutomaticUpgrade(t *testing.T) {
 		var prevIPVersionPrefix string
 		var newIPVersionPrefix string
 
-		prevCSVVersion = CKClusterServiceVersion(noAdditionalConditions, ns)().Spec.Version
+		prevCSVVersion = clusterServiceVersion(noAdditionalConditions, ns)().Spec.Version
 		prevIPVersionPrefix = fmt.Sprintf("%d.%d", prevCSVVersion.Version.Major, prevCSVVersion.Version.Minor)
 
-		Expect(OperatorPod(ns)).ToNot(BeNil())
-
-		Expect(Kamel("run", "-n", ns, "files/yaml.yaml").Execute()).To(Succeed())
-		Eventually(IntegrationPodPhase(ns, "yaml"), TestTimeoutMedium).Should(Equal(v1.PodRunning))
+		// Check the operator pod is running
+		Eventually(OperatorPodPhase(ns), TestTimeoutMedium).Should(Equal(corev1.PodRunning))
 
+		// Check the IntegrationPlatform has been reconciled
 		Eventually(PlatformVersion(ns)).Should(ContainSubstring(prevIPVersionPrefix))
-		Expect(IntegrationVersion(ns, "yaml")()).To(ContainSubstring(prevIPVersionPrefix))
 
-		t.Run("OLM upgrade", func(t *testing.T) {
+		name := "yaml"
+		Expect(Kamel("run", "-n", ns, "files/yaml.yaml").Execute()).To(Succeed())
+		Eventually(IntegrationPodPhase(ns, name), TestTimeoutMedium).Should(Equal(corev1.PodRunning))
+		// Check the Integration runs correctly
+		Eventually(IntegrationCondition(ns, name, v1.IntegrationConditionReady), TestTimeoutShort).Should(Equal(corev1.ConditionTrue))
+		// Check the Integration version matches that of the current operator
+		Expect(IntegrationVersion(ns, name)()).To(ContainSubstring(prevIPVersionPrefix))
 
-			//invoke OLM upgrade
-			Expect(createCatalogSource(CATALOG_SOURCE_NAME, newIIB, ns)).To(Succeed())
+		t.Run("OLM upgrade", func(t *testing.T) {
+			// Trigger Camel K operator upgrade by updating the CatalogSource with the new index image
+			Expect(createOrUpdateCatalogSource(ns, catalogSourceName, newIIB)).To(Succeed())
 
-			// previous CSV is REPLACING
-			Eventually(CKClusterServiceVersionPhase(func(csv v1alpha1.ClusterServiceVersion) bool {
+			// Check the previous CSV is being replaced
+			Eventually(clusterServiceVersionPhase(func(csv olm.ClusterServiceVersion) bool {
 				return csv.Spec.Version.Version.String() == prevCSVVersion.Version.String()
-			}, ns), TestTimeoutMedium).Should(Equal(v1alpha1.CSVPhaseReplacing))
+			}, ns), TestTimeoutMedium).Should(Equal(olm.CSVPhaseReplacing))
 
-			// the new version is installed
-			Eventually(CKClusterServiceVersionPhase(func(csv v1alpha1.ClusterServiceVersion) bool {
+			// The new CSV is installed
+			Eventually(clusterServiceVersionPhase(func(csv olm.ClusterServiceVersion) bool {
 				return csv.Spec.Version.Version.String() != prevCSVVersion.Version.String()
-			}, ns), TestTimeoutMedium).Should(Equal(v1alpha1.CSVPhaseSucceeded))
+			}, ns), TestTimeoutMedium).Should(Equal(olm.CSVPhaseSucceeded))
 
-			// the old version is gone
-			Eventually(CKClusterServiceVersion(func(csv v1alpha1.ClusterServiceVersion) bool {
+			// The old CSV is gone
+			Eventually(clusterServiceVersion(func(csv olm.ClusterServiceVersion) bool {
 				return csv.Spec.Version.Version.String() == prevCSVVersion.Version.String()
 			}, ns), TestTimeoutMedium).Should(BeNil())
 
-			newCSVVersion = CKClusterServiceVersion(noAdditionalConditions, ns)().Spec.Version
+			newCSVVersion = clusterServiceVersion(noAdditionalConditions, ns)().Spec.Version
 			newIPVersionPrefix = fmt.Sprintf("%d.%d", newCSVVersion.Version.Major, newCSVVersion.Version.Minor)
 
 			Expect(prevCSVVersion.Version.String()).NotTo(Equal(newCSVVersion.Version.String()))
 
-			Eventually(OperatorPodPhase(ns)).Should(Equal(v1.PodRunning))
+			Eventually(OperatorPodPhase(ns), TestTimeoutMedium).Should(Equal(corev1.PodRunning))
 
+			// Check the IntegrationPlatform has been reconciled
 			Eventually(PlatformVersion(ns)).Should(ContainSubstring(newIPVersionPrefix))
 		})
 
 		t.Run("Integration upgrade", func(t *testing.T) {
-
 			// Clear the KAMEL_BIN environment variable so that the current version is used from now on
 			Expect(os.Setenv("KAMEL_BIN", "")).To(Succeed())
 
-			Expect(IntegrationVersion(ns, "yaml")()).To(ContainSubstring(prevIPVersionPrefix))
-			Expect(Kamel("rebuild", "yaml", "-n", ns).Execute()).To(Succeed())
+			// Check the Integration is still running the old version
+			Expect(IntegrationVersion(ns, name)()).To(ContainSubstring(prevIPVersionPrefix))
+
+			// Rebuild the Integration
+			Expect(Kamel("rebuild", name, "-n", ns).Execute()).To(Succeed())
 
-			Eventually(IntegrationVersion(ns, "yaml"), TestTimeoutMedium).Should(ContainSubstring(newIPVersionPrefix))
-			Eventually(IntegrationPodPhase(ns, "yaml")).Should(Equal(v1.PodRunning))
+			// Check the Integration runs correctly
+			Eventually(IntegrationPodPhase(ns, name)).Should(Equal(corev1.PodRunning))
+			// Check the Integration has upgraded
+			Eventually(IntegrationVersion(ns, name), TestTimeoutMedium).Should(ContainSubstring(newIPVersionPrefix))
 
 			// Clean up
 			Expect(Kamel("delete", "--all", "-n", ns).Execute()).To(Succeed())
 			Expect(Kamel("uninstall", "-n", ns).Execute()).To(Succeed())
 		})
-
 	})
 }
 
-func createCatalogSource(name string, image string, ns string) error {
-	catalogSource := v1alpha1.CatalogSource{
-		TypeMeta: metav1.TypeMeta{
-			Kind:       "CatalogSource",
-			APIVersion: v1alpha1.SchemeGroupVersion.String(),
-		},
+func createOrUpdateCatalogSource(ns, name, image string) error {
+	catalogSource := &olm.CatalogSource{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: ns,
 			Name:      name,
 		},
 	}
-	key := ctrl.ObjectKey{
-		Namespace: ns,
-		Name:      name,
-	}
-	if err := TestClient().Get(TestContext, key, &catalogSource); errors.IsNotFound(err) {
-		catalogSource.Spec = v1alpha1.CatalogSourceSpec{
+
+	_, err := ctrlutil.CreateOrUpdate(TestContext, TestClient(), catalogSource, func() error {
+		catalogSource.Spec = olm.CatalogSourceSpec{
 			Image:       image,
 			SourceType:  "grpc",
 			DisplayName: "OLM upgrade test Catalog",
 			Publisher:   "grpc",
 		}
-		return TestClient().Create(TestContext, &catalogSource)
-	} else {
-		catalogSource.Spec.Image = image
-		return TestClient().Update(TestContext, &catalogSource)
-	}
+		return nil
+	})
 
+	return err
 }
diff --git a/e2e/upgrade/util.go b/e2e/upgrade/util.go
new file mode 100644
index 0000000..0b6d667
--- /dev/null
+++ b/e2e/upgrade/util.go
@@ -0,0 +1,92 @@
+// +build integration
+
+// To enable compilation of this file in Goland, go to "Settings -> Go -> Vendoring & Build Tags -> Custom Tags" and add "integration"
+
+/*
+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.
+*/
+
+package common
+
+import (
+	"strings"
+	"unsafe"
+
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	ctrl "sigs.k8s.io/controller-runtime/pkg/client"
+
+	olm "github.com/operator-framework/api/pkg/operators/v1alpha1"
+
+	. "github.com/apache/camel-k/e2e/support"
+	"github.com/apache/camel-k/pkg/util/log"
+)
+
+func clusterServiceVersion(conditions func(olm.ClusterServiceVersion) bool, ns string) func() *olm.ClusterServiceVersion {
+	return func() *olm.ClusterServiceVersion {
+		lst := olm.ClusterServiceVersionList{}
+		if err := TestClient().List(TestContext, &lst, ctrl.InNamespace(ns)); err != nil {
+			panic(err)
+		}
+		for _, s := range lst.Items {
+			if strings.Contains(s.Name, "camel-k") && conditions(s) {
+				return &s
+			}
+		}
+		return nil
+	}
+}
+
+func clusterServiceVersionPhase(conditions func(olm.ClusterServiceVersion) bool, ns string) func() olm.ClusterServiceVersionPhase {
+	return func() olm.ClusterServiceVersionPhase {
+		if csv := clusterServiceVersion(conditions, ns)(); csv != nil && unsafe.Sizeof(csv.Status) > 0 {
+			return csv.Status.Phase
+		}
+		return ""
+	}
+}
+
+func catalogSource(ns, name string) func() *olm.CatalogSource {
+	return func() *olm.CatalogSource {
+		cs := &olm.CatalogSource{
+			TypeMeta: metav1.TypeMeta{
+				Kind:       "CatalogSource",
+				APIVersion: olm.SchemeGroupVersion.String(),
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: ns,
+				Name:      name,
+			},
+		}
+		if err := TestClient().Get(TestContext, ctrl.ObjectKeyFromObject(cs), cs); err != nil && errors.IsNotFound(err) {
+			return nil
+		} else if err != nil {
+			log.Errorf(err, "Error while retrieving CatalogSource %s", name)
+			return nil
+		}
+		return cs
+	}
+}
+
+func catalogSourcePhase(ns, name string) func() string {
+	return func() string {
+		if source := catalogSource(ns, name)(); source != nil && source.Status.GRPCConnectionState != nil {
+			return catalogSource(ns, name)().Status.GRPCConnectionState.LastObservedState
+		}
+		return ""
+	}
+}