Merge pull request #14314 from ChengyuZhu6/handler-platform

cri: resolve runtime handlers for image pulls
This commit is contained in:
Akihiro Suda authored and GitHub committed 2026-10-27 21:50:50 +00:00
commit 43aa4ea134
12 files changed
+485 -114

No files matched your search

+11
View File
@@ -26,6 +26,7 @@ import (
"golang.org/x/sync/semaphore"
"github.com/containerd/errdefs"
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/containerd/containerd/v2/core/content"
"github.com/containerd/containerd/v2/core/images"
@@ -65,6 +66,16 @@ func NewTransferService(cs content.Store, is images.Store, tc TransferConfig) tr
return ts
}
// SupportsUnpack reports whether the transfer service is configured to unpack
// the given platform into the given snapshotter.
func (ts *localTransferService) SupportsUnpack(ctx context.Context, platform ocispec.Platform, snapshotter string) bool {
matched, _ := getSupportedPlatform(ctx, transfer.UnpackConfiguration{
Platform: platform,
Snapshotter: snapshotter,
}, ts.config.UnpackPlatforms)
return matched
}
func (ts *localTransferService) Transfer(ctx context.Context, src any, dest any, opts ...transfer.Opt) error {
topts := &transfer.Config{}
for _, opt := range opts {
+2
View File
@@ -43,6 +43,8 @@ const (
ImageLabelKey = criContainerdPrefix + ".image"
// ImageLabelValue is the label value indicating the image is managed by cri plugin.
ImageLabelValue = "managed"
// ImagePlatformLabelKey is the label key indicating the platform selected when pulling the image.
ImagePlatformLabelKey = criContainerdPrefix + ".image.platform"
// PinnedImageLabelKey is the label value indicating the image is pinned.
PinnedImageLabelKey = criContainerdPrefix + ".pinned"
// PinnedImageLabelValue is the label value indicating the image is pinned.
@@ -398,6 +398,9 @@ func (s *fakeImageService) PullImage(context.Context, string, func(string) (stri
func (s *fakeImageService) UpdateRuntime(runtimeName string, imagePlatform *images.ImagePlatform) {
}
func (s *fakeImageService) UpdateDefaultRuntimeName(runtimeName string) {
}
func patchExceptedWithState(expected *runtime.ContainerStatus, state runtime.ContainerState) {
expected.State = state
switch state {
+16 -1
View File
@@ -41,8 +41,23 @@ func (c *CRIImageService) CheckImages(ctx context.Context) error {
var wg sync.WaitGroup
for _, i := range cImages {
wg.Go(func() {
name := i.Name()
platform, err := imagePlatformFromLabels(i.Labels())
if err != nil {
log.G(ctx).WithError(err).Errorf("Failed to get platform for image %q", name)
return
}
if platform == nil {
platform = platforms.Default()
} else {
i, err = c.client.GetImageWithPlatform(ctx, name, platform)
if err != nil {
log.G(ctx).WithError(err).Errorf("Failed to get image %q for its platform", name)
return
}
}
// TODO: Check platform/snapshot combination. Snapshot check should come first
ok, _, _, _, err := images.Check(ctx, i.ContentStore(), i.Target(), platforms.Default())
ok, _, _, _, err := images.Check(ctx, i.ContentStore(), i.Target(), platform)
if err != nil {
log.G(ctx).WithError(err).Errorf("Failed to check image content readiness for %q", i.Name())
return
+109 -54
View File
@@ -161,10 +161,9 @@ func (c *CRIImageService) PullImage(ctx context.Context, name string, credential
}
// CRI requires the request to be rejected if the runtime handler is unknown.
if runtimeHandler != "" {
if _, ok := c.runtimePlatforms[runtimeHandler]; !ok {
return "", fmt.Errorf("unknown runtime handler %q: %w", runtimeHandler, errdefs.ErrInvalidArgument)
}
runtime, err := c.resolveRequestRuntimeHandler(ctx, sandboxConfig, runtimeHandler)
if err != nil {
return "", err
}
imagePullProgressTimeout, err := time.ParseDuration(c.config.ImagePullProgressTimeout)
@@ -172,10 +171,8 @@ func (c *CRIImageService) PullImage(ctx context.Context, name string, credential
return "", fmt.Errorf("failed to parse image_pull_progress_timeout %q: %w", c.config.ImagePullProgressTimeout, err)
}
snapshotter, err := c.snapshotterFromPodSandboxConfig(ctx, ref, sandboxConfig, runtimeHandler)
if err != nil {
return "", err
}
snapshotter := runtime.Snapshotter
platform := runtime.Platform
span.SetAttributes(
tracing.Attribute("image.ref", ref),
@@ -187,14 +184,35 @@ func (c *CRIImageService) PullImage(ctx context.Context, name string, credential
//
// Transfer service does not currently support all the CRI image config options.
// TODO: Add support for DisableSnapshotAnnotations, DiscardUnpackedLayers, ImagePullWithSyncFs and unpackDuplicationSuppressor
//
// The transfer service only unpacks the platform/snapshotter combinations it is
// configured with, so fall back to the local pull when it can't unpack this one.
// A nil platform means no platform is configured for the runtime, which the
// pull paths resolve to the default spec. Check the same platform they use,
// so the fallback also applies to runtimes which only set a snapshotter.
unpackPlatform := platforms.DefaultSpec()
if platform != nil {
unpackPlatform = *platform
}
if platform != nil {
labels[crilabels.ImagePlatformLabelKey] = platforms.FormatAll(unpackPlatform)
}
useLocalPull := c.config.UseLocalImagePull
if !useLocalPull {
if ts, ok := c.transferrer.(unpackSupportChecker); ok && !ts.SupportsUnpack(ctx, unpackPlatform, snapshotter) {
log.G(ctx).Infof("Transfer service cannot unpack %s with snapshotter %q, using local pull",
platforms.Format(unpackPlatform), snapshotter)
useLocalPull = true
}
}
var (
image containerd.Image
bytesPulled uint64
)
if c.config.UseLocalImagePull {
image, bytesPulled, err = c.pullImageWithLocalPull(ctx, ref, credentials, snapshotter, labels, imagePullProgressTimeout)
if useLocalPull {
image, bytesPulled, err = c.pullImageWithLocalPull(ctx, ref, credentials, snapshotter, platform, labels, imagePullProgressTimeout)
} else {
image, bytesPulled, err = c.pullImageWithTransferService(ctx, ref, credentials, snapshotter, labels, imagePullProgressTimeout)
image, bytesPulled, err = c.pullImageWithTransferService(ctx, ref, credentials, snapshotter, platform, labels, imagePullProgressTimeout)
}
if err != nil {
@@ -222,8 +240,14 @@ func (c *CRIImageService) PullImage(ctx context.Context, name string, credential
// No need to use `updateImage`, because the image reference must
// have been managed by the cri plugin.
// TODO: Use image service directly
if err := c.imageStore.Update(ctx, r); err != nil {
return "", fmt.Errorf("failed to update image store %q: %w", r, err)
var updateErr error
if platform != nil {
updateErr = c.imageStore.UpdateWithPlatform(ctx, r, platforms.Only(*platform))
} else {
updateErr = c.imageStore.Update(ctx, r)
}
if updateErr != nil {
return "", fmt.Errorf("failed to update image store %q: %w", r, updateErr)
}
}
@@ -255,6 +279,7 @@ func (c *CRIImageService) pullImageWithLocalPull(
ref string,
credentials func(string) (string, string, error),
snapshotter string,
platform *imagespec.Platform,
labels map[string]string,
imagePullProgressTimeout time.Duration,
) (containerd.Image, uint64, error) {
@@ -281,6 +306,10 @@ func (c *CRIImageService) pullImageWithLocalPull(
}),
}
if platform != nil {
pullOpts = append(pullOpts, containerd.WithPlatformMatcher(platforms.Only(*platform)))
}
pullOpts = append(pullOpts, c.encryptedImagesPullOpts()...)
if !c.config.DisableSnapshotAnnotations {
pullOpts = append(pullOpts,
@@ -313,6 +342,7 @@ func (c *CRIImageService) pullImageWithTransferService(
ref string,
credentials func(string) (string, string, error),
snapshotter string,
platform *imagespec.Platform,
labels map[string]string,
imagePullProgressTimeout time.Duration,
) (containerd.Image, uint64, error) {
@@ -322,9 +352,13 @@ func (c *CRIImageService) pullImageWithTransferService(
transferProgressReporter := newTransferProgressReporter(ref, rcancel, imagePullProgressTimeout)
// Set image store opts
unpackPlatform := platforms.DefaultSpec()
if platform != nil {
unpackPlatform = *platform
}
sopts := []transferimage.StoreOpt{
transferimage.WithPlatforms(platforms.DefaultSpec()),
transferimage.WithUnpack(platforms.DefaultSpec(), snapshotter),
transferimage.WithPlatforms(unpackPlatform),
transferimage.WithUnpack(unpackPlatform, snapshotter),
transferimage.WithImageLabels(labels),
}
@@ -353,7 +387,12 @@ func (c *CRIImageService) pullImageWithTransferService(
}
// Image should be pulled, unpacked and present in containerd image store at this moment
image, err := c.client.GetImage(ctx, ref)
var image containerd.Image
if platform != nil {
image, err = c.client.GetImageWithPlatform(ctx, ref, platforms.Only(*platform))
} else {
image, err = c.client.GetImage(ctx, ref)
}
if err != nil {
return nil, 0, fmt.Errorf("failed to get image %q from containerd image store: %w", ref, err)
}
@@ -434,6 +473,9 @@ func (c *CRIImageService) createOrUpdateImageReference(ctx context.Context, name
labels[crilabels.PinnedImageLabelKey] == crilabels.PinnedImageLabelValue {
fieldpaths = append(fieldpaths, "labels."+crilabels.PinnedImageLabelKey)
}
if oldImg.Labels[crilabels.ImagePlatformLabelKey] != labels[crilabels.ImagePlatformLabelKey] {
fieldpaths = append(fieldpaths, "labels."+crilabels.ImagePlatformLabelKey)
}
if oldImg.Target.Digest == img.Target.Digest && len(fieldpaths) < 2 {
return nil
}
@@ -452,6 +494,18 @@ func (c *CRIImageService) getLabels(ctx context.Context, name string) map[string
return labels
}
func imagePlatformFromLabels(labels map[string]string) (platforms.MatchComparer, error) {
value := labels[crilabels.ImagePlatformLabelKey]
if value == "" {
return nil, nil
}
platform, err := platforms.Parse(value)
if err != nil {
return nil, err
}
return platforms.Only(platform), nil
}
// UpdateImage updates image store to reflect the newest state of an image reference
// in containerd. If the reference is not managed by the cri plugin, the function also
// generates necessary metadata for the image and make it managed.
@@ -464,14 +518,28 @@ func (c *CRIImageService) UpdateImage(ctx context.Context, r string) error {
}
// If the image is not found, we should continue updating the cache,
// so that the image can be removed from the cache.
if err := c.imageStore.Update(ctx, r); err != nil {
if err := c.imageStore.UpdateWithPlatform(ctx, r, nil); err != nil {
return fmt.Errorf("update image store for %q: %w", r, err)
}
return nil
}
platform, err := imagePlatformFromLabels(img.Labels())
if err != nil {
return fmt.Errorf("get platform for image %q: %w", r, err)
}
if platform != nil {
img, err = c.client.GetImageWithPlatform(ctx, r, platform)
if err != nil {
return fmt.Errorf("get image by reference and platform: %w", err)
}
}
labels := img.Labels()
criLabels := c.getLabels(ctx, r)
if value := labels[crilabels.ImagePlatformLabelKey]; value != "" {
criLabels[crilabels.ImagePlatformLabelKey] = value
}
for key, value := range criLabels {
if labels[key] != value {
// Make sure the image has the image id as its unique
@@ -484,7 +552,7 @@ func (c *CRIImageService) UpdateImage(ctx context.Context, r string) error {
if err := c.createOrUpdateImageReference(ctx, id, img.Target(), criLabels); err != nil {
return fmt.Errorf("create image id reference %q: %w", id, err)
}
if err := c.imageStore.Update(ctx, id); err != nil {
if err := c.imageStore.UpdateWithPlatform(ctx, id, platform); err != nil {
return fmt.Errorf("update image store for %q: %w", id, err)
}
// The image id is ready, add the label to mark the image as managed.
@@ -494,7 +562,7 @@ func (c *CRIImageService) UpdateImage(ctx context.Context, r string) error {
break
}
}
if err := c.imageStore.Update(ctx, r); err != nil {
if err := c.imageStore.UpdateWithPlatform(ctx, r, platform); err != nil {
return fmt.Errorf("update image store for %q: %w", r, err)
}
return nil
@@ -870,49 +938,36 @@ func (rt *pullRequestReporterRoundTripper) RoundTrip(req *http.Request) (*http.R
return resp, err
}
// snapshotterFromPodSandboxConfig returns the snapshotter to use for the given
// runtime handler. If a runtime-specific snapshotter is configured, it will be
// returned; otherwise the default snapshotter is used.
//
// The runtimeHandler parameter (from CRI PullImageRequest, available since cri-api v0.29.0)
// takes precedence. If empty, we fall back to the experimental annotation for backward
// compatibility with clients that don't pass the runtime handler.
// resolveRequestRuntimeHandler resolves the request field or falls back to the
// deprecated annotation. The request field always takes precedence.
//
// Deprecated: The annotation-based fallback (io.containerd.cri.runtime-handler) is
// deprecated and will be removed in containerd 2.5.
//
// See https://github.com/containerd/containerd/issues/6657
func (c *CRIImageService) snapshotterFromPodSandboxConfig(ctx context.Context, imageRef string,
s *runtime.PodSandboxConfig, runtimeHandler string) (string, error) {
snapshotter := c.config.Snapshotter
if s == nil {
return snapshotter, nil
}
// If runtimeHandler parameter is empty, fall back to annotation for backward compatibility
if runtimeHandler == "" {
if s.Annotations == nil {
return snapshotter, nil
}
var ok bool
runtimeHandler, ok = s.Annotations[annotations.RuntimeHandler]
if !ok {
return snapshotter, nil
}
log.G(ctx).Warnf("Using deprecated annotation %q for runtime handler. "+
"This will be removed in a future release (2.5). "+
"Please update your CRI client to pass runtime handler in PullImageRequest.",
annotations.RuntimeHandler)
}
if c.runtimePlatforms != nil {
if p, ok := c.runtimePlatforms[runtimeHandler]; ok && p != nil && p.Snapshotter != snapshotter {
snapshotter = p.Snapshotter
log.G(ctx).Infof("experimental: PullImage %q for runtime %s, using snapshotter %s", imageRef, runtimeHandler, snapshotter)
func (c *CRIImageService) resolveRequestRuntimeHandler(ctx context.Context,
s *runtime.PodSandboxConfig, handler string) (resolvedRuntimeHandler, error) {
if handler == "" && s != nil {
if h, ok := s.Annotations[annotations.RuntimeHandler]; ok {
log.G(ctx).Warnf("Using deprecated annotation %q for runtime handler. "+
"This will be removed in a future release (2.5). "+
"Please update your CRI client to pass runtime handler in PullImageRequest.",
annotations.RuntimeHandler)
if _, ok := c.runtimePlatforms[h]; ok {
handler = h
} else if h != "" {
log.G(ctx).Warnf("Ignoring unknown runtime handler %q from annotation %q", h, annotations.RuntimeHandler)
}
}
}
return c.resolveRuntimeHandler(handler)
}
return snapshotter, nil
// unpackSupportChecker is implemented by transfer services that can report whether a
// platform can be unpacked into a snapshotter. Transfer services which don't implement
// it keep being used as before.
type unpackSupportChecker interface {
SupportsUnpack(context.Context, imagespec.Platform, string) bool
}
type criCredentials struct {
+241 -32
View File
@@ -27,13 +27,15 @@ import (
"time"
"github.com/containerd/errdefs"
"github.com/containerd/platforms"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
"github.com/containerd/platforms"
containerd "github.com/containerd/containerd/v2/client"
ctrdimages "github.com/containerd/containerd/v2/core/images"
"github.com/containerd/containerd/v2/core/transfer"
transferimage "github.com/containerd/containerd/v2/core/transfer/image"
"github.com/containerd/containerd/v2/internal/cri/annotations"
criconfig "github.com/containerd/containerd/v2/internal/cri/config"
"github.com/containerd/containerd/v2/internal/cri/labels"
@@ -381,39 +383,64 @@ func TestEncryptedImagePullOpts(t *testing.T) {
}
}
func TestSnapshotterFromPodSandboxConfig(t *testing.T) {
func TestResolveRequestRuntimeHandler(t *testing.T) {
defaultSnapshotter := "native"
runtimeSnapshotter := "devmapper"
runtimePlatform := ocispec.Platform{OS: "linux", Architecture: "arm64", Variant: "v8"}
tests := []struct {
desc string
podSandboxConfig *runtime.PodSandboxConfig
runtimeHandler string
defaultRuntimeName string
expectedSnapshotter string
expectedPlatform *ocispec.Platform
expectedErr bool
}{
{
desc: "should return default snapshotter for nil podSandboxConfig",
runtimeHandler: "",
desc: "should use default runtime handler for nil podSandboxConfig",
expectedSnapshotter: defaultSnapshotter,
},
{
desc: "should return default snapshotter for empty runtimeHandler",
desc: "should use default runtime handler for empty runtimeHandler",
podSandboxConfig: &runtime.PodSandboxConfig{},
runtimeHandler: "",
expectedSnapshotter: defaultSnapshotter,
},
{
desc: "should return default snapshotter for runtime not found",
podSandboxConfig: &runtime.PodSandboxConfig{},
runtimeHandler: "runtime-not-exists",
expectedSnapshotter: defaultSnapshotter,
},
{
desc: "should return snapshotter for existing runtime",
desc: "should use snapshotter for existing runtime",
podSandboxConfig: &runtime.PodSandboxConfig{},
runtimeHandler: "existing-runtime",
expectedSnapshotter: runtimeSnapshotter,
},
{
desc: "should honor explicit runtime handler with nil podSandboxConfig",
runtimeHandler: "existing-runtime",
expectedSnapshotter: runtimeSnapshotter,
},
{
desc: "should use platform configured for the runtime handler",
runtimeHandler: "platform-runtime",
expectedSnapshotter: runtimeSnapshotter,
expectedPlatform: &runtimePlatform,
},
{
desc: "should resolve default runtime handler without configured platform",
podSandboxConfig: &runtime.PodSandboxConfig{},
defaultRuntimeName: "existing-runtime",
expectedSnapshotter: runtimeSnapshotter,
},
{
desc: "should resolve default runtime handler with configured platform",
podSandboxConfig: &runtime.PodSandboxConfig{},
defaultRuntimeName: "platform-runtime",
expectedSnapshotter: runtimeSnapshotter,
expectedPlatform: &runtimePlatform,
},
{
desc: "should reject unknown runtime handler",
podSandboxConfig: &runtime.PodSandboxConfig{},
runtimeHandler: "runtime-not-exists",
expectedErr: true,
},
{
desc: "should fall back to annotation when runtimeHandler is empty",
podSandboxConfig: &runtime.PodSandboxConfig{
@@ -421,44 +448,50 @@ func TestSnapshotterFromPodSandboxConfig(t *testing.T) {
annotations.RuntimeHandler: "existing-runtime",
},
},
runtimeHandler: "",
expectedSnapshotter: runtimeSnapshotter,
},
{
desc: "should prefer runtimeHandler parameter over annotation",
desc: "should ignore unknown runtime handler from annotation",
podSandboxConfig: &runtime.PodSandboxConfig{
Annotations: map[string]string{
annotations.RuntimeHandler: "runtime-not-exists",
},
},
runtimeHandler: "existing-runtime",
expectedSnapshotter: runtimeSnapshotter,
},
{
desc: "should return default when annotation has unknown runtime and runtimeHandler is empty",
podSandboxConfig: &runtime.PodSandboxConfig{
Annotations: map[string]string{
annotations.RuntimeHandler: "runtime-not-exists",
},
},
runtimeHandler: "",
expectedSnapshotter: defaultSnapshotter,
},
{
desc: "should prefer runtimeHandler parameter over annotation",
runtimeHandler: "platform-runtime",
podSandboxConfig: &runtime.PodSandboxConfig{
Annotations: map[string]string{
annotations.RuntimeHandler: "runtime-not-exists",
},
},
expectedSnapshotter: runtimeSnapshotter,
expectedPlatform: &runtimePlatform,
},
}
for _, tt := range tests {
t.Run(tt.desc, func(t *testing.T) {
cri, _ := newTestCRIService()
cri.config.Snapshotter = defaultSnapshotter
cri.UpdateDefaultRuntimeName(tt.defaultRuntimeName)
cri.runtimePlatforms["existing-runtime"] = &ImagePlatform{
Platform: platforms.DefaultSpec(),
Snapshotter: runtimeSnapshotter,
}
snapshotter, err := cri.snapshotterFromPodSandboxConfig(context.Background(), "test-image", tt.podSandboxConfig, tt.runtimeHandler)
assert.Equal(t, tt.expectedSnapshotter, snapshotter)
if tt.expectedErr {
assert.Error(t, err)
cri.runtimePlatforms["platform-runtime"] = &ImagePlatform{
Platform: &runtimePlatform,
Snapshotter: runtimeSnapshotter,
}
h, err := cri.resolveRequestRuntimeHandler(context.Background(), tt.podSandboxConfig, tt.runtimeHandler)
if tt.expectedErr {
assert.ErrorIs(t, err, errdefs.ErrInvalidArgument)
return
}
assert.NoError(t, err)
assert.Equal(t, tt.expectedSnapshotter, h.Snapshotter)
assert.Equal(t, tt.expectedPlatform, h.Platform)
})
}
}
@@ -470,6 +503,140 @@ func TestPullImageRuntimeHandler(t *testing.T) {
assert.ErrorIs(t, err, errdefs.ErrInvalidArgument)
}
func TestRuntimeHandlerPullBackends(t *testing.T) {
platform := ocispec.Platform{
OS: "windows",
Architecture: "amd64",
OSVersion: "10.0.20348",
OSFeatures: []string{"win32k"},
}
t.Run("local pull", func(t *testing.T) {
client := &pullTestClient{
pull: func(_ context.Context, _ string, opts ...containerd.RemoteOpt) (containerd.Image, error) {
remote := &containerd.RemoteContext{}
for _, opt := range opts {
require.NoError(t, opt(nil, remote))
}
require.Equal(t, "native", remote.Snapshotter)
require.True(t, remote.PlatformMatcher.Match(platform))
return nil, errdefs.ErrUnavailable
},
}
c := &CRIImageService{
client: client,
config: criconfig.ImageConfig{DisableSnapshotAnnotations: true},
}
_, _, err := c.pullImageWithLocalPull(context.Background(), "example.com/test:latest", nil, "native", &platform, nil, time.Minute)
require.ErrorIs(t, err, errdefs.ErrUnavailable)
})
t.Run("transfer pull", func(t *testing.T) {
transferrer := &pullTestTransferrer{
transfer: func(_ context.Context, _, destination any, _ ...transfer.Opt) error {
store := destination.(*transferimage.Store)
require.Equal(t, []ocispec.Platform{platform}, store.Platforms())
require.Equal(t, platform, store.UnpackPlatforms()[0].Platform)
require.Equal(t, "native", store.UnpackPlatforms()[0].Snapshotter)
return nil
},
}
client := &pullTestClient{
getImageWithPlatform: func(_ context.Context, ref string, matcher platforms.MatchComparer) (containerd.Image, error) {
require.True(t, matcher.Match(platform))
return containerd.NewImageWithPlatform(&containerd.Client{}, ctrdimages.Image{Name: ref}, matcher), nil
},
}
c := &CRIImageService{client: client, transferrer: transferrer}
image, _, err := c.pullImageWithTransferService(context.Background(), "example.com/test:latest", nil, "native", &platform, nil, time.Minute)
require.NoError(t, err)
require.True(t, image.Platform().Match(platform))
})
t.Run("unsupported unpack", func(t *testing.T) {
localCalled := false
client := &pullTestClient{
pull: func(_ context.Context, _ string, opts ...containerd.RemoteOpt) (containerd.Image, error) {
localCalled = true
remote := &containerd.RemoteContext{}
for _, opt := range opts {
require.NoError(t, opt(nil, remote))
}
require.Equal(t, platforms.FormatAll(platform), remote.Labels[labels.ImagePlatformLabelKey])
return nil, errdefs.ErrUnavailable
},
}
transferrer := &pullTestTransferrer{
supportsUnpack: func(p ocispec.Platform, snapshotter string) bool {
require.Equal(t, platform, p)
require.Equal(t, "native", snapshotter)
return false
},
transfer: func(context.Context, any, any, ...transfer.Opt) error {
t.Fatal("transfer should not be called")
return nil
},
}
c := &CRIImageService{
client: client,
transferrer: transferrer,
runtimePlatforms: map[string]*ImagePlatform{"native": {Snapshotter: "native", Platform: &platform}},
config: criconfig.ImageConfig{ImagePullProgressTimeout: "1m", DisableSnapshotAnnotations: true},
}
_, err := c.PullImage(context.Background(), "example.com/test:latest", nil, nil, "native")
require.ErrorIs(t, err, errdefs.ErrUnavailable)
require.True(t, localCalled)
})
}
func TestPersistedImagePlatform(t *testing.T) {
platform := ocispec.Platform{
OS: "windows",
Architecture: "amd64",
OSVersion: "10.0.20348",
OSFeatures: []string{"win32k"},
}
ref := "example.com/test:latest"
image := containerd.NewImage(&containerd.Client{}, ctrdimages.Image{
Name: ref,
Labels: map[string]string{labels.ImagePlatformLabelKey: platforms.FormatAll(platform)},
})
t.Run("event update", func(t *testing.T) {
client := &pullTestClient{
getImage: func(context.Context, string) (containerd.Image, error) {
return image, nil
},
getImageWithPlatform: func(_ context.Context, _ string, matcher platforms.MatchComparer) (containerd.Image, error) {
require.True(t, matcher.Match(platform))
return nil, errdefs.ErrUnavailable
},
}
c := &CRIImageService{client: client}
require.ErrorIs(t, c.UpdateImage(context.Background(), ref), errdefs.ErrUnavailable)
})
t.Run("recovery", func(t *testing.T) {
called := false
client := &pullTestClient{
listImages: func(context.Context, ...string) ([]containerd.Image, error) {
return []containerd.Image{image}, nil
},
getImageWithPlatform: func(_ context.Context, _ string, matcher platforms.MatchComparer) (containerd.Image, error) {
called = true
require.True(t, matcher.Match(platform))
return nil, errdefs.ErrUnavailable
},
}
c := &CRIImageService{client: client}
require.NoError(t, c.CheckImages(context.Background()))
require.True(t, called)
})
}
func TestImageGetLabels(t *testing.T) {
criService, _ := newTestCRIService()
@@ -953,6 +1120,48 @@ func (fn roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error)
return fn(req)
}
type pullTestClient struct {
listImages func(context.Context, ...string) ([]containerd.Image, error)
getImage func(context.Context, string) (containerd.Image, error)
pull func(context.Context, string, ...containerd.RemoteOpt) (containerd.Image, error)
getImageWithPlatform func(context.Context, string, platforms.MatchComparer) (containerd.Image, error)
}
func (c *pullTestClient) ListImages(ctx context.Context, filters ...string) ([]containerd.Image, error) {
if c.listImages != nil {
return c.listImages(ctx, filters...)
}
return nil, nil
}
func (c *pullTestClient) GetImage(ctx context.Context, ref string) (containerd.Image, error) {
if c.getImage != nil {
return c.getImage(ctx, ref)
}
return nil, errdefs.ErrNotImplemented
}
func (c *pullTestClient) GetImageWithPlatform(ctx context.Context, ref string, platform platforms.MatchComparer) (containerd.Image, error) {
return c.getImageWithPlatform(ctx, ref, platform)
}
func (c *pullTestClient) Pull(ctx context.Context, ref string, opts ...containerd.RemoteOpt) (containerd.Image, error) {
return c.pull(ctx, ref, opts...)
}
type pullTestTransferrer struct {
transfer func(context.Context, any, any, ...transfer.Opt) error
supportsUnpack func(ocispec.Platform, string) bool
}
func (t *pullTestTransferrer) Transfer(ctx context.Context, source, destination any, opts ...transfer.Opt) error {
return t.transfer(ctx, source, destination, opts...)
}
func (t *pullTestTransferrer) SupportsUnpack(_ context.Context, platform ocispec.Platform, snapshotter string) bool {
return t.supportsUnpack(platform, snapshotter)
}
// fakeObserver records every Observe call so tests can assert both the
// presence and the value of observations without touching the real histogram.
type fakeObserver struct {
+64 -9
View File
@@ -18,8 +18,11 @@ package images
import (
"context"
"fmt"
"time"
"github.com/containerd/errdefs"
containerd "github.com/containerd/containerd/v2/client"
"github.com/containerd/containerd/v2/core/content"
"github.com/containerd/containerd/v2/core/images"
@@ -42,12 +45,26 @@ import (
type imageClient interface {
ListImages(context.Context, ...string) ([]containerd.Image, error)
GetImage(context.Context, string) (containerd.Image, error)
GetImageWithPlatform(context.Context, string, platforms.MatchComparer) (containerd.Image, error)
Pull(context.Context, string, ...containerd.RemoteOpt) (containerd.Image, error)
}
type platformImageClient struct {
*containerd.Client
}
func (c *platformImageClient) GetImageWithPlatform(ctx context.Context, ref string, platform platforms.MatchComparer) (containerd.Image, error) {
i, err := c.ImageService().Get(ctx, ref)
if err != nil {
return nil, err
}
return containerd.NewImageWithPlatform(c.Client, i, platform), nil
}
type ImagePlatform struct {
Snapshotter string
Platform imagespec.Platform
// Platform is nil when no platform is explicitly configured.
Platform *imagespec.Platform
}
type CRIImageService struct {
@@ -65,6 +82,9 @@ type CRIImageService struct {
imageFSPaths map[string]string
// runtimePlatforms maps runtime handlers to their configured image platforms.
runtimePlatforms map[string]*ImagePlatform
// defaultRuntimeName is the runtime used when a CRI request has an empty
// runtime handler.
defaultRuntimeName string
// imageStore stores all resources associated with images.
imageStore *imagestore.Store
// snapshotStore stores information of all snapshots.
@@ -80,6 +100,11 @@ type CRIImageService struct {
downloadLimiter *semaphore.Weighted
}
type resolvedRuntimeHandler struct {
Platform *imagespec.Platform
Snapshotter string
}
type GRPCCRIImageService struct {
*CRIImageService
}
@@ -95,7 +120,7 @@ type CRIImageServiceOptions struct {
Snapshotters map[string]snapshots.Snapshotter
Client imageClient
Client *containerd.Client
Transferrer transfer.Transferrer
}
@@ -118,7 +143,7 @@ func NewService(config criconfig.ImageConfig, options *CRIImageServiceOptions) (
svc := CRIImageService{
config: config,
images: options.Images,
client: options.Client,
client: &platformImageClient{Client: options.Client},
imageStore: imagestore.NewStore(options.Images, options.Content, platforms.Default()),
imageFSPaths: options.ImageFSPaths,
runtimePlatforms: options.RuntimePlatforms,
@@ -139,16 +164,17 @@ func NewService(config criconfig.ImageConfig, options *CRIImageServiceOptions) (
return &svc, nil
}
// UpdateRuntime records a runtime handler and its snapshotter, if one is configured.
// It is called by the main CRI plugin after both image and runtime plugins are initialized.
// imagePlatform is nil when the runtime has no snapshotter configured.
// UpdateRuntime records a runtime handler and its image settings.
func (c *CRIImageService) UpdateRuntime(runtimeName string, imagePlatform *ImagePlatform) {
if c.runtimePlatforms == nil {
c.runtimePlatforms = make(map[string]*ImagePlatform)
}
// Don't override if already configured
if _, exists := c.runtimePlatforms[runtimeName]; exists {
log.L.Debugf("Runtime %q already configured, not overriding", runtimeName)
// Fill missing settings without overriding image configuration.
if existing, exists := c.runtimePlatforms[runtimeName]; exists {
if existing != nil && existing.Snapshotter == "" && imagePlatform != nil {
existing.Snapshotter = imagePlatform.Snapshotter
}
log.L.Debugf("Runtime %q already configured, preserving image settings", runtimeName)
return
}
c.runtimePlatforms[runtimeName] = imagePlatform
@@ -157,6 +183,35 @@ func (c *CRIImageService) UpdateRuntime(runtimeName string, imagePlatform *Image
}
}
// UpdateDefaultRuntimeName records the runtime used for CRI requests with an empty
// runtime handler.
func (c *CRIImageService) UpdateDefaultRuntimeName(runtimeName string) {
c.defaultRuntimeName = runtimeName
}
func (c *CRIImageService) resolveRuntimeHandler(handler string) (resolvedRuntimeHandler, error) {
if handler != "" {
if _, ok := c.runtimePlatforms[handler]; !ok {
return resolvedRuntimeHandler{}, fmt.Errorf("unknown runtime handler %q: %w", handler, errdefs.ErrInvalidArgument)
}
}
name := handler
if name == "" {
name = c.defaultRuntimeName
}
h := resolvedRuntimeHandler{
Snapshotter: c.config.Snapshotter,
}
if p, ok := c.runtimePlatforms[name]; ok && p != nil {
h.Platform = p.Platform
if p.Snapshotter != "" {
h.Snapshotter = p.Snapshotter
}
}
return h, nil
}
// LocalResolve resolves image reference locally and returns corresponding image metadata. It
// returns errdefs.ErrNotFound if the reference doesn't exist.
func (c *CRIImageService) LocalResolve(refOrID string) (imagestore.Image, error) {
@@ -55,6 +55,17 @@ var testImageConfig = criconfig.ImageConfig{
},
}
func TestUpdateRuntimeMergesSnapshotter(t *testing.T) {
c, _ := newTestCRIService()
c.runtimePlatforms["runtime"] = &ImagePlatform{}
c.UpdateRuntime("runtime", &ImagePlatform{Snapshotter: "native"})
assert.Equal(t, "native", c.runtimePlatforms["runtime"].Snapshotter)
c.runtimePlatforms["runtime"] = &ImagePlatform{Snapshotter: "overlayfs"}
c.UpdateRuntime("runtime", &ImagePlatform{Snapshotter: "native"})
assert.Equal(t, "overlayfs", c.runtimePlatforms["runtime"].Snapshotter)
}
func TestLocalResolve(t *testing.T) {
image := imagestore.Image{
ID: "sha256:c75bebcdd211f41b3a460c7bf82970ed6c75acaab9cd4c9a4e125b03ca113799",
+1
View File
@@ -116,6 +116,7 @@ type ImageService interface {
Config() criconfig.ImageConfig
UpdateRuntime(runtimeName string, imagePlatform *images.ImagePlatform)
UpdateDefaultRuntimeName(runtimeName string)
}
// criService implements CRIService.
+16 -6
View File
@@ -97,6 +97,12 @@ func NewStore(img Getter, provider content.InfoReaderProvider, platform platform
// Update updates cache for a reference.
func (s *Store) Update(ctx context.Context, ref string) error {
return s.UpdateWithPlatform(ctx, ref, nil)
}
// UpdateWithPlatform updates cache for a reference using the given platform matcher.
// A nil matcher keeps using the store default.
func (s *Store) UpdateWithPlatform(ctx context.Context, ref string, platform platforms.MatchComparer) error {
s.lock.Lock()
defer s.lock.Unlock()
@@ -107,7 +113,7 @@ func (s *Store) Update(ctx context.Context, ref string) error {
var img *Image
if err == nil {
img, err = s.getImage(ctx, i)
img, err = s.getImage(ctx, i, platform)
if err != nil {
return fmt.Errorf("get image info from containerd: %w", err)
}
@@ -146,20 +152,24 @@ func (s *Store) update(ref string, img *Image) error {
return s.store.add(*img)
}
// getImage gets image information from containerd for current platform.
func (s *Store) getImage(ctx context.Context, i images.Image) (*Image, error) {
diffIDs, err := i.RootFS(ctx, s.provider, s.platform)
// getImage gets image information from containerd for the given platform matcher.
// A nil matcher keeps using the store default.
func (s *Store) getImage(ctx context.Context, i images.Image, platform platforms.MatchComparer) (*Image, error) {
if platform == nil {
platform = s.platform
}
diffIDs, err := i.RootFS(ctx, s.provider, platform)
if err != nil {
return nil, fmt.Errorf("get image diffIDs: %w", err)
}
chainID := imageidentity.ChainID(diffIDs)
size, err := usage.CalculateImageUsage(ctx, i, s.provider, usage.WithManifestLimit(s.platform, 1), usage.WithManifestUsage())
size, err := usage.CalculateImageUsage(ctx, i, s.provider, usage.WithManifestLimit(platform, 1), usage.WithManifestUsage())
if err != nil {
return nil, fmt.Errorf("get image compressed resource size: %w", err)
}
desc, err := i.Config(ctx, s.provider, s.platform)
desc, err := i.Config(ctx, s.provider, platform)
if err != nil {
return nil, fmt.Errorf("get image config descriptor: %w", err)
}
+1 -1
View File
@@ -94,12 +94,12 @@ func initCRIService(ic *plugin.InitContext) (any, error) {
if err := validateRuntimePlatforms(runtimeConfig.Runtimes, imageConfig.RuntimePlatforms); err != nil {
return nil, err
}
imageSvc.UpdateDefaultRuntimeName(runtimeConfig.DefaultRuntimeName)
for runtimeName, rt := range runtimeConfig.Runtimes {
var imagePlatform *images.ImagePlatform
if rt.Snapshotter != "" {
imagePlatform = &images.ImagePlatform{
Snapshotter: rt.Snapshotter,
Platform: platforms.DefaultSpec(),
}
}
imageSvc.UpdateRuntime(runtimeName, imagePlatform)
+10 -11
View File
@@ -37,6 +37,7 @@ import (
"github.com/containerd/platforms"
"github.com/containerd/plugin"
"github.com/containerd/plugin/registry"
imagespec "github.com/opencontainers/image-spec/specs-go/v1"
)
func init() {
@@ -137,27 +138,25 @@ func init() {
log.L.Infof("Get image filesystem path %q for snapshotter %q", options.ImageFSPaths[defaultSnapshotter], defaultSnapshotter)
for runtimeName, rp := range config.RuntimePlatforms {
snapshotter := rp.Snapshotter
if snapshotter == "" {
snapshotter = defaultSnapshotter
if rp.Snapshotter != "" {
if _, ok := options.ImageFSPaths[rp.Snapshotter]; !ok {
options.ImageFSPaths[rp.Snapshotter] = snapshotRoot(rp.Snapshotter)
log.L.Infof("Get image filesystem path %q for snapshotter %q", options.ImageFSPaths[rp.Snapshotter], rp.Snapshotter)
}
}
if _, ok := options.ImageFSPaths[snapshotter]; !ok {
options.ImageFSPaths[snapshotter] = snapshotRoot(snapshotter)
log.L.Infof("Get image filesystem path %q for snapshotter %q", options.ImageFSPaths[snapshotter], snapshotter)
}
platform := platforms.DefaultSpec()
// Left unset when runtime_platforms doesn't configure a platform.
var platform *imagespec.Platform
if rp.Platform != "" {
p, err := platforms.Parse(rp.Platform)
if err != nil {
return nil, fmt.Errorf("unable to parse platform %q: %w", rp.Platform, err)
}
platform = p
platform = &p
}
options.RuntimePlatforms[runtimeName] = &images.ImagePlatform{
Snapshotter: snapshotter,
Snapshotter: rp.Snapshotter,
Platform: platform,
}
}