Fix advertise metrics on events (#1144)

Author: phillebaba

pkg/metrics/metrics.go

@@ -29,10 +29,6 @@ var (
 		Name: "spegel_resolve_duration_seconds",
 		Help: "The duration for router to resolve a peer.",
 	}, []string{"router"})
-	AdvertisedImages = prometheus.NewGaugeVec(prometheus.GaugeOpts{
-		Name: "spegel_advertised_images",
-		Help: "Number of images advertised to be available.",
-	}, []string{"registry"})
 	AdvertisedImageTags = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 		Name: "spegel_advertised_image_tags",
 		Help: "Number of image tags advertised to be available.",
@@ -41,19 +37,18 @@ var (
 		Name: "spegel_advertised_image_digests",
 		Help: "Number of image digests advertised to be available.",
 	}, []string{"registry"})
-	AdvertisedKeys = prometheus.NewGaugeVec(prometheus.GaugeOpts{
-		Name: "spegel_advertised_keys",
-		Help: "Number of keys advertised to be available.",
+	AdvertisedContentDigests = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "spegel_advertised_content_digests",
+		Help: "Number of content digests advertised to be available.",
 	}, []string{"registry"})
 )
 
 func Register() {
 	DefaultRegisterer.MustRegister(MirrorRequestsTotal)
 	DefaultRegisterer.MustRegister(MirrorLastSuccessTimestamp)
 	DefaultRegisterer.MustRegister(ResolveDurHistogram)
-	DefaultRegisterer.MustRegister(AdvertisedImages)
 	DefaultRegisterer.MustRegister(AdvertisedImageTags)
 	DefaultRegisterer.MustRegister(AdvertisedImageDigests)
-	DefaultRegisterer.MustRegister(AdvertisedKeys)
+	DefaultRegisterer.MustRegister(AdvertisedContentDigests)
 	httpx.RegisterMetrics(DefaultRegisterer)
 }

pkg/oci/containerd.go

@@ -455,7 +455,7 @@ func (c *Containerd) handleEvent(ctx context.Context, envelope events.Envelope,
 			return fmt.Errorf("manifest with digest %s still exists", img.Digest.String())
 		}, retryOpts...)
 		if err != nil {
-			logr.FromContextOrDiscard(ctx).Error(err, "image manifest has not been deleted")
+			return nil, fmt.Errorf("image manifest has not been deleted: %w", err)
 		}
 		// Create delete events for contents that has been removed.
 		events := []OCIEvent{}

pkg/state/state.go

@@ -53,11 +53,8 @@ func Track(ctx context.Context, ociStore oci.Store, router routing.Router, opts
 		if ok {
 			keys = append(keys, tagName)
 			metrics.AdvertisedImageTags.WithLabelValues(img.Registry).Inc()
-			metrics.AdvertisedKeys.WithLabelValues(img.Registry).Inc()
 		}
-		metrics.AdvertisedImages.WithLabelValues(img.Registry).Inc()
 		metrics.AdvertisedImageDigests.WithLabelValues(img.Registry).Inc()
-		metrics.AdvertisedKeys.WithLabelValues(img.Registry).Inc()
 	}
 	contents, err := ociStore.ListContent(ctx)
 	if err != nil {
@@ -69,7 +66,7 @@ func Track(ctx context.Context, ociStore oci.Store, router routing.Router, opts
 			continue
 		}
 		for _, ref := range refs {
-			metrics.AdvertisedKeys.WithLabelValues(ref.Registry).Inc()
+			metrics.AdvertisedContentDigests.WithLabelValues(ref.Registry).Inc()
 		}
 		keys = append(keys, refs[0].Digest.String())
 	}
@@ -79,8 +76,7 @@ func Track(ctx context.Context, ociStore oci.Store, router routing.Router, opts
 	}
 
 	// Watch for OCI events.
-	log := logr.FromContextOrDiscard(ctx)
-	log.Info("waiting for store events")
+	logr.FromContextOrDiscard(ctx).Info("waiting for store events")
 	for {
 		select {
 		case <-ctx.Done():
@@ -89,25 +85,38 @@ func Track(ctx context.Context, ociStore oci.Store, router routing.Router, opts
 			if !ok {
 				return errors.New("event channel closed")
 			}
-			log.Info("OCI event", "ref", event.Reference.String(), "type", event.Type)
-			err := handleEvent(ctx, router, event)
+			err := handleEvent(ctx, router, event, cfg.Filters)
 			if err != nil {
-				log.Error(err, "could not handle event")
+				logr.FromContextOrDiscard(ctx).Error(err, "could not handle event")
 				continue
 			}
 		}
 	}
 }
 
-func handleEvent(ctx context.Context, router routing.Router, event oci.OCIEvent) error {
+func handleEvent(ctx context.Context, router routing.Router, event oci.OCIEvent, filters []oci.Filter) error {
+	if oci.MatchesFilter(event.Reference, filters) {
+		return nil
+	}
+	logr.FromContextOrDiscard(ctx).Info("OCI event", "ref", event.Reference.String(), "type", event.Type)
 	switch event.Type {
 	case oci.CreateEvent:
+		if event.Reference.Tag != "" {
+			metrics.AdvertisedImageTags.WithLabelValues(event.Reference.Registry).Inc()
+		} else {
+			metrics.AdvertisedContentDigests.WithLabelValues(event.Reference.Registry).Inc()
+		}
 		err := router.Advertise(ctx, []string{event.Reference.Identifier()})
 		if err != nil {
 			return err
 		}
 		return nil
 	case oci.DeleteEvent:
+		if event.Reference.Tag != "" {
+			metrics.AdvertisedImageTags.WithLabelValues(event.Reference.Registry).Dec()
+		} else {
+			metrics.AdvertisedContentDigests.WithLabelValues(event.Reference.Registry).Dec()
+		}
 		err := router.Withdraw(ctx, []string{event.Reference.Identifier()})
 		if err != nil {
 			return err

test/integration/containerd/containerd_test.go

@@ -157,7 +157,6 @@ func TestContainerdPull(t *testing.T) {
 				}
 
 				t.Log("Pulling image with CRI", benchmarkImg.String())
-				fmt.Println(benchmarkImg.String())
 				_, err = imageClient.PullImage(t.Context(), &runtimeapi.PullImageRequest{Image: &runtimeapi.ImageSpec{Image: benchmarkImg.String()}})
 				require.NoError(t, err)
 				ensureEvents(t, eventCh, expectedCreateEvents)