Repository: spegel-org/spegel (go)

Issue:
Subscribe to Containerd content instead of images
### Describe the problem to be solved

Currently an images is not advertised until the whole image has been completely pulled and committed by to the snapshotter. This means that large images will delay the advertising of other layers. We should instead subscribe to content changes and advertise the content as fast as it is fetched to disk. This will hopefully solve latency issues when multiple nodes are pulling at the same time.

### Proposed solution to the problem

_No response_

Pull request description:
Add support for content create events
This change adds support for Containerd content events that was introduced in Containerd v2.1.0. Advertising on content create rather than image create will make layers available as soon as they are present on disk.

Fixes #502 

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 43ec7b7..eefa030 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - [#877](https://github.com/spegel-org/spegel/pull/877) Add support for www authenticate header.
 - [#878](https://github.com/spegel-org/spegel/pull/878) Add dial timeout configuration in Containerd mirror configuration.
+- [#889](https://github.com/spegel-org/spegel/pull/889) Add support for content create events.
 
 ### Changed
 
diff --git a/pkg/oci/containerd.go b/pkg/oci/containerd.go
index f7699c2..720b81b 100644
--- a/pkg/oci/containerd.go
+++ b/pkg/oci/containerd.go
@@ -19,6 +19,7 @@ import (
 	"github.com/containerd/containerd/v2/client"
 	"github.com/containerd/containerd/v2/core/content"
 	"github.com/containerd/containerd/v2/core/events"
+	"github.com/containerd/containerd/v2/pkg/filters"
 	"github.com/containerd/containerd/v2/pkg/labels"
 	"github.com/containerd/errdefs"
 	"github.com/containerd/typeurl/v2"
@@ -36,16 +37,43 @@ const (
 	backupDir = "_backup"
 )
 
+type Feature uint32
+
+const (
+	FeatureConfigCheck Feature = 1 << iota
+	FeatureContentEvent
+)
+
+func (f *Feature) Set(feat Feature) {
+	*f |= feat
+}
+
+func (f Feature) Has(feat Feature) bool {
+	return f&feat != 0
+}
+
+func (f Feature) String() string {
+	feats := []string{}
+	if f.Has(FeatureConfigCheck) {
+		feats = append(feats, "ConfigCheck")
+	}
+	if f.Has(FeatureContentEvent) {
+		feats = append(feats, "ContentEvent")
+	}
+	return strings.Join(feats, "|")
+}
+
 var _ Store = &Containerd{}
 
 type Containerd struct {
 	contentPath        string
+	registryConfigPath string
 	client             *client.Client
 	clientGetter       func() (*client.Client, error)
-	imageFilter        string
-	eventFilter        string
-	contentFilter      string
-	registryConfigPath string
+	features           *Feature
+	imageFilter        []string
+	eventFilter        []string
+	contentFilter      []string
 }
 
 type Option func(*Containerd)
@@ -99,41 +127,71 @@ func (c *Containerd) Verify(ctx context.Context) error {
 		return errors.New("could not reach Containerd service")
 	}
 
+	feats, err := c.Features(ctx)
+	if err != nil {
+		return err
+	}
+	if !feats.Has(FeatureConfigCheck) {
+		log.Info("skipping verification of Containerd configuration")
+		return nil
+	}
 	grpcConn, ok := client.Conn().(*grpc.ClientConn)
 	if !ok {
 		return errors.New("client connection is not grpc")
 	}
 	srv := runtimeapi.NewRuntimeServiceClient(grpcConn)
-	versionResp, err := srv.Version(ctx, &runtimeapi.VersionRequest{})
+	statusResp, err := srv.Status(ctx, &runtimeapi.StatusRequest{Verbose: true})
 	if err != nil {
 		return err
 	}
-	ok, err = canVerifyContainerdConfiguration(versionResp.RuntimeVersion)
+	err = verifyStatusResponse(statusResp, c.registryConfigPath)
 	if err != nil {
-		return fmt.Errorf("could not check Containerd version %s: %w", versionResp.RuntimeVersion, err)
+		return err
 	}
-	if !ok {
-		log.Info("skipping verification of Containerd configuration", "version", versionResp.RuntimeVersion)
-		return nil
+	return nil
+}
+
+func (c *Containerd) Features(ctx context.Context) (Feature, error) {
+	if c.features != nil {
+		return *c.features, nil
 	}
 
-	statusResp, err := srv.Status(ctx, &runtimeapi.StatusRequest{Verbose: true})
+	log := logr.FromContextOrDiscard(ctx)
+	client, err := c.Client()
 	if err != nil {
-		return err
+		return 0, err
 	}
-	err = verifyStatusResponse(statusResp, c.registryConfigPath)
+	grpcConn, ok := client.Conn().(*grpc.ClientConn)
+	if !ok {
+		return 0, errors.New("client connection is not grpc")
+	}
+	srv := runtimeapi.NewRuntimeServiceClient(grpcConn)
+	versionResp, err := srv.Version(ctx, &runtimeapi.VersionRequest{})
 	if err != nil {
-		return err
+		return 0, err
 	}
-	return nil
+	feats, err := featuresForVersion(versionResp.RuntimeVersion)
+	if err != nil {
+		return 0, err
+	}
+	log.Info("setting features for Containerd version", "features", feats.String(), "version", versionResp.String())
+	c.features = &feats
+	return feats, nil
 }
 
-func canVerifyContainerdConfiguration(version string) (bool, error) {
+func featuresForVersion(version string) (Feature, error) {
 	v, err := utilversion.Parse(version)
 	if err != nil {
-		return false, err
+		return 0, fmt.Errorf("could not parse version %s: %w", version, err)
+	}
+	feats := Feature(0)
+	if v.LessThan(utilversion.MustParse("2.0")) {
+		feats.Set(FeatureConfigCheck)
+	}
+	if v.AtLeast(utilversion.MustParse("2.1")) {
+		feats.Set(FeatureContentEvent)
 	}
-	return v.LessThan(utilversion.MustParse("2.0")), nil
+	return feats, nil
 }
 
 func verifyStatusResponse(resp *runtimeapi.StatusResponse, configPath string) error {
@@ -184,7 +242,7 @@ func (c *Containerd) Subscribe(ctx context.Context) (<-chan OCIEvent, error) {
 
 	ctx, cancel := context.WithCancel(ctx)
 	eventCh := make(chan OCIEvent)
-	envelopeCh, cErrCh := client.EventService().Subscribe(ctx, c.eventFilter)
+	envelopeCh, cErrCh := client.EventService().Subscribe(ctx, c.eventFilter...)
 	go func() {
 		defer close(eventCh)
 		for {
@@ -218,7 +276,7 @@ func (c *Containerd) ListImages(ctx context.Context) ([]Image, error) {
 	if err != nil {
 		return nil, err
 	}
-	cImgs, err := client.ListImages(ctx, c.imageFilter)
+	cImgs, err := client.ListImages(ctx, c.imageFilter...)
 	if err != nil {
 		return nil, err
 	}
@@ -259,7 +317,7 @@ func (c *Containerd) ListContents(ctx context.Context) ([]Content, error) {
 		}
 		contents = append(contents, content)
 		return nil
-	}, c.contentFilter)
+	}, c.contentFilter...)
 	if err != nil {
 		return nil, err
 	}
@@ -333,6 +391,11 @@ func (c *Containerd) GetBlob(ctx context.Context, dgst digest.Digest) (io.ReadSe
 }
 
 func (c *Containerd) convertEvent(ctx context.Context, envelope events.Envelope) ([]OCIEvent, error) {
+	client, err := c.Client()
+	if err != nil {
+		return nil, err
+	}
+
 	if envelope.Event == nil {
 		return nil, errors.New("envelope event cannot be nil")
 	}
@@ -341,11 +404,24 @@ func (c *Containerd) convertEvent(ctx context.Context, envelope events.Envelope)
 		return nil, fmt.Errorf("failed to unmarshal envelope event: %w", err)
 	}
 	switch e := evt.(type) {
-	case *eventtypes.ImageCreate:
-		// Containerd creates an image create event for image config which we ignore.
-		if err := digest.Digest(e.GetName()).Validate(); err == nil {
+	case *eventtypes.ContentCreate:
+		filter, err := filters.ParseAll(c.contentFilter...)
+		if err != nil {
+			return nil, err
+		}
+		dgst, err := digest.Parse(e.GetDigest())
+		if err != nil {
+			return nil, err
+		}
+		info, err := client.ContentStore().Info(ctx, dgst)
+		if err != nil {
+			return nil, err
+		}
+		if filter.Match(content.AdaptInfo(info)) {
 			return nil, nil
 		}
+		return []OCIEvent{{Type: CreateEvent, Key: e.GetDigest()}}, nil
+	case *eventtypes.ImageCreate:
 		img, err := ParseImage(e.GetName())
 		if err != nil {
 			return nil, err
@@ -354,7 +430,14 @@ func (c *Containerd) convertEvent(ctx context.Context, envelope events.Envelope)
 		if img.Digest == "" {
 			return []OCIEvent{{Type: CreateEvent, Key: e.GetName()}}, nil
 		}
-		// Walk the image to return events for all of the layers.
+		// If Containerd supports content events we can skip walking the image.
+		feats, err := c.Features(ctx)
+		if err != nil {
+			return nil, err
+		}
+		if feats.Has(FeatureContentEvent) {
+			return nil, nil
+		}
 		dgsts, err := WalkImage(ctx, c, img)
 		if err != nil {
 			return nil, fmt.Errorf("could not get digests for image %s: %w", img.String(), err)
@@ -365,10 +448,6 @@ func (c *Containerd) convertEvent(ctx context.Context, envelope events.Envelope)
 		}
 		return events, nil
 	case *eventtypes.ImageDelete:
-		// Ignore delete event created for image config.
-		if err := digest.Digest(e.GetName()).Validate(); err == nil {
-			return nil, nil
-		}
 		return []OCIEvent{{Type: DeleteEvent, Key: e.GetName()}}, nil
 	default:
 		return nil, errors.New("unsupported event type")
@@ -386,7 +465,7 @@ func parseContentRegistries(l map[string]string) []string {
 	return registries
 }
 
-func createFilters(mirroredRegistries []url.URL) (string, string, string) {
+func createFilters(mirroredRegistries []url.URL) ([]string, []string, []string) {
 	registryHosts := []string{}
 	for _, registry := range mirroredRegistries {
 		registryHosts = append(registryHosts, strings.ReplaceAll(registry.Host, `.`, `\\.`))
@@ -402,7 +481,7 @@ func createFilters(mirroredRegistries []url.URL) (string, string, string) {
 	for _, registry := range mirroredRegistries {
 		contentFilters = append(contentFilters, fmt.Sprintf(`labels."%s.%s"~="^."`, labels.LabelDistributionSource, registry.Host))
 	}
-	return imageFilter, eventFilter, strings.Join(contentFilters, " ")
+	return []string{imageFilter}, []string{eventFilter, `topic~="/content/create"`}, contentFilters
 }
 
 // Refer to containerd registry configuration documentation for more information about required configuration.


Test patch:
diff --git a/pkg/oci/containerd_test.go b/pkg/oci/containerd_test.go
index 7b79971..eda7122 100644
--- a/pkg/oci/containerd_test.go
+++ b/pkg/oci/containerd_test.go
@@ -9,6 +9,7 @@ import (
 	"path/filepath"
 	"testing"
 
+	"github.com/containerd/containerd/v2/pkg/filters"
 	"github.com/go-logr/logr"
 	"github.com/stretchr/testify/require"
 	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
@@ -28,47 +29,6 @@ func TestNewContainerd(t *testing.T) {
 	require.Equal(t, "local", c.contentPath)
 }
 
-func TestCanVerifyContainerdConfiguration(t *testing.T) {
-	t.Parallel()
-
-	tests := []struct {
-		version  string
-		expected bool
-	}{
-		{
-			version:  "v2.0.2",
-			expected: false,
-		},
-		{
-			version:  "2.1.4",
-			expected: false,
-		},
-		{
-			version:  "v1.7.27",
-			expected: true,
-		},
-		{
-			version:  "1.6.0",
-			expected: true,
-		},
-	}
-	for _, tt := range tests {
-		// Testing with a suffix is important as some Linux distributions will modify the version
-		// with a non Semver compliant modification. Even if the version is supposed to comply with
-		// semver that may not always be the case.
-		for _, suffix := range []string{"", "~ds1"} {
-			version := tt.version + suffix
-			t.Run(version, func(t *testing.T) {
-				t.Parallel()
-
-				ok, err := canVerifyContainerdConfiguration(tt.version)
-				require.NoError(t, err)
-				require.Equal(t, tt.expected, ok)
-			})
-		}
-	}
-}
-
 func TestVerifyStatusResponse(t *testing.T) {
 	t.Parallel()
 
@@ -244,29 +204,79 @@ func TestParseContentRegistries(t *testing.T) {
 	}
 }
 
+func TestFeaturesForVersion(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		version          string
+		expectedString   string
+		expectedFeatures []Feature
+	}{
+		{
+			version:          "v2.0.2",
+			expectedFeatures: []Feature{},
+			expectedString:   "",
+		},
+		{
+			version:          "2.1.0",
+			expectedFeatures: []Feature{FeatureContentEvent},
+			expectedString:   "ContentEvent",
+		},
+		{
+			version:          "v1.7.27",
+			expectedFeatures: []Feature{FeatureConfigCheck},
+			expectedString:   "ConfigCheck",
+		},
+		{
+			version:          "1.6.0",
+			expectedFeatures: []Feature{FeatureConfigCheck},
+			expectedString:   "ConfigCheck",
+		},
+	}
+	for _, tt := range tests {
+		// Testing with a suffix is important as some Linux distributions will modify the version
+		// with a non Semver compliant modification. Even if the version is supposed to comply with
+		// semver that may not always be the case.
+		for _, suffix := range []string{"", "~ds1"} {
+			version := tt.version + suffix
+			t.Run(version, func(t *testing.T) {
+				t.Parallel()
+
+				feats, err := featuresForVersion(tt.version)
+				require.NoError(t, err)
+				for _, feat := range tt.expectedFeatures {
+					ok := feats.Has(feat)
+					require.True(t, ok)
+				}
+				require.Equal(t, tt.expectedString, feats.String())
+			})
+		}
+	}
+}
+
 func TestCreateFilter(t *testing.T) {
 	t.Parallel()
 
 	tests := []struct {
 		name                  string
-		expectedImageFilter   string
-		expectedEventFilter   string
-		expectedContentFilter string
+		expectedImageFilter   []string
+		expectedEventFilter   []string
+		expectedContentFilter []string
 		registries            []string
 	}{
 		{
 			name:                  "with registry filtering",
 			registries:            []string{"https://docker.io", "https://gcr.io"},
-			expectedImageFilter:   `name~="^(docker\\.io|gcr\\.io)/"`,
-			expectedEventFilter:   `topic~="/images/create|/images/delete",event.name~="^(docker\\.io|gcr\\.io)/"`,
-			expectedContentFilter: `labels."containerd.io/distribution.source.docker.io"~="^." labels."containerd.io/distribution.source.gcr.io"~="^."`,
+			expectedImageFilter:   []string{`name~="^(docker\\.io|gcr\\.io)/"`},
+			expectedEventFilter:   []string{`topic~="/images/create|/images/delete",event.name~="^(docker\\.io|gcr\\.io)/"`, `topic~="/content/create"`},
+			expectedContentFilter: []string{`labels."containerd.io/distribution.source.docker.io"~="^."`, `labels."containerd.io/distribution.source.gcr.io"~="^."`},
 		},
 		{
 			name:                  "without registry filtering",
 			registries:            []string{},
-			expectedImageFilter:   `name~="^.+/"`,
-			expectedEventFilter:   `topic~="/images/create|/images/delete",event.name~="^.+/"`,
-			expectedContentFilter: "",
+			expectedImageFilter:   []string{`name~="^.+/"`},
+			expectedEventFilter:   []string{`topic~="/images/create|/images/delete",event.name~="^.+/"`, `topic~="/content/create"`},
+			expectedContentFilter: []string{},
 		},
 	}
 	for _, tt := range tests {
@@ -274,9 +284,16 @@ func TestCreateFilter(t *testing.T) {
 			t.Parallel()
 
 			imageFilter, eventFilter, contentFilter := createFilters(stringListToUrlList(t, tt.registries))
+
 			require.Equal(t, tt.expectedImageFilter, imageFilter)
+			_, err := filters.ParseAll(imageFilter...)
+			require.NoError(t, err)
 			require.Equal(t, tt.expectedEventFilter, eventFilter)
+			_, err = filters.ParseAll(eventFilter...)
+			require.NoError(t, err)
 			require.Equal(t, tt.expectedContentFilter, contentFilter)
+			_, err = filters.ParseAll(contentFilter...)
+			require.NoError(t, err)
 		})
 	}
 }