Repository: spegel-org/spegel
Pull request #880 (2025-05-14 14:33:04)

Refactor store advertisement to list content
The current implementation of state tracking will list images and then walk the image digests to get all of the digests linked to it. This has worked well for a long time. The solution is overly complicated for the end result that we want to achieve. The walking of all the manifests can be resource consuming and can be error prone as new OCI artifact types emerge.

This new solution instead lists all of the image tags, and then all of the contents in the OCI store, which are then advertised. The solution is a lot more simpler, and does not explicitly require content to be linked to an image.

This work is also required in preparation for the work related to capturing content create events.

Part of #502 


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index c2a3645..06bbf29 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - [#871](https://github.com/spegel-org/spegel/pull/871) Implement OCI client and refactor debug web pulling.
 - [#873](https://github.com/spegel-org/spegel/pull/873) Refactor web to use internal mux router.
 - [#875](https://github.com/spegel-org/spegel/pull/875) Change debug unit formatting and add totals.
+- [#880](https://github.com/spegel-org/spegel/pull/880) Refactor store advertisement to list content.
 
 ### Deprecated
 
diff --git a/pkg/oci/containerd.go b/pkg/oci/containerd.go
index f0b54b0..710b7e0 100644
--- a/pkg/oci/containerd.go
+++ b/pkg/oci/containerd.go
@@ -18,6 +18,7 @@ import (
 	eventtypes "github.com/containerd/containerd/api/events"
 	"github.com/containerd/containerd/v2/client"
 	"github.com/containerd/containerd/v2/core/content"
+	"github.com/containerd/containerd/v2/pkg/labels"
 	"github.com/containerd/errdefs"
 	"github.com/containerd/typeurl/v2"
 	"github.com/go-logr/logr"
@@ -40,8 +41,9 @@ type Containerd struct {
 	contentPath        string
 	client             *client.Client
 	clientGetter       func() (*client.Client, error)
-	listFilter         string
+	imageFilter        string
 	eventFilter        string
+	contentFilter      string
 	registryConfigPath string
 }
 
@@ -54,13 +56,14 @@ func WithContentPath(path string) Option {
 }
 
 func NewContainerd(sock, namespace, registryConfigPath string, mirroredRegistries []url.URL, opts ...Option) (*Containerd, error) {
-	listFilter, eventFilter := createFilters(mirroredRegistries)
+	imageFilter, eventFilter, contentFilter := createFilters(mirroredRegistries)
 	c := &Containerd{
 		clientGetter: func() (*client.Client, error) {
 			return client.New(sock, client.WithDefaultNamespace(namespace))
 		},
-		listFilter:         listFilter,
+		imageFilter:        imageFilter,
 		eventFilter:        eventFilter,
+		contentFilter:      contentFilter,
 		registryConfigPath: registryConfigPath,
 	}
 	for _, opt := range opts {
@@ -224,7 +227,7 @@ func (c *Containerd) ListImages(ctx context.Context) ([]Image, error) {
 	if err != nil {
 		return nil, err
 	}
-	cImgs, err := client.ListImages(ctx, c.listFilter)
+	cImgs, err := client.ListImages(ctx, c.imageFilter)
 	if err != nil {
 		return nil, err
 	}
@@ -251,6 +254,27 @@ func (c *Containerd) Resolve(ctx context.Context, ref string) (digest.Digest, er
 	return cImg.Target().Digest, nil
 }
 
+func (c *Containerd) ListContents(ctx context.Context) ([]Content, error) {
+	client, err := c.Client()
+	if err != nil {
+		return nil, err
+	}
+	contents := []Content{}
+	err = client.ContentStore().Walk(ctx, func(i content.Info) error {
+		registries := parseContentRegistries(i.Labels)
+		content := Content{
+			Digest:     i.Digest,
+			Registires: registries,
+		}
+		contents = append(contents, content)
+		return nil
+	}, c.contentFilter)
+	if err != nil {
+		return nil, err
+	}
+	return contents, nil
+}
+
 func (c *Containerd) Size(ctx context.Context, dgst digest.Digest) (int64, error) {
 	client, err := c.Client()
 	if err != nil {
@@ -317,6 +341,17 @@ func (c *Containerd) GetBlob(ctx context.Context, dgst digest.Digest) (io.ReadSe
 	}, nil
 }
 
+func parseContentRegistries(l map[string]string) []string {
+	registries := []string{}
+	for k := range l {
+		if !strings.HasPrefix(k, labels.LabelDistributionSource) {
+			continue
+		}
+		registries = append(registries, strings.TrimPrefix(k, labels.LabelDistributionSource+"."))
+	}
+	return registries
+}
+
 func getEventImage(e typeurl.Any) (string, EventType, error) {
 	if e == nil {
 		return "", "", errors.New("any cannot be nil")
@@ -337,19 +372,23 @@ func getEventImage(e typeurl.Any) (string, EventType, error) {
 	}
 }
 
-func createFilters(mirroredRegistries []url.URL) (string, string) {
+func createFilters(mirroredRegistries []url.URL) (string, string, string) {
 	registryHosts := []string{}
 	for _, registry := range mirroredRegistries {
 		registryHosts = append(registryHosts, strings.ReplaceAll(registry.Host, `.`, `\\.`))
 	}
-	listFilter := fmt.Sprintf(`name~="^(%s)/"`, strings.Join(registryHosts, "|"))
+	imageFilter := fmt.Sprintf(`name~="^(%s)/"`, strings.Join(registryHosts, "|"))
 	if len(registryHosts) == 0 {
 		// Filter images that do not have a registry in it's reference,
 		// as we cant mirror images without registries.
-		listFilter = `name~="^.+/"`
+		imageFilter = `name~="^.+/"`
+	}
+	eventFilter := fmt.Sprintf(`topic~="/images/create|/images/update|/images/delete",event.%s`, imageFilter)
+	contentFilters := []string{}
+	for _, registry := range mirroredRegistries {
+		contentFilters = append(contentFilters, fmt.Sprintf(`labels."%s.%s"~="^."`, labels.LabelDistributionSource, registry.Host))
 	}
-	eventFilter := fmt.Sprintf(`topic~="/images/create|/images/update|/images/delete",event.%s`, listFilter)
-	return listFilter, eventFilter
+	return imageFilter, eventFilter, strings.Join(contentFilters, " ")
 }
 
 // Refer to containerd registry configuration documentation for more information about required configuration.
diff --git a/pkg/oci/memory.go b/pkg/oci/memory.go
index 8d69f14..7b7f9fb 100644
--- a/pkg/oci/memory.go
+++ b/pkg/oci/memory.go
@@ -58,6 +58,17 @@ func (m *Memory) Resolve(ctx context.Context, ref string) (digest.Digest, error)
 	return dgst, nil
 }
 
+func (m *Memory) ListContents(ctx context.Context) ([]Content, error) {
+	m.mx.RLock()
+	defer m.mx.RUnlock()
+
+	contents := []Content{}
+	for k := range m.blobs {
+		contents = append(contents, Content{Digest: k})
+	}
+	return contents, nil
+}
+
 func (m *Memory) Size(ctx context.Context, dgst digest.Digest) (int64, error) {
 	m.mx.RLock()
 	defer m.mx.RUnlock()
diff --git a/pkg/oci/oci.go b/pkg/oci/oci.go
index 8ec1329..52b3677 100644
--- a/pkg/oci/oci.go
+++ b/pkg/oci/oci.go
@@ -30,6 +30,11 @@ type ImageEvent struct {
 	Type  EventType
 }
 
+type Content struct {
+	Digest     digest.Digest
+	Registires []string
+}
+
 type Store interface {
 	// Name returns the name of the store implementation.
 	Name() string
@@ -47,6 +52,9 @@ type Store interface {
 	// The ref is expected to be in the format `registry/name:tag`.
 	Resolve(ctx context.Context, ref string) (digest.Digest, error)
 
+	// ListContents returns a list of all the contents.
+	ListContents(ctx context.Context) ([]Content, error)
+
 	// Size returns the content byte size for the given digest.
 	// Will return ErrNotFound if the digest cannot be found.
 	Size(ctx context.Context, dgst digest.Digest) (int64, error)
diff --git a/pkg/state/state.go b/pkg/state/state.go
index 73232ea..5b596f5 100644
--- a/pkg/state/state.go
+++ b/pkg/state/state.go
@@ -31,8 +31,9 @@ func Track(ctx context.Context, ociStore oci.Store, router routing.Router, resol
 		case <-ctx.Done():
 			return nil
 		case <-tickerCh:
-			log.Info("running scheduled image state update")
-			if err := all(ctx, ociStore, router, resolveLatestTag); err != nil {
+			log.Info("running tick state update")
+			err := tick(ctx, ociStore, router, resolveLatestTag)
+			if err != nil {
 				log.Error(err, "received errors when updating all images")
 				continue
 			}
@@ -54,41 +55,61 @@ func Track(ctx context.Context, ociStore oci.Store, router routing.Router, resol
 	}
 }
 
-func all(ctx context.Context, ociStore oci.Store, router routing.Router, resolveLatestTag bool) error {
-	log := logr.FromContextOrDiscard(ctx).V(4)
+func tick(ctx context.Context, ociStore oci.Store, router routing.Router, resolveLatest bool) error {
+	advertisedImages := map[string]float64{}
+	advertisedImageDigests := map[string]float64{}
+	advertisedImageTags := map[string]float64{}
+	advertisedKeys := map[string]float64{}
+
 	imgs, err := ociStore.ListImages(ctx)
 	if err != nil {
 		return err
 	}
-
-	// TODO: Update metrics on subscribed events. This will require keeping state in memory to know about key count changes.
-	metrics.AdvertisedKeys.Reset()
-	metrics.AdvertisedImages.Reset()
-	metrics.AdvertisedImageTags.Reset()
-	metrics.AdvertisedImageDigests.Reset()
-	errs := []error{}
-	targets := map[string]any{}
 	for _, img := range imgs {
-		_, skipDigests := targets[img.Digest.String()]
-		// Handle the list re-sync as update events; this will also prevent the
-		// update function from setting metrics values.
-		event := oci.ImageEvent{Image: img, Type: oci.UpdateEvent}
-		log.Info("sync image event", "image", event.Image.String(), "type", event.Type)
-		keyTotal, err := update(ctx, ociStore, router, event, skipDigests, resolveLatestTag)
-		if err != nil {
-			errs = append(errs, err)
+		advertisedImages[img.Registry] += 1
+		advertisedImageDigests[img.Registry] += 1
+		if !resolveLatest && img.IsLatestTag() {
 			continue
 		}
-		targets[img.Digest.String()] = nil
-		metrics.AdvertisedKeys.WithLabelValues(img.Registry).Add(float64(keyTotal))
-		metrics.AdvertisedImages.WithLabelValues(img.Registry).Add(1)
-		if img.Tag == "" {
-			metrics.AdvertisedImageDigests.WithLabelValues(event.Image.Registry).Add(1)
-		} else {
-			metrics.AdvertisedImageTags.WithLabelValues(event.Image.Registry).Add(1)
+		tagName, ok := img.TagName()
+		if !ok {
+			continue
+		}
+		err := router.Advertise(ctx, []string{tagName})
+		if err != nil {
+			return err
+		}
+		advertisedImageTags[img.Registry] += 1
+		advertisedKeys[img.Registry] += 1
+	}
+
+	contents, err := ociStore.ListContents(ctx)
+	if err != nil {
+		return err
+	}
+	for _, content := range contents {
+		err := router.Advertise(ctx, []string{content.Digest.String()})
+		if err != nil {
+			return err
+		}
+		for _, registry := range content.Registires {
+			advertisedKeys[registry] += 1
 		}
 	}
-	return errors.Join(errs...)
+
+	for k, v := range advertisedImages {
+		metrics.AdvertisedImages.WithLabelValues(k).Set(v)
+	}
+	for k, v := range advertisedImageDigests {
+		metrics.AdvertisedImageDigests.WithLabelValues(k).Set(v)
+	}
+	for k, v := range advertisedImageTags {
+		metrics.AdvertisedImageTags.WithLabelValues(k).Set(v)
+	}
+	for k, v := range advertisedKeys {
+		metrics.AdvertisedKeys.WithLabelValues(k).Set(v)
+	}
+	return nil
 }
 
 func update(ctx context.Context, ociStore oci.Store, router routing.Router, event oci.ImageEvent, skipDigests, resolveLatestTag bool) (int, error) {


Test patch:
diff --git a/pkg/oci/containerd_test.go b/pkg/oci/containerd_test.go
index 5e3b79c..df77c4c 100644
--- a/pkg/oci/containerd_test.go
+++ b/pkg/oci/containerd_test.go
@@ -206,35 +206,79 @@ func TestBackupConfig(t *testing.T) {
 	require.Len(t, files, 1)
 }
 
+func TestParseContentRegistries(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name     string
+		labels   map[string]string
+		expected []string
+	}{
+		{
+			name:     "no labels",
+			labels:   map[string]string{},
+			expected: []string{},
+		},
+		{
+			name: "one matching",
+			labels: map[string]string{
+				"containerd.io/distribution.source.docker.io": "library/alpine",
+			},
+			expected: []string{"docker.io"},
+		},
+		{
+			name: "multiple matching",
+			labels: map[string]string{
+				"containerd.io/distribution.source.example.com": "foo",
+				"containerd.io/distribution.source.ghcr.io":     "spegel-org/spegel",
+			},
+			expected: []string{"ghcr.io", "example.com"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(t.Name(), func(t *testing.T) {
+			t.Parallel()
+
+			registries := parseContentRegistries(tt.labels)
+			require.ElementsMatch(t, tt.expected, registries)
+		})
+	}
+}
+
 func TestCreateFilter(t *testing.T) {
 	t.Parallel()
 
 	tests := []struct {
-		name                string
-		expectedListFilter  string
-		expectedEventFilter string
-		registries          []string
+		name                  string
+		expectedImageFilter   string
+		expectedEventFilter   string
+		expectedContentFilter string
+		registries            []string
 	}{
 		{
-			name:                "with registry filtering",
-			registries:          []string{"https://docker.io", "https://gcr.io"},
-			expectedListFilter:  `name~="^(docker\\.io|gcr\\.io)/"`,
-			expectedEventFilter: `topic~="/images/create|/images/update|/images/delete",event.name~="^(docker\\.io|gcr\\.io)/"`,
+			name:                  "with registry filtering",
+			registries:            []string{"https://docker.io", "https://gcr.io"},
+			expectedImageFilter:   `name~="^(docker\\.io|gcr\\.io)/"`,
+			expectedEventFilter:   `topic~="/images/create|/images/update|/images/delete",event.name~="^(docker\\.io|gcr\\.io)/"`,
+			expectedContentFilter: `labels."containerd.io/distribution.source.docker.io"~="^." labels."containerd.io/distribution.source.gcr.io"~="^."`,
 		},
 		{
-			name:                "without registry filtering",
-			registries:          []string{},
-			expectedListFilter:  `name~="^.+/"`,
-			expectedEventFilter: `topic~="/images/create|/images/update|/images/delete",event.name~="^.+/"`,
+			name:                  "without registry filtering",
+			registries:            []string{},
+			expectedImageFilter:   `name~="^.+/"`,
+			expectedEventFilter:   `topic~="/images/create|/images/update|/images/delete",event.name~="^.+/"`,
+			expectedContentFilter: "",
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			t.Parallel()
 
-			listFilter, eventFilter := createFilters(stringListToUrlList(t, tt.registries))
-			require.Equal(t, tt.expectedListFilter, listFilter)
+			imageFilter, eventFilter, contentFilter := createFilters(stringListToUrlList(t, tt.registries))
+			require.Equal(t, tt.expectedImageFilter, imageFilter)
 			require.Equal(t, tt.expectedEventFilter, eventFilter)
+			require.Equal(t, tt.expectedContentFilter, contentFilter)
 		})
 	}
 }