Make storage file writes atomic

Author: hiddeco

controllers/helmchart_controller.go

@@ -17,8 +17,10 @@ limitations under the License.
 package controllers
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net/url"
 	"os"
@@ -255,12 +257,8 @@ func (r *HelmChartReconciler) reconcileFromHelmRepository(ctx context.Context,
 		return sourcev1.HelmChartNotReady(chart, sourcev1.ChartPullFailedReason, err.Error()), err
 	}
 
-	chartBytes, err := ioutil.ReadAll(res)
-	if err != nil {
-		return sourcev1.HelmChartNotReady(chart, sourcev1.ChartPullFailedReason, err.Error()), err
-	}
-
-	sum := r.Storage.Checksum(chartBytes)
+	var buf bytes.Buffer
+	sum := r.Storage.Checksum(io.TeeReader(res, &buf))
 	artifact := r.Storage.ArtifactFor(chart.Kind, chart.GetObjectMeta(),
 		fmt.Sprintf("%s-%s-%s.tgz", cv.Name, cv.Version, sum), cv.Version, sum)
 
@@ -280,7 +278,7 @@ func (r *HelmChartReconciler) reconcileFromHelmRepository(ctx context.Context,
 	defer unlock()
 
 	// save artifact to storage
-	err = r.Storage.WriteFile(artifact, chartBytes)
+	err = r.Storage.AtomicWriteFile(artifact, &buf, 0644)
 	if err != nil {
 		err = fmt.Errorf("unable to write chart file: %w", err)
 		return sourcev1.HelmChartNotReady(chart, sourcev1.StorageOperationFailedReason, err.Error()), err

controllers/helmrepository_controller.go

@@ -17,6 +17,7 @@ limitations under the License.
 package controllers
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"io/ioutil"
@@ -212,23 +213,23 @@ func (r *HelmRepositoryReconciler) reconcile(ctx context.Context, repository sou
 		return sourcev1.HelmRepositoryNotReady(repository, sourcev1.IndexationFailedReason, err.Error()), err
 	}
 
-	data, err := ioutil.ReadAll(res)
+	b, err := ioutil.ReadAll(res)
 	if err != nil {
 		return sourcev1.HelmRepositoryNotReady(repository, sourcev1.IndexationFailedReason, err.Error()), err
 	}
 
-	i := &repo.IndexFile{}
-	if err := yaml.Unmarshal(data, i); err != nil {
+	i := repo.IndexFile{}
+	if err := yaml.Unmarshal(b, &i); err != nil {
 		return sourcev1.HelmRepositoryNotReady(repository, sourcev1.IndexationFailedReason, err.Error()), err
 	}
 	i.SortEntries()
 
-	index, err := yaml.Marshal(i)
+	b, err = yaml.Marshal(&i)
 	if err != nil {
 		return sourcev1.HelmRepositoryNotReady(repository, sourcev1.IndexationFailedReason, err.Error()), err
 	}
 
-	sum := r.Storage.Checksum(index)
+	sum := r.Storage.Checksum(bytes.NewReader(b))
 	artifact := r.Storage.ArtifactFor(repository.Kind, repository.ObjectMeta.GetObjectMeta(),
 		fmt.Sprintf("index-%s.yaml", sum), i.Generated.Format(time.RFC3339Nano), sum)
 
@@ -248,7 +249,7 @@ func (r *HelmRepositoryReconciler) reconcile(ctx context.Context, repository sou
 	defer unlock()
 
 	// save artifact to storage
-	err = r.Storage.WriteFile(artifact, index)
+	err = r.Storage.AtomicWriteFile(artifact, bytes.NewReader(b), 0644)
 	if err != nil {
 		err = fmt.Errorf("unable to write repository index file: %w", err)
 		return sourcev1.HelmRepositoryNotReady(repository, sourcev1.StorageOperationFailedReason, err.Error()), err

controllers/helmrepository_controller_test.go

@@ -140,7 +140,7 @@ var _ = Describe("HelmRepositoryReconciler", func() {
 			Eventually(func() error {
 				r := &sourcev1.HelmRepository{}
 				return k8sClient.Get(context.Background(), key, r)
-			}).ShouldNot(Succeed())
+			}, timeout, interval).ShouldNot(Succeed())
 
 			exists := func(path string) bool {
 				// wait for tmp sync on macOS

controllers/storage.go

@@ -36,6 +36,7 @@ import (
 	"github.com/fluxcd/pkg/lockedfile"
 
 	sourcev1 "github.com/fluxcd/source-controller/api/v1alpha1"
+	"github.com/fluxcd/source-controller/internal/fs"
 )
 
 const (
@@ -95,7 +96,8 @@ func (s *Storage) RemoveAll(artifact sourcev1.Artifact) error {
 	return os.RemoveAll(dir)
 }
 
-// RemoveAllButCurrent removes all files for the given artifact base dir excluding the current one
+// RemoveAllButCurrent removes all files for the given v1alpha1.Artifact base dir,
+// excluding the current one.
 func (s *Storage) RemoveAllButCurrent(artifact sourcev1.Artifact) error {
 	localPath := s.LocalPath(artifact)
 	dir := filepath.Dir(localPath)
@@ -120,8 +122,8 @@ func (s *Storage) RemoveAllButCurrent(artifact sourcev1.Artifact) error {
 	return nil
 }
 
-// ArtifactExist returns a boolean indicating whether the artifact exists in storage and is a
-// regular file.
+// ArtifactExist returns a boolean indicating whether the v1alpha1.Artifact exists in storage
+// and is a regular file.
 func (s *Storage) ArtifactExist(artifact sourcev1.Artifact) bool {
 	fi, err := os.Lstat(s.LocalPath(artifact))
 	if err != nil {
@@ -130,9 +132,10 @@ func (s *Storage) ArtifactExist(artifact sourcev1.Artifact) bool {
 	return fi.Mode().IsRegular()
 }
 
-// Archive creates a tar.gz to the artifact path from the given dir excluding any VCS specific
-// files and directories, or any of the excludes defined in the excludeFiles.
-func (s *Storage) Archive(artifact sourcev1.Artifact, dir string, spec sourcev1.GitRepositorySpec) error {
+// Archive atomically creates a tar.gz to the v1alpha1.Artifact path from the given dir,
+// excluding any VCS specific files and directories, or any of the excludes defined in
+// the excludeFiles.
+func (s *Storage) Archive(artifact sourcev1.Artifact, dir string, spec sourcev1.GitRepositorySpec) (err error) {
 	if _, err := os.Stat(dir); err != nil {
 		return err
 	}
@@ -141,22 +144,53 @@ func (s *Storage) Archive(artifact sourcev1.Artifact, dir string, spec sourcev1.
 	if err != nil {
 		return err
 	}
-
 	matcher := gitignore.NewMatcher(ps)
 
-	gzFile, err := os.Create(s.LocalPath(artifact))
+	localPath := s.LocalPath(artifact)
+	tmpGzFile, err := ioutil.TempFile(filepath.Split(localPath))
 	if err != nil {
 		return err
 	}
-	defer gzFile.Close()
-
-	gw := gzip.NewWriter(gzFile)
-	defer gw.Close()
+	tmpName := tmpGzFile.Name()
+	defer func() {
+		if err != nil {
+			os.Remove(tmpName)
+		}
+	}()
 
+	gw := gzip.NewWriter(tmpGzFile)
 	tw := tar.NewWriter(gw)
-	defer tw.Close()
+	if err := writeToArchiveExcludeMatches(dir, matcher, tw); err != nil {
+		tw.Close()
+		gw.Close()
+		tmpGzFile.Close()
+		return err
+	}
+
+	if err := tw.Close(); err != nil {
+		gw.Close()
+		tmpGzFile.Close()
+		return err
+	}
+	if err := gw.Close(); err != nil {
+		tmpGzFile.Close()
+		return err
+	}
+	if err := tmpGzFile.Close(); err != nil {
+		return err
+	}
+
+	if err := os.Chmod(tmpName, 0644); err != nil {
+		return err
+	}
 
-	return filepath.Walk(dir, func(p string, fi os.FileInfo, err error) error {
+	return fs.RenameWithFallback(tmpName, localPath)
+}
+
+// writeToArchiveExcludeMatches walks over the given dir and writes any regular file that does
+// not match the given gitignore.Matcher.
+func writeToArchiveExcludeMatches(dir string, matcher gitignore.Matcher, writer *tar.Writer) error {
+	fn := func(p string, fi os.FileInfo, err error) error {
 		if err != nil {
 			return err
 		}
@@ -187,33 +221,48 @@ func (s *Storage) Archive(artifact sourcev1.Artifact, dir string, spec sourcev1.
 		}
 		header.Name = relFilePath
 
-		if err := tw.WriteHeader(header); err != nil {
+		if err := writer.WriteHeader(header); err != nil {
 			return err
 		}
 
 		f, err := os.Open(p)
 		if err != nil {
+			f.Close()
 			return err
 		}
-		if _, err := io.Copy(tw, f); err != nil {
+		if _, err := io.Copy(writer, f); err != nil {
 			f.Close()
 			return err
 		}
 		return f.Close()
-	})
+	}
+	return filepath.Walk(dir, fn)
 }
 
-// WriteFile writes the given bytes to the artifact path if the checksum differs
-func (s *Storage) WriteFile(artifact sourcev1.Artifact, data []byte) error {
+// AtomicWriteFile atomically writes a file to the v1alpha1.Artifact Path.
+func (s *Storage) AtomicWriteFile(artifact sourcev1.Artifact, reader io.Reader, mode os.FileMode) (err error) {
 	localPath := s.LocalPath(artifact)
-	sum := s.Checksum(data)
-	if file, err := os.Stat(localPath); !os.IsNotExist(err) && !file.IsDir() {
-		if fb, err := ioutil.ReadFile(localPath); err == nil && sum == s.Checksum(fb) {
-			return nil
+	tmpFile, err := ioutil.TempFile(filepath.Split(localPath))
+	if err != nil {
+		return err
+	}
+	tmpName := tmpFile.Name()
+	defer func() {
+		if err != nil {
+			os.Remove(tmpName)
 		}
+	}()
+	if _, err := io.Copy(tmpFile, reader); err != nil {
+		tmpFile.Close()
+		return err
 	}
-
-	return ioutil.WriteFile(localPath, data, 0644)
+	if err := tmpFile.Close(); err != nil {
+		return err
+	}
+	if err := os.Chmod(tmpName, mode); err != nil {
+		return err
+	}
+	return fs.RenameWithFallback(tmpName, localPath)
 }
 
 // Symlink creates or updates a symbolic link for the given artifact
@@ -241,12 +290,14 @@ func (s *Storage) Symlink(artifact sourcev1.Artifact, linkName string) (string,
 	return url, nil
 }
 
-// Checksum returns the SHA1 checksum for the given bytes as a string
-func (s *Storage) Checksum(b []byte) string {
-	return fmt.Sprintf("%x", sha1.Sum(b))
+// Checksum returns the SHA1 checksum for the data of the given io.Reader as a string.
+func (s *Storage) Checksum(reader io.Reader) string {
+	h := sha1.New()
+	_, _ = io.Copy(h, reader)
+	return fmt.Sprintf("%x", h.Sum(nil))
 }
 
-// Lock creates a file lock for the given artifact
+// Lock creates a file lock for the given v1alpha1.Artifact.
 func (s *Storage) Lock(artifact sourcev1.Artifact) (unlock func(), err error) {
 	lockFile := s.LocalPath(artifact) + ".lock"
 	mutex := lockedfile.MutexAt(lockFile)
@@ -262,6 +313,8 @@ func (s *Storage) LocalPath(artifact sourcev1.Artifact) string {
 	return filepath.Join(s.BasePath, artifact.Path)
 }
 
+// getPatterns collects ignore patterns from the given reader and returns them
+// as a gitignore.Pattern slice.
 func getPatterns(reader io.Reader, path []string) []gitignore.Pattern {
 	var ps []gitignore.Pattern
 	scanner := bufio.NewScanner(reader)

internal/fs/LICENSE

@@ -0,0 +1,27 @@
+Copyright (c) 2014 The Go Authors. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+   * Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+   * Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following disclaimer
+in the documentation and/or other materials provided with the
+distribution.
+   * Neither the name of Google Inc. nor the names of its
+contributors may be used to endorse or promote products derived from
+this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.