storage: flush/sync contents to disk on file close

Behavior controlled by configuration (default=off)
It is a trade-off between performance and consistency.

References:
[1] https://github.com/golang/go/issues/20599

Signed-off-by: Ramkumar Chinchani <rchincha@cisco.com>

pkg/storage/s3/s3_test.go

@@ -53,7 +53,7 @@ func skipIt(t *testing.T) {
 func createMockStorage(rootDir string, store driver.StorageDriver) storage.ImageStore {
 	log := log.Logger{Logger: zerolog.New(os.Stdout)}
 	metrics := monitoring.NewMetricsServer(false, log)
-	il := s3.NewImageStore(rootDir, false, false, log, metrics, store)
+	il := s3.NewImageStore(rootDir, false, false, false, log, metrics, store)
 
 	return il
 }
@@ -86,7 +86,7 @@ func createObjectsStore(rootDir string) (driver.StorageDriver, storage.ImageStor
 
 	log := log.Logger{Logger: zerolog.New(os.Stdout)}
 	metrics := monitoring.NewMetricsServer(false, log)
-	il := s3.NewImageStore(rootDir, false, false, log, metrics, store)
+	il := s3.NewImageStore(rootDir, false, false, false, log, metrics, store)
 
 	return store, il, err
 }

pkg/storage/s3/storage.go

@@ -63,7 +63,7 @@ func (is *ObjectStorage) DirExists(d string) bool {
 
 // NewObjectStorage returns a new image store backed by cloud storages.
 // see https://github.com/docker/docker.github.io/tree/master/registry/storage-drivers
-func NewImageStore(rootDir string, gc bool, dedupe bool, log zlog.Logger, metrics monitoring.MetricServer,
+func NewImageStore(rootDir string, gc, dedupe, commit bool, log zlog.Logger, metrics monitoring.MetricServer,
 	store driver.StorageDriver) storage.ImageStore {
 	imgStore := &ObjectStorage{
 		rootDir:           rootDir,

pkg/storage/scrub_test.go

@@ -36,7 +36,7 @@ func TestCheckAllBlobsIntegrity(t *testing.T) {
 
 	metrics := monitoring.NewMetricsServer(false, log)
 
-	imgStore := storage.NewImageStore(dir, true, true, log, metrics)
+	imgStore := storage.NewImageStore(dir, true, true, true, log, metrics)
 
 	Convey("Scrub only one repo", t, func(c C) {
 		// initialize repo

pkg/storage/storage_fs.go

@@ -29,6 +29,7 @@ import (
 	zerr "zotregistry.io/zot/errors"
 	"zotregistry.io/zot/pkg/extensions/monitoring"
 	zlog "zotregistry.io/zot/pkg/log"
+	"zotregistry.io/zot/pkg/test"
 )
 
 const (
@@ -61,6 +62,7 @@ type ImageStoreFS struct {
 	cache       *Cache
 	gc          bool
 	dedupe      bool
+	commit      bool
 	log         zerolog.Logger
 	metrics     monitoring.MetricServer
 }
@@ -103,7 +105,8 @@ func (sc StoreController) GetImageStore(name string) ImageStore {
 }
 
 // NewImageStore returns a new image store backed by a file storage.
-func NewImageStore(rootDir string, gc bool, dedupe bool, log zlog.Logger, metrics monitoring.MetricServer) ImageStore {
+func NewImageStore(rootDir string, gc, dedupe, commit bool,
+	log zlog.Logger, metrics monitoring.MetricServer) ImageStore {
 	if _, err := os.Stat(rootDir); os.IsNotExist(err) {
 		if err := os.MkdirAll(rootDir, DefaultDirPerms); err != nil {
 			log.Error().Err(err).Str("rootDir", rootDir).Msg("unable to create root dir")
@@ -118,6 +121,7 @@ func NewImageStore(rootDir string, gc bool, dedupe bool, log zlog.Logger, metric
 		blobUploads: make(map[string]BlobUpload),
 		gc:          gc,
 		dedupe:      dedupe,
+		commit:      commit,
 		log:         log.With().Caller().Logger(),
 		metrics:     metrics,
 	}
@@ -203,7 +207,7 @@ func (is *ImageStoreFS) initRepo(name string) error {
 			is.log.Panic().Err(err).Msg("unable to marshal JSON")
 		}
 
-		if err := ioutil.WriteFile(ilPath, buf, DefaultFilePerms); err != nil {
+		if err := is.writeFile(ilPath, buf); err != nil {
 			is.log.Error().Err(err).Str("file", ilPath).Msg("unable to write file")
 
 			return err
@@ -221,7 +225,7 @@ func (is *ImageStoreFS) initRepo(name string) error {
 			is.log.Panic().Err(err).Msg("unable to marshal JSON")
 		}
 
-		if err := ioutil.WriteFile(indexPath, buf, DefaultFilePerms); err != nil {
+		if err := is.writeFile(indexPath, buf); err != nil {
 			is.log.Error().Err(err).Str("file", indexPath).Msg("unable to write file")
 
 			return err
@@ -660,7 +664,7 @@ func (is *ImageStoreFS) PutImageManifest(repo string, reference string, mediaTyp
 	_ = ensureDir(dir, is.log)
 	file := path.Join(dir, mDigest.Encoded())
 
-	if err := ioutil.WriteFile(file, body, DefaultFilePerms); err != nil {
+	if err := is.writeFile(file, body); err != nil {
 		is.log.Error().Err(err).Str("file", file).Msg("unable to write")
 
 		return "", err
@@ -678,7 +682,7 @@ func (is *ImageStoreFS) PutImageManifest(repo string, reference string, mediaTyp
 		return "", err
 	}
 
-	if err := ioutil.WriteFile(file, buf, DefaultFilePerms); err != nil {
+	if err := is.writeFile(file, buf); err != nil {
 		is.log.Error().Err(err).Str("file", file).Msg("unable to write")
 
 		return "", err
@@ -781,7 +785,7 @@ func (is *ImageStoreFS) DeleteImageManifest(repo string, reference string) error
 		return err
 	}
 
-	if err := ioutil.WriteFile(file, buf, DefaultFilePerms); err != nil {
+	if err := is.writeFile(file, buf); err != nil {
 		return err
 	}
 
@@ -884,17 +888,20 @@ func (is *ImageStoreFS) PutBlobChunkStreamed(repo string, uuid string, body io.R
 		return -1, zerr.ErrUploadNotFound
 	}
 
-	file, err := os.OpenFile(
-		blobUploadPath,
-		os.O_WRONLY|os.O_CREATE,
-		DefaultFilePerms,
-	)
+	file, err := os.OpenFile(blobUploadPath, os.O_WRONLY|os.O_CREATE, DefaultFilePerms)
 	if err != nil {
 		is.log.Error().Err(err).Msg("failed to open file")
 
 		return -1, err
 	}
-	defer file.Close()
+
+	defer func() {
+		if is.commit {
+			_ = file.Sync()
+		}
+
+		_ = file.Close()
+	}()
 
 	if _, err := file.Seek(0, io.SeekEnd); err != nil {
 		is.log.Error().Err(err).Msg("failed to seek file")
@@ -929,17 +936,20 @@ func (is *ImageStoreFS) PutBlobChunk(repo string, uuid string, from int64, to in
 		return -1, zerr.ErrBadUploadRange
 	}
 
-	file, err := os.OpenFile(
-		blobUploadPath,
-		os.O_WRONLY|os.O_CREATE,
-		DefaultFilePerms,
-	)
+	file, err := os.OpenFile(blobUploadPath, os.O_WRONLY|os.O_CREATE, DefaultFilePerms)
 	if err != nil {
 		is.log.Error().Err(err).Msg("failed to open file")
 
 		return -1, err
 	}
-	defer file.Close()
+
+	defer func() {
+		if is.commit {
+			_ = file.Sync()
+		}
+
+		_ = file.Close()
+	}()
 
 	if _, err := file.Seek(from, io.SeekStart); err != nil {
 		is.log.Error().Err(err).Msg("failed to seek file")
@@ -1077,7 +1087,13 @@ func (is *ImageStoreFS) FullBlobUpload(repo string, body io.Reader, digest strin
 		return "", -1, zerr.ErrUploadNotFound
 	}
 
-	defer blobFile.Close()
+	defer func() {
+		if is.commit {
+			_ = blobFile.Sync()
+		}
+
+		_ = blobFile.Close()
+	}()
 
 	digester := sha256.New()
 	mw := io.MultiWriter(blobFile, digester)
@@ -1514,6 +1530,30 @@ func (is *ImageStoreFS) GetReferrers(repo, digest string, mediaType string) ([]a
 	return result, nil
 }
 
+func (is *ImageStoreFS) writeFile(filename string, data []byte) error {
+	if !is.commit {
+		return ioutil.WriteFile(filename, data, DefaultFilePerms)
+	}
+
+	fhandle, err := os.OpenFile(filename, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, DefaultFilePerms)
+	if err != nil {
+		return err
+	}
+
+	_, err = fhandle.Write(data)
+
+	if err1 := test.Error(fhandle.Sync()); err1 != nil && err == nil {
+		err = err1
+		is.log.Error().Err(err).Str("filename", filename).Msg("unable to sync file")
+	}
+
+	if err1 := test.Error(fhandle.Close()); err1 != nil && err == nil {
+		err = err1
+	}
+
+	return err
+}
+
 func IsSupportedMediaType(mediaType string) bool {
 	return mediaType == ispec.MediaTypeImageManifest ||
 		mediaType == artifactspec.MediaTypeArtifactManifest

pkg/storage/storage_fs_test.go

@@ -34,7 +34,7 @@ func TestStorageFSAPIs(t *testing.T) {
 
 	log := log.Logger{Logger: zerolog.New(os.Stdout)}
 	metrics := monitoring.NewMetricsServer(false, log)
-	imgStore := storage.NewImageStore(dir, true, true, log, metrics)
+	imgStore := storage.NewImageStore(dir, true, true, true, log, metrics)
 
 	Convey("Repo layout", t, func(c C) {
 		repoName := "test"
@@ -171,7 +171,7 @@ func TestDedupeLinks(t *testing.T) {
 
 	log := log.Logger{Logger: zerolog.New(os.Stdout)}
 	metrics := monitoring.NewMetricsServer(false, log)
-	imgStore := storage.NewImageStore(dir, true, true, log, metrics)
+	imgStore := storage.NewImageStore(dir, true, true, true, log, metrics)
 
 	Convey("Dedupe", t, func(c C) {
 		// manifest1
@@ -311,7 +311,7 @@ func TestDedupe(t *testing.T) {
 
 			log := log.Logger{Logger: zerolog.New(os.Stdout)}
 			metrics := monitoring.NewMetricsServer(false, log)
-			il := storage.NewImageStore(dir, true, true, log, metrics)
+			il := storage.NewImageStore(dir, true, true, true, log, metrics)
 
 			So(il.DedupeBlob("", "", ""), ShouldNotBeNil)
 		})
@@ -330,9 +330,9 @@ func TestNegativeCases(t *testing.T) {
 		log := log.Logger{Logger: zerolog.New(os.Stdout)}
 		metrics := monitoring.NewMetricsServer(false, log)
 
-		So(storage.NewImageStore(dir, true, true, log, metrics), ShouldNotBeNil)
+		So(storage.NewImageStore(dir, true, true, true, log, metrics), ShouldNotBeNil)
 		if os.Geteuid() != 0 {
-			So(storage.NewImageStore("/deadBEEF", true, true, log, metrics), ShouldBeNil)
+			So(storage.NewImageStore("/deadBEEF", true, true, true, log, metrics), ShouldBeNil)
 		}
 	})
 
@@ -345,7 +345,7 @@ func TestNegativeCases(t *testing.T) {
 
 		log := log.Logger{Logger: zerolog.New(os.Stdout)}
 		metrics := monitoring.NewMetricsServer(false, log)
-		imgStore := storage.NewImageStore(dir, true, true, log, metrics)
+		imgStore := storage.NewImageStore(dir, true, true, true, log, metrics)
 
 		err = os.Chmod(dir, 0o000) // remove all perms
 		if err != nil {
@@ -384,7 +384,7 @@ func TestNegativeCases(t *testing.T) {
 
 		log := log.Logger{Logger: zerolog.New(os.Stdout)}
 		metrics := monitoring.NewMetricsServer(false, log)
-		imgStore := storage.NewImageStore(dir, true, true, log, metrics)
+		imgStore := storage.NewImageStore(dir, true, true, true, log, metrics)
 
 		So(imgStore, ShouldNotBeNil)
 		So(imgStore.InitRepo("test"), ShouldBeNil)
@@ -502,7 +502,7 @@ func TestNegativeCases(t *testing.T) {
 
 		log := log.Logger{Logger: zerolog.New(os.Stdout)}
 		metrics := monitoring.NewMetricsServer(false, log)
-		imgStore := storage.NewImageStore(dir, true, true, log, metrics)
+		imgStore := storage.NewImageStore(dir, true, true, true, log, metrics)
 
 		So(imgStore, ShouldNotBeNil)
 		So(imgStore.InitRepo("test"), ShouldBeNil)
@@ -529,7 +529,7 @@ func TestNegativeCases(t *testing.T) {
 
 		log := log.Logger{Logger: zerolog.New(os.Stdout)}
 		metrics := monitoring.NewMetricsServer(false, log)
-		imgStore := storage.NewImageStore(dir, true, true, log, metrics)
+		imgStore := storage.NewImageStore(dir, true, true, true, log, metrics)
 
 		So(imgStore, ShouldNotBeNil)
 		So(imgStore.InitRepo("test"), ShouldBeNil)
@@ -574,7 +574,7 @@ func TestNegativeCases(t *testing.T) {
 
 		log := log.Logger{Logger: zerolog.New(os.Stdout)}
 		metrics := monitoring.NewMetricsServer(false, log)
-		imgStore := storage.NewImageStore(dir, true, true, log, metrics)
+		imgStore := storage.NewImageStore(dir, true, true, true, log, metrics)
 
 		So(imgStore, ShouldNotBeNil)
 		So(imgStore.InitRepo("test"), ShouldBeNil)
@@ -636,7 +636,7 @@ func TestNegativeCases(t *testing.T) {
 
 		log := log.Logger{Logger: zerolog.New(os.Stdout)}
 		metrics := monitoring.NewMetricsServer(false, log)
-		imgStore := storage.NewImageStore(dir, true, true, log, metrics)
+		imgStore := storage.NewImageStore(dir, true, true, true, log, metrics)
 
 		upload, err := imgStore.NewBlobUpload("dedupe1")
 		So(err, ShouldBeNil)
@@ -788,6 +788,55 @@ func TestHardLink(t *testing.T) {
 	})
 }
 
+func TestWriteFile(t *testing.T) {
+	Convey("writeFile with commit", t, func() {
+		dir, err := ioutil.TempDir("", "oci-repo-test")
+		So(err, ShouldBeNil)
+		defer os.RemoveAll(dir)
+
+		log := log.Logger{Logger: zerolog.New(os.Stdout)}
+		metrics := monitoring.NewMetricsServer(false, log)
+		imgStore := storage.NewImageStore(dir, true, true, true, log, metrics)
+
+		Convey("Failure path1", func() {
+			injected := test.InjectFailure(0)
+
+			err := imgStore.InitRepo("repo1")
+			if injected {
+				So(err, ShouldNotBeNil)
+			} else {
+				So(err, ShouldBeNil)
+			}
+		})
+
+		Convey("Failure path2", func() {
+			injected := test.InjectFailure(1)
+
+			err := imgStore.InitRepo("repo2")
+			if injected {
+				So(err, ShouldNotBeNil)
+			} else {
+				So(err, ShouldBeNil)
+			}
+		})
+	})
+
+	Convey("writeFile without commit", t, func() {
+		dir, err := ioutil.TempDir("", "oci-repo-test")
+		So(err, ShouldBeNil)
+		defer os.RemoveAll(dir)
+
+		log := log.Logger{Logger: zerolog.New(os.Stdout)}
+		metrics := monitoring.NewMetricsServer(false, log)
+		imgStore := storage.NewImageStore(dir, true, true, false, log, metrics)
+
+		Convey("Failure path not reached", func() {
+			err := imgStore.InitRepo("repo1")
+			So(err, ShouldBeNil)
+		})
+	})
+}
+
 func randSeq(n int) string {
 	letters := []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
 

pkg/storage/storage_test.go

@@ -72,7 +72,7 @@ func createObjectsStore(rootDir string) (driver.StorageDriver, storage.ImageStor
 	log := log.Logger{Logger: zerolog.New(os.Stdout)}
 	metrics := monitoring.NewMetricsServer(false, log)
 
-	il := s3.NewImageStore(rootDir, false, false, log, metrics, store)
+	il := s3.NewImageStore(rootDir, false, false, false, log, metrics, store)
 
 	return store, il, err
 }
@@ -120,7 +120,7 @@ func TestStorageAPIs(t *testing.T) {
 
 				log := log.Logger{Logger: zerolog.New(os.Stdout)}
 				metrics := monitoring.NewMetricsServer(false, log)
-				imgStore = storage.NewImageStore(dir, true, true, log, metrics)
+				imgStore = storage.NewImageStore(dir, true, true, true, log, metrics)
 			}
 
 			Convey("Repo layout", t, func(c C) {
@@ -711,11 +711,11 @@ func TestStorageHandler(t *testing.T) {
 				metrics := monitoring.NewMetricsServer(false, log)
 
 				// Create ImageStore
-				firstStore = storage.NewImageStore(firstRootDir, false, false, log, metrics)
+				firstStore = storage.NewImageStore(firstRootDir, false, false, false, log, metrics)
 
-				secondStore = storage.NewImageStore(secondRootDir, false, false, log, metrics)
+				secondStore = storage.NewImageStore(secondRootDir, false, false, false, log, metrics)
 
-				thirdStore = storage.NewImageStore(thirdRootDir, false, false, log, metrics)
+				thirdStore = storage.NewImageStore(thirdRootDir, false, false, false, log, metrics)
 			}
 
 			Convey("Test storage handler", t, func() {