From c37bd10f8cfc4cb6df5513881eed53e6c5f0be09 Mon Sep 17 00:00:00 2001 From: Derek McGowan Date: Tue, 9 Aug 2016 11:55:17 -0700 Subject: [PATCH] Update layer store to sync transaction files before committing Fixes case where shutdown occurs before content is synced to disked on layer creation. This case can leave the layer store in an bad state and require manual recovery. This change ensures all files are synced to disk before a layer is committed. Any shutdown that occurs will only cause the layer to not show up but will allow it to be repulled or recreated without error. Added generic io logic to ioutils package to abstract it out of the layer store package. Signed-off-by: Derek McGowan --- layer/filestore.go | 28 +++++----- pkg/ioutils/fswriters.go | 80 +++++++++++++++++++++++++++++ pkg/ioutils/fswriters_test.go | 97 ++++++++++++++++++++++++++++++++++- 3 files changed, 189 insertions(+), 16 deletions(-) diff --git a/layer/filestore.go b/layer/filestore.go index 6361af647c..42b45556e3 100644 --- a/layer/filestore.go +++ b/layer/filestore.go @@ -34,7 +34,7 @@ type fileMetadataStore struct { type fileMetadataTransaction struct { store *fileMetadataStore - root string + ws *ioutils.AtomicWriteSet } // NewFSMetadataStore returns an instance of a metadata store @@ -71,33 +71,32 @@ func (fms *fileMetadataStore) StartTransaction() (MetadataTransaction, error) { if err := os.MkdirAll(tmpDir, 0755); err != nil { return nil, err } - - td, err := ioutil.TempDir(tmpDir, "layer-") + ws, err := ioutils.NewAtomicWriteSet(tmpDir) if err != nil { return nil, err } - // Create a new tempdir + return &fileMetadataTransaction{ store: fms, - root: td, + ws: ws, }, nil } func (fm *fileMetadataTransaction) SetSize(size int64) error { content := fmt.Sprintf("%d", size) - return ioutil.WriteFile(filepath.Join(fm.root, "size"), []byte(content), 0644) + return fm.ws.WriteFile("size", []byte(content), 0644) } func (fm *fileMetadataTransaction) SetParent(parent ChainID) error { - return ioutil.WriteFile(filepath.Join(fm.root, "parent"), []byte(digest.Digest(parent).String()), 0644) + return fm.ws.WriteFile("parent", []byte(digest.Digest(parent).String()), 0644) } func (fm *fileMetadataTransaction) SetDiffID(diff DiffID) error { - return ioutil.WriteFile(filepath.Join(fm.root, "diff"), []byte(digest.Digest(diff).String()), 0644) + return fm.ws.WriteFile("diff", []byte(digest.Digest(diff).String()), 0644) } func (fm *fileMetadataTransaction) SetCacheID(cacheID string) error { - return ioutil.WriteFile(filepath.Join(fm.root, "cache-id"), []byte(cacheID), 0644) + return fm.ws.WriteFile("cache-id", []byte(cacheID), 0644) } func (fm *fileMetadataTransaction) SetDescriptor(ref distribution.Descriptor) error { @@ -105,11 +104,11 @@ func (fm *fileMetadataTransaction) SetDescriptor(ref distribution.Descriptor) er if err != nil { return err } - return ioutil.WriteFile(filepath.Join(fm.root, "descriptor.json"), jsonRef, 0644) + return fm.ws.WriteFile("descriptor.json", jsonRef, 0644) } func (fm *fileMetadataTransaction) TarSplitWriter(compressInput bool) (io.WriteCloser, error) { - f, err := os.OpenFile(filepath.Join(fm.root, "tar-split.json.gz"), os.O_TRUNC|os.O_CREATE|os.O_WRONLY, 0644) + f, err := fm.ws.FileWriter("tar-split.json.gz", os.O_TRUNC|os.O_CREATE|os.O_WRONLY, 0644) if err != nil { return nil, err } @@ -131,15 +130,16 @@ func (fm *fileMetadataTransaction) Commit(layer ChainID) error { if err := os.MkdirAll(filepath.Dir(finalDir), 0755); err != nil { return err } - return os.Rename(fm.root, finalDir) + + return fm.ws.Commit(finalDir) } func (fm *fileMetadataTransaction) Cancel() error { - return os.RemoveAll(fm.root) + return fm.ws.Cancel() } func (fm *fileMetadataTransaction) String() string { - return fm.root + return fm.ws.String() } func (fms *fileMetadataStore) GetSize(layer ChainID) (int64, error) { diff --git a/pkg/ioutils/fswriters.go b/pkg/ioutils/fswriters.go index 6dc50a03dc..a56c462651 100644 --- a/pkg/ioutils/fswriters.go +++ b/pkg/ioutils/fswriters.go @@ -80,3 +80,83 @@ func (w *atomicFileWriter) Close() (retErr error) { } return nil } + +// AtomicWriteSet is used to atomically write a set +// of files and ensure they are visible at the same time. +// Must be committed to a new directory. +type AtomicWriteSet struct { + root string +} + +// NewAtomicWriteSet creates a new atomic write set to +// atomically create a set of files. The given directory +// is used as the base directory for storing files before +// commit. If no temporary directory is given the system +// default is used. +func NewAtomicWriteSet(tmpDir string) (*AtomicWriteSet, error) { + td, err := ioutil.TempDir(tmpDir, "write-set-") + if err != nil { + return nil, err + } + + return &AtomicWriteSet{ + root: td, + }, nil +} + +// WriteFile writes a file to the set, guaranteeing the file +// has been synced. +func (ws *AtomicWriteSet) WriteFile(filename string, data []byte, perm os.FileMode) error { + f, err := ws.FileWriter(filename, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, perm) + if err != nil { + return err + } + n, err := f.Write(data) + if err == nil && n < len(data) { + err = io.ErrShortWrite + } + if err1 := f.Close(); err == nil { + err = err1 + } + return err +} + +type syncFileCloser struct { + *os.File +} + +func (w syncFileCloser) Close() error { + err := w.File.Sync() + if err1 := w.File.Close(); err == nil { + err = err1 + } + return err +} + +// FileWriter opens a file writer inside the set. The file +// should be synced and closed before calling commit. +func (ws *AtomicWriteSet) FileWriter(name string, flag int, perm os.FileMode) (io.WriteCloser, error) { + f, err := os.OpenFile(filepath.Join(ws.root, name), flag, perm) + if err != nil { + return nil, err + } + return syncFileCloser{f}, nil +} + +// Cancel cancels the set and removes all temporary data +// created in the set. +func (ws *AtomicWriteSet) Cancel() error { + return os.RemoveAll(ws.root) +} + +// Commit moves all created files to the target directory. The +// target directory must not exist and the parent of the target +// directory must exist. +func (ws *AtomicWriteSet) Commit(target string) error { + return os.Rename(ws.root, target) +} + +// String returns the location the set is writing to. +func (ws *AtomicWriteSet) String() string { + return ws.root +} diff --git a/pkg/ioutils/fswriters_test.go b/pkg/ioutils/fswriters_test.go index 470ca1a6f4..c4d1419306 100644 --- a/pkg/ioutils/fswriters_test.go +++ b/pkg/ioutils/fswriters_test.go @@ -5,9 +5,21 @@ import ( "io/ioutil" "os" "path/filepath" + "runtime" "testing" ) +var ( + testMode os.FileMode = 0640 +) + +func init() { + // Windows does not support full Linux file mode + if runtime.GOOS == "windows" { + testMode = 0666 + } +} + func TestAtomicWriteToFile(t *testing.T) { tmpDir, err := ioutil.TempDir("", "atomic-writers-test") if err != nil { @@ -16,7 +28,7 @@ func TestAtomicWriteToFile(t *testing.T) { defer os.RemoveAll(tmpDir) expected := []byte("barbaz") - if err := AtomicWriteFile(filepath.Join(tmpDir, "foo"), expected, 0666); err != nil { + if err := AtomicWriteFile(filepath.Join(tmpDir, "foo"), expected, testMode); err != nil { t.Fatalf("Error writing to file: %v", err) } @@ -33,7 +45,88 @@ func TestAtomicWriteToFile(t *testing.T) { if err != nil { t.Fatalf("Error statting file: %v", err) } - if expected := os.FileMode(0666); st.Mode() != expected { + if expected := os.FileMode(testMode); st.Mode() != expected { t.Fatalf("Mode mismatched, expected %o, got %o", expected, st.Mode()) } } + +func TestAtomicWriteSetCommit(t *testing.T) { + tmpDir, err := ioutil.TempDir("", "atomic-writerset-test") + if err != nil { + t.Fatalf("Error when creating temporary directory: %s", err) + } + defer os.RemoveAll(tmpDir) + + if err := os.Mkdir(filepath.Join(tmpDir, "tmp"), 0700); err != nil { + t.Fatalf("Error creating tmp directory: %s", err) + } + + targetDir := filepath.Join(tmpDir, "target") + ws, err := NewAtomicWriteSet(filepath.Join(tmpDir, "tmp")) + if err != nil { + t.Fatalf("Error creating atomic write set: %s", err) + } + + expected := []byte("barbaz") + if err := ws.WriteFile("foo", expected, testMode); err != nil { + t.Fatalf("Error writing to file: %v", err) + } + + if _, err := ioutil.ReadFile(filepath.Join(targetDir, "foo")); err == nil { + t.Fatalf("Expected error reading file where should not exist") + } + + if err := ws.Commit(targetDir); err != nil { + t.Fatalf("Error committing file: %s", err) + } + + actual, err := ioutil.ReadFile(filepath.Join(targetDir, "foo")) + if err != nil { + t.Fatalf("Error reading from file: %v", err) + } + + if bytes.Compare(actual, expected) != 0 { + t.Fatalf("Data mismatch, expected %q, got %q", expected, actual) + } + + st, err := os.Stat(filepath.Join(targetDir, "foo")) + if err != nil { + t.Fatalf("Error statting file: %v", err) + } + if expected := os.FileMode(testMode); st.Mode() != expected { + t.Fatalf("Mode mismatched, expected %o, got %o", expected, st.Mode()) + } + +} + +func TestAtomicWriteSetCancel(t *testing.T) { + tmpDir, err := ioutil.TempDir("", "atomic-writerset-test") + if err != nil { + t.Fatalf("Error when creating temporary directory: %s", err) + } + defer os.RemoveAll(tmpDir) + + if err := os.Mkdir(filepath.Join(tmpDir, "tmp"), 0700); err != nil { + t.Fatalf("Error creating tmp directory: %s", err) + } + + ws, err := NewAtomicWriteSet(filepath.Join(tmpDir, "tmp")) + if err != nil { + t.Fatalf("Error creating atomic write set: %s", err) + } + + expected := []byte("barbaz") + if err := ws.WriteFile("foo", expected, testMode); err != nil { + t.Fatalf("Error writing to file: %v", err) + } + + if err := ws.Cancel(); err != nil { + t.Fatalf("Error committing file: %s", err) + } + + if _, err := ioutil.ReadFile(filepath.Join(tmpDir, "target", "foo")); err == nil { + t.Fatalf("Expected error reading file where should not exist") + } else if !os.IsNotExist(err) { + t.Fatalf("Unexpected error reading file: %s", err) + } +}