Fix cache walker to only walk each path once

7 files changed, 59 insertions, 41 deletions

diff --git a/changelogs/unreleased/po-cache-walker-one-per-path.yml b/changelogs/unreleased/po-cache-walker-one-per-path.yml
new file mode 100644
index 000000000..e97d8d4be
--- /dev/null
+++ b/changelogs/unreleased/po-cache-walker-one-per-path.yml
@@ -0,0 +1,5 @@
+---
+title: Fix cache walker to only walk each path once
+merge_request: 1769
+author:
+type: fixed
diff --git a/internal/cache/keyer.go b/internal/cache/keyer.go
index 86957c3fe..350c13851 100644
--- a/internal/cache/keyer.go
+++ b/internal/cache/keyer.go
@@ -200,6 +200,11 @@ func newPendingLease(repo *gitalypb.Repository) (string, error) {
 		return "", err
 	}
 
+	lPath := latestPath(repoStatePath)
+	if err := os.Remove(lPath); err != nil && !os.IsNotExist(err) {
+		return "", err
+	}
+
 	pDir := pendingDir(repoStatePath)
 	if err := os.MkdirAll(pDir, 0755); err != nil {
 		return "", err
@@ -207,6 +212,7 @@ func newPendingLease(repo *gitalypb.Repository) (string, error) {
 
 	f, err := ioutil.TempFile(pDir, "")
 	if err != nil {
+		err = fmt.Errorf("creating pending lease failed: %v", err)
 		return "", err
 	}
 
diff --git a/internal/cache/prometheus.go b/internal/cache/prometheus.go
index 40f5927e0..ee3a7b0a1 100644
--- a/internal/cache/prometheus.go
+++ b/internal/cache/prometheus.go
@@ -2,7 +2,6 @@ package cache
 
 import (
 	"github.com/prometheus/client_golang/prometheus"
-	"gitlab.com/gitlab-org/gitaly/internal/config"
 )
 
 var (
@@ -61,19 +60,17 @@ var (
 			Help: "Total number of errors during diskcache filesystem walks",
 		},
 	)
-	walkerEmptyDirTotal = prometheus.NewCounterVec(
+	walkerEmptyDirTotal = prometheus.NewCounter(
 		prometheus.CounterOpts{
 			Name: "gitaly_diskcache_walker_empty_dir_total",
 			Help: "Total number of empty directories encountered",
 		},
-		[]string{"storage"},
 	)
-	walkerEmptyDirRemovalTotal = prometheus.NewCounterVec(
+	walkerEmptyDirRemovalTotal = prometheus.NewCounter(
 		prometheus.CounterOpts{
 			Name: "gitaly_diskcache_walker_empty_dir_removal_total",
 			Help: "Total number of empty directories removed",
 		},
-		[]string{"storage"},
 	)
 )
 
@@ -110,6 +107,6 @@ var (
 	countWalkRemoval     = func() { walkerRemovalTotal.Inc() }
 	countWalkCheck       = func() { walkerCheckTotal.Inc() }
 	countWalkError       = func() { walkerErrorTotal.Inc() }
-	countEmptyDir        = func(s config.Storage) { walkerEmptyDirTotal.With(prometheus.Labels{"storage": s.Name}).Inc() }
-	countEmptyDirRemoval = func(s config.Storage) { walkerEmptyDirRemovalTotal.With(prometheus.Labels{"storage": s.Name}).Inc() }
+	countEmptyDir        = func() { walkerEmptyDirTotal.Inc() }
+	countEmptyDirRemoval = func() { walkerEmptyDirRemovalTotal.Inc() }
 )
diff --git a/internal/cache/walker.go b/internal/cache/walker.go
index bdc13b634..4408cadba 100644
--- a/internal/cache/walker.go
+++ b/internal/cache/walker.go
@@ -9,7 +9,6 @@ import (
 	"io/ioutil"
 	"os"
 	"path/filepath"
-	"sync"
 	"time"
 
 	"github.com/sirupsen/logrus"
@@ -27,7 +26,7 @@ func logWalkErr(err error, path, msg string) {
 		Warn(msg)
 }
 
-func cleanWalk(s config.Storage, path string) error {
+func cleanWalk(path string) error {
 	defer time.Sleep(100 * time.Microsecond) // relieve pressure
 
 	countWalkCheck()
@@ -44,7 +43,7 @@ func cleanWalk(s config.Storage, path string) error {
 		ePath := filepath.Join(path, e.Name())
 
 		if e.IsDir() {
-			if err := cleanWalk(s, ePath); err != nil {
+			if err := cleanWalk(ePath); err != nil {
 				return err
 			}
 			continue
@@ -76,7 +75,7 @@ func cleanWalk(s config.Storage, path string) error {
 	}
 
 	if len(files) == 0 {
-		countEmptyDir(s)
+		countEmptyDir()
 		if err := os.Remove(path); err != nil {
 			if os.IsNotExist(err) {
 				return nil
@@ -84,7 +83,7 @@ func cleanWalk(s config.Storage, path string) error {
 			logWalkErr(err, path, "unable to remove empty directory")
 			return err
 		}
-		countEmptyDirRemoval(s)
+		countEmptyDirRemoval()
 		countWalkRemoval()
 	}
 
@@ -93,13 +92,15 @@ func cleanWalk(s config.Storage, path string) error {
 
 const cleanWalkFrequency = 10 * time.Minute
 
-func walkLoop(s config.Storage, walkPath string) {
-	logrus.WithField("storage", s.Name).Infof("Starting file walker for %s", walkPath)
+func walkLoop(walkPath string) {
+	logger := logrus.WithField("path", walkPath)
+	logger.Infof("Starting file walker for %s", walkPath)
+
 	walkTick := time.NewTicker(cleanWalkFrequency)
 	dontpanic.GoForever(time.Minute, func() {
 		for {
-			if err := cleanWalk(s, walkPath); err != nil {
-				logrus.WithField("storage", s.Name).Error(err)
+			if err := cleanWalk(walkPath); err != nil {
+				logger.Error(err)
 			}
 
 			<-walkTick.C
@@ -107,13 +108,13 @@ func walkLoop(s config.Storage, walkPath string) {
 	})
 }
 
-func startCleanWalker(storage config.Storage) {
+func startCleanWalker(storagePath string) {
 	if disableWalker {
 		return
 	}
 
-	walkLoop(storage, tempdir.CacheDir(storage))
-	walkLoop(storage, tempdir.StateDir(storage))
+	walkLoop(tempdir.AppendCacheDir(storagePath))
+	walkLoop(tempdir.AppendStateDir(storagePath))
 }
 
 var (
@@ -123,15 +124,15 @@ var (
 
 // moveAndClear will move the cache to the storage location's
 // temporary folder, and then remove its contents asynchronously
-func moveAndClear(storage config.Storage) error {
+func moveAndClear(storagePath string) error {
 	if disableMoveAndClear {
 		return nil
 	}
 
-	logger := logrus.WithField("storage", storage.Name)
+	logger := logrus.WithField("path", storagePath)
 	logger.Info("clearing disk cache object folder")
 
-	tempPath := tempdir.TempDir(storage)
+	tempPath := tempdir.AppendTempDir(storagePath)
 	if err := os.MkdirAll(tempPath, 0755); err != nil {
 		return err
 	}
@@ -142,7 +143,7 @@ func moveAndClear(storage config.Storage) error {
 	}
 
 	logger.Infof("moving disk cache object folder to %s", tmpDir)
-	cachePath := tempdir.CacheDir(storage)
+	cachePath := tempdir.AppendCacheDir(storagePath)
 	if err := os.Rename(cachePath, filepath.Join(tmpDir, "moved")); err != nil {
 		if os.IsNotExist(err) {
 			logger.Info("disk cache object folder doesn't exist, no need to remove")
@@ -165,21 +166,18 @@ func moveAndClear(storage config.Storage) error {
 }
 
 func init() {
-	oncePerStorage := map[string]*sync.Once{}
-	var err error
-
 	config.RegisterHook(func(cfg config.Cfg) error {
+		pathSet := map[string]struct{}{}
 		for _, storage := range cfg.Storages {
-			if _, ok := oncePerStorage[storage.Name]; !ok {
-				oncePerStorage[storage.Name] = new(sync.Once)
+			pathSet[storage.Path] = struct{}{}
+		}
+
+		for sPath := range pathSet {
+			if err := moveAndClear(sPath); err != nil {
+				return err
 			}
-			oncePerStorage[storage.Name].Do(func() {
-				if err = moveAndClear(storage); err != nil {
-					return
-				}
-				startCleanWalker(storage)
-			})
+			startCleanWalker(sPath)
 		}
-		return err
+		return nil
 	})
 }
diff --git a/internal/cache/walker_internal_test.go b/internal/cache/walker_internal_test.go
index c39ac548b..bd5d3e768 100644
--- a/internal/cache/walker_internal_test.go
+++ b/internal/cache/walker_internal_test.go
@@ -11,11 +11,10 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"gitlab.com/gitlab-org/gitaly/internal/config"
 )
 
 func TestCleanWalkDirNotExists(t *testing.T) {
-	err := cleanWalk(config.Storage{}, "/path/that/does/not/exist")
+	err := cleanWalk("/path/that/does/not/exist")
 	assert.NoError(t, err, "cleanWalk returned an error for a non existing directory")
 }
 
@@ -47,7 +46,7 @@ func TestCleanWalkEmptyDirs(t *testing.T) {
 		}
 	}
 
-	require.NoError(t, cleanWalk(config.Storage{}, tmp))
+	require.NoError(t, cleanWalk(tmp))
 
 	actual := findFiles(t, tmp)
 	expect := `.
diff --git a/internal/cache/walker_test.go b/internal/cache/walker_test.go
index d18bcf5c3..50e4a55c5 100644
--- a/internal/cache/walker_test.go
+++ b/internal/cache/walker_test.go
@@ -163,6 +163,7 @@ func pollCountersUntil(t testing.TB, expectRemovals int) {
 		if cache.ExportMockRemovalCounter.Count() == expectRemovals {
 			break
 		}
+		t.Log(cache.ExportMockRemovalCounter.Count())
 		time.Sleep(time.Millisecond)
 	}
 }
diff --git a/internal/tempdir/tempdir.go b/internal/tempdir/tempdir.go
index 55e3be5e5..ba2481e85 100644
--- a/internal/tempdir/tempdir.go
+++ b/internal/tempdir/tempdir.go
@@ -41,13 +41,25 @@ const (
 )
 
 // CacheDir returns the path to the cache dir for a storage location
-func CacheDir(storage config.Storage) string { return filepath.Join(storage.Path, cachePrefix) }
+func CacheDir(storage config.Storage) string { return AppendCacheDir(storage.Path) }
+
+// AppendCacheDir will append the cache directory convention to the storage path
+// provided
+func AppendCacheDir(storagePath string) string { return filepath.Join(storagePath, cachePrefix) }
 
 // StateDir returns the path to the state dir for a storage location
-func StateDir(storage config.Storage) string { return filepath.Join(storage.Path, statePrefix) }
+func StateDir(storage config.Storage) string { return AppendStateDir(storage.Path) }
+
+// AppendStateDir will append the state directory convention to the storage path
+// provided
+func AppendStateDir(storagePath string) string { return filepath.Join(storagePath, statePrefix) }
 
 // TempDir returns the path to the temp dir for a storage location
-func TempDir(storage config.Storage) string { return filepath.Join(storage.Path, tmpRootPrefix) }
+func TempDir(storage config.Storage) string { return AppendTempDir(storage.Path) }
+
+// AppendTempDir will append the temp directory convention to the storage path
+// provided
+func AppendTempDir(storagePath string) string { return filepath.Join(storagePath, tmpRootPrefix) }
 
 // ForDeleteAllRepositories returns a temporary directory for the given storage. It is not context-scoped but it will get removed eventuall (after MaxAge).
 func ForDeleteAllRepositories(storageName string) (string, error) {