feat(metrics): Emit metrics for CatalogSource state

This PR introduces a prometheus gauage vector `catalogsource_ready`
that is emitted by each CatalogSource to indicate the connectivity.State
of the CatalogSource object. A value of 1 for the vector indicates that
the CatalogSource object is in a READY state, while a value of 0
indicates that the CatalogSource object is in one of IDLE/CONNECTING/
TRANSIENT_FAILURE/SHUTDOWN state. If/When the CatalogSource object
eventually reaches a READY state, the value for the vector is set
to 0.

Signed-off-by: Anik Bhattacharjee <[email protected]>

Author: anik120

pkg/controller/operators/catalog/operator.go

@@ -396,6 +396,7 @@ func (o *Operator) syncSourceState(state grpc.SourceState) {
 	o.sourcesLastUpdate.Set(o.now().Time)
 
 	o.logger.Infof("state.Key.Namespace=%s state.Key.Name=%s state.State=%s", state.Key.Namespace, state.Key.Name, state.State.String())
+	metrics.RegisterCatalogSourceState(state.Key.Name, state.Key.Namespace, state.State)
 
 	switch state.State {
 	case connectivity.Ready:
@@ -513,6 +514,8 @@ func (o *Operator) handleCatSrcDeletion(obj interface{}) {
 		o.logger.WithError(err).Warn("error closing client")
 	}
 	o.logger.WithField("source", sourceKey).Info("removed client for deleted catalogsource")
+
+	metrics.DeleteCatalogSourceStateMetric(catsrc.GetName(), catsrc.GetNamespace())
 }
 
 func validateSourceType(logger *logrus.Entry, in *v1alpha1.CatalogSource) (out *v1alpha1.CatalogSource, continueSync bool, _ error) {

pkg/metrics/metrics.go

@@ -4,6 +4,7 @@ import (
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"google.golang.org/grpc/connectivity"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 
@@ -141,6 +142,14 @@ var (
 		},
 	)
 
+	catalogSourceReady = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "catalogSource_ready",
+			Help: "State of a CatalogSource. 1 indicates that the CatalogSource is in a READY state. 0 indicates CatalogSource is in a Non READY state.",
+		},
+		[]string{NAMESPACE_LABEL, NAME_LABEL},
+	)
+
 	// exported since it's not handled by HandleMetrics
 	CSVUpgradeCount = prometheus.NewCounter(
 		prometheus.CounterOpts{
@@ -206,6 +215,7 @@ func RegisterCatalog() {
 	prometheus.MustRegister(installPlanCount)
 	prometheus.MustRegister(subscriptionCount)
 	prometheus.MustRegister(catalogSourceCount)
+	prometheus.MustRegister(catalogSourceReady)
 	prometheus.MustRegister(SubscriptionSyncCount)
 	prometheus.MustRegister(dependencyResolutionSummary)
 }
@@ -214,6 +224,19 @@ func CounterForSubscription(name, installedCSV, channelName, packageName, planAp
 	return SubscriptionSyncCount.WithLabelValues(name, installedCSV, channelName, packageName, planApprovalStrategy)
 }
 
+func RegisterCatalogSourceState(name, namespace string, state connectivity.State) {
+	switch state {
+	case connectivity.Ready:
+		catalogSourceReady.WithLabelValues(namespace, name).Set(1)
+	default:
+		catalogSourceReady.WithLabelValues(namespace, name).Set(0)
+	}
+}
+
+func DeleteCatalogSourceStateMetric(name, namespace string) {
+	catalogSourceReady.DeleteLabelValues(namespace, name)
+}
+
 func DeleteCSVMetric(oldCSV *olmv1alpha1.ClusterServiceVersion) {
 	// Delete the old CSV metrics
 	csvAbnormal.DeleteLabelValues(oldCSV.Namespace, oldCSV.Name, oldCSV.Spec.Version.String(), string(oldCSV.Status.Phase), string(oldCSV.Status.Reason))

test/e2e/like_metric_matcher_test.go

@@ -51,6 +51,10 @@ func WithName(name string) MetricPredicate {
 	return WithLabel("name", name)
 }
 
+func WithNamespace(namespace string) MetricPredicate {
+	return WithLabel("namespace", namespace)
+}
+
 func WithChannel(channel string) MetricPredicate {
 	return WithLabel("channel", channel)
 }

test/e2e/metrics_e2e_test.go

@@ -5,19 +5,23 @@ package e2e
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"regexp"
 	"strings"
 
+	"github.com/blang/semver/v4"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	io_prometheus_client "github.com/prometheus/client_model/go"
 	"github.com/prometheus/common/expfmt"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/net"
 
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/clientset/versioned"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorclient"
 	"github.com/operator-framework/operator-lifecycle-manager/test/e2e/ctx"
 )
@@ -266,6 +270,109 @@ var _ = Describe("Metrics are generated for OLM managed resources", func() {
 			})
 		})
 	})
+
+	Context("Metrics emitted by CatalogSources", func() {
+		When("A valid CatalogSource object is created", func() {
+			var (
+				name        = "metrics-catsrc-valid"
+				cleanup     func()
+				cleanupDone = false
+			)
+			BeforeEach(func() {
+				mainPackageName := genName("nginx-")
+
+				mainPackageStable := fmt.Sprintf("%s-stable", mainPackageName)
+
+				stableChannel := "stable"
+
+				mainCRD := newCRD(genName("ins-"))
+				mainCSV := newCSV(mainPackageStable, testNamespace, "", semver.MustParse("0.1.0"), []apiextensions.CustomResourceDefinition{mainCRD}, nil, nil)
+
+				mainManifests := []registry.PackageManifest{
+					{
+						PackageName: mainPackageName,
+						Channels: []registry.PackageChannel{
+							{Name: stableChannel, CurrentCSVName: mainPackageStable},
+						},
+						DefaultChannelName: stableChannel,
+					},
+				}
+				_, cleanup = createInternalCatalogSource(c, crc, name, testNamespace, mainManifests, []apiextensions.CustomResourceDefinition{mainCRD}, []v1alpha1.ClusterServiceVersion{mainCSV})
+			})
+			AfterEach(func() {
+				if !cleanupDone {
+					cleanup()
+				}
+			})
+			It("emits metrics for the catalogSource", func() {
+				Eventually(func() []Metric {
+					return getMetricsFromPod(c, getPodWithLabel(c, "app=catalog-operator"), "8081")
+				}).Should(And(
+					ContainElement(LikeMetric(
+						WithFamily("catalog_source_count"),
+						WithValueGreaterThan(0),
+					)),
+					ContainElement(LikeMetric(
+						WithFamily("catalogSource_ready"),
+						WithName(name),
+						WithNamespace(testNamespace),
+						WithValue(1),
+					)),
+				))
+			})
+			When("The CatalogSource object is deleted", func() {
+				BeforeEach(func() {
+					cleanup()
+					cleanupDone = true
+				})
+				It("deletes the metrics for the CatalogSource", func() {
+					Eventually(func() []Metric {
+						return getMetricsFromPod(c, getPodWithLabel(c, "app=catalog-operator"), "8081")
+					}).Should(And(
+						Not(ContainElement(LikeMetric(
+							WithFamily("catalogSource_ready"),
+							WithName(name),
+							WithNamespace(testNamespace),
+						)))))
+				})
+			})
+		})
+
+		When("A CatalogSource object is in an invalid state", func() {
+			var (
+				name    = "metrics-catsrc-invalid"
+				cleanup func()
+			)
+			BeforeEach(func() {
+				_, cleanup = createInvalidGRPCCatalogSource(crc, name, testNamespace)
+			})
+			AfterEach(func() {
+				cleanup()
+			})
+			It("emits metrics for the CatlogSource with a Value greater than 0", func() {
+				Eventually(func() []Metric {
+					return getMetricsFromPod(c, getPodWithLabel(c, "app=catalog-operator"), "8081")
+				}).Should(And(
+					ContainElement(LikeMetric(
+						WithFamily("catalogSource_ready"),
+						WithName(name),
+						WithNamespace(testNamespace),
+						WithValue(0),
+					)),
+				))
+				Consistently(func() []Metric {
+					return getMetricsFromPod(c, getPodWithLabel(c, "app=catalog-operator"), "8081")
+				}, "3m").Should(And(
+					ContainElement(LikeMetric(
+						WithFamily("catalogSource_ready"),
+						WithName(name),
+						WithNamespace(testNamespace),
+						WithValue(0),
+					)),
+				))
+			})
+		})
+	})
 })
 
 func getPodWithLabel(client operatorclient.ClientInterface, label string) *corev1.Pod {

test/e2e/util_test.go

@@ -521,6 +521,30 @@ func buildServiceAccountCleanupFunc(t GinkgoTInterface, c operatorclient.ClientI
 	}
 }
 
+func createInvalidGRPCCatalogSource(crc versioned.Interface, name, namespace string) (*v1alpha1.CatalogSource, cleanupFunc) {
+
+	catalogSource := &v1alpha1.CatalogSource{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       v1alpha1.CatalogSourceKind,
+			APIVersion: v1alpha1.CatalogSourceCRDAPIVersion,
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Spec: v1alpha1.CatalogSourceSpec{
+			SourceType: "grpc",
+			Image:      "localhost:0/not/exists:catsrc",
+		},
+	}
+
+	ctx.Ctx().Logf("Creating catalog source %s in namespace %s...", name, namespace)
+	catalogSource, err := crc.OperatorsV1alpha1().CatalogSources(namespace).Create(context.TODO(), catalogSource, metav1.CreateOptions{})
+	Expect(err).ToNot(HaveOccurred())
+	ctx.Ctx().Logf("Catalog source %s created", name)
+	return catalogSource, buildCatalogSourceCleanupFunc(crc, namespace, catalogSource)
+}
+
 func createInternalCatalogSource(c operatorclient.ClientInterface, crc versioned.Interface, name, namespace string, manifests []registry.PackageManifest, crds []apiextensions.CustomResourceDefinition, csvs []v1alpha1.ClusterServiceVersion) (*v1alpha1.CatalogSource, cleanupFunc) {
 	configMap, configMapCleanup := createConfigMapForCatalogData(c, name, namespace, manifests, crds, csvs)
 
@@ -539,7 +563,6 @@ func createInternalCatalogSource(c operatorclient.ClientInterface, crc versioned
 			ConfigMap:  configMap.GetName(),
 		},
 	}
-	catalogSource.SetNamespace(namespace)
 
 	ctx.Ctx().Logf("Creating catalog source %s in namespace %s...", name, namespace)
 	catalogSource, err := crc.OperatorsV1alpha1().CatalogSources(namespace).Create(context.TODO(), catalogSource, metav1.CreateOptions{})