Correct clean existing cache files during startup (treeverse#1237)

arielshaqed · web-flow · commit bfc4503fa0ea · 2021-01-19T16:26:45.000+02:00
* Correct clean existing cache files during startup Also improve warnings on immediate rejection from cache. Fixes treeverse#1183. * [CR] Use tfs.eviction member where possible * [CR] Fix workspace cleanup on startup * [CR] Keep deleting files when ristretto cache.Set() fails A synchronous failure is due to the ristretto worker goroutine not being set up correctly (possibly during cache shutdown), so onReject might not be called. But continue to log nicely, of course, to debug such weird behaviour. * Keep evidence of failed tier_fs tests This is a temporary directory that is left behind on test failure. In CI it does not matter because the container dies anyway; on a developer machine it helps. Also improve test error detection in TestStartup.
diff --git a/pyramid/eviction.go b/pyramid/eviction.go
@@ -4,7 +4,7 @@ import (
 	"fmt"
 
 	"github.com/dgraph-io/ristretto"
-	params "github.com/treeverse/lakefs/pyramid/params"
+	"github.com/treeverse/lakefs/pyramid/params"
 )
 
 // nolint: unused
diff --git a/pyramid/params/params.go b/pyramid/params/params.go
@@ -11,11 +11,11 @@ type RelativePath string
 // Eviction abstracts eviction control.
 type Eviction interface {
 	// Touch indicates the eviction that the file has been used now
-	Touch(rPath RelativePath)
+	Touch(path RelativePath)
 
 	// Store orders the eviction to Store the path.
 	// returns true iff the eviction accepted the path.
-	Store(rPath RelativePath, filesize int64) bool
+	Store(path RelativePath, filesize int64) bool
 }
 
 // LocalDiskParams is pyramid.FS params that are identical for all file-systems
diff --git a/pyramid/tier_fs.go b/pyramid/tier_fs.go
@@ -48,7 +48,7 @@ func NewFS(c *params.InstanceParams) (FS, error) {
 		return nil, fmt.Errorf("creating base dir: %s - %w", fsLocalBaseDir, err)
 	}
 
-	tierFS := &TierFS{
+	tfs := &TierFS{
 		adapter:        c.Adapter,
 		fsName:         c.FSName,
 		logger:         c.Logger,
@@ -59,44 +59,43 @@ func NewFS(c *params.InstanceParams) (FS, error) {
 	}
 	if c.Eviction == nil {
 		var err error
-		c.Eviction, err = newRistrettoEviction(c.AllocatedBytes(), tierFS.removeFromLocal)
+		c.Eviction, err = newRistrettoEviction(c.AllocatedBytes(), tfs.removeFromLocal)
 		if err != nil {
 			return nil, fmt.Errorf("creating eviction control: %w", err)
 		}
 	}
 
-	tierFS.eviction = c.Eviction
-	if err := handleExistingFiles(tierFS.eviction, fsLocalBaseDir); err != nil {
+	tfs.eviction = c.Eviction
+	if err := tfs.handleExistingFiles(); err != nil {
 		return nil, fmt.Errorf("handling existing files: %w", err)
 	}
 
-	return tierFS, nil
+	return tfs, nil
 }
 
 // handleExistingFiles should only be called during init of the TierFS.
 // It does 2 things:
 // 1. Adds stored files to the eviction control
 // 2. Remove workspace directories and all its content if it
 //	  exist under the namespace dir.
-func handleExistingFiles(eviction params.Eviction, fsLocalBaseDir string) error {
-	if err := filepath.Walk(fsLocalBaseDir, func(rPath string, info os.FileInfo, err error) error {
+func (tfs *TierFS) handleExistingFiles() error {
+	if err := filepath.Walk(tfs.fsLocalBaseDir, func(p string, info os.FileInfo, err error) error {
 		if err != nil {
 			return err
 		}
 		if info.IsDir() {
 			if info.Name() == workspaceDir {
 				// skipping workspaces and saving them for later delete
-				if err := os.RemoveAll(rPath); err != nil {
+				if err := os.RemoveAll(p); err != nil {
 					return fmt.Errorf("removing dir: %w", err)
 				}
 				return filepath.SkipDir
 			}
 			return nil
 		}
 
-		if err := storeLocalFile(rPath, info.Size(), eviction); err != nil {
-			return err
-		}
+		rPath := strings.TrimPrefix(p, tfs.fsLocalBaseDir)
+		tfs.storeLocalFile(params.RelativePath(rPath), info.Size())
 		return nil
 	}); err != nil {
 		return fmt.Errorf("walking the fs dir: %w", err)
@@ -121,7 +120,7 @@ func (tfs *TierFS) removeFromLocalInternal(rPath params.RelativePath) {
 		return
 	}
 
-	if err := tfs.syncDir.deleteDirRecIfEmpty(path.Dir(p)); err != nil {
+	if err := tfs.syncDir.deleteDirRecIfEmpty(path.Dir(string(rPath))); err != nil {
 		tfs.logger.WithError(err).Error("Failed deleting empty dir")
 		errorsTotal.WithLabelValues(tfs.fsName, "DirRemoval")
 	}
@@ -240,10 +239,16 @@ func (tfs *TierFS) openFile(fileRef localFileRef, fh *os.File) (*ROFile, error)
 	}
 
 	if !tfs.eviction.Store(fileRef.fsRelativePath, stat.Size()) {
-		// This is where we get less strict.
-		// Ideally, newly fetched file will never be rejected by the cache.
-		// But if it did, we prefer to serve the file and delete it.
-		// When the user will close the file, the file will be deleted from the disk too.
+		tfs.logger.WithFields(logging.Fields{
+			"namespace": fileRef.namespace,
+			"file":      fileRef.filename,
+			"full_path": fileRef.fullPath,
+		}).Info("stored file immediately rejected from cache (delete but continue)")
+
+		// A rare occurrence, (currently) happens when Ristretto cache is not set up
+		// to perform any caching.  So be less strict: prefer to serve the file and
+		// delete it from the cache. It will be removed from disk when the last
+		// surviving file descriptor -- returned from this function -- is closed.
 		if err := os.Remove(fileRef.fullPath); err != nil {
 			return nil, err
 		}
@@ -334,17 +339,6 @@ func (tfs *TierFS) openWithLock(ctx context.Context, fileRef localFileRef) (*os.
 	return fh, nil
 }
 
-func storeLocalFile(rPath string, size int64, eviction params.Eviction) error {
-	relativePath := params.RelativePath(rPath)
-	if !eviction.Store(relativePath, size) {
-		err := os.Remove(rPath)
-		if err != nil {
-			return fmt.Errorf("removing file: %w", err)
-		}
-	}
-	return nil
-}
-
 func validateFilename(filename string) error {
 	if strings.HasPrefix(filename, workspaceDir+string(os.PathSeparator)) {
 		return errPathInWorkspace
@@ -363,13 +357,36 @@ type localFileRef struct {
 	fsRelativePath params.RelativePath
 }
 
+func (tfs *TierFS) storeLocalFile(rPath params.RelativePath, size int64) {
+	if !tfs.eviction.Store(rPath, size) {
+		// Rejected from cache, so deleted.  This is safe, but can only happen when
+		// the cache size was lowered -- so warn.
+		tfs.logger.WithFields(logging.Fields{
+			"path": rPath,
+			"size": size,
+		}).Warn("existing file immediately rejected from cache on startup (safe if cache size changed; continue)")
+
+		// A rare occurrence, (currently) happens when Ristretto cache is not set up
+		// to perform any caching.  So be less strict: prefer to serve the file and
+		// delete it from the cache. It will be removed from disk when the last
+		// surviving file descriptor -- returned from this function -- is closed.
+		if err := os.Remove(string(rPath)); err != nil {
+			tfs.logger.WithFields(logging.Fields{
+				"path": rPath,
+				"size": size,
+			}).Error("failed to delete immediately-rejected existing file from cache on startup")
+			return
+		}
+	}
+}
+
 func (tfs *TierFS) newLocalFileRef(namespace, nsPath, filename string) localFileRef {
-	relative := path.Join(nsPath, filename)
+	rPath := path.Join(nsPath, filename)
 	return localFileRef{
 		namespace:      namespace,
 		filename:       filename,
-		fsRelativePath: params.RelativePath(relative),
-		fullPath:       path.Join(tfs.fsLocalBaseDir, relative),
+		fsRelativePath: params.RelativePath(rPath),
+		fullPath:       path.Join(tfs.fsLocalBaseDir, rPath),
 	}
 }
 
diff --git a/pyramid/tier_fs_test.go b/pyramid/tier_fs_test.go
@@ -14,6 +14,7 @@ import (
 	"testing"
 
 	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/treeverse/lakefs/block/mem"
 	"github.com/treeverse/lakefs/logging"
@@ -73,8 +74,12 @@ func TestStartup(t *testing.T) {
 	// cleanup
 	baseDir := path.Join(os.TempDir(), fsName)
 	defer func() {
+		if t.Failed() {
+			// Leave behind the evidence.
+			return
+		}
 		if err := os.RemoveAll(baseDir); err != nil {
-			t.Fatal("Remove all filed under", baseDir, err)
+			t.Fatal("Remove all files under", baseDir, err)
 		}
 	}()
 
@@ -112,16 +117,18 @@ func TestStartup(t *testing.T) {
 	}
 
 	dir, err := os.Open(workspacePath)
-	require.Nil(t, dir)
-	require.True(t, os.IsNotExist(err))
+	assert.Nil(t, dir, "expected to fail to open %s", workspacePath)
+	// os.IsNotExist does not look as hard to errors.Is; for errors returned directly from
+	// package os this does not matter.
+	assert.Error(t, err, os.ErrNotExist, "expected %s not to exist", workspacePath)
 
 	f, err := localFS.Open(ctx, namespace, filename)
 	defer func() { _ = f.Close() }()
-	require.NoError(t, err)
+	assert.NoError(t, err)
 
 	data, err := ioutil.ReadAll(f)
-	require.NoError(t, err)
-	require.Equal(t, content, data)
+	assert.NoError(t, err)
+	assert.Equal(t, content, data)
 }
 
 func testEviction(t *testing.T, namespaces ...string) {

Original file line number	Diff line number	Diff line change
`@@ -4,7 +4,7 @@ import (`
`4`	`4`	`"fmt"`
`5`	`5`
`6`	`6`	`"github.com/dgraph-io/ristretto"`
`7`		`- params "github.com/treeverse/lakefs/pyramid/params"`
	`7`	`+ "github.com/treeverse/lakefs/pyramid/params"`
`8`	`8`	`)`
`9`	`9`
`10`	`10`	`// nolint: unused`