diff --git a/dev-shell.sh b/dev-shell.sh index dc2a40f..13bfaf5 100755 --- a/dev-shell.sh +++ b/dev-shell.sh @@ -1,7 +1,7 @@ down() { docker compose --env-file=test/compose/.env.test -f test/compose/docker-compose.yml down --remove-orphans # remove the docker runtime and compose app store volumes - docker volume rm compose_docker-data compose_docker-runtime compose_reset-apps \ + docker volume rm compose_docker-data compose_docker-runtime \ compose_publisher-data compose_publisher-runtime } diff --git a/docs/containerd-install-linkin-progress-plan-2026-10-19.md b/docs/containerd-install-linkin-progress-plan-2026-10-19.md new file mode 100644 index 0000000..4be67c2 --- /dev/null +++ b/docs/containerd-install-linkin-progress-plan-2026-10-19.md @@ -0,0 +1,26 @@ +# Install into containerd: link the layers in + +The containerd install (`containerd-install-plan-2026-10-08.md`) no longer copies the +compressed layers from composectl's store into containerd's content store. Each layer +is hardlinked instead, so an install writes nothing but the snapshots. + +## How + +1. Check that the store blob exists and has the layer's size; otherwise the install + fails. +2. Hardlink it to `/blobs//`. A file already there is + containerd's own blob or an earlier link and is used as it is. +3. Register it under the install lease. With containerd's default `shared` content + policy a writer for a blob already at its path copies nothing and does not hash it. + The blob was verified when it was pulled, and the unpack checks every extracted + layer against the image's diff IDs. If containerd's blob cleanup removed the file + in between, the layer is linked and registered once more; a second miss fails the + install. + +Deleting the lease lets containerd's GC unlink its name only; the store keeps its own. + +## Limit + +The store and the content store must be on one filesystem: a link across filesystems +fails the install, naming both paths. The test environment keeps its store on the test +dockerd's data volume for that. diff --git a/pkg/compose/ctrd_install.go b/pkg/compose/ctrd_install.go index 45db92f..3d560bf 100644 --- a/pkg/compose/ctrd_install.go +++ b/pkg/compose/ctrd_install.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "io/fs" "os" "path/filepath" "strings" @@ -25,7 +26,8 @@ import ( ) // Install of app images straight into the containerd behind dockerd: the compressed layers are -// streamed under a short lease and dropped by containerd's GC once the lease is deleted. +// hardlinked into containerd's content store under a short lease and dropped by containerd's GC +// once the lease is deleted. const ( // MinDockerVersionForCtrdInstall is the oldest dockerd on which `docker image ls` lists an image @@ -39,6 +41,9 @@ const ( gcRefManifestLabel = "containerd.io/gc.ref.content.m.0" // recordCleanupTimeout bounds the removal of partial image records after a failed write. recordCleanupTimeout = 30 * time.Second + // ingestAbortAttempts and ingestAbortDelay bound the retries of a failed ingest abort. + ingestAbortAttempts = 20 + ingestAbortDelay = 50 * time.Millisecond ) type ( @@ -165,7 +170,7 @@ func pluginRoot(ctx context.Context, c *containerd.Client, pluginType, id string return resp.Plugins[0].Exports["root"], nil } -// installImagesCtrd installs the images of one app into dockerd's containerd. Under a lease it streams +// installImagesCtrd installs the images of one app into dockerd's containerd. Under a lease it links // the layers that have no snapshot yet, unpacks every image and syncs the snapshotter's filesystem, // and only then creates the image records: a record never points at a snapshot that a power cut // can still lose. Deleting the lease lets containerd's GC drop the compressed layers. @@ -234,9 +239,10 @@ func deleteInstallLeases(ctx context.Context, ls leases.Manager) error { return nil } -// installImage streams what image needs and unpacks it. If a base snapshot went away between the -// scan and the unpack (for example `docker image prune` removed the image that held it), the unpack -// fails on the missing layer blob; then every layer is streamed and the unpack is run once more. +// installImage puts what the image needs into the content store and unpacks it. If a base snapshot +// went away between the scan and the unpack (for example `docker image prune` removed the image that +// held it), the unpack fails on the missing layer blob; then every layer is linked and the unpack is +// run once more. func (c *ctrdClient) installImage(ctx context.Context, img *ctrdImage, blobsRoot string, platform platforms.MatchComparer, report ProgressCallbackFn) error { first, err := c.firstMissingSnapshot(ctx, img.chainIDs) @@ -244,8 +250,8 @@ func (c *ctrdClient) installImage(ctx context.Context, img *ctrdImage, blobsRoot return fmt.Errorf("failed to scan the snapshots: %w", err) } reportImageState(report, ImageLoadStateImageLoading, img.ref) - if err := c.streamImage(ctx, img, blobsRoot, first); err != nil { - return fmt.Errorf("failed to stream the image to containerd: %w", err) + if err := c.putImage(ctx, img, blobsRoot, first); err != nil { + return fmt.Errorf("failed to put the image into containerd: %w", err) } err = c.unpack(ctx, img, platform) if err == nil { @@ -254,17 +260,17 @@ func (c *ctrdClient) installImage(ctx context.Context, img *ctrdImage, blobsRoot if !errdefs.IsNotFound(err) { return fmt.Errorf("failed to unpack: %w", err) } - if err := c.streamLayers(ctx, img.layers, blobsRoot); err != nil { - return fmt.Errorf("failed to stream every layer to containerd after a snapshot went away: %w", err) + if err := c.linkLayers(ctx, img.layers, blobsRoot); err != nil { + return fmt.Errorf("failed to link every layer into containerd after a snapshot went away: %w", err) } if err := c.unpack(ctx, img, platform); err != nil { - return fmt.Errorf("failed to unpack after streaming every layer: %w", err) + return fmt.Errorf("failed to unpack after linking every layer: %w", err) } return nil } // firstMissingSnapshot returns the index of the first layer whose snapshot does not exist; every -// layer after it has to be streamed and unpacked too, because a snapshot needs its parent. +// layer after it has to be linked and unpacked too, because a snapshot needs its parent. func (c *ctrdClient) firstMissingSnapshot(ctx context.Context, chainIDs []digest.Digest) (int, error) { sn := c.SnapshotService(c.snapshotter) for i, chainID := range chainIDs { @@ -279,11 +285,11 @@ func (c *ctrdClient) firstMissingSnapshot(ctx context.Context, chainIDs []digest return len(chainIDs), nil } -// streamImage writes the layers from first on, the config, the manifest and, when the image is -// pinned by one, the index. The manifest gets no layer labels: only the lease holds the layers. -// WriteBlob leaves the labels of a blob that already exists as they are. -func (c *ctrdClient) streamImage(ctx context.Context, img *ctrdImage, blobsRoot string, first int) error { - if err := c.streamLayers(ctx, img.layers[first:], blobsRoot); err != nil { +// putImage puts the layers from first on, the config, the manifest and, when the image is pinned +// by one, the index into the content store. The manifest gets no layer labels: only the lease holds +// the layers. WriteBlob leaves the labels of a blob that already exists as they are. +func (c *ctrdClient) putImage(ctx context.Context, img *ctrdImage, blobsRoot string, first int) error { + if err := c.linkLayers(ctx, img.layers[first:], blobsRoot); err != nil { return err } if err := c.writeBlob(ctx, blobsRoot, img.config, nil); err != nil { @@ -306,15 +312,130 @@ func (c *ctrdClient) streamImage(ctx context.Context, img *ctrdImage, blobsRoot return err } -func (c *ctrdClient) streamLayers(ctx context.Context, layers []ocispec.Descriptor, blobsRoot string) error { +// linkLayers puts the layers into the content store without copying them: each store blob is +// hardlinked to the path the content store keeps it at, and then registered in place. +func (c *ctrdClient) linkLayers(ctx context.Context, layers []ocispec.Descriptor, blobsRoot string) error { for _, layer := range layers { - if err := c.writeBlob(ctx, blobsRoot, layer, nil); err != nil { + if err := linkLayer(c.contentRoot, blobsRoot, layer); err != nil { return err } + if err := c.registerLayer(ctx, blobsRoot, layer); err != nil { + return err + } + } + return nil +} + +// registerLayer records a layer whose file is at the content store's path for it, without reading +// it: the writer the content store opens for such a file owns no ingest and reports the layer's +// size as its offset, and is committed as it is. A writer at any other offset means the file is +// not there (containerd's blob cleanup removes a file it has no record of): its ingest is +// discarded, the layer is linked again and the registration is tried once more; a second miss +// fails the install rather than copy the layer. +func (c *ctrdClient) registerLayer(ctx context.Context, blobsRoot string, desc ocispec.Descriptor) error { + cs := c.ContentStore() + ref := "composectl-" + desc.Digest.String() + for attempt := 1; ; attempt++ { + w, err := content.OpenWriter(ctx, cs, content.WithRef(ref), content.WithDescriptor(desc)) + if errdefs.IsAlreadyExists(err) { + return nil + } + if err != nil { + return fmt.Errorf("failed to open a content writer for layer %s: %w", desc.Digest, err) + } + st, err := w.Status() + if err == nil && st.Offset == desc.Size { + err = w.Commit(ctx, desc.Size, desc.Digest) + _ = w.Close() + if err != nil && !errdefs.IsAlreadyExists(err) { + return fmt.Errorf("failed to register layer %s: %w", desc.Digest, err) + } + return nil + } + _ = w.Close() + if err != nil { + return fmt.Errorf("failed to read the content writer status of layer %s: %w", desc.Digest, err) + } + if err := abortIngest(ctx, cs, ref); err != nil { + return fmt.Errorf("failed to discard the ingest of layer %s: %w", desc.Digest, err) + } + if attempt == 2 { + return fmt.Errorf("layer %s cannot be registered in containerd's content store without a copy:"+ + " its link is removed before it is recorded", desc.Digest) + } + if err := linkLayer(c.contentRoot, blobsRoot, desc); err != nil { + return err + } + } +} + +// abortIngest discards the ingest of ref, retrying up to ingestAbortAttempts times, ingestAbortDelay +// apart, while the abort fails: the writer's Close only tells containerd to stop, so its side may +// still be writing the ingest's status when the abort removes the ingest directory, which then +// fails as not empty. A cancelled context ends the retries with its error. +func abortIngest(ctx context.Context, cs content.IngestManager, ref string) error { + for attempt := 1; ; attempt++ { + err := cs.Abort(ctx, ref) + if err == nil || errdefs.IsNotFound(err) { + return nil + } + if attempt == ingestAbortAttempts { + return err + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(ingestAbortDelay): + } + } +} + +// linkFile makes the hardlink of a store blob in the content store; a variable so that tests can +// make it fail. +var linkFile = os.Link + +// linkLayer checks that the store blob of a layer is there with the layer's size and hardlinks it +// to the path the content store keeps that blob at. containerd registers a file that is already at +// that path without copying or hashing it (the "shared" content sharing policy), so the write that +// follows moves no bytes; the blob was verified when it was pulled, and the unpack checks every +// extracted layer against the image's diff IDs. A file already at that path is containerd's own +// blob or a link an interrupted install left, and is used as it is; a link that cannot be made +// for any other reason fails the install. +func linkLayer(contentRoot, blobsRoot string, desc ocispec.Descriptor) error { + src := filepath.Join(blobsRoot, desc.Digest.Encoded()) + if err := checkBlob(src, desc); err != nil { + return err + } + dir := filepath.Join(contentRoot, "blobs", desc.Digest.Algorithm().String()) + if err := os.MkdirAll(dir, 0700); err != nil { + return fmt.Errorf("failed to create the content store directory of layer %s: %w", desc.Digest, err) + } + err := linkFile(src, filepath.Join(dir, desc.Digest.Encoded())) + switch { + case err == nil || errors.Is(err, fs.ErrExist): + return nil + case errors.Is(err, syscall.EXDEV): + return fmt.Errorf("cannot link layer %s from %s into containerd's content store %s: both must be on one filesystem", + desc.Digest, blobsRoot, contentRoot) + default: + return fmt.Errorf("failed to link layer %s into containerd's content store: %w", desc.Digest, err) + } +} + +// checkBlob checks that the store blob is there and has the size its descriptor promises. +func checkBlob(path string, desc ocispec.Descriptor) error { + fi, err := os.Stat(path) + if err != nil { + return fmt.Errorf("failed to find blob %s in the store: %w", desc.Digest, err) + } + if fi.Size() != desc.Size { + return fmt.Errorf("blob %s in the store has %d bytes, expected %d", desc.Digest, fi.Size(), desc.Size) } return nil } +// writeBlob writes the store blob of desc to the content store; a blob the content store has +// already is left as it is. func (c *ctrdClient) writeBlob(ctx context.Context, blobsRoot string, desc ocispec.Descriptor, labels map[string]string) error { f, err := os.Open(filepath.Join(blobsRoot, desc.Digest.Encoded())) if err != nil { diff --git a/pkg/compose/ctrd_install_rig_test.go b/pkg/compose/ctrd_install_rig_test.go index 6559aba..9882ac3 100644 --- a/pkg/compose/ctrd_install_rig_test.go +++ b/pkg/compose/ctrd_install_rig_test.go @@ -5,18 +5,26 @@ package compose_test // (built by `make`) is not there, as under `make test-unit`. import ( + "bytes" "context" "fmt" "os" "os/exec" + "path/filepath" + "strings" + "syscall" "testing" "time" + "github.com/containerd/containerd/content" "github.com/containerd/containerd/errdefs" "github.com/containerd/containerd/images" + "github.com/containerd/containerd/leases" "github.com/containerd/containerd/platforms" "github.com/foundriesio/composeapp/pkg/compose" f "github.com/foundriesio/composeapp/test/fixtures" + "github.com/opencontainers/go-digest" + ocispec "github.com/opencontainers/image-spec/specs-go/v1" ) const derivedAppDef = ` @@ -44,7 +52,7 @@ func ctrdRig(t *testing.T) *compose.CtrdClient { } // TestCtrdInstallRetriesAfterBaseSnapshotRemoval: the base snapshot found by the scan is removed -// before the unpack, as `docker image prune` could do, and the install streams every layer and +// before the unpack, as `docker image prune` could do, and the install links every layer and // unpacks again. The removal runs in the install's progress callback, which installImagesCtrd // calls synchronously after the scan and before the layers are streamed. func TestCtrdInstallRetriesAfterBaseSnapshotRemoval(t *testing.T) { @@ -106,3 +114,272 @@ func TestCtrdInstallRetriesAfterBaseSnapshotRemoval(t *testing.T) { out, err := exec.Command("docker", "run", "--rm", "--pull", "never", "--network", "none", derivedRef, "echo", "ok").CombinedOutput() f.Checkf(t, err, "docker run %s: %s", derivedRef, out) } + +// linkRig publishes and pulls a one-image app with a fresh layer and returns the app, its tree and +// its layer descriptors. +func linkRig(t *testing.T, cfg *compose.Config) (*f.App, compose.App, []ocispec.Descriptor) { + t.Helper() + ctx := context.Background() + store := compose.NewStoreBlobProvider(cfg.GetBlobsRoot()) + app := f.NewApp(t, fmt.Sprintf(derivedAppDef, f.PublishDerivedImage(t, "ghcr.io/foundriesio/busybox:1.36", 3))) + app.Publish(t) + app.Pull(t) + tree, err := cfg.AppLoader.LoadAppTree(ctx, store, platforms.OnlyStrict(cfg.Platform), app.PublishedUri) + f.Check(t, err) + layers, err := compose.LayerDescriptorsOf(ctx, tree, store) + f.Check(t, err) + return app, tree, layers[0] +} + +// storeCopyIn copies the store's blobs into a new store root under dir and returns a config that +// uses it. +func storeCopyIn(t *testing.T, cfg *compose.Config, dir string) *compose.Config { + t.Helper() + root, err := os.MkdirTemp(dir, "composectl-store-") + f.Check(t, err) + t.Cleanup(func() { os.RemoveAll(root) }) + copyCfg := *cfg + copyCfg.StoreRoot = root + f.Check(t, os.MkdirAll(copyCfg.GetBlobsRoot(), 0700)) + blobs, err := os.ReadDir(cfg.GetBlobsRoot()) + f.Check(t, err) + for _, b := range blobs { + data, err := os.ReadFile(filepath.Join(cfg.GetBlobsRoot(), b.Name())) + f.Check(t, err) + f.Check(t, os.WriteFile(filepath.Join(copyCfg.GetBlobsRoot(), b.Name()), data, 0600)) + } + return ©Cfg +} + +// layersToRegister returns the layers whose snapshot does not exist yet, from the first one on: the +// install leaves the ones below it alone, so only these reach the content store. +func layersToRegister(t *testing.T, c *compose.CtrdClient, cfg *compose.Config, tree compose.App, layers []ocispec.Descriptor) []ocispec.Descriptor { + t.Helper() + ctx := context.Background() + chains, err := compose.ChainIDsOf(ctx, tree, compose.NewStoreBlobProvider(cfg.GetBlobsRoot())) + f.Check(t, err) + sn := c.SnapshotService(compose.CtrdSnapshotter(c)) + for i, chainID := range chains[0] { + if _, err := sn.Stat(ctx, chainID); errdefs.IsNotFound(err) { + return layers[i:] + } + } + return nil +} + +func deviceOf(t *testing.T, path string) uint64 { + t.Helper() + var st syscall.Stat_t + f.Check(t, syscall.Stat(path, &st)) + return uint64(st.Dev) +} + +// deleteImage removes both records the install creates for the app's image, the pinned digest +// reference and the short tag reference, so that containerd's GC drops its snapshots. +func deleteImage(t *testing.T, c *compose.CtrdClient, tree compose.App) { + t.Helper() + ref := tree.GetComposeRoot().Children[0].Ref() + parsed, err := compose.ParseImageRef(ref) + f.Check(t, err) + for _, name := range []string{ref, parsed.GetTagRef()} { + f.Check(t, c.ImageService().Delete(context.Background(), name, images.SynchronousDelete())) + } +} + +func contentBlobPath(c *compose.CtrdClient, d digest.Digest) string { + return filepath.Join(compose.CtrdContentRoot(c), "blobs", d.Algorithm().String(), d.Encoded()) +} + +func statNlink(t *testing.T, path string) (os.FileInfo, uint64) { + t.Helper() + fi, err := os.Stat(path) + f.Check(t, err) + return fi, uint64(fi.Sys().(*syscall.Stat_t).Nlink) +} + +// TestCtrdInstallLinksLayers: every layer blob is a second name of the store blob while the install +// lease exists, and the content root holds no copy of it afterwards. One layer is linked before the +// install, as an interrupted install leaves it, and is registered as is. +func TestCtrdInstallLinksLayers(t *testing.T) { + c := ctrdRig(t) + cfg := f.NewTestConfig(t) + app, tree, layers := linkRig(t, cfg) + defer app.Remove(t) + layers = layersToRegister(t, c, cfg, tree, layers) + if len(layers) == 0 { + t.Fatal("every layer of the fresh image already has a snapshot") + } + storePath := func(d digest.Digest) string { return filepath.Join(cfg.GetBlobsRoot(), d.Encoded()) } + f.Check(t, os.MkdirAll(filepath.Dir(contentBlobPath(c, layers[0].Digest)), 0700)) + f.Check(t, os.Link(storePath(layers[0].Digest), contentBlobPath(c, layers[0].Digest))) + + checked := false + atLoaded := func(p *compose.LoadImageProgress) { + if p.State != compose.ImageLoadStateImageLoaded { + return + } + checked = true + for _, l := range layers { + src, nlink := statNlink(t, storePath(l.Digest)) + dst, _ := statNlink(t, contentBlobPath(c, l.Digest)) + if nlink != 2 || !os.SameFile(src, dst) { + t.Errorf("layer %s: nlink %d, same file %v; expected a link into the content root", l.Digest, nlink, os.SameFile(src, dst)) + } + } + } + f.Check(t, compose.InstallImagesCtrd(context.Background(), c, tree, cfg, atLoaded)) + defer deleteImage(t, c, tree) + if !checked { + t.Fatal("the image was never reported loaded") + } + for _, l := range layers { + if _, nlink := statNlink(t, storePath(l.Digest)); nlink != 1 { + t.Errorf("layer %s still has %d names after the lease was deleted", l.Digest, nlink) + } + if _, err := os.Stat(contentBlobPath(c, l.Digest)); !errdefs.IsNotFound(err) && !os.IsNotExist(err) { + t.Errorf("layer %s is still in the content root: %v", l.Digest, err) + } + } +} + +// TestCtrdInstallFailsAcrossFilesystems: with the store on another filesystem the link fails and +// so does the install, naming both directories, without a copy of any layer in the content root; +// the next install from a store on the content root's filesystem removes the lease it left. +func TestCtrdInstallFailsAcrossFilesystems(t *testing.T) { + c := ctrdRig(t) + ctx := context.Background() + cfg := f.NewTestConfig(t) + app, tree, layers := linkRig(t, cfg) + defer app.Remove(t) + other := "/dev/shm" + if _, err := os.Stat(other); err != nil || deviceOf(t, other) == deviceOf(t, compose.CtrdContentRoot(c)) { + t.Skip("no filesystem other than the content root's is available") + } + otherCfg := storeCopyIn(t, cfg, other) + layers = layersToRegister(t, c, otherCfg, tree, layers) + if len(layers) == 0 { + t.Fatal("every layer of the fresh image already has a snapshot") + } + + err := compose.InstallImagesCtrd(ctx, c, tree, otherCfg, nil) + if err == nil || !strings.Contains(err.Error(), otherCfg.GetBlobsRoot()) || !strings.Contains(err.Error(), compose.CtrdContentRoot(c)) { + t.Fatalf("expected the install to fail naming the store and the content root, got %v", err) + } + for _, l := range layers { + if _, err := os.Stat(contentBlobPath(c, l.Digest)); !os.IsNotExist(err) { + t.Errorf("layer %s is in the content root after the failed install: %v", l.Digest, err) + } + } + installLeases := func() (n int) { + all, err := c.LeasesService().List(ctx) + f.Check(t, err) + for _, l := range all { + if strings.HasPrefix(l.ID, "composectl/install/") { + n++ + } + } + return n + } + if installLeases() != 1 { + t.Fatalf("expected the failed install to leave its lease, found %d", installLeases()) + } + f.Check(t, compose.InstallImagesCtrd(ctx, c, tree, cfg, nil)) + defer deleteImage(t, c, tree) + if installLeases() != 0 { + t.Fatalf("expected the next install to remove the lease, found %d", installLeases()) + } +} + +// TestCtrdLinkedBlobAndGC: containerd's blob-directory cleanup runs after a GC that removed content. +// A linked file it finds before the writer is opened is removed, and the registration links it again +// rather than copy it; once the writer is open its expected digest keeps the file; after the commit +// the record keeps it. +func TestCtrdLinkedBlobAndGC(t *testing.T) { + c := ctrdRig(t) + ctx := context.Background() + cfg := f.NewTestConfig(t) + app, _, layers := linkRig(t, cfg) + defer app.Remove(t) + cs, ls := c.ContentStore(), c.LeasesService() + layer := layers[len(layers)-1] + src := filepath.Join(cfg.GetBlobsRoot(), layer.Digest.Encoded()) + dst := contentBlobPath(c, layer.Digest) + ref := "composectl-" + layer.Digest.String() + f.Check(t, os.MkdirAll(filepath.Dir(dst), 0700)) + + // forceGC commits a throwaway blob under its own lease and deletes that lease, which removes + // the blob's record and so triggers the blob-directory cleanup. + forceGC := func() { + lease, err := ls.Create(ctx, leases.WithRandomID()) + f.Check(t, err) + data := []byte(fmt.Sprintf("throwaway %d", time.Now().UnixNano())) + f.Check(t, content.WriteBlob(leases.WithLease(ctx, lease.ID), cs, "throwaway", bytes.NewReader(data), + ocispec.Descriptor{Digest: digest.FromBytes(data), Size: int64(len(data))})) + f.Check(t, ls.Delete(ctx, lease, leases.SynchronousDelete)) + } + waitGone := func(path string) bool { + for i := 0; i < 50; i++ { + if _, err := os.Stat(path); os.IsNotExist(err) { + return true + } + time.Sleep(100 * time.Millisecond) + } + return false + } + lease, err := ls.Create(ctx, leases.WithRandomID()) + f.Check(t, err) + defer func() { _ = ls.Delete(ctx, lease, leases.SynchronousDelete) }() + lctx := leases.WithLease(ctx, lease.ID) + + // (a) cleanup before the writer: the link goes; the registration finds an ordinary writer, + // discards its ingest, links again and registers the link, copying nothing. + f.Check(t, os.Link(src, dst)) + forceGC() + if !waitGone(dst) { + t.Fatal("cleanup did not remove the unreferenced link") + } + time.Sleep(time.Second) // the cleanup pass that removed the link may still be walking the directory + f.Check(t, compose.RegisterLayer(lctx, c, cfg.GetBlobsRoot(), layer)) + srcInfo, _ := statNlink(t, src) + if dstInfo, nlink := statNlink(t, dst); nlink != 2 || !os.SameFile(srcInfo, dstInfo) { + t.Fatalf("expected the layer linked again after cleanup removed the link, nlink %d same file %v", nlink, os.SameFile(srcInfo, dstInfo)) + } + if _, err := cs.Info(ctx, layer.Digest); err != nil { + t.Fatalf("the relinked blob has no record: %v", err) + } + if statuses, err := cs.ListStatuses(ctx, "ref=="+ref); err != nil || len(statuses) != 0 { + t.Fatalf("expected no ingest left behind, got %v, %v", statuses, err) + } + f.Check(t, cs.Delete(ctx, layer.Digest)) + forceGC() + if !waitGone(dst) { + t.Fatal("the deleted blob was not cleaned up") + } + + // (b) cleanup between the writer and the commit: the expected digest keeps the file. + f.Check(t, os.Link(src, dst)) + w, err := content.OpenWriter(lctx, cs, content.WithRef("composectl-"+layer.Digest.String()), content.WithDescriptor(layer)) + f.Check(t, err) + st, err := w.Status() + f.Check(t, err) + if st.Offset != layer.Size { + t.Fatalf("expected a shared writer at offset %d, got %d", layer.Size, st.Offset) + } + forceGC() + time.Sleep(time.Second) + if _, nlink := statNlink(t, dst); nlink != 2 { + t.Fatalf("the linked blob of an open writer was removed, nlink %d", nlink) + } + f.Check(t, w.Commit(lctx, layer.Size, layer.Digest)) + w.Close() + + // (c) cleanup after the commit: the record keeps the file. + forceGC() + time.Sleep(time.Second) + if _, nlink := statNlink(t, dst); nlink != 2 { + t.Fatalf("the committed linked blob was removed, nlink %d", nlink) + } + if _, err := cs.Info(ctx, layer.Digest); err != nil { + t.Fatalf("the linked blob has no record: %v", err) + } +} diff --git a/pkg/compose/ctrd_install_test.go b/pkg/compose/ctrd_install_test.go index 4448741..f6a1a02 100644 --- a/pkg/compose/ctrd_install_test.go +++ b/pkg/compose/ctrd_install_test.go @@ -3,13 +3,18 @@ package compose import ( "context" "errors" + "os" + "path/filepath" "strings" + "syscall" "testing" + "github.com/containerd/containerd/content" "github.com/containerd/containerd/errdefs" "github.com/containerd/containerd/images" "github.com/containerd/containerd/leases" "github.com/opencontainers/go-digest" + ocispec "github.com/opencontainers/image-spec/specs-go/v1" ) func TestInstallLeaseID(t *testing.T) { @@ -121,3 +126,153 @@ func TestPutImageRecordsRollsBackOnCancel(t *testing.T) { t.Fatal("the rollback ran on the cancelled context") } } + +func writeStoreBlob(t *testing.T, blobsRoot string, data []byte) ocispec.Descriptor { + t.Helper() + d := digest.FromBytes(data) + if err := os.MkdirAll(blobsRoot, 0700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(blobsRoot, d.Encoded()), data, 0600); err != nil { + t.Fatal(err) + } + return ocispec.Descriptor{Digest: d, Size: int64(len(data))} +} + +func TestLinkLayerLinksVerifiedBlob(t *testing.T) { + root := t.TempDir() + blobsRoot := filepath.Join(root, "store") + contentRoot := filepath.Join(root, "content") + desc := writeStoreBlob(t, blobsRoot, []byte("layer bytes")) + + if err := linkLayer(contentRoot, blobsRoot, desc); err != nil { + t.Fatal(err) + } + src, err := os.Stat(filepath.Join(blobsRoot, desc.Digest.Encoded())) + if err != nil { + t.Fatal(err) + } + dst, err := os.Stat(filepath.Join(contentRoot, "blobs", "sha256", desc.Digest.Encoded())) + if err != nil { + t.Fatalf("the blob was not linked into the content root: %v", err) + } + if !os.SameFile(src, dst) { + t.Fatal("the content root blob is not a link to the store blob") + } + // A second install finds the link in place and leaves it alone. + if err := linkLayer(contentRoot, blobsRoot, desc); err != nil { + t.Fatalf("linking over an existing link must not fail: %v", err) + } +} + +func TestLinkLayerRejectsWrongSize(t *testing.T) { + root := t.TempDir() + blobsRoot := filepath.Join(root, "store") + contentRoot := filepath.Join(root, "content") + desc := writeStoreBlob(t, blobsRoot, []byte("layer bytes")) + desc.Size++ + + if err := linkLayer(contentRoot, blobsRoot, desc); err == nil || !strings.Contains(err.Error(), desc.Digest.String()) { + t.Fatalf("expected a size mismatch naming %s, got %v", desc.Digest, err) + } + if _, err := os.Stat(filepath.Join(contentRoot, "blobs")); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("nothing must be linked for a blob of the wrong size, stat: %v", err) + } +} + +func TestLinkLayerFailsOnAnotherFilesystem(t *testing.T) { + root := t.TempDir() + blobsRoot := filepath.Join(root, "store") + contentRoot := filepath.Join(root, "content") + desc := writeStoreBlob(t, blobsRoot, []byte("layer bytes")) + + orig := linkFile + t.Cleanup(func() { linkFile = orig }) + linkFile = func(string, string) error { return &os.LinkError{Op: "link", Err: syscall.EXDEV} } + + err := linkLayer(contentRoot, blobsRoot, desc) + if err == nil || !strings.Contains(err.Error(), blobsRoot) || !strings.Contains(err.Error(), contentRoot) { + t.Fatalf("expected a failure naming the store and the content root, got %v", err) + } + if _, err := os.Stat(filepath.Join(contentRoot, "blobs", "sha256", desc.Digest.Encoded())); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("no link expected, stat: %v", err) + } +} + +func TestLinkLayerFailsOnMissingBlob(t *testing.T) { + root := t.TempDir() + desc := ocispec.Descriptor{Digest: digest.FromString("absent"), Size: 6} + err := linkLayer(filepath.Join(root, "content"), filepath.Join(root, "store"), desc) + if !errors.Is(err, os.ErrNotExist) || !strings.Contains(err.Error(), desc.Digest.String()) { + t.Fatalf("expected a not-found failure naming the layer, got %v", err) + } +} + +// fakeIngests is a content.IngestManager whose Abort returns err for the first failures calls, as +// containerd does while the server side of a closed writer still writes the ingest's status, and +// calls cancel, when set, on the first call. +type fakeIngests struct { + failures int + err error + cancel context.CancelFunc + aborts int +} + +func (s *fakeIngests) Status(context.Context, string) (content.Status, error) { + return content.Status{}, errdefs.ErrNotFound +} +func (s *fakeIngests) ListStatuses(context.Context, ...string) ([]content.Status, error) { + return nil, nil +} +func (s *fakeIngests) Abort(context.Context, string) error { + s.aborts++ + if s.cancel != nil { + s.cancel() + } + if s.aborts <= s.failures { + return s.err + } + return nil +} + +func TestAbortIngestRetriesWhileTheAbortFails(t *testing.T) { + ingests := &fakeIngests{failures: 2, err: errors.New("unlinkat ingest: directory not empty")} + if err := abortIngest(context.Background(), ingests, "ref"); err != nil { + t.Fatalf("the abort failed after the retries: %v", err) + } + if ingests.aborts != 3 { + t.Fatalf("expected 3 abort calls, got %d", ingests.aborts) + } +} + +func TestAbortIngestGivesUpAfterTheBoundedRetries(t *testing.T) { + ingests := &fakeIngests{failures: ingestAbortAttempts + 1, err: errors.New("unlinkat ingest: directory not empty")} + if err := abortIngest(context.Background(), ingests, "ref"); !errors.Is(err, ingests.err) { + t.Fatalf("expected the last abort error, got %v", err) + } + if ingests.aborts != ingestAbortAttempts { + t.Fatalf("expected %d abort calls, got %d", ingestAbortAttempts, ingests.aborts) + } +} + +func TestAbortIngestOfUnknownRefSucceeds(t *testing.T) { + ingests := &fakeIngests{failures: 1, err: errdefs.ErrNotFound} + if err := abortIngest(context.Background(), ingests, "ref"); err != nil { + t.Fatalf("an unknown ingest must not fail the abort: %v", err) + } + if ingests.aborts != 1 { + t.Fatalf("expected 1 abort call, got %d", ingests.aborts) + } +} + +func TestAbortIngestStopsWhenCancelled(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + ingests := &fakeIngests{failures: ingestAbortAttempts + 1, err: errors.New("unlinkat ingest: directory not empty"), cancel: cancel} + if err := abortIngest(ctx, ingests, "ref"); !errors.Is(err, context.Canceled) { + t.Fatalf("expected the cancellation to be returned, got %v", err) + } + if ingests.aborts != 1 { + t.Fatalf("expected the retries to stop at the cancellation, got %d abort calls", ingests.aborts) + } +} diff --git a/pkg/compose/export_test.go b/pkg/compose/export_test.go index 82039ba..6ec50db 100644 --- a/pkg/compose/export_test.go +++ b/pkg/compose/export_test.go @@ -1,6 +1,10 @@ package compose -import "context" +import ( + "context" + + ocispec "github.com/opencontainers/image-spec/specs-go/v1" +) // Exported for the tests in package compose_test that drive the containerd install directly. @@ -11,6 +15,10 @@ var ( InstallImagesCtrd = installImagesCtrd ) +func RegisterLayer(ctx context.Context, c *CtrdClient, blobsRoot string, desc ocispec.Descriptor) error { + return c.registerLayer(ctx, blobsRoot, desc) +} + func CtrdSnapshotter(c *CtrdClient) string { return c.snapshotter } @@ -31,3 +39,20 @@ func ChainIDsOf(ctx context.Context, app App, provider BlobProvider) ([][]string } return out, nil } + +func CtrdContentRoot(c *CtrdClient) string { + return c.contentRoot +} + +// LayerDescriptorsOf returns the layer descriptors of each app image, in the app's image order. +func LayerDescriptorsOf(ctx context.Context, app App, provider BlobProvider) ([][]ocispec.Descriptor, error) { + imgs, err := ctrdImagesOf(ctx, app, provider) + if err != nil { + return nil, err + } + var out [][]ocispec.Descriptor + for _, img := range imgs { + out = append(out, img.layers) + } + return out, nil +} diff --git a/test/compose/docker-compose.yml b/test/compose/docker-compose.yml index b2a70e5..e9084ad 100644 --- a/test/compose/docker-compose.yml +++ b/test/compose/docker-compose.yml @@ -24,7 +24,6 @@ services: - docker-runtime:/var/run/docker - docker-data:/var/lib/docker - ${REG_DIR}/daemon.json:/etc/docker/daemon.json:ro - - reset-apps:/var/sota/reset-apps # Plays the role of the FoundriesFactory CI when tests publish apps: test images are pulled into it # and the layer sizes are read from its overlay2 store, whatever image store the dockerd above uses. @@ -54,7 +53,6 @@ services: - ${BIN_DIR}:${SRC_DIR}/bin - docker-runtime:/var/run/docker - docker-data:/var/lib/docker - - reset-apps:/var/sota/reset-apps - publisher-runtime:/var/run/docker-publisher - publisher-data:/var/lib/docker-publisher:ro working_dir: ${SRC_DIR} @@ -68,7 +66,8 @@ services: - DOCKER_HOST=unix:///var/run/docker/docker.sock - COMPOSECTL_EXE=${BIN_DIR}/composectl - SRC_DIR=${SRC_DIR} - - STOREROOT=/var/sota/reset-apps + # On the mount of dockerd's containerd, whose content store hardlinks the store's layers + - STOREROOT=/var/lib/docker/reset-apps - COMPOSEROOT=/var/sota/compose-apps - DOCKERD_FIO_PATCH=${DOCKERD_FIO_PATCH} - PUBLISHER_DOCKER_HOST=unix:///var/run/docker-publisher/docker.sock @@ -77,6 +76,5 @@ services: volumes: docker-data: docker-runtime: - reset-apps: publisher-data: publisher-runtime: diff --git a/test/fixtures/composectl_cmds.go b/test/fixtures/composectl_cmds.go index 4729fa6..011b69f 100644 --- a/test/fixtures/composectl_cmds.go +++ b/test/fixtures/composectl_cmds.go @@ -23,7 +23,8 @@ import ( ) const ( - AppStoreRoot = "/var/sota/reset-apps" + // On the mount of the test dockerd's containerd: an install hardlinks the store's layers into it. + AppStoreRoot = "/var/lib/docker/reset-apps" AppComposeRootRoot = "/var/sota/compose-apps" )