Fix issue where by metrics _could_ be registered *after* the pod has been deleted

Author: David Collom

cmd/app/app.go

@@ -98,7 +98,7 @@ func NewCommand(ctx context.Context) *cobra.Command {
 				log.Fatal("Unable to set up ready check:", err)
 			}
 
-			metricsServer := metrics.New(log, ctrmetrics.Registry)
+			metricsServer := metrics.New(log, ctrmetrics.Registry, mgr.GetCache())
 
 			opts.Client.Transport = transport.Chain(
 				cleanhttp.DefaultTransport(),

pkg/controller/pod_controller_test.go

@@ -34,7 +34,11 @@ func init() {
 
 func TestNewController(t *testing.T) {
 	kubeClient := fake.NewFakeClient()
-	metrics := &metrics.Metrics{}
+	metrics := metrics.New(
+		logrus.NewEntry(logrus.StandardLogger()),
+		prometheus.NewRegistry(),
+		kubeClient,
+	)
 	imageClient := &client.Client{}
 
 	controller := NewPodReconciler(5*time.Minute, metrics, imageClient, kubeClient, testLogger, true)
@@ -78,6 +82,7 @@ func TestReconcile(t *testing.T) {
 			metrics := metrics.New(
 				logrus.NewEntry(logrus.StandardLogger()),
 				prometheus.NewRegistry(),
+				kubeClient,
 			)
 			controller := NewPodReconciler(5*time.Minute, metrics, imageClient, kubeClient, testLogger, true)
 			controller.RequeueDuration = 5 * time.Minute
@@ -110,7 +115,11 @@ func TestReconcile(t *testing.T) {
 }
 func TestSetupWithManager(t *testing.T) {
 	kubeClient := fake.NewClientBuilder().Build()
-	metrics := &metrics.Metrics{}
+	metrics := metrics.New(
+		logrus.NewEntry(logrus.StandardLogger()),
+		prometheus.NewRegistry(),
+		kubeClient,
+	)
 	imageClient := &client.Client{}
 	controller := NewPodReconciler(5*time.Minute, metrics, imageClient, kubeClient, testLogger, true)
 

pkg/controller/pod_sync_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	"github.com/prometheus/client_golang/prometheus"
 
@@ -24,8 +25,13 @@ import (
 // Test for the sync method.
 func TestController_Sync(t *testing.T) {
 	t.Parallel()
+
 	log := logrus.NewEntry(logrus.New())
-	metrics := metrics.New(log, prometheus.NewRegistry())
+	metrics := metrics.New(
+		logrus.NewEntry(logrus.StandardLogger()),
+		prometheus.NewRegistry(),
+		fake.NewFakeClient(),
+	)
 	imageClient := &client.Client{}
 	searcher := search.New(log, 5*time.Minute, version.New(log, imageClient, 5*time.Minute))
 	checker := checker.New(searcher)
@@ -60,7 +66,7 @@ func TestController_Sync(t *testing.T) {
 func TestController_SyncContainer(t *testing.T) {
 	t.Parallel()
 	log := logrus.NewEntry(logrus.New())
-	metrics := metrics.New(log, prometheus.NewRegistry())
+	metrics := metrics.New(log, prometheus.NewRegistry(), fake.NewFakeClient())
 	imageClient := &client.Client{}
 	searcher := search.New(log, 5*time.Minute, version.New(log, imageClient, 5*time.Minute))
 	checker := checker.New(searcher)
@@ -92,7 +98,7 @@ func TestController_SyncContainer(t *testing.T) {
 func TestController_CheckContainer(t *testing.T) {
 	t.Parallel()
 	log := logrus.NewEntry(logrus.New())
-	metrics := metrics.New(log, prometheus.NewRegistry())
+	metrics := metrics.New(log, prometheus.NewRegistry(), fake.NewFakeClient())
 	imageClient := &client.Client{}
 	searcher := search.New(log, 5*time.Minute, version.New(log, imageClient, 5*time.Minute))
 	checker := checker.New(searcher)
@@ -122,7 +128,7 @@ func TestController_SyncContainer_NoVersionFound(t *testing.T) {
 	t.Parallel()
 
 	log := logrus.NewEntry(logrus.New())
-	metrics := metrics.New(log, prometheus.NewRegistry())
+	metrics := metrics.New(log, prometheus.NewRegistry(), fake.NewFakeClient())
 	imageClient := &client.Client{}
 	searcher := search.New(log, 5*time.Minute, version.New(log, imageClient, 5*time.Minute))
 	checker := checker.New(searcher)

pkg/metrics/metrics.go

@@ -1,11 +1,16 @@
 package metrics
 
 import (
+	"context"
 	"sync"
 	"time"
 
 	"github.com/sirupsen/logrus"
 
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	k8sclient "sigs.k8s.io/controller-runtime/pkg/client"
+
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -22,13 +27,16 @@ type Metrics struct {
 	containerImageDuration *prometheus.GaugeVec
 	containerImageErrors   *prometheus.CounterVec
 
+	cache k8sclient.Reader
+
 	// Contains all metrics for the roundtripper
 	roundTripper *RoundTripper
 
 	mu sync.Mutex
 }
 
-func New(log *logrus.Entry, reg ctrmetrics.RegistererGatherer) *Metrics {
+// func New(log *logrus.Entry, reg ctrmetrics.RegistererGatherer, kubeClient k8sclient.Client) *Metrics {
+func New(log *logrus.Entry, reg ctrmetrics.RegistererGatherer, cache k8sclient.Reader) *Metrics {
 	// Attempt to register, but ignore errors
 	_ = reg.Register(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}))
 	_ = reg.Register(collectors.NewGoCollector())
@@ -63,7 +71,9 @@ func New(log *logrus.Entry, reg ctrmetrics.RegistererGatherer) *Metrics {
 	)
 
 	return &Metrics{
-		log:                    log.WithField("module", "metrics"),
+		log:   log.WithField("module", "metrics"),
+		cache: cache,
+
 		registry:               reg,
 		containerImageVersion:  containerImageVersion,
 		containerImageDuration: containerImageDuration,
@@ -125,6 +135,11 @@ func (m *Metrics) RegisterImageDuration(namespace, pod, container, image string,
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
+	if !m.PodExists(context.Background(), namespace, pod) {
+		m.log.WithField("metric", "RegisterImageDuration").Warnf("pod %s/%s not found, not registering error", namespace, pod)
+		return
+	}
+
 	m.containerImageDuration.WithLabelValues(
 		namespace, pod, container, image,
 	).Set(time.Since(startTime).Seconds())
@@ -134,6 +149,11 @@ func (m *Metrics) ReportError(namespace, pod, container, imageURL string) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
+	if !m.PodExists(context.Background(), namespace, pod) {
+		m.log.WithField("metric", "ReportError").Warnf("pod %s/%s not found, not registering error", namespace, pod)
+		return
+	}
+
 	m.containerImageErrors.WithLabelValues(
 		namespace, pod, container, imageURL,
 	).Inc()
@@ -157,3 +177,10 @@ func (m *Metrics) buildPartialLabels(namespace, pod string) prometheus.Labels {
 		"pod":       pod,
 	}
 }
+
+// This _should_ leverage the Controllers Cache
+func (m *Metrics) PodExists(ctx context.Context, ns, name string) bool {
+	pod := &corev1.Pod{}
+	err := m.cache.Get(ctx, types.NamespacedName{Name: name, Namespace: ns}, pod)
+	return err == nil && pod.GetDeletionTimestamp() == nil
+}

pkg/metrics/metrics_test.go

@@ -1,17 +1,30 @@
 package metrics
 
 import (
+	"context"
 	"fmt"
 	"testing"
+	"time"
 
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	types "k8s.io/apimachinery/pkg/types"
 )
 
+var fakek8s = fake.NewFakeClient()
+
 func TestCache(t *testing.T) {
-	m := New(logrus.NewEntry(logrus.New()), prometheus.NewRegistry())
+	m := New(logrus.NewEntry(logrus.New()), prometheus.NewRegistry(), fakek8s)
 
 	for i, typ := range []string{"init", "container"} {
 		version := fmt.Sprintf("0.1.%d", i)
@@ -20,7 +33,9 @@ func TestCache(t *testing.T) {
 
 	for i, typ := range []string{"init", "container"} {
 		version := fmt.Sprintf("0.1.%d", i)
-		mt, _ := m.containerImageVersion.GetMetricWith(m.buildLabels("namespace", "pod", "container", typ, "url", version, version))
+		mt, _ := m.containerImageVersion.GetMetricWith(
+			m.buildLabels("namespace", "pod", "container", typ, "url", version, version),
+		)
 		count := testutil.ToFloat64(mt)
 		assert.Equal(t, count, float64(1), "Expected to get a metric for containerImageVersion")
 	}
@@ -30,15 +45,17 @@ func TestCache(t *testing.T) {
 	}
 	for i, typ := range []string{"init", "container"} {
 		version := fmt.Sprintf("0.1.%d", i)
-		mt, _ := m.containerImageVersion.GetMetricWith(m.buildLabels("namespace", "pod", "container", typ, "url", version, version))
+		mt, _ := m.containerImageVersion.GetMetricWith(
+			m.buildLabels("namespace", "pod", "container", typ, "url", version, version),
+		)
 		count := testutil.ToFloat64(mt)
-		assert.Equal(t, count, float64(0), "Expected to get a metric for containerImageVersion")
+		assert.Equal(t, count, float64(0), "Expected NOT to get a metric for containerImageVersion")
 	}
 }
 
 // TestErrorsReporting verifies that the error metric increments correctly
 func TestErrorsReporting(t *testing.T) {
-	m := New(logrus.NewEntry(logrus.New()), prometheus.NewRegistry())
+	m := New(logrus.NewEntry(logrus.New()), prometheus.NewRegistry(), fakek8s)
 
 	// Reset the metrics before testing
 	m.containerImageErrors.Reset()
@@ -57,6 +74,16 @@ func TestErrorsReporting(t *testing.T) {
 
 	for i, tc := range testCases {
 		t.Run(fmt.Sprintf("Case %d", i+1), func(t *testing.T) {
+			err := fakek8s.DeleteAllOf(context.Background(), &corev1.Pod{})
+			require.NoError(t, err)
+
+			// We need to ensure that the pod Exists!
+			err = fakek8s.Create(context.Background(), &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{Name: tc.pod, Namespace: tc.namespace},
+				Spec:       corev1.PodSpec{Containers: []corev1.Container{{Name: tc.container, Image: tc.image}}},
+			})
+			require.NoError(t, err)
+
 			// Report an error
 			m.ReportError(tc.namespace, tc.pod, tc.container, tc.image)
 
@@ -75,3 +102,66 @@ func TestErrorsReporting(t *testing.T) {
 		})
 	}
 }
+
+func Test_Metrics_SkipOnDeletedPod(t *testing.T) {
+	scheme := runtime.NewScheme()
+	_ = corev1.AddToScheme(scheme)
+
+	// Step 1: Create fake client with Pod
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "mypod",
+			Namespace: "default",
+			UID:       types.UID("test-uid"),
+		},
+	}
+	client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(pod).Build()
+
+	// Step 2: Create Metrics with fake registry
+	reg := prometheus.NewRegistry()
+	log := logrus.NewEntry(logrus.New())
+	metrics := New(log, reg, client)
+
+	// verify Pod exists!
+	require.True(t,
+		metrics.PodExists(context.Background(), "default", "mypod"),
+		"Pod should exist at this point!",
+	)
+
+	// Register some metrics....
+	metrics.RegisterImageDuration("default", "mypod", "mycontainer", "nginx:latest", time.Now())
+
+	// Step 3: Simulate a Delete occuring, Whilst still Reconciling...
+	_ = client.Delete(context.Background(), pod)
+	metrics.RemovePod("default", "mypod")
+
+	// Step 4: Validate that all metrics have been removed...
+	metricFamilies, err := reg.Gather()
+	assert.NoError(t, err)
+	for _, mf := range metricFamilies {
+		assert.NotContains(t, *mf.Name, "is_latest_version", "Should not have been found: %+v", mf)
+		assert.NotContains(t, *mf.Name, "image_lookup_duration", "Should not have been found: %+v", mf)
+		assert.NotContains(t, *mf.Name, "image_failures_total", "Should not have been found: %+v", mf)
+	}
+
+	// Register Error _after_ sync has completed!
+	metrics.ReportError("default", "mypod", "mycontianer", "nginx:latest")
+
+	// Step 5: Attempt to register metrics (should not register anything)
+	require.False(t,
+		metrics.PodExists(context.Background(), "default", "mypod"),
+		"Pod should NOT exist at this point!",
+	)
+
+	metrics.RegisterImageDuration("default", "mypod", "mycontainer", "nginx:latest", time.Now())
+	metrics.ReportError("default", "mypod", "mycontianer", "nginx:latest")
+
+	// Step 6: Gather metrics and assert none were registered
+	metricFamilies, err = reg.Gather()
+	assert.NoError(t, err)
+	for _, mf := range metricFamilies {
+		assert.NotContains(t, *mf.Name, "is_latest_version", "Should not have been found: %+v", mf)
+		assert.NotContains(t, *mf.Name, "image_lookup_duration", "Should not have been found: %+v", mf)
+		assert.NotContains(t, *mf.Name, "image_failures_total", "Should not have been found: %+v", mf)
+	}
+}

pkg/metrics/roundtripper_test.go

@@ -164,7 +164,7 @@ func TestRoundTripper(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			metricsServer := New(log, prometheus.NewRegistry())
+			metricsServer := New(log, prometheus.NewRegistry(), fakek8s)
 			server := httptest.NewServer(tt.handler)
 			defer server.Close()