Centralize safe cache writes (#4132)

* centralize all writing of the index.json to one place

Signed-off-by: Avi Deitcher <avi@deitcher.net>

* create filelock utility

Signed-off-by: Avi Deitcher <avi@deitcher.net>

* leverage file locks for cache index.json

Signed-off-by: Avi Deitcher <avi@deitcher.net>

---------

Signed-off-by: Avi Deitcher <avi@deitcher.net>

src/cmd/linuxkit/cache/cacheindex.go

@@ -0,0 +1,71 @@
+// ALL writes to index.json at the root of the cache directory
+// must be done through calls in this file. This is to ensure that it always does
+// proper locking.
+package cache
+
+import (
+	"errors"
+	"fmt"
+	"path/filepath"
+
+	v1 "github.com/google/go-containerregistry/pkg/v1"
+	"github.com/google/go-containerregistry/pkg/v1/match"
+	"github.com/linuxkit/linuxkit/src/cmd/linuxkit/util"
+	imagespec "github.com/opencontainers/image-spec/specs-go/v1"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	indexFile = "index.json"
+)
+
+// DescriptorWrite writes a descriptor to the cache index; it validates that it has a name
+// and replaces any existing one
+func (p *Provider) DescriptorWrite(image string, desc v1.Descriptor) error {
+	if image == "" {
+		return errors.New("cannot write descriptor without reference name")
+	}
+	if desc.Annotations == nil {
+		desc.Annotations = map[string]string{}
+	}
+	desc.Annotations[imagespec.AnnotationRefName] = image
+	log.Debugf("writing descriptor for image %s", image)
+
+	// get our lock
+	lock, err := util.Lock(filepath.Join(p.dir, indexFile))
+	if err != nil {
+		return fmt.Errorf("unable to lock cache index for writing descriptor for %s: %v", image, err)
+	}
+	defer func() {
+		if err := lock.Unlock(); err != nil {
+			log.Errorf("unable to close lock for cache index after writing descriptor for %s: %v", image, err)
+		}
+	}()
+
+	// do we update an existing one? Or create a new one?
+	if err := p.cache.RemoveDescriptors(match.Name(image)); err != nil {
+		return fmt.Errorf("unable to remove old descriptors for %s: %v", image, err)
+	}
+
+	if err := p.cache.AppendDescriptor(desc); err != nil {
+		return fmt.Errorf("unable to append new descriptor for %s: %v", image, err)
+	}
+
+	return nil
+}
+
+// RemoveDescriptors removes all descriptors that match the provided matcher.
+// It does so in a parallel-access-safe way
+func (p *Provider) RemoveDescriptors(matcher match.Matcher) error {
+	// get our lock
+	lock, err := util.Lock(filepath.Join(p.dir, indexFile))
+	if err != nil {
+		return fmt.Errorf("unable to lock cache index for removing descriptor for %v: %v", matcher, err)
+	}
+	defer func() {
+		if err := lock.Unlock(); err != nil {
+			log.Errorf("unable to close lock for cache index after writing descriptor for %v: %v", matcher, err)
+		}
+	}()
+	return p.cache.RemoveDescriptors(matcher)
+}

src/cmd/linuxkit/cache/open.go

@@ -2,9 +2,12 @@ package cache
 
 import (
 	"fmt"
+	"path/filepath"
 
 	"github.com/google/go-containerregistry/pkg/v1/empty"
 	"github.com/google/go-containerregistry/pkg/v1/layout"
+	"github.com/linuxkit/linuxkit/src/cmd/linuxkit/util"
+	log "github.com/sirupsen/logrus"
 )
 
 // Get get or initialize the cache
@@ -12,6 +15,15 @@ func Get(cache string) (layout.Path, error) {
 	// initialize the cache path if needed
 	p, err := layout.FromPath(cache)
 	if err != nil {
+		lock, err := util.Lock(filepath.Join(cache, indexFile))
+		if err != nil {
+			return "", fmt.Errorf("unable to lock cache index for writing descriptor for new cache: %v", err)
+		}
+		defer func() {
+			if err := lock.Unlock(); err != nil {
+				log.Errorf("unable to close lock for cache index after writing descriptor for new cache: %v", err)
+			}
+		}()
 		p, err = layout.Write(cache, empty.Index)
 		if err != nil {
 			return p, fmt.Errorf("could not initialize cache at path %s: %v", cache, err)

src/cmd/linuxkit/cache/provider.go

@@ -10,6 +10,7 @@ import (
 type Provider struct {
 	cache layout.Path
 	store content.Store
+	dir   string
 }
 
 // NewProvider create a new CacheProvider based in the provided directory
@@ -22,5 +23,5 @@ func NewProvider(dir string) (*Provider, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &Provider{p, store}, nil
+	return &Provider{p, store, dir}, nil
 }

src/cmd/linuxkit/cache/pull.go

@@ -9,8 +9,6 @@ import (
 	"github.com/google/go-containerregistry/pkg/authn"
 	namepkg "github.com/google/go-containerregistry/pkg/name"
 	v1 "github.com/google/go-containerregistry/pkg/v1"
-	"github.com/google/go-containerregistry/pkg/v1/layout"
-	"github.com/google/go-containerregistry/pkg/v1/match"
 	"github.com/google/go-containerregistry/pkg/v1/partial"
 	"github.com/google/go-containerregistry/pkg/v1/remote"
 	"github.com/google/go-containerregistry/pkg/v1/validate"
@@ -182,11 +180,6 @@ func (p *Provider) Pull(name string, withArchReferences bool) error {
 		return fmt.Errorf("error getting manifest for trusted image %s: %v", name, err)
 	}
 
-	// use the original image name in the annotation
-	annotations := map[string]string{
-		imagespec.AnnotationRefName: fullname,
-	}
-
 	// first attempt as an index
 	ii, err := desc.ImageIndex()
 	if err == nil {
@@ -195,7 +188,7 @@ func (p *Provider) Pull(name string, withArchReferences bool) error {
 		if err := p.cache.WriteIndex(ii); err != nil {
 			return fmt.Errorf("unable to write index: %v", err)
 		}
-		if err := p.DescriptorWrite(&v1ref, desc.Descriptor); err != nil {
+		if err := p.DescriptorWrite(v1ref.String(), desc.Descriptor); err != nil {
 			return fmt.Errorf("unable to write index descriptor to cache: %v", err)
 		}
 		if withArchReferences {
@@ -206,11 +199,10 @@ func (p *Provider) Pull(name string, withArchReferences bool) error {
 			for _, m := range im.Manifests {
 				if m.MediaType.IsImage() && m.Platform != nil && m.Platform.Architecture != unknown && m.Platform.OS != unknown {
 					archSpecific := fmt.Sprintf("%s-%s", ref.String(), m.Platform.Architecture)
-					archRef, err := reference.Parse(archSpecific)
-					if err != nil {
+					if _, err := reference.Parse(archSpecific); err != nil {
 						return fmt.Errorf("unable to parse arch-specific reference %s: %v", archSpecific, err)
 					}
-					if err := p.DescriptorWrite(&archRef, m); err != nil {
+					if err := p.DescriptorWrite(archSpecific, m); err != nil {
 						return fmt.Errorf("unable to write index descriptor to cache: %v", err)
 					}
 				}
@@ -224,9 +216,12 @@ func (p *Provider) Pull(name string, withArchReferences bool) error {
 			return fmt.Errorf("provided image is neither an image nor an index: %s", name)
 		}
 		log.Debugf("ImageWrite retrieved %s is image, saving", fullname)
-		if err = p.cache.ReplaceImage(im, match.Name(fullname), layout.WithAnnotations(annotations)); err != nil {
+		if err = p.cache.WriteImage(im); err != nil {
 			return fmt.Errorf("unable to save image to cache: %v", err)
 		}
+		if err = p.DescriptorWrite(fullname, desc.Descriptor); err != nil {
+			return fmt.Errorf("unable to write updated descriptor to cache: %v", err)
+		}
 	}
 
 	return nil

src/cmd/linuxkit/cache/push.go

@@ -12,7 +12,6 @@ import (
 
 	"github.com/google/go-containerregistry/pkg/v1/mutate"
 	"github.com/linuxkit/linuxkit/src/cmd/linuxkit/util"
-	imagespec "github.com/opencontainers/image-spec/specs-go/v1"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -131,11 +130,7 @@ func (p *Provider) Push(name, remoteName string, withArchSpecificTags, override
 					// it might not have existed, so we can add it locally
 					// use the original image name in the annotation
 					desc := m.DeepCopy()
-					if desc.Annotations == nil {
-						desc.Annotations = map[string]string{}
-					}
-					desc.Annotations[imagespec.AnnotationRefName] = archTag
-					if err := p.cache.AppendDescriptor(*desc); err != nil {
+					if err := p.DescriptorWrite(archTag, *desc); err != nil {
 						return fmt.Errorf("error appending descriptor for %s to layout index: %v", archTag, err)
 					}
 					img, err = p.cache.Image(m.Digest)

src/cmd/linuxkit/cache/remove.go

@@ -65,7 +65,7 @@ func (p *Provider) Remove(name string) error {
 			log.Warnf("unable to remove blob %s for %s: %v", blob, name, err)
 		}
 	}
-	return p.cache.RemoveDescriptors(match.Name(name))
+	return p.RemoveDescriptors(match.Name(name))
 }
 
 func blobsForImage(img v1.Image) ([]v1.Hash, error) {

src/cmd/linuxkit/cache/write.go

@@ -14,7 +14,6 @@ import (
 	"github.com/google/go-containerregistry/pkg/authn"
 	"github.com/google/go-containerregistry/pkg/name"
 	v1 "github.com/google/go-containerregistry/pkg/v1"
-	"github.com/google/go-containerregistry/pkg/v1/layout"
 	"github.com/google/go-containerregistry/pkg/v1/match"
 	"github.com/google/go-containerregistry/pkg/v1/partial"
 	"github.com/google/go-containerregistry/pkg/v1/remote"
@@ -77,11 +76,6 @@ func (p *Provider) ImagePull(ref *reference.Spec, platforms []imagespec.Platform
 		return fmt.Errorf("error getting manifest for image %s: %v", pullImageName, err)
 	}
 
-	// use the original image name in the annotation
-	annotations := map[string]string{
-		imagespec.AnnotationRefName: image,
-	}
-
 	// first attempt as an index
 	ii, err := desc.ImageIndex()
 	if err == nil {
@@ -120,7 +114,7 @@ func (p *Provider) ImagePull(ref *reference.Spec, platforms []imagespec.Platform
 		if err := p.cache.WriteIndex(ii); err != nil {
 			return fmt.Errorf("unable to write index: %v", err)
 		}
-		if err := p.DescriptorWrite(ref, desc.Descriptor); err != nil {
+		if err := p.DescriptorWrite(ref.String(), desc.Descriptor); err != nil {
 			return fmt.Errorf("unable to write index descriptor to cache: %v", err)
 		}
 	} else {
@@ -131,8 +125,11 @@ func (p *Provider) ImagePull(ref *reference.Spec, platforms []imagespec.Platform
 			return fmt.Errorf("provided image is neither an image nor an index: %s", image)
 		}
 		log.Debugf("ImageWrite retrieved %s is image, saving", pullImageName)
-		if err = p.cache.ReplaceImage(im, match.Name(image), layout.WithAnnotations(annotations)); err != nil {
-			return fmt.Errorf("unable to save image to cache: %v", err)
+		if err := p.cache.WriteImage(im); err != nil {
+			return fmt.Errorf("error writing image %s to cache: %v", pullImageName, err)
+		}
+		if err := p.DescriptorWrite(image, desc.Descriptor); err != nil {
+			return fmt.Errorf("unable to write image descriptor to cache: %v", err)
 		}
 	}
 	return nil
@@ -208,20 +205,11 @@ func (p *Provider) ImageLoad(r io.Reader) ([]v1.Descriptor, error) {
 		// each of these is either an image or an index
 		// either way, it gets added directly to the linuxkit cache index.
 		for _, desc := range im.Manifests {
-			if imgName, ok := desc.Annotations[images.AnnotationImageName]; ok {
-				// remove the old descriptor, if it exists
-				if err := p.cache.RemoveDescriptors(match.Name(imgName)); err != nil {
-					return nil, fmt.Errorf("unable to remove old descriptors for %s: %v", imgName, err)
+			imgName, ok := desc.Annotations[images.AnnotationImageName]
+			if ok {
+				if err := p.DescriptorWrite(imgName, desc); err != nil {
+					return nil, fmt.Errorf("error writing descriptor for %s: %v", imgName, err)
 				}
-				// save the image name under our proper annotation
-				if desc.Annotations == nil {
-					desc.Annotations = map[string]string{}
-				}
-				desc.Annotations[imagespec.AnnotationRefName] = imgName
-			}
-			log.Debugf("appending descriptor %#v", desc)
-			if err := p.cache.AppendDescriptor(desc); err != nil {
-				return nil, fmt.Errorf("error appending descriptor to layout index: %v", err)
 			}
 			descs = append(descs, desc)
 		}
@@ -366,9 +354,6 @@ func (p *Provider) IndexWrite(ref *reference.Spec, descriptors ...v1.Descriptor)
 		return fmt.Errorf("error writing new index to json: %v", err)
 	}
 	// finally update the descriptor in the root
-	if err := p.cache.RemoveDescriptors(match.Name(image)); err != nil {
-		return fmt.Errorf("unable to remove old descriptor from index.json: %v", err)
-	}
 	desc := v1.Descriptor{
 		MediaType: types.OCIImageIndex,
 		Size:      size,
@@ -377,36 +362,7 @@ func (p *Provider) IndexWrite(ref *reference.Spec, descriptors ...v1.Descriptor)
 			imagespec.AnnotationRefName: image,
 		},
 	}
-	if err := p.cache.AppendDescriptor(desc); err != nil {
-		return fmt.Errorf("unable to append new descriptor to index.json: %v", err)
-	}
-
-	return nil
-}
-
-// DescriptorWrite writes a descriptor to the cache index; it validates that it has a name
-// and replaces any existing one
-func (p *Provider) DescriptorWrite(ref *reference.Spec, desc v1.Descriptor) error {
-	if ref == nil {
-		return errors.New("cannot write descriptor without reference name")
-	}
-	image := ref.String()
-	if desc.Annotations == nil {
-		desc.Annotations = map[string]string{}
-	}
-	desc.Annotations[imagespec.AnnotationRefName] = image
-	log.Debugf("writing descriptor for image %s", image)
-
-	// do we update an existing one? Or create a new one?
-	if err := p.cache.RemoveDescriptors(match.Name(image)); err != nil {
-		return fmt.Errorf("unable to remove old descriptors for %s: %v", image, err)
-	}
-
-	if err := p.cache.AppendDescriptor(desc); err != nil {
-		return fmt.Errorf("unable to append new descriptor for %s: %v", image, err)
-	}
-
-	return nil
+	return p.DescriptorWrite(ref.String(), desc)
 }
 
 func (p *Provider) ImageInCache(ref *reference.Spec, trustedRef, architecture string) (bool, error) {

src/cmd/linuxkit/pkglib/build.go

@@ -561,7 +561,7 @@ func (p Pkg) Build(bos ...BuildOpt) error {
 	if err != nil {
 		return err
 	}
-	if err := c.DescriptorWrite(&ref, *desc); err != nil {
+	if err := c.DescriptorWrite(fullRelTag, *desc); err != nil {
 		return err
 	}
 	if err := c.Push(fullRelTag, "", bo.manifest, true); err != nil {

src/cmd/linuxkit/pkglib/build_test.go

@@ -407,13 +407,12 @@ func (c *cacheMocker) Push(name, remoteName string, withManifest, override bool)
 	return nil
 }
 
-func (c *cacheMocker) DescriptorWrite(ref *reference.Spec, desc v1.Descriptor) error {
+func (c *cacheMocker) DescriptorWrite(image string, desc v1.Descriptor) error {
 	if !c.enabledDescriptorWrite {
 		return errors.New("descriptor disabled")
 	}
 	var (
-		image = ref.String()
-		im    = v1.IndexManifest{
+		im = v1.IndexManifest{
 			MediaType:     types.OCIImageIndex,
 			Manifests:     []v1.Descriptor{desc},
 			SchemaVersion: 2,

src/cmd/linuxkit/spec/cache.go

@@ -37,7 +37,7 @@ type CacheProvider interface {
 	ImageLoad(r io.Reader) ([]v1.Descriptor, error)
 	// DescriptorWrite writes a descriptor to the cache index; it validates that it has a name
 	// and replaces any existing one
-	DescriptorWrite(ref *reference.Spec, descriptors v1.Descriptor) error
+	DescriptorWrite(image string, descriptors v1.Descriptor) error
 	// Push an image along with a multi-arch index from local cache to remote registry.
 	// name is the name as referenced in the local cache, remoteName is the name to give it remotely.
 	// If remoteName is empty, it is the same as name.

src/cmd/linuxkit/util/filelock.go

@@ -0,0 +1,9 @@
+package util
+
+import (
+	"os"
+)
+
+type FileLock struct {
+	file *os.File
+}

src/cmd/linuxkit/util/filelock_other.go

@@ -0,0 +1,19 @@
+//go:build !unix
+
+package util
+
+// Lock opens the file (creating it if needed) and sets an exclusive lock.
+// Returns a FileLock that can later be unlocked.
+func Lock(path string) (*FileLock, error) {
+	return &FileLock{}, nil
+}
+
+// Unlock releases the lock and closes the file.
+func (l *FileLock) Unlock() error {
+	return nil
+}
+
+// CheckLock attempts to detect if the file is locked by another process.
+func CheckLock(path string) (locked bool, holderPID int, err error) {
+	return false, 0, nil
+}

src/cmd/linuxkit/util/filelock_unix.go

@@ -0,0 +1,101 @@
+//go:build unix
+
+package util
+
+import (
+	"fmt"
+	"io"
+	"os"
+
+	"golang.org/x/sys/unix"
+)
+
+// Lock opens the file (creating it if needed) and sets an exclusive lock.
+// Returns a FileLock that can later be unlocked.
+func Lock(path string) (*FileLock, error) {
+	f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0644)
+	if err != nil {
+		return nil, fmt.Errorf("open file: %w", err)
+	}
+
+	flock := unix.Flock_t{
+		Type:   unix.F_WRLCK,
+		Whence: int16(io.SeekStart),
+		Start:  0,
+		Len:    0,
+	}
+
+	if err := unix.FcntlFlock(f.Fd(), unix.F_SETLKW, &flock); err != nil {
+		_ = f.Close()
+		return nil, fmt.Errorf("set lock: %w", err)
+	}
+
+	return &FileLock{file: f}, nil
+}
+
+// Unlock releases the lock and closes the file.
+func (l *FileLock) Unlock() error {
+	flock := unix.Flock_t{
+		Type:   unix.F_UNLCK,
+		Whence: int16(io.SeekStart),
+		Start:  0,
+		Len:    0,
+	}
+	if err := unix.FcntlFlock(l.file.Fd(), unix.F_SETLKW, &flock); err != nil {
+		return fmt.Errorf("unlock: %w", err)
+	}
+	return l.file.Close()
+}
+
+// CheckLock attempts to detect if the file is locked by another process.
+func CheckLock(path string) (locked bool, holderPID int, err error) {
+	f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0644)
+	if err != nil {
+		return false, 0, fmt.Errorf("open file: %w", err)
+	}
+	defer func() {
+		_ = f.Close()
+	}()
+
+	check := unix.Flock_t{
+		Type:   unix.F_WRLCK,
+		Whence: int16(io.SeekStart),
+		Start:  0,
+		Len:    0,
+	}
+
+	if err := unix.FcntlFlock(f.Fd(), unix.F_GETLK, &check); err != nil {
+		return false, 0, fmt.Errorf("get lock: %w", err)
+	}
+
+	if check.Type == unix.F_UNLCK {
+		return false, 0, nil
+	}
+	return true, int(check.Pid), nil
+}
+
+// WaitUnlocked waits until the file is unlocked by another process, and uses it for reading but not writing.
+func WaitUnlocked(path string) error {
+	f, err := os.OpenFile(path, os.O_RDONLY, 0644)
+	if err != nil {
+		return fmt.Errorf("open file: %w", err)
+	}
+	defer func() {
+		_ = f.Close()
+	}()
+
+	flock := unix.Flock_t{
+		Type:   unix.F_RDLCK,
+		Whence: int16(io.SeekStart),
+		Start:  0,
+		Len:    0,
+	}
+
+	if err := unix.FcntlFlock(f.Fd(), unix.F_SETLKW, &flock); err != nil {
+		_ = f.Close()
+		return fmt.Errorf("set lock: %w", err)
+	}
+	fileRef := &FileLock{file: f}
+	_ = fileRef.Unlock()
+	return nil
+}