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

@@ -45,34 +45,37 @@ func (s *subscriptionSyncer) now() *metav1.Time {
 // Sync reconciles Subscription events by invoking a sequence of reconcilers, passing the result of each
 // successful reconciliation as an argument to its successor.
 func (s *subscriptionSyncer) Sync(ctx context.Context, event kubestate.ResourceEvent) error {
-	res := &v1alpha1.Subscription{}
-	if err := scheme.Convert(event.Resource(), res, nil); err != nil {
+	sub := &v1alpha1.Subscription{}
+	if err := scheme.Convert(event.Resource(), sub, nil); err != nil {
 		return err
 	}
 
-	s.recordMetrics(res)
+	metrics.EmitSubMetric(sub)
 
 	logger := s.logger.WithFields(logrus.Fields{
-		"reconciling": fmt.Sprintf("%T", res),
-		"selflink":    res.GetSelfLink(),
+		"reconciling": fmt.Sprintf("%T", sub),
+		"selflink":    sub.GetSelfLink(),
 		"event":       event.Type(),
 	})
 	logger.Info("syncing")
 
 	// Enter initial state based on subscription and event type
 	// TODO: Consider generalizing initial generic add, update, delete transitions in the kubestate package.
 	// 		 Possibly make a resource event aware bridge between Sync and reconciler.
-	initial := NewSubscriptionState(res.DeepCopy())
+	initialSubState := NewSubscriptionState(sub.DeepCopy())
 	switch event.Type() {
 	case kubestate.ResourceAdded:
-		initial = initial.Add()
+		initialSubState = initialSubState.Add()
 	case kubestate.ResourceUpdated:
-		initial = initial.Update()
+		initialSubState = initialSubState.Update()
 	case kubestate.ResourceDeleted:
-		initial = initial.Delete()
+		{
+			initialSubState = initialSubState.Delete()
+			metrics.DeleteSubsMetric(sub)
+		}
 	}
 
-	reconciled, err := s.reconcilers.Reconcile(ctx, initial)
+	reconciled, err := s.reconcilers.Reconcile(ctx, initialSubState)
 	if err != nil {
 		logger.WithError(err).Warn("an error was encountered during reconciliation")
 		return err
@@ -85,15 +88,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/controller/operators/catalog/subscription/syncer_test.go

@@ -6,11 +6,19 @@ import (
 
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/kubestate"
 )
 
+const (
+	name             = "test-subscription"
+	packageName      = "test-package"
+	channel          = "test-channel"
+	installedCSVName = "test-csv"
+)
+
 func TestSync(t *testing.T) {
 	type fields struct {
 		syncer kubestate.Syncer
@@ -39,7 +47,22 @@ func TestSync(t *testing.T) {
 			args: args{
 				event: kubestate.NewResourceEvent(
 					kubestate.ResourceAdded,
-					&v1alpha1.Subscription{},
+					&v1alpha1.Subscription{
+						TypeMeta: metav1.TypeMeta{
+							Kind:       v1alpha1.SubscriptionKind,
+							APIVersion: v1alpha1.SubscriptionCRDAPIVersion,
+						},
+						ObjectMeta: metav1.ObjectMeta{
+							Name: name,
+						},
+						Spec: &v1alpha1.SubscriptionSpec{
+							Package: packageName,
+							Channel: channel,
+						},
+						Status: v1alpha1.SubscriptionStatus{
+							InstalledCSV: installedCSVName,
+						},
+					},
 				),
 			},
 			want: want{

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"
@@ -217,3 +218,11 @@ 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) {
+	SubscriptionSyncCount.WithLabelValues(sub.GetName(), sub.Status.InstalledCSV, sub.Spec.Channel, sub.Spec.Package).Inc()
+}
+
+func DeleteSubsMetric(sub *olmv1alpha1.Subscription) {
+	SubscriptionSyncCount.DeleteLabelValues(sub.GetName(), sub.Status.InstalledCSV, sub.Spec.Channel, sub.Spec.Package)
+}

test/e2e/metrics_e2e_test.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorclient"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/metrics"
 )
 
 var _ = Describe("Metrics", func() {
@@ -65,6 +66,22 @@ var _ = Describe("Metrics", func() {
 			ContainSubstring("csv_abnormal{name=\"%s\"", failingCSV.Name),
 			ContainSubstring("csv_succeeded{name=\"%s\"", failingCSV.Name),
 		))
+
+		subscriptionCleanup := createSubscription(GinkgoT(), crc, testNamespace, "metric-subscription", testPackageName, stableChannel, v1alpha1.ApprovalManual)
+
+		// Verify metrics have been emitted for subscription
+		Expect(getMetricsFromPod(c, getCatalogPod(c), "8081")).To(And(
+			ContainSubstring("subscription_sync_total"),
+			ContainSubstring(fmt.Sprintf("%s=\"%s\"", metrics.NAME_LABEL, "metric-subscription")),
+			ContainSubstring(fmt.Sprintf("%s=\"%s\"", metrics.CHANNEL_LABEL, stableChannel)),
+			ContainSubstring(fmt.Sprintf("%s=\"%s\"", metrics.PACKAGE_LABEL, testPackageName)),
+		))
+
+		subscriptionCleanup()
+
+		Expect(getMetricsFromPod(c, getCatalogPod(c), "8081")).ToNot(And(
+			ContainSubstring("subscription_sync_total{name=\"metric-subscription\""),
+		))
 	})
 })
 
@@ -80,6 +97,18 @@ func getOLMPod(client operatorclient.ClientInterface) *corev1.Pod {
 	return &podList.Items[0]
 }
 
+func getCatalogPod(client operatorclient.ClientInterface) *corev1.Pod {
+	listOptions := metav1.ListOptions{LabelSelector: "app=catalog-operator"}
+	var podList *corev1.PodList
+	Eventually(func() (err error) {
+		podList, err = client.KubernetesInterface().CoreV1().Pods(operatorNamespace).List(context.TODO(), listOptions)
+		return
+	}).Should(Succeed(), "Failed to list OLM pods")
+	Expect(len(podList.Items)).To(Equal(1))
+
+	return &podList.Items[0]
+}
+
 func getMetricsFromPod(client operatorclient.ClientInterface, pod *corev1.Pod, port string) string {
 	By(fmt.Sprintf("querying pod %s/%s", pod.GetNamespace(), pod.GetName()))
 

test/e2e/subscription_e2e_test.go

@@ -1802,7 +1802,7 @@ func createSubscription(t GinkgoTInterface, crc versioned.Interface, namespace,
 	}
 
 	subscription, err := crc.OperatorsV1alpha1().Subscriptions(namespace).Create(context.TODO(), subscription, metav1.CreateOptions{})
-	require.NoError(t, err)
+	Expect(err).ToNot(HaveOccurred())
 	return buildSubscriptionCleanupFunc(crc, subscription)
 }