diff --git a/daemon/containerd/fake_service_test.go b/daemon/containerd/fake_service_test.go new file mode 100644 index 0000000000..f615021cb5 --- /dev/null +++ b/daemon/containerd/fake_service_test.go @@ -0,0 +1,290 @@ +package containerd + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/containerd/containerd" + "github.com/containerd/containerd/content" + "github.com/containerd/containerd/leases" + "github.com/containerd/containerd/metadata" + "github.com/containerd/containerd/snapshots" + cerrdefs "github.com/containerd/errdefs" + "github.com/docker/docker/container" + daemonevents "github.com/docker/docker/daemon/events" + "github.com/opencontainers/go-digest" + ocispec "github.com/opencontainers/image-spec/specs-go/v1" + "gotest.tools/v3/assert" +) + +func fakeImageService(t testing.TB, ctx context.Context, cs content.Store) *ImageService { + snapshotter := &testSnapshotterService{} + + mdb := newTestDB(ctx, t) + + snapshotters := map[string]snapshots.Snapshotter{ + containerd.DefaultSnapshotter: snapshotter, + } + + service := &ImageService{ + images: metadata.NewImageStore(mdb), + containers: container.NewMemoryStore(), + content: cs, + eventsService: daemonevents.New(), + snapshotterServices: snapshotters, + snapshotter: containerd.DefaultSnapshotter, + } + + // containerd.Image gets the services directly from containerd.Client + // so we need to create a "fake" containerd.Client with the test services. + c8dCli, err := containerd.New("", containerd.WithServices( + containerd.WithImageStore(service.images), + containerd.WithContentStore(cs), + containerd.WithSnapshotters(snapshotters), + containerd.WithLeasesService(noopLeasesManager{}), + )) + assert.NilError(t, err) + + service.client = c8dCli + return service +} + +type noopLeasesManager struct{} + +func (noopLeasesManager) Create(context.Context, ...leases.Opt) (leases.Lease, error) { + return leases.Lease{}, nil +} + +func (noopLeasesManager) Delete(context.Context, leases.Lease, ...leases.DeleteOpt) error { + return nil +} + +func (noopLeasesManager) List(context.Context, ...string) ([]leases.Lease, error) { + return nil, nil +} + +func (noopLeasesManager) AddResource(context.Context, leases.Lease, leases.Resource) error { + return nil +} + +func (noopLeasesManager) DeleteResource(context.Context, leases.Lease, leases.Resource) error { + return nil +} + +func (noopLeasesManager) ListResources(context.Context, leases.Lease) ([]leases.Resource, error) { + return nil, nil +} + +type blobsDirContentStore struct { + blobs string +} + +type fileReaderAt struct { + *os.File +} + +func (f *fileReaderAt) Size() int64 { + fi, err := f.Stat() + if err != nil { + return -1 + } + return fi.Size() +} + +func (s *blobsDirContentStore) ReaderAt(ctx context.Context, desc ocispec.Descriptor) (content.ReaderAt, error) { + p := filepath.Join(s.blobs, desc.Digest.Encoded()) + r, err := os.Open(p) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, fmt.Errorf("%w: %s", cerrdefs.ErrNotFound, desc.Digest) + } + return nil, err + } + return &fileReaderAt{r}, nil +} + +func (s *blobsDirContentStore) Writer(ctx context.Context, opts ...content.WriterOpt) (content.Writer, error) { + return nil, fmt.Errorf("read-only") +} + +func (s *blobsDirContentStore) Status(ctx context.Context, _ string) (content.Status, error) { + return content.Status{}, fmt.Errorf("not implemented") +} + +func (s *blobsDirContentStore) Delete(ctx context.Context, dgst digest.Digest) error { + p := filepath.Join(s.blobs, dgst.Encoded()) + return os.Remove(p) +} + +func (s *blobsDirContentStore) ListStatuses(ctx context.Context, filters ...string) ([]content.Status, error) { + return nil, nil +} + +func (s *blobsDirContentStore) Abort(ctx context.Context, ref string) error { + return fmt.Errorf("not implemented") +} + +func (s *blobsDirContentStore) Walk(ctx context.Context, fn content.WalkFunc, filters ...string) error { + entries, err := os.ReadDir(s.blobs) + if err != nil { + return err + } + + for _, e := range entries { + if e.IsDir() { + continue + } + + d := digest.FromString(e.Name()) + if d == "" { + continue + } + + stat, err := e.Info() + if err != nil { + return err + } + + if err := fn(content.Info{Digest: d, Size: stat.Size()}); err != nil { + return err + } + } + + return nil +} + +func (s *blobsDirContentStore) Info(ctx context.Context, dgst digest.Digest) (content.Info, error) { + f, err := os.Open(filepath.Join(s.blobs, dgst.Encoded())) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return content.Info{}, fmt.Errorf("%w: %s", cerrdefs.ErrNotFound, dgst) + } + return content.Info{}, err + } + defer f.Close() + + stat, err := f.Stat() + if err != nil { + return content.Info{}, err + } + + return content.Info{ + Digest: dgst, + Size: stat.Size(), + }, nil +} + +func (s *blobsDirContentStore) Update(ctx context.Context, info content.Info, fieldpaths ...string) (content.Info, error) { + return content.Info{}, fmt.Errorf("read-only") +} + +// delayedStore is a content store wrapper that adds a constant delay to all +// operations in order to imitate gRPC overhead. +// +// The delay is constant to make the benchmark results more reproducible +// Since content store may be accessed concurrently random delay would be +// order-dependent. +type delayedStore struct { + store content.Store + overhead time.Duration +} + +func (s *delayedStore) delay() { + time.Sleep(s.overhead) +} + +func (s *delayedStore) ReaderAt(ctx context.Context, desc ocispec.Descriptor) (content.ReaderAt, error) { + s.delay() + return s.store.ReaderAt(ctx, desc) +} + +func (s *delayedStore) Writer(ctx context.Context, opts ...content.WriterOpt) (content.Writer, error) { + s.delay() + return s.store.Writer(ctx, opts...) +} + +func (s *delayedStore) Status(ctx context.Context, st string) (content.Status, error) { + s.delay() + return s.store.Status(ctx, st) +} + +func (s *delayedStore) Delete(ctx context.Context, dgst digest.Digest) error { + s.delay() + return s.store.Delete(ctx, dgst) +} + +func (s *delayedStore) ListStatuses(ctx context.Context, filters ...string) ([]content.Status, error) { + s.delay() + return s.store.ListStatuses(ctx, filters...) +} + +func (s *delayedStore) Abort(ctx context.Context, ref string) error { + s.delay() + return s.store.Abort(ctx, ref) +} + +func (s *delayedStore) Walk(ctx context.Context, fn content.WalkFunc, filters ...string) error { + s.delay() + return s.store.Walk(ctx, fn, filters...) +} + +func (s *delayedStore) Info(ctx context.Context, dgst digest.Digest) (content.Info, error) { + s.delay() + return s.store.Info(ctx, dgst) +} + +func (s *delayedStore) Update(ctx context.Context, info content.Info, fieldpaths ...string) (content.Info, error) { + s.delay() + return s.store.Update(ctx, info, fieldpaths...) +} + +type memoryLabelStore struct { + mu sync.Mutex + labels map[digest.Digest]map[string]string +} + +// Get returns all the labels for the given digest +func (s *memoryLabelStore) Get(dgst digest.Digest) (map[string]string, error) { + s.mu.Lock() + labels := s.labels[dgst] + s.mu.Unlock() + return labels, nil +} + +// Set sets all the labels for a given digest +func (s *memoryLabelStore) Set(dgst digest.Digest, labels map[string]string) error { + s.mu.Lock() + if s.labels == nil { + s.labels = make(map[digest.Digest]map[string]string) + } + s.labels[dgst] = labels + s.mu.Unlock() + return nil +} + +// Update replaces the given labels for a digest, +// a key with an empty value removes a label. +func (s *memoryLabelStore) Update(dgst digest.Digest, update map[string]string) (map[string]string, error) { + s.mu.Lock() + defer s.mu.Unlock() + + labels, ok := s.labels[dgst] + if !ok { + labels = map[string]string{} + } + for k, v := range update { + labels[k] = v + } + if s.labels == nil { + s.labels = map[digest.Digest]map[string]string{} + } + s.labels[dgst] = labels + + return labels, nil +} diff --git a/daemon/containerd/image_exporter.go b/daemon/containerd/image_exporter.go index a27fd2c6fe..1fd877e047 100644 --- a/daemon/containerd/image_exporter.go +++ b/daemon/containerd/image_exporter.go @@ -256,8 +256,6 @@ func (i *ImageService) LoadImage(ctx context.Context, inTar io.ReadCloser, platf opts := []containerd.ImportOpt{ containerd.WithImportPlatform(pm), - containerd.WithSkipMissing(), - // Create an additional image with dangling name for imported images... containerd.WithDigestRef(danglingImageName), // ... but only if they don't have a name or it's invalid. @@ -270,13 +268,51 @@ func (i *ImageService) LoadImage(ctx context.Context, inTar io.ReadCloser, platf }), } + if platform == nil { + // Allow variants to be missing if no specific platform is requested. + opts = append(opts, containerd.WithSkipMissing()) + } + imgs, err := i.client.Import(ctx, decompressed, opts...) if err != nil { + if platform != nil { + p := platforms.FormatAll(*platform) + log.G(ctx).WithFields(log.Fields{"error": err, "platform": p}).Debug("failed to import image to containerd") + + // Note: ErrEmptyWalk will not be returned in most cases as + // index.json will contain a descriptor of the actual OCI index or + // Docker manifest list, so the walk is never empty. + // Even in case of a single-platform image, the manifest descriptor + // doesn't have a platform set, so it won't be filtered out by the + // FilterPlatform containerd handler. + if errors.Is(err, containerdimages.ErrEmptyWalk) { + return errdefs.NotFound(errors.Wrapf(err, "requested platform (%s) not found", p)) + } + if cerrdefs.IsNotFound(err) { + return errdefs.NotFound(errors.Wrapf(err, "requested platform (%s) found, but some content is missing", p)) + } + } log.G(ctx).WithError(err).Debug("failed to import image to containerd") return errdefs.System(err) } + if platform != nil { + // Verify that the requested platform is available for the loaded images. + // While the ideal behavior here would be to verify whether the input + // archive actually supplied them, we're not able to determine that + // as the imported index is not returned by the import operation. + if err := i.verifyImagesProvidePlatform(ctx, imgs, *platform, pm); err != nil { + return err + } + } + progress := streamformatter.NewStdoutWriter(outStream) + // Unpack only an image of the host platform + unpackPm := i.hostPlatformMatcher() + // If a load of specific platform is requested, unpack it + if platform != nil { + unpackPm = pm + } for _, img := range imgs { name := img.Name @@ -310,8 +346,7 @@ func (i *ImageService) LoadImage(ctx context.Context, inTar io.ReadCloser, platf return nil } - // Only unpack the image if it matches the host platform - if !i.hostPlatformMatcher().Match(imgPlat) { + if !unpackPm.Match(imgPlat) { return nil } @@ -342,3 +377,57 @@ func (i *ImageService) LoadImage(ctx context.Context, inTar io.ReadCloser, platf return nil } + +// verifyImagesProvidePlatform checks if the requested platform is loaded. +// If the requested platform is not loaded, it returns an error. +func (i *ImageService) verifyImagesProvidePlatform(ctx context.Context, imgs []containerdimages.Image, platform ocispec.Platform, pm platforms.Matcher) error { + if len(imgs) == 0 { + return errdefs.NotFound(fmt.Errorf("no images providing the requested platform %s found", platforms.FormatAll(platform))) + } + var incompleteImgs []string + for _, img := range imgs { + hasRequestedPlatform := false + err := i.walkImageManifests(ctx, img, func(platformImg *ImageManifest) error { + imgPlat, err := platformImg.ImagePlatform(ctx) + if err != nil { + if cerrdefs.IsNotFound(err) { + return nil + } + return errors.Wrapf(err, "failed to determine image platform") + } + + if !pm.Match(imgPlat) { + return nil + } + available, err := platformImg.CheckContentAvailable(ctx) + if err != nil { + return errors.Wrapf(err, "failed to determine image content availability for platform %s", platforms.FormatAll(platform)) + } + + if available { + hasRequestedPlatform = true + return nil + } + return nil + }) + if err != nil { + return errdefs.System(err) + } + if !hasRequestedPlatform { + incompleteImgs = append(incompleteImgs, imageFamiliarName(img)) + } + } + + msg := "" + switch len(incompleteImgs) { + case 0: + // Success - All images provide the requested platform. + return nil + case 1: + msg = "image %s was loaded, but doesn't provide the requested platform (%s)" + default: + msg = "images [%s] were loaded, but don't provide the requested platform (%s)" + } + + return errdefs.NotFound(fmt.Errorf(msg, strings.Join(incompleteImgs, ", "), platforms.FormatAll(platform))) +} diff --git a/daemon/containerd/image_list_test.go b/daemon/containerd/image_list_test.go index 6a8a420cda..a1f0418fd9 100644 --- a/daemon/containerd/image_list_test.go +++ b/daemon/containerd/image_list_test.go @@ -2,10 +2,7 @@ package containerd import ( "context" - "errors" - "fmt" "math/rand" - "os" "path/filepath" "slices" "sort" @@ -13,18 +10,13 @@ import ( "testing" "time" - "github.com/containerd/containerd" "github.com/containerd/containerd/content" "github.com/containerd/containerd/images" - "github.com/containerd/containerd/metadata" "github.com/containerd/containerd/namespaces" - "github.com/containerd/containerd/snapshots" - cerrdefs "github.com/containerd/errdefs" "github.com/containerd/log/logtest" "github.com/containerd/platforms" imagetypes "github.com/docker/docker/api/types/image" "github.com/docker/docker/container" - daemonevents "github.com/docker/docker/daemon/events" "github.com/docker/docker/internal/testutils/specialimage" "github.com/opencontainers/go-digest" ocispec "github.com/opencontainers/image-spec/specs-go/v1" @@ -340,197 +332,3 @@ func TestImageList(t *testing.T) { }) } } - -func fakeImageService(t testing.TB, ctx context.Context, cs content.Store) *ImageService { - snapshotter := &testSnapshotterService{} - - mdb := newTestDB(ctx, t) - - snapshotters := map[string]snapshots.Snapshotter{ - containerd.DefaultSnapshotter: snapshotter, - } - - service := &ImageService{ - images: metadata.NewImageStore(mdb), - containers: container.NewMemoryStore(), - content: cs, - eventsService: daemonevents.New(), - snapshotterServices: snapshotters, - snapshotter: containerd.DefaultSnapshotter, - } - - // containerd.Image gets the services directly from containerd.Client - // so we need to create a "fake" containerd.Client with the test services. - c8dCli, err := containerd.New("", containerd.WithServices( - containerd.WithImageStore(service.images), - containerd.WithContentStore(cs), - containerd.WithSnapshotters(snapshotters), - )) - assert.NilError(t, err) - - service.client = c8dCli - return service -} - -type blobsDirContentStore struct { - blobs string -} - -type fileReaderAt struct { - *os.File -} - -func (f *fileReaderAt) Size() int64 { - fi, err := f.Stat() - if err != nil { - return -1 - } - return fi.Size() -} - -func (s *blobsDirContentStore) ReaderAt(ctx context.Context, desc ocispec.Descriptor) (content.ReaderAt, error) { - p := filepath.Join(s.blobs, desc.Digest.Encoded()) - r, err := os.Open(p) - if err != nil { - if errors.Is(err, os.ErrNotExist) { - return nil, fmt.Errorf("%w: %s", cerrdefs.ErrNotFound, desc.Digest) - } - return nil, err - } - return &fileReaderAt{r}, nil -} - -func (s *blobsDirContentStore) Writer(ctx context.Context, opts ...content.WriterOpt) (content.Writer, error) { - return nil, fmt.Errorf("read-only") -} - -func (s *blobsDirContentStore) Status(ctx context.Context, _ string) (content.Status, error) { - return content.Status{}, fmt.Errorf("not implemented") -} - -func (s *blobsDirContentStore) Delete(ctx context.Context, dgst digest.Digest) error { - p := filepath.Join(s.blobs, dgst.Encoded()) - return os.Remove(p) -} - -func (s *blobsDirContentStore) ListStatuses(ctx context.Context, filters ...string) ([]content.Status, error) { - return nil, nil -} - -func (s *blobsDirContentStore) Abort(ctx context.Context, ref string) error { - return fmt.Errorf("not implemented") -} - -func (s *blobsDirContentStore) Walk(ctx context.Context, fn content.WalkFunc, filters ...string) error { - entries, err := os.ReadDir(s.blobs) - if err != nil { - return err - } - - for _, e := range entries { - if e.IsDir() { - continue - } - - d := digest.FromString(e.Name()) - if d == "" { - continue - } - - stat, err := e.Info() - if err != nil { - return err - } - - if err := fn(content.Info{Digest: d, Size: stat.Size()}); err != nil { - return err - } - } - - return nil -} - -func (s *blobsDirContentStore) Info(ctx context.Context, dgst digest.Digest) (content.Info, error) { - f, err := os.Open(filepath.Join(s.blobs, dgst.Encoded())) - if err != nil { - if errors.Is(err, os.ErrNotExist) { - return content.Info{}, fmt.Errorf("%w: %s", cerrdefs.ErrNotFound, dgst) - } - return content.Info{}, err - } - defer f.Close() - - stat, err := f.Stat() - if err != nil { - return content.Info{}, err - } - - return content.Info{ - Digest: dgst, - Size: stat.Size(), - }, nil -} - -func (s *blobsDirContentStore) Update(ctx context.Context, info content.Info, fieldpaths ...string) (content.Info, error) { - return content.Info{}, fmt.Errorf("read-only") -} - -// delayedStore is a content store wrapper that adds a constant delay to all -// operations in order to imitate gRPC overhead. -// -// The delay is constant to make the benchmark results more reproducible -// Since content store may be accessed concurrently random delay would be -// order-dependent. -type delayedStore struct { - store content.Store - overhead time.Duration -} - -func (s *delayedStore) delay() { - time.Sleep(s.overhead) -} - -func (s *delayedStore) ReaderAt(ctx context.Context, desc ocispec.Descriptor) (content.ReaderAt, error) { - s.delay() - return s.store.ReaderAt(ctx, desc) -} - -func (s *delayedStore) Writer(ctx context.Context, opts ...content.WriterOpt) (content.Writer, error) { - s.delay() - return s.store.Writer(ctx, opts...) -} - -func (s *delayedStore) Status(ctx context.Context, st string) (content.Status, error) { - s.delay() - return s.store.Status(ctx, st) -} - -func (s *delayedStore) Delete(ctx context.Context, dgst digest.Digest) error { - s.delay() - return s.store.Delete(ctx, dgst) -} - -func (s *delayedStore) ListStatuses(ctx context.Context, filters ...string) ([]content.Status, error) { - s.delay() - return s.store.ListStatuses(ctx, filters...) -} - -func (s *delayedStore) Abort(ctx context.Context, ref string) error { - s.delay() - return s.store.Abort(ctx, ref) -} - -func (s *delayedStore) Walk(ctx context.Context, fn content.WalkFunc, filters ...string) error { - s.delay() - return s.store.Walk(ctx, fn, filters...) -} - -func (s *delayedStore) Info(ctx context.Context, dgst digest.Digest) (content.Info, error) { - s.delay() - return s.store.Info(ctx, dgst) -} - -func (s *delayedStore) Update(ctx context.Context, info content.Info, fieldpaths ...string) (content.Info, error) { - s.delay() - return s.store.Update(ctx, info, fieldpaths...) -} diff --git a/daemon/containerd/image_load_test.go b/daemon/containerd/image_load_test.go new file mode 100644 index 0000000000..f68b2c680a --- /dev/null +++ b/daemon/containerd/image_load_test.go @@ -0,0 +1,111 @@ +package containerd + +import ( + "bytes" + "context" + "math/rand" + "os" + "path/filepath" + "testing" + + "github.com/containerd/containerd/content" + "github.com/containerd/containerd/content/local" + "github.com/containerd/containerd/namespaces" + "github.com/containerd/platforms" + "github.com/docker/docker/errdefs" + "github.com/docker/docker/internal/testutils/specialimage" + "github.com/docker/docker/pkg/archive" + ocispec "github.com/opencontainers/image-spec/specs-go/v1" + "gotest.tools/v3/assert" + is "gotest.tools/v3/assert/cmp" +) + +func TestImageLoadMissing(t *testing.T) { + linuxAmd64 := ocispec.Platform{OS: "linux", Architecture: "amd64"} + linuxArm64 := ocispec.Platform{OS: "linux", Architecture: "arm64"} + linuxArmv5 := ocispec.Platform{OS: "linux", Architecture: "arm", Variant: "v5"} + + ctx := namespaces.WithNamespace(context.TODO(), "testing-"+t.Name()) + + store, err := local.NewLabeledStore(t.TempDir(), &memoryLabelStore{}) + assert.NilError(t, err) + + imgSvc := fakeImageService(t, ctx, store) + // Mock the daemon platform. + imgSvc.defaultPlatformOverride = platforms.Only(linuxAmd64) + + tryLoad := func(ctx context.Context, t *testing.T, dir string, platform ocispec.Platform) error { + tarRc, err := archive.Tar(dir, archive.Uncompressed) + assert.NilError(t, err) + defer tarRc.Close() + + buf := bytes.Buffer{} + + defer func() { + t.Log(buf.String()) + }() + + return imgSvc.LoadImage(ctx, tarRc, &platform, &buf, true) + } + + clearStore := func(ctx context.Context, t *testing.T) { + assert.NilError(t, store.Walk(ctx, func(info content.Info) error { + return store.Delete(ctx, info.Digest) + }), "failed to delete all content") + } + + t.Run("empty index", func(t *testing.T) { + imgDataDir := t.TempDir() + _, err := specialimage.EmptyIndex(imgDataDir) + assert.NilError(t, err) + + err = tryLoad(ctx, t, imgDataDir, linuxAmd64) + assert.Check(t, is.Error(err, "image emptyindex:latest was loaded, but doesn't provide the requested platform (linux/amd64)")) + assert.Check(t, is.ErrorType(err, errdefs.IsNotFound)) + }) + clearStore(ctx, t) + + t.Run("single platform", func(t *testing.T) { + imgDataDir := t.TempDir() + r := rand.NewSource(0x9127371238) + _, err := specialimage.RandomSinglePlatform(imgDataDir, linuxAmd64, r) + assert.NilError(t, err) + + err = tryLoad(ctx, t, imgDataDir, linuxArm64) + assert.Check(t, is.ErrorContains(err, "doesn't provide the requested platform (linux/arm64)")) + assert.Check(t, is.ErrorType(err, errdefs.IsNotFound)) + }) + + clearStore(ctx, t) + + t.Run("2 platform image", func(t *testing.T) { + imgDataDir := t.TempDir() + _, mfstDescs, err := specialimage.MultiPlatform(imgDataDir, "multiplatform:latest", []ocispec.Platform{linuxAmd64, linuxArm64}) + assert.NilError(t, err) + + t.Run("platform not included in index", func(t *testing.T) { + err = tryLoad(ctx, t, imgDataDir, linuxArmv5) + assert.Check(t, is.Error(err, "image multiplatform:latest was loaded, but doesn't provide the requested platform (linux/arm/v5)")) + assert.Check(t, is.ErrorType(err, errdefs.IsNotFound)) + }) + + clearStore(ctx, t) + + t.Run("platform blobs missing", func(t *testing.T) { + // Assumption: arm64 image is second in the index (implementation detail of specialimage.MultiPlatform) + mfstDesc := mfstDescs[1] + assert.Assert(t, mfstDesc.Platform.Architecture == linuxArm64.Architecture) + assert.Assert(t, mfstDesc.Platform.Variant == linuxArm64.Variant) + + t.Log(mfstDesc.Digest) + + // Delete arm64 manifest + mfstPath := filepath.Join(imgDataDir, "blobs/sha256", mfstDesc.Digest.Encoded()) + assert.NilError(t, os.Remove(mfstPath)) + + err = tryLoad(ctx, t, imgDataDir, linuxArm64) + assert.Check(t, is.ErrorContains(err, "requested platform (linux/arm64) found, but some content is missing")) + assert.Check(t, is.ErrorType(err, errdefs.IsNotFound)) + }) + }) +}