Bug 1822396: Delete subscription metric when an operator is uninstalled

When an operator was subscribed to using a Subscription Object, the subscription_sync_total
metric was emitted whenever the Subscription Object was created/updated/deleted. This PR
updates that behaviour to emit the metric only when the Subscription object is created/updated,
and deletes the time series for that particular subscription when the subscription object is
deleted.

Author: anik120

pkg/controller/operators/catalog/subscription/syncer.go

@@ -50,7 +50,7 @@ func (s *subscriptionSyncer) Sync(ctx context.Context, event kubestate.ResourceE
 		return err
 	}
 
-	s.recordMetrics(res)
+	metrics.EmitSubMetric(res)
 
 	logger := s.logger.WithFields(logrus.Fields{
 		"reconciling": fmt.Sprintf("%T", res),
@@ -68,8 +68,10 @@ func (s *subscriptionSyncer) Sync(ctx context.Context, event kubestate.ResourceE
 		initial = initial.Add()
 	case kubestate.ResourceUpdated:
 		initial = initial.Update()
+		metrics.UpdateSubsSyncCounterStorage(res)
 	case kubestate.ResourceDeleted:
 		initial = initial.Delete()
+		metrics.DeleteSubsMetric(res)
 	}
 
 	reconciled, err := s.reconcilers.Reconcile(ctx, initial)
@@ -85,15 +87,6 @@ func (s *subscriptionSyncer) Sync(ctx context.Context, event kubestate.ResourceE
 	return nil
 }
 
-func (s *subscriptionSyncer) recordMetrics(sub *v1alpha1.Subscription) {
-	// sub.Spec is not a required field.
-	if sub.Spec == nil {
-		return
-	}
-
-	metrics.CounterForSubscription(sub.GetName(), sub.Status.InstalledCSV, sub.Spec.Channel, sub.Spec.Package).Inc()
-}
-
 func (s *subscriptionSyncer) Notify(event kubestate.ResourceEvent) {
 	s.notify(event)
 }

pkg/metrics/metrics.go

@@ -2,6 +2,7 @@ package metrics
 
 import (
 	"context"
+
 	"github.com/prometheus/client_golang/prometheus"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -168,8 +169,19 @@ var (
 		},
 		[]string{NAMESPACE_LABEL, NAME_LABEL, VERSION_LABEL, PHASE_LABEL, REASON_LABEL},
 	)
+
+	// subscriptionSyncCounters keeps a record of the promethues counters emitted by
+	// Subscription objects. The key of a record is the Subscription name, while the value
+	//  is struct containing label values used in the counter
+	subscriptionSyncCounters = make(map[string]subscriptionSyncLabelValues)
 )
 
+type subscriptionSyncLabelValues struct {
+	installedCSV string
+	pkg          string
+	channel      string
+}
+
 func RegisterOLM() {
 	prometheus.MustRegister(csvCount)
 	prometheus.MustRegister(csvSucceeded)
@@ -217,3 +229,43 @@ func EmitCSVMetric(oldCSV *olmv1alpha1.ClusterServiceVersion, newCSV *olmv1alpha
 		csvAbnormal.WithLabelValues(newCSV.Namespace, newCSV.Name, newCSV.Spec.Version.String(), string(newCSV.Status.Phase), string(newCSV.Status.Reason)).Set(1)
 	}
 }
+
+func EmitSubMetric(sub *olmv1alpha1.Subscription) {
+	if sub.Spec == nil {
+		return
+	}
+	SubscriptionSyncCount.WithLabelValues(sub.GetName(), sub.Status.InstalledCSV, sub.Spec.Channel, sub.Spec.Package).Inc()
+	if _, present := subscriptionSyncCounters[sub.GetName()]; !present {
+		subscriptionSyncCounters[sub.GetName()] = subscriptionSyncLabelValues{
+			installedCSV: sub.Status.InstalledCSV,
+			pkg:          sub.Spec.Package,
+			channel:      sub.Spec.Channel,
+		}
+	}
+}
+
+func DeleteSubsMetric(sub *olmv1alpha1.Subscription) {
+	if sub.Spec == nil {
+		return
+	}
+	SubscriptionSyncCount.DeleteLabelValues(sub.GetName(), sub.Status.InstalledCSV, sub.Spec.Channel, sub.Spec.Package)
+}
+
+func UpdateSubsSyncCounterStorage(sub *olmv1alpha1.Subscription) {
+	if sub.Spec == nil {
+		return
+	}
+	counterValues := subscriptionSyncCounters[sub.GetName()]
+
+	if sub.Spec.Channel != counterValues.channel ||
+		sub.Spec.Package != counterValues.pkg ||
+		sub.Status.InstalledCSV != counterValues.installedCSV {
+
+		// Delete metric will label values of old Subscription first
+		SubscriptionSyncCount.DeleteLabelValues(sub.GetName(), counterValues.installedCSV, counterValues.channel, counterValues.pkg)
+
+		counterValues.installedCSV = sub.Status.InstalledCSV
+		counterValues.pkg = sub.Spec.Package
+		counterValues.channel = sub.Spec.Channel
+	}
+}

test/e2e/metrics_e2e_test.go

@@ -5,20 +5,23 @@ package e2e
 import (
 	"context"
 	"fmt"
+	"regexp"
+	"time"
+
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/net"
-	"regexp"
 
 	"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/lib/operatorclient"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/metrics"
 	"github.com/operator-framework/operator-lifecycle-manager/test/e2e/ctx"
 )
 
-var _ = Describe("Metrics are generated for OLM pod", func() {
+var _ = Describe("Metrics are generated for OLM managed resources", func() {
 
 	var (
 		c   operatorclient.ClientInterface
@@ -69,7 +72,7 @@ var _ = Describe("Metrics are generated for OLM pod", func() {
 			It("generates csv_abnormal metric for OLM pod", func() {
 
 				// Verify metrics have been emitted for packageserver csv
-				Expect(getMetricsFromPod(c, getOLMPod(c), "8081")).To(And(
+				Expect(getMetricsFromPod(c, getPodWithLabel(c, "app=olm-operator"), "8081")).To(And(
 					ContainSubstring("csv_abnormal"),
 					ContainSubstring(fmt.Sprintf("name=\"%s\"", failingCSV.Name)),
 					ContainSubstring("phase=\"Failed\""),
@@ -91,18 +94,89 @@ var _ = Describe("Metrics are generated for OLM pod", func() {
 
 				It("deletes its associated CSV metrics", func() {
 					// Verify that when the csv has been deleted, it deletes the corresponding CSV metrics
-					Expect(getMetricsFromPod(c, getOLMPod(c), "8081")).ToNot(And(
+					Expect(getMetricsFromPod(c, getPodWithLabel(c, "app=olm-operator"), "8081")).ToNot(And(
 						ContainSubstring("csv_abnormal{name=\"%s\"", failingCSV.Name),
 						ContainSubstring("csv_succeeded{name=\"%s\"", failingCSV.Name),
 					))
 				})
 			})
 		})
 	})
+
+	Context("Subscription Metric", func() {
+		var (
+			subscriptionCleanup cleanupFunc
+			subscription        *v1alpha1.Subscription
+		)
+		When("A subscription object is created", func() {
+
+			BeforeEach(func() {
+				subscriptionCleanup, _ = createSubscription(GinkgoT(), crc, testNamespace, "metric-subscription-for-create", testPackageName, stableChannel, v1alpha1.ApprovalManual)
+			})
+
+			It("generates subscription_sync_total metric", func() {
+
+				// Verify metrics have been emitted for subscription
+				Eventually(func() string {
+					return getMetricsFromPod(c, getPodWithLabel(c, "app=catalog-operator"), "8081")
+				}, time.Minute, 5*time.Second).Should(And(
+					ContainSubstring("subscription_sync_total"),
+					ContainSubstring(fmt.Sprintf("%s=\"%s\"", metrics.NAME_LABEL, "metric-subscription-for-create")),
+					ContainSubstring(fmt.Sprintf("%s=\"%s\"", metrics.CHANNEL_LABEL, stableChannel)),
+					ContainSubstring(fmt.Sprintf("%s=\"%s\"", metrics.PACKAGE_LABEL, testPackageName))))
+			})
+			if subscriptionCleanup != nil {
+				subscriptionCleanup()
+			}
+		})
+		When("A subscription object is updated", func() {
+
+			BeforeEach(func() {
+				subscriptionCleanup, subscription = createSubscription(GinkgoT(), crc, testNamespace, "metric-subscription-for-update", testPackageName, stableChannel, v1alpha1.ApprovalManual)
+				subscription.Spec.Channel = "beta"
+				updateSubscription(GinkgoT(), crc, subscription)
+			})
+
+			It("deletes the old Subscription metric and emits the new metric", func() {
+				Eventually(func() string {
+					return getMetricsFromPod(c, getPodWithLabel(c, "app=catalog-operator"), "8081")
+				}, time.Minute, 5*time.Second).ShouldNot(And(
+					ContainSubstring("subscription_sync_total{name=\"metric-subscription-for-update\""),
+					ContainSubstring(fmt.Sprintf("%s=\"%s\"", metrics.CHANNEL_LABEL, stableChannel))))
+
+				Eventually(func() string {
+					return getMetricsFromPod(c, getPodWithLabel(c, "app=catalog-operator"), "8081")
+				}, time.Minute, 5*time.Second).Should(And(
+					ContainSubstring("subscription_sync_total"),
+					ContainSubstring(fmt.Sprintf("%s=\"%s\"", metrics.NAME_LABEL, "metric-subscription-for-update")),
+					ContainSubstring(fmt.Sprintf("%s=\"%s\"", metrics.CHANNEL_LABEL, "beta")),
+					ContainSubstring(fmt.Sprintf("%s=\"%s\"", metrics.PACKAGE_LABEL, testPackageName))))
+			})
+			if subscriptionCleanup != nil {
+				subscriptionCleanup()
+			}
+		})
+
+		When("A subscription object is deleted", func() {
+
+			BeforeEach(func() {
+				subscriptionCleanup, subscription = createSubscription(GinkgoT(), crc, testNamespace, "metric-subscription-for-delete", testPackageName, stableChannel, v1alpha1.ApprovalManual)
+				if subscriptionCleanup != nil {
+					subscriptionCleanup()
+				}
+			})
+
+			It("deletes the Subscription metric", func() {
+				Eventually(func() string {
+					return getMetricsFromPod(c, getPodWithLabel(c, "app=catalog-operator"), "8081")
+				}, time.Minute, 5*time.Second).ShouldNot(ContainSubstring("subscription_sync_total{name=\"metric-subscription-for-update\""))
+			})
+		})
+	})
 })
 
-func getOLMPod(client operatorclient.ClientInterface) *corev1.Pod {
-	listOptions := metav1.ListOptions{LabelSelector: "app=olm-operator"}
+func getPodWithLabel(client operatorclient.ClientInterface, label string) *corev1.Pod {
+	listOptions := metav1.ListOptions{LabelSelector: label}
 	var podList *corev1.PodList
 	Eventually(func() (err error) {
 		podList, err = client.KubernetesInterface().CoreV1().Pods(operatorNamespace).List(context.TODO(), listOptions)

test/e2e/subscription_e2e_test.go

@@ -51,7 +51,7 @@ var _ = Describe("Subscription", func() {
 		}()
 		require.NoError(GinkgoT(), initCatalog(GinkgoT(), c, crc))
 
-		cleanup := createSubscription(GinkgoT(), crc, testNamespace, testSubscriptionName, testPackageName, betaChannel, v1alpha1.ApprovalAutomatic)
+		cleanup, _ := createSubscription(GinkgoT(), crc, testNamespace, testSubscriptionName, testPackageName, betaChannel, v1alpha1.ApprovalAutomatic)
 		defer cleanup()
 
 		subscription, err := fetchSubscription(crc, testNamespace, testSubscriptionName, subscriptionStateAtLatestChecker)
@@ -80,7 +80,7 @@ var _ = Describe("Subscription", func() {
 		_, err := createCSV(c, crc, stableCSV, testNamespace, false, false)
 		require.NoError(GinkgoT(), err)
 
-		subscriptionCleanup := createSubscription(GinkgoT(), crc, testNamespace, testSubscriptionName, testPackageName, alphaChannel, v1alpha1.ApprovalAutomatic)
+		subscriptionCleanup, _ := createSubscription(GinkgoT(), crc, testNamespace, testSubscriptionName, testPackageName, alphaChannel, v1alpha1.ApprovalAutomatic)
 		defer subscriptionCleanup()
 
 		subscription, err := fetchSubscription(crc, testNamespace, testSubscriptionName, subscriptionStateAtLatestChecker)
@@ -188,7 +188,7 @@ var _ = Describe("Subscription", func() {
 		}()
 		require.NoError(GinkgoT(), initCatalog(GinkgoT(), c, crc))
 
-		subscriptionCleanup := createSubscription(GinkgoT(), crc, testNamespace, "manual-subscription", testPackageName, stableChannel, v1alpha1.ApprovalManual)
+		subscriptionCleanup, _ := createSubscription(GinkgoT(), crc, testNamespace, "manual-subscription", testPackageName, stableChannel, v1alpha1.ApprovalManual)
 		defer subscriptionCleanup()
 
 		subscription, err := fetchSubscription(crc, testNamespace, "manual-subscription", subscriptionStateUpgradePendingChecker)
@@ -1783,7 +1783,7 @@ func buildSubscriptionCleanupFunc(crc versioned.Interface, subscription *v1alpha
 	}
 }
 
-func createSubscription(t GinkgoTInterface, crc versioned.Interface, namespace, name, packageName, channel string, approval v1alpha1.Approval) cleanupFunc {
+func createSubscription(t GinkgoTInterface, crc versioned.Interface, namespace, name, packageName, channel string, approval v1alpha1.Approval) (cleanupFunc, *v1alpha1.Subscription) {
 	subscription := &v1alpha1.Subscription{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       v1alpha1.SubscriptionKind,
@@ -1803,8 +1803,13 @@ func createSubscription(t GinkgoTInterface, crc versioned.Interface, namespace,
 	}
 
 	subscription, err := crc.OperatorsV1alpha1().Subscriptions(namespace).Create(context.TODO(), subscription, metav1.CreateOptions{})
-	require.NoError(t, err)
-	return buildSubscriptionCleanupFunc(crc, subscription)
+	Expect(err).ToNot(HaveOccurred())
+	return buildSubscriptionCleanupFunc(crc, subscription), subscription
+}
+
+func updateSubscription(t GinkgoTInterface, crc versioned.Interface, subscription *v1alpha1.Subscription) {
+	_, err := crc.OperatorsV1alpha1().Subscriptions(subscription.GetNamespace()).Update(context.TODO(), subscription, metav1.UpdateOptions{})
+	Expect(err).ToNot(HaveOccurred())
 }
 
 func createSubscriptionForCatalog(crc versioned.Interface, namespace, name, catalog, packageName, channel, startingCSV string, approval v1alpha1.Approval) cleanupFunc {