Move file operations into internal/atomicfiles

Author: tchap

pkg/operator/staticpod/certsyncpod/certsync_controller.go

@@ -1,9 +1,7 @@
 package certsyncpod
 
 import (
-	"bytes"
 	"context"
-	"fmt"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -19,8 +17,8 @@ import (
 
 	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
-	"github.com/openshift/library-go/pkg/operator/staticpod"
 	"github.com/openshift/library-go/pkg/operator/staticpod/controller/installer"
+	"github.com/openshift/library-go/pkg/operator/staticpod/internal/atomicfiles"
 )
 
 type CertSyncController struct {
@@ -154,7 +152,7 @@ func (c *CertSyncController) sync(ctx context.Context, syncCtx factory.SyncConte
 			continue
 		}
 
-		errors = append(errors, writeFiles(&realFS, c.eventRecorder, "configmap", configMap.ObjectMeta, contentDir, data, 0644))
+		errors = append(errors, writeFiles(c.eventRecorder, "configmap", configMap.ObjectMeta, contentDir, data, 0644))
 	}
 
 	klog.Infof("Syncing secrets: %v", c.secrets)
@@ -239,104 +237,20 @@ func (c *CertSyncController) sync(ctx context.Context, syncCtx factory.SyncConte
 			continue
 		}
 
-		errors = append(errors, writeFiles(&realFS, c.eventRecorder, "secret", secret.ObjectMeta, contentDir, data, 0600))
+		errors = append(errors, writeFiles(c.eventRecorder, "secret", secret.ObjectMeta, contentDir, data, 0600))
 	}
 	return utilerrors.NewAggregate(errors)
 }
 
-type fileSystem struct {
-	MkdirAll              func(path string, perm os.FileMode) error
-	MkdirTemp             func(dir, pattern string) (string, error)
-	RemoveAll             func(path string) error
-	WriteFile             func(name string, data []byte, perm os.FileMode) error
-	SwapDirectoriesAtomic func(dirA, dirB string) error
-	HashDirectory         func(path string) ([]byte, error)
-}
-
-var realFS = fileSystem{
-	MkdirAll:              os.MkdirAll,
-	MkdirTemp:             os.MkdirTemp,
-	RemoveAll:             os.RemoveAll,
-	WriteFile:             os.WriteFile,
-	SwapDirectoriesAtomic: staticpod.SwapDirectoriesAtomic,
-	HashDirectory:         hashDirectory,
-}
-
 func writeFiles[C string | []byte](
-	fs *fileSystem, eventRecorder events.Recorder,
+	eventRecorder events.Recorder,
 	typeName string, o metav1.ObjectMeta,
 	targetDir string, files map[string]C, filePerm os.FileMode,
 ) error {
-	// We are doing to prepare a tmp directory and write all files into that directory.
-	// Then we are going to atomically swap the new data directory for the old one.
-	// This is currently implemented as really atomically exchanging directories.
-	//
-	// The same goal of atomic swap could be implemented using symlinks much like AtomicWriter does in
-	// https://github.com/kubernetes/kubernetes/blob/v1.34.0/pkg/volume/util/atomic_writer.go#L58
-	// The reason we don't do that is that we already have a directory populated and watched that needs to we swapped,
-	// in other words, it's for compatibility reasons. And if we were to migrate to the symlink approach,
-	// we would anyway need to atomically turn the current data directory to a symlink.
-	// This would all just increase complexity and require atomic swap on the OS level anyway.
-
-	// In case the target directory does not exist, create it so that the directory not existing is not a special case.
-	klog.Infof("Ensuring content directory %q exists ...", targetDir)
-	if err := fs.MkdirAll(targetDir, 0755); err != nil && !os.IsExist(err) {
-		eventRecorder.Warningf("CertificateUpdateFailed", "Failed creating content directory for %s: %s/%s: %v", typeName, o.Namespace, o.Name, err)
-		return err
-	}
-
-	// We make sure the target directory is unchanged while we prepare the switch by computing the directory hash.
-	klog.Infof("Hashing current content directory %q ...", targetDir)
-	targetDirHashBefore, err := fs.HashDirectory(targetDir)
-	if err != nil {
-		eventRecorder.Warningf("CertificateUpdateFailed", "Failed to hash current content directory for %s: %s/%s: %v", typeName, o.Namespace, o.Name, err)
-		return err
-	}
-
-	// Create a tmp source directory to be swapped.
-	klog.Infof("Creating temporary directory to swap for %q ...", targetDir)
-	tmpDir, err := fs.MkdirTemp(filepath.Dir(targetDir), filepath.Base(targetDir)+"-*")
-	if err != nil {
-		eventRecorder.Warningf("CertificateUpdateFailed", "Failed to create temporary directory for %s: %s/%s: %v", typeName, o.Namespace, o.Name, err)
+	if err := atomicfiles.WriteFiles(typeName, o, targetDir, files, filePerm); err != nil {
+		eventRecorder.Warning("CertificateUpdateFailed", err.Error())
 		return err
 	}
-	defer func() {
-		if err := fs.RemoveAll(tmpDir); err != nil {
-			klog.Errorf("Failed to remove temporary directory %q during cleanup: %v", tmpDir, err)
-		}
-	}()
-
-	// Populate the tmp directory with files.
-	for filename, content := range files {
-		fullFilename := filepath.Join(tmpDir, filename)
-		klog.Infof("Writing %s manifest %q ...", typeName, fullFilename)
-
-		if err := fs.WriteFile(fullFilename, []byte(content), filePerm); err != nil {
-			eventRecorder.Warningf("CertificateUpdateFailed", "Failed writing file for %s: %s/%s: %v", typeName, o.Namespace, o.Name, err)
-			return err
-		}
-	}
-
-	// Make sure the target directory hasn't changed in the meantime.
-	klog.Infof("Hashing current content directory %q again and ensuring it's unchanged ...", targetDir)
-	targetDirHashAfter, err := fs.HashDirectory(targetDir)
-	if err != nil {
-		eventRecorder.Warningf("CertificateUpdateFailed", "Failed to hash current content directory for %s: %s/%s: %v", typeName, o.Namespace, o.Name, err)
-		return err
-	}
-	if !bytes.Equal(targetDirHashBefore, targetDirHashAfter) {
-		eventRecorder.Warningf("CertificateUpdateFailed", "Content directory changed while preparing to apply an update for %s: %s/%s", typeName, o.Namespace, o.Name)
-		klog.Warningf("Content directory changed while preparing to apply an update: %q", targetDir)
-		return fmt.Errorf("content directory changed while preparing to apply an update: %q", targetDir)
-	}
-
-	// Swap directories atomically.
-	klog.Infof("Atomically swapping target directory %q with temporary directory %q for %s: %s/%s ...", targetDir, tmpDir, typeName, o.Namespace, o.Name)
-	if err := fs.SwapDirectoriesAtomic(targetDir, tmpDir); err != nil {
-		eventRecorder.Warningf("CertificateUpdateFailed", "Failed to swap target directory %q with temporary directory %q for %s: %s/%s: %v", targetDir, tmpDir, typeName, o.Namespace, o.Name, err)
-		return err
-	}
-
 	eventRecorder.Eventf("CertificateUpdated", "Wrote updated %s: %s/%s", typeName, o.Namespace, o.Name)
 	return nil
 }

pkg/operator/staticpod/certsyncpod/certsync_controller_linux_test.go

@@ -85,7 +85,7 @@ func TestDynamicCertificates(t *testing.T) {
 	}
 	err = wait.PollUntilContextCancel(ctx, 250*time.Millisecond, true, func(ctx context.Context) (bool, error) {
 		// Replace the secret directory.
-		if err := writeFiles(&realFS, recorder, typeName, om, secretDir, files, 0600); err != nil {
+		if err := writeFiles(recorder, typeName, om, secretDir, files, 0600); err != nil {
 			t.Errorf("Failed to write files: %v", err)
 			return false, err
 		}

pkg/operator/staticpod/file_utils_linux.go renamed to pkg/operator/staticpod/internal/atomicfiles/swap_directories_linux.go

@@ -1,6 +1,6 @@
 //go:build linux
 
-package staticpod
+package atomicfiles
 
 import (
 	"fmt"
@@ -9,10 +9,10 @@ import (
 	"golang.org/x/sys/unix"
 )
 
-// SwapDirectoriesAtomic can be used to swap two directories atomically.
+// SwapDirectories can be used to swap two directories atomically.
 //
 // This function requires absolute paths and will return an error if that's not the case.
-func SwapDirectoriesAtomic(dirA, dirB string) error {
+func SwapDirectories(dirA, dirB string) error {
 	if !filepath.IsAbs(dirA) {
 		return fmt.Errorf("not an absolute path: %q", dirA)
 	}

pkg/operator/staticpod/file_utils_linux_test.go renamed to pkg/operator/staticpod/internal/atomicfiles/swap_directories_linux_test.go

@@ -1,6 +1,6 @@
 //go:build linux
 
-package staticpod
+package atomicfiles
 
 import (
 	"bytes"
@@ -13,7 +13,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 )
 
-func TestSwapDirectoriesAtomic(t *testing.T) {
+func TestSwapDirectories(t *testing.T) {
 	expectNoError := func(t *testing.T, err error) {
 		t.Helper()
 		if err != nil {
@@ -129,7 +129,7 @@ func TestSwapDirectoriesAtomic(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			pathA, pathB := tc.setup(t, t.TempDir())
-			tc.checkResult(t, pathA, pathB, SwapDirectoriesAtomic(pathA, pathB))
+			tc.checkResult(t, pathA, pathB, SwapDirectories(pathA, pathB))
 		})
 	}
 }

pkg/operator/staticpod/file_utils_other.go renamed to pkg/operator/staticpod/internal/atomicfiles/swap_directories_other.go

@@ -1,6 +1,6 @@
 //go:build !linux
 
-package staticpod
+package atomicfiles
 
 import (
 	"fmt"
@@ -9,11 +9,11 @@ import (
 	"path/filepath"
 )
 
-// SwapDirectoriesAtomic swaps two directories, but NOT atomically in this case.
+// SwapDirectories swaps two directories, but NOT atomically in this case.
 // Atomic implementation is only available for Linux.
 // This function is essentially a mock for tests, and it simply uses os.Rename.
 // In case there is any error, the swapping process is left in an inconsistent state.
-func SwapDirectoriesAtomic(dirA, dirB string) error {
+func SwapDirectories(dirA, dirB string) error {
 	// Still retain the constraints as in the Linux implementation.
 	if !filepath.IsAbs(dirA) {
 		return fmt.Errorf("not an absolute path: %q", dirA)

pkg/operator/staticpod/internal/atomicfiles/write_files.go

@@ -0,0 +1,85 @@
+package atomicfiles
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/klog/v2"
+)
+
+// WriteFiles can be used to atomically write given files into the target directory.
+// This is done by populating a temporary directory, then atomically swapping it with the target directory.
+func WriteFiles[C string | []byte](
+	typeName string, o metav1.ObjectMeta, targetDir string, files map[string]C, filePerm os.FileMode,
+) error {
+	return writeFiles(&realFS, typeName, o, targetDir, files, filePerm)
+}
+
+type fileSystem struct {
+	MkdirAll              func(path string, perm os.FileMode) error
+	MkdirTemp             func(dir, pattern string) (string, error)
+	RemoveAll             func(path string) error
+	WriteFile             func(name string, data []byte, perm os.FileMode) error
+	SwapDirectoriesAtomic func(dirA, dirB string) error
+}
+
+var realFS = fileSystem{
+	MkdirAll:              os.MkdirAll,
+	MkdirTemp:             os.MkdirTemp,
+	RemoveAll:             os.RemoveAll,
+	WriteFile:             os.WriteFile,
+	SwapDirectoriesAtomic: SwapDirectories,
+}
+
+func writeFiles[C string | []byte](
+	fs *fileSystem, typeName string, o metav1.ObjectMeta,
+	targetDir string, files map[string]C, filePerm os.FileMode,
+) error {
+	// We are doing to prepare a tmp directory and write all files into that directory.
+	// Then we are going to atomically swap the new data directory for the old one.
+	// This is currently implemented as really atomically exchanging directories.
+	//
+	// The same goal of atomic swap could be implemented using symlinks much like AtomicWriter does in
+	// https://github.com/kubernetes/kubernetes/blob/v1.34.0/pkg/volume/util/atomic_writer.go#L58
+	// The reason we don't do that is that we already have a directory populated and watched that needs to we swapped,
+	// in other words, it's for compatibility reasons. And if we were to migrate to the symlink approach,
+	// we would anyway need to atomically turn the current data directory to a symlink.
+	// This would all just increase complexity and require atomic swap on the OS level anyway.
+
+	// In case the target directory does not exist, create it so that the directory not existing is not a special case.
+	klog.Infof("Ensuring content directory %q exists ...", targetDir)
+	if err := fs.MkdirAll(targetDir, 0755); err != nil && !os.IsExist(err) {
+		return fmt.Errorf("failed creating content directory for %s: %s/%s: %w", typeName, o.Namespace, o.Name, err)
+	}
+
+	// Create a tmp source directory to be swapped.
+	klog.Infof("Creating temporary directory to swap for %q ...", targetDir)
+	tmpDir, err := fs.MkdirTemp(filepath.Dir(targetDir), filepath.Base(targetDir)+"-*")
+	if err != nil {
+		return fmt.Errorf("failed creating temporary directory for %s: %s/%s: %w", typeName, o.Namespace, o.Name, err)
+	}
+	defer func() {
+		if err := fs.RemoveAll(tmpDir); err != nil {
+			klog.Errorf("Failed to remove temporary directory %q during cleanup: %v", tmpDir, err)
+		}
+	}()
+
+	// Populate the tmp directory with files.
+	for filename, content := range files {
+		fullFilename := filepath.Join(tmpDir, filename)
+		klog.Infof("Writing %s manifest %q ...", typeName, fullFilename)
+
+		if err := fs.WriteFile(fullFilename, []byte(content), filePerm); err != nil {
+			return fmt.Errorf("failed writing file for %s: %s/%s: %w", typeName, o.Namespace, o.Name, err)
+		}
+	}
+
+	// Swap directories atomically.
+	klog.Infof("Atomically swapping target directory %q with temporary directory %q for %s: %s/%s ...", targetDir, tmpDir, typeName, o.Namespace, o.Name)
+	if err := fs.SwapDirectoriesAtomic(targetDir, tmpDir); err != nil {
+		return fmt.Errorf("failed swapping target directory %q with temporary directory %q for %s: %s/%s: %w", targetDir, tmpDir, typeName, o.Namespace, o.Name, err)
+	}
+	return nil
+}

pkg/operator/staticpod/certsyncpod/certsync_controller_test.go renamed to pkg/operator/staticpod/internal/atomicfiles/write_files_test.go

@@ -1,4 +1,4 @@
-package certsyncpod
+package atomicfiles
 
 import (
 	"errors"
@@ -8,8 +8,6 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
-
-	"github.com/openshift/library-go/pkg/operator/events/eventstesting"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -289,27 +287,10 @@ func TestWriteFiles(t *testing.T) {
 			}
 			return fs
 		}),
-		errorTestCase("directory unchanged on content directory hash mismatch", func() *fileSystem {
-			fs := newRealFS()
-			fs.HashDirectory = hashDirectoryFuncWithReturnValues("hash", "mismatch")
-			return fs
-		}),
-		errorTestCase("directory unchanged on content directory initial hash error", func() *fileSystem {
-			fs := newRealFS()
-			fs.HashDirectory = hashDirectoryFuncWithReturnValues("error: nuked")
-			return fs
-		}),
-		errorTestCase("directory unchanged on content directory subsequent hash error", func() *fileSystem {
-			fs := newRealFS()
-			fs.HashDirectory = hashDirectoryFuncWithReturnValues("hash", "error: nuked")
-			return fs
-		}),
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			recorder := eventstesting.NewTestingEventRecorder(t)
-
 			// Write the current directory contents.
 			contentDir := filepath.Join(t.TempDir(), "secrets", om.Name)
 			if tc.directoryData != nil {
@@ -326,7 +307,7 @@ func TestWriteFiles(t *testing.T) {
 			}
 
 			// Replace with the object data.
-			err := writeFiles(tc.newFS(), recorder, typeName, om, contentDir, tc.objectData, 0600)
+			err := writeFiles(tc.newFS(), typeName, om, contentDir, tc.objectData, 0600)
 
 			// Check the resulting state.
 			tc.checkState(t, contentDir, tc.directoryData, tc.objectData, err)