Disk cache object walker

The disk cache object walker is a background worker that continually
walks through the directory structure for the objects stored by the
disk cache (keyed by the LeaseKeyer). For every file encountered,
the walker will delete the file if stale (i.e. older than one hour).

5 files changed, 242 insertions, 4 deletions

diff --git a/changelogs/unreleased/po-diskcache-walker.yml b/changelogs/unreleased/po-diskcache-walker.yml
new file mode 100644
index 000000000..f48bf1d3f
--- /dev/null
+++ b/changelogs/unreleased/po-diskcache-walker.yml
@@ -0,0 +1,5 @@
+---
+title: Disk cache object walker
+merge_request: 1362
+author:
+type: added
diff --git a/internal/cache/export_test.go b/internal/cache/export_test.go
new file mode 100644
index 000000000..cb1f65b67
--- /dev/null
+++ b/internal/cache/export_test.go
@@ -0,0 +1,31 @@
+package cache
+
+import "sync"
+
+var (
+	ExportMockRemovalCounter = new(mockCounter)
+	ExportMockCheckCounter   = new(mockCounter)
+)
+
+type mockCounter struct {
+	sync.RWMutex
+	count int
+}
+
+func (mc *mockCounter) Add(n int) {
+	mc.Lock()
+	mc.count += n
+	mc.Unlock()
+}
+
+func (mc *mockCounter) Count() int {
+	mc.RLock()
+	defer mc.RUnlock()
+	return mc.count
+}
+
+func init() {
+	// override counter functions with our mocked version
+	countWalkRemoval = func() { ExportMockRemovalCounter.Add(1) }
+	countWalkCheck = func() { ExportMockCheckCounter.Add(1) }
+}
diff --git a/internal/cache/prometheus.go b/internal/cache/prometheus.go
index 45672f24b..f4b7c9a11 100644
--- a/internal/cache/prometheus.go
+++ b/internal/cache/prometheus.go
@@ -34,6 +34,18 @@ var (
 		},
 		[]string{"error"},
 	)
+	walkerCheckTotal = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "gitaly_diskcache_walker_check_total",
+			Help: "Total number of events during diskcache filesystem walks",
+		},
+	)
+	walkerRemovalTotal = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "gitaly_diskcache_walker_removal_total",
+			Help: "Total number of events during diskcache filesystem walks",
+		},
+	)
 )
 
 func init() {
@@ -42,6 +54,8 @@ func init() {
 	prometheus.MustRegister(bytesStoredtotals)
 	prometheus.MustRegister(bytesFetchedtotals)
 	prometheus.MustRegister(errTotal)
+	prometheus.MustRegister(walkerCheckTotal)
+	prometheus.MustRegister(walkerRemovalTotal)
 }
 
 func countErr(err error) error {
@@ -54,7 +68,11 @@ func countErr(err error) error {
 	return err
 }
 
-func countRequest()             { requestTotals.Inc() }
-func countMiss()                { missTotals.Inc() }
-func countWriteBytes(n float64) { bytesStoredtotals.Add(n) }
-func countReadBytes(n float64)  { bytesFetchedtotals.Add(n) }
+var (
+	countRequest     = func() { requestTotals.Inc() }
+	countMiss        = func() { missTotals.Inc() }
+	countWriteBytes  = func(n float64) { bytesStoredtotals.Add(n) }
+	countReadBytes   = func(n float64) { bytesFetchedtotals.Add(n) }
+	countWalkRemoval = func() { walkerRemovalTotal.Inc() }
+	countWalkCheck   = func() { walkerCheckTotal.Inc() }
+)
diff --git a/internal/cache/walker.go b/internal/cache/walker.go
new file mode 100644
index 000000000..a782ebd6b
--- /dev/null
+++ b/internal/cache/walker.go
@@ -0,0 +1,74 @@
+// Package cache supplies background workers for periodically cleaning the
+// cache folder on all storages listed in the config file. Upon configuration
+// validation, one worker will be started for each storage. The worker will
+// walk the cache directory tree and remove any files older than one hour. The
+// worker will walk the cache directory every ten minutes.
+package cache
+
+import (
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	"gitlab.com/gitlab-org/gitaly/internal/config"
+	"gitlab.com/gitlab-org/gitaly/internal/tempdir"
+)
+
+func cleanWalk(storagePath string) error {
+	cachePath := filepath.Join(storagePath, tempdir.CachePrefix)
+
+	return filepath.Walk(cachePath, func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+
+		if info.IsDir() {
+			return nil
+		}
+
+		countWalkCheck()
+
+		threshold := time.Now().Add(-1 * staleAge)
+		if info.ModTime().After(threshold) {
+			return nil
+		}
+
+		if err := os.Remove(path); err != nil {
+			if os.IsNotExist(err) {
+				// race condition: another file walker on the same storage may
+				// have deleted the file already
+				return nil
+			}
+			return err
+		}
+
+		countWalkRemoval()
+
+		return nil
+	})
+}
+
+const cleanWalkFrequency = 10 * time.Minute
+
+func startCleanWalker(storage config.Storage) {
+	logrus.WithField("storage", storage.Name).Info("Starting disk cache object walker")
+	walkTick := time.NewTicker(cleanWalkFrequency)
+	go func() {
+		for {
+			if err := cleanWalk(storage.Path); err != nil {
+				logrus.WithField("storage", storage.Name).Error(err)
+			}
+			<-walkTick.C
+		}
+	}()
+}
+
+func init() {
+	config.RegisterHook(func() error {
+		for _, storage := range config.Config.Storages {
+			startCleanWalker(storage)
+		}
+		return nil
+	})
+}
diff --git a/internal/cache/walker_test.go b/internal/cache/walker_test.go
new file mode 100644
index 000000000..ef6f9b9e3
--- /dev/null
+++ b/internal/cache/walker_test.go
@@ -0,0 +1,110 @@
+package cache_test
+
+import (
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gitlab.com/gitlab-org/gitaly/internal/cache"
+	"gitlab.com/gitlab-org/gitaly/internal/config"
+	"gitlab.com/gitlab-org/gitaly/internal/tempdir"
+)
+
+func TestDiskCacheObjectWalker(t *testing.T) {
+	tmpPath, err := ioutil.TempDir("", t.Name())
+	require.NoError(t, err)
+	defer func() { require.NoError(t, os.RemoveAll(tmpPath)) }()
+
+	oldStorages := config.Config.Storages
+	config.Config.Storages = []config.Storage{
+		{
+			Name: t.Name(),
+			Path: tmpPath,
+		},
+	}
+	defer func() { config.Config.Storages = oldStorages }()
+
+	satisfyConfigValidation(tmpPath)
+
+	var shouldExist, shouldNotExist []string
+
+	for _, tt := range []struct {
+		name          string
+		age           time.Duration
+		expectRemoval bool
+	}{
+		{"0f/oldey", time.Hour, true},
+		{"90/n00b", time.Minute, false},
+		{"2b/ancient", 24 * time.Hour, true},
+		{"cd/baby", time.Second, false},
+	} {
+		path := filepath.Join(tmpPath, tempdir.CachePrefix, tt.name)
+		require.NoError(t, os.MkdirAll(filepath.Dir(path), 0755))
+
+		f, err := os.Create(path)
+		require.NoError(t, err)
+		require.NoError(t, f.Close())
+
+		require.NoError(t, os.Chtimes(path, time.Now(), time.Now().Add(-1*tt.age)))
+
+		if tt.expectRemoval {
+			shouldNotExist = append(shouldNotExist, path)
+		} else {
+			shouldExist = append(shouldExist, path)
+		}
+	}
+
+	expectChecks := cache.ExportMockCheckCounter.Count() + 4
+	expectRemovals := cache.ExportMockRemovalCounter.Count() + 2
+
+	require.NoError(t, config.Validate()) // triggers walker
+
+	pollCountersUntil(t, expectChecks, expectRemovals)
+
+	for _, p := range shouldExist {
+		assert.FileExists(t, p)
+	}
+
+	for _, p := range shouldNotExist {
+		_, err := os.Stat(p)
+		require.True(t, os.IsNotExist(err), "expected %s not to exist", p)
+	}
+}
+
+// satisfyConfigValidation puts garbage values in the config file to satisfy
+// validation
+func satisfyConfigValidation(tmpPath string) {
+	config.Config.ListenAddr = "meow"
+	config.Config.GitlabShell = config.GitlabShell{
+		Dir: tmpPath,
+	}
+	config.Config.Ruby = config.Ruby{
+		Dir: tmpPath,
+	}
+}
+
+func pollCountersUntil(t testing.TB, expectChecks, expectRemovals int) {
+	// poll injected mock prometheus counters until expected events occur
+	timeout := time.After(time.Second)
+	for {
+		select {
+		case <-timeout:
+			t.Fatalf(
+				"timed out polling prometheus stats; checks: %d removals: %d",
+				cache.ExportMockCheckCounter.Count(),
+				cache.ExportMockRemovalCounter.Count(),
+			)
+		default:
+			// keep on truckin'
+		}
+		if cache.ExportMockCheckCounter.Count() == expectChecks &&
+			cache.ExportMockRemovalCounter.Count() == expectRemovals {
+			break
+		}
+		time.Sleep(time.Millisecond)
+	}
+}