cri: unpack images with per-layer labels for runtime-specific snapshotters

Remote/proxy snapshotters like nydus need per-layer annotations on each
snapshot (cri.image-ref, cri.layer-digest, cri.manifest-digest,
cri.image-layers) so they can lazily fetch content inside the guest VM.
During a normal PullImage, these annotations are set by
AppendInfoHandlerWrapper and flow through the core/unpack.Unpacker to
each layer's Prepare/Commit call.

However, when an image is already present for one snapshotter (e.g.,
overlayfs) and needs to be used with a different one (e.g., nydus for
Kata), no pull occurs.  The image must be unpacked into the target
snapshotter with the correct per-layer labels.

Replace the image.Unpack() fallback in customopts.WithNewSnapshot with
unpackImage, which leverages the existing core/unpack.Unpacker and wraps
the image handler with AppendInfoHandlerWrapper when snapshot annotations
are enabled (!DisableSnapshotAnnotations).  This reuses the same unpack
machinery as PullImage, including retry handling, parallel layer
support, and deduplication.

Signed-off-by: Fabiano Fidêncio <ffidencio@nvidia.com>

integration/runtime_handler_unpack_labels_linux_test.go

@@ -0,0 +1,144 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package integration
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/containerd/errdefs"
+	"github.com/opencontainers/image-spec/identity"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	criruntime "k8s.io/cri-api/pkg/apis/runtime/v1"
+
+	containerd "github.com/containerd/containerd/v2/client"
+	"github.com/containerd/containerd/v2/integration/images"
+	snpkg "github.com/containerd/containerd/v2/pkg/snapshotters"
+	"github.com/containerd/containerd/v2/plugins"
+)
+
+func TestRuntimeHandlerUnpackWithSnapshotLabels(t *testing.T) {
+	workDir := t.TempDir()
+	cfgPath := filepath.Join(workDir, "config.toml")
+	cfg := `
+version = 3
+
+[plugins.'io.containerd.cri.v1.images']
+  snapshotter = "overlayfs"
+  disable_snapshot_annotations = false
+
+[plugins.'io.containerd.cri.v1.runtime'.containerd]
+  default_runtime_name = "runc"
+
+[plugins.'io.containerd.cri.v1.runtime'.containerd.runtimes.runc]
+  runtime_type = "io.containerd.runc.v2"
+  snapshotter = "overlayfs"
+
+[plugins.'io.containerd.cri.v1.runtime'.containerd.runtimes.erofs]
+  runtime_type = "io.containerd.runc.v2"
+  snapshotter = "erofs"
+`
+	require.NoError(t, os.WriteFile(cfgPath, []byte(cfg), 0o600))
+
+	ctrd := newCtrdProc(t, *containerdBin, workDir, nil)
+	require.NoError(t, ctrd.isReady())
+
+	rSvc := ctrd.criRuntimeService(t)
+	iSvc := ctrd.criImageService(t)
+
+	ctrdClient, err := containerd.New(ctrd.grpcAddress(), containerd.WithDefaultNamespace(k8sNamespace))
+	require.NoError(t, err)
+
+	t.Cleanup(func() {
+		if t.Failed() {
+			t.Log("Dumping containerd config and logs due to test failure")
+			dumpFileContent(t, ctrd.configPath())
+			dumpFileContent(t, ctrd.logPath())
+		}
+		assert.NoError(t, ctrdClient.Close())
+		cleanupPods(t, rSvc)
+		assert.NoError(t, ctrd.kill(syscall.SIGTERM))
+		assert.NoError(t, ctrd.wait(5*time.Minute))
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	resp, err := ctrdClient.IntrospectionService().Plugins(ctx, fmt.Sprintf("type==%s,id==%s", plugins.SnapshotPlugin, "erofs"))
+	require.NoError(t, err)
+	if len(resp.Plugins) == 0 {
+		t.Skip("erofs snapshotter plugin is not registered")
+	}
+	if initErr := resp.Plugins[0].InitErr; initErr != nil {
+		t.Skipf("erofs snapshotter plugin is not ready: %s", initErr.Message)
+	}
+
+	nginxImage := images.Get(images.Nginx)
+	pullImagesByCRI(t, iSvc, nginxImage)
+
+	img, err := ctrdClient.GetImage(context.Background(), nginxImage)
+	require.NoError(t, err)
+	diffIDs, err := img.RootFS(context.Background())
+	require.NoError(t, err)
+	chainIDs := identity.ChainIDs(diffIDs)
+
+	// First pod uses default runtime handler (overlayfs). No containers created.
+	sb1Cfg := PodSandboxConfig("overlay-pod", "runtime-handler-unpack")
+	_, err = rSvc.RunPodSandbox(sb1Cfg, "")
+	require.NoError(t, err)
+
+	// Image is pulled with overlayfs; nginx snapshots should not exist on erofs yet.
+	erofsSn := ctrdClient.SnapshotService("erofs")
+	for _, chainID := range chainIDs {
+		_, err := erofsSn.Stat(context.Background(), chainID.String())
+		assert.Truef(t, errdefs.IsNotFound(err), "expected no erofs snapshot for chainID %s before erofs container creation, got err=%v", chainID, err)
+	}
+
+	// Second pod uses erofs runtime handler. Creating nginx container should trigger
+	// automatic unpack for erofs with snapshot labels.
+	sb2Cfg := PodSandboxConfig("erofs-pod", "runtime-handler-unpack")
+	sb2ID, err := rSvc.RunPodSandbox(sb2Cfg, "erofs")
+	require.NoError(t, err)
+	cn2Cfg := ContainerConfig("erofs-container", nginxImage, WithCommand("sleep", "1d"))
+	cn2ID, err := rSvc.CreateContainer(sb2ID, cn2Cfg, sb2Cfg)
+	require.NoError(t, err)
+
+	for _, chainID := range chainIDs {
+		snInfo, err := erofsSn.Stat(context.Background(), chainID.String())
+		require.NoErrorf(t, err, "failed to stat erofs snapshot for chainID %s", chainID)
+		require.NotNil(t, snInfo.Labels)
+		assert.NotEmpty(t, snInfo.Labels[snpkg.TargetRefLabel], "missing %s on chainID %s", snpkg.TargetRefLabel, chainID)
+		assert.NotEmpty(t, snInfo.Labels[snpkg.TargetManifestDigestLabel], "missing %s on chainID %s", snpkg.TargetManifestDigestLabel, chainID)
+		assert.NotEmpty(t, snInfo.Labels[snpkg.TargetLayerDigestLabel], "missing %s on chainID %s", snpkg.TargetLayerDigestLabel, chainID)
+		assert.NotEmpty(t, snInfo.Labels[snpkg.TargetImageLayersLabel], "missing %s on chainID %s", snpkg.TargetImageLayersLabel, chainID)
+	}
+
+	// Make sure the second pod really used the erofs runtime handler path.
+	sb2Status, err := rSvc.PodSandboxStatus(sb2ID)
+	require.NoError(t, err)
+	assert.Equal(t, "erofs", sb2Status.RuntimeHandler)
+
+	// Container should be created.
+	cn2Status, err := rSvc.ContainerStatus(cn2ID)
+	require.NoError(t, err)
+	assert.Equal(t, criruntime.ContainerState_CONTAINER_CREATED, cn2Status.State)
+}

internal/cri/opts/container.go

@@ -24,19 +24,24 @@ import (
 	"strings"
 
 	"github.com/containerd/continuity/fs"
+	"github.com/containerd/errdefs"
+	"github.com/containerd/log"
+	"github.com/containerd/platforms"
 	imagespec "github.com/opencontainers/image-spec/specs-go/v1"
+	"golang.org/x/sync/semaphore"
 
 	containerd "github.com/containerd/containerd/v2/client"
 	"github.com/containerd/containerd/v2/core/containers"
+	"github.com/containerd/containerd/v2/core/images"
 	"github.com/containerd/containerd/v2/core/mount"
 	"github.com/containerd/containerd/v2/core/snapshots"
-	"github.com/containerd/errdefs"
-	"github.com/containerd/log"
+	"github.com/containerd/containerd/v2/core/unpack"
+	snpkg "github.com/containerd/containerd/v2/pkg/snapshotters"
 )
 
 // WithNewSnapshot wraps `containerd.WithNewSnapshot` so that if creating the
 // snapshot fails we make sure the image is actually unpacked and retry.
-func WithNewSnapshot(id string, i containerd.Image, opts ...snapshots.Opt) containerd.NewContainerOpts {
+func WithNewSnapshot(id string, i containerd.Image, appendSnapshotLabels bool, opts ...snapshots.Opt) containerd.NewContainerOpts {
 	f := containerd.WithNewSnapshot(id, i, opts...)
 	return func(ctx context.Context, client *containerd.Client, c *containers.Container) error {
 		if err := f(ctx, client, c); err != nil {
@@ -44,7 +49,7 @@ func WithNewSnapshot(id string, i containerd.Image, opts ...snapshots.Opt) conta
 				return err
 			}
 
-			if err := i.Unpack(ctx, c.Snapshotter); err != nil {
+			if err := unpackImage(ctx, client, i, c.Snapshotter, appendSnapshotLabels); err != nil {
 				return fmt.Errorf("error unpacking image: %w", err)
 			}
 			return f(ctx, client, c)
@@ -53,6 +58,55 @@ func WithNewSnapshot(id string, i containerd.Image, opts ...snapshots.Opt) conta
 	}
 }
 
+func unpackImage(ctx context.Context, client *containerd.Client, i containerd.Image, snapshotter string, appendSnapshotLabels bool) error {
+	ctx, done, err := client.WithLease(ctx)
+	if err != nil {
+		return err
+	}
+	defer done(ctx)
+
+	matcher := platforms.Default()
+
+	capabilities, err := client.GetSnapshotterCapabilities(ctx, snapshotter)
+	if err != nil {
+		return err
+	}
+
+	u, err := unpack.NewUnpacker(
+		ctx,
+		i.ContentStore(),
+		unpack.WithUnpackPlatform(unpack.Platform{
+			Platform:                matcher,
+			SnapshotterKey:          snapshotter,
+			Snapshotter:             client.SnapshotService(snapshotter),
+			Applier:                 client.DiffService(),
+			SnapshotterCapabilities: capabilities,
+		}),
+		unpack.WithUnpackLimiter(semaphore.NewWeighted(3)),
+	)
+	if err != nil {
+		return fmt.Errorf("unable to initialize unpacker: %w", err)
+	}
+
+	childrenHandler := images.ChildrenHandler(i.ContentStore())
+	childrenHandler = images.FilterPlatforms(childrenHandler, matcher)
+	childrenHandler = images.LimitManifests(childrenHandler, matcher, 1)
+
+	var h images.Handler = childrenHandler
+	if appendSnapshotLabels {
+		h = snpkg.AppendInfoHandlerWrapper(i.Name())(h)
+	}
+
+	if err := images.Dispatch(ctx, u.Unpack(h), nil, i.Target()); err != nil {
+		_, _ = u.Wait()
+		return err
+	}
+	if _, err := u.Wait(); err != nil {
+		return err
+	}
+	return nil
+}
+
 // WithVolumes copies ownership of volume in rootfs to its corresponding host path.
 // It doesn't update runtime spec.
 // The passed in map is a host path to container path map for all volumes.

internal/cri/server/container_create.go

@@ -357,7 +357,7 @@ func (c *criService) createContainer(r *createContainerRequest) (_ string, retEr
 		// the runtime (runc) a chance to modify (e.g. to create mount
 		// points corresponding to spec.Mounts) before making the
 		// rootfs readonly (requested by spec.Root.Readonly).
-		customopts.WithNewSnapshot(r.containerID, *r.containerdImage, sOpts...),
+		customopts.WithNewSnapshot(r.containerID, *r.containerdImage, !c.ImageService.Config().DisableSnapshotAnnotations, sOpts...),
 	}
 	if len(volumeMounts) > 0 {
 		mountMap := make(map[string]string)

internal/cri/server/podsandbox/sandbox_run.go

@@ -194,7 +194,7 @@ func (c *Controller) Start(ctx context.Context, id string) (cin sandbox.Controll
 
 	opts := []containerd.NewContainerOpts{
 		containerd.WithSnapshotter(sandboxSnapshotter),
-		customopts.WithNewSnapshot(id, pauseImage, snapshotterOpt...),
+		customopts.WithNewSnapshot(id, pauseImage, !c.imageConfig.DisableSnapshotAnnotations, snapshotterOpt...),
 		containerd.WithSpec(spec, specOpts...),
 		containerd.WithContainerLabels(sandboxLabels),
 		containerd.WithContainerExtension(crilabels.SandboxMetadataExtension, &metadata),