diff --git a/engine/repositories/cleaner_test.go b/engine/repositories/cleaner_test.go new file mode 100644 index 0000000000..e9e8fdfa6d --- /dev/null +++ b/engine/repositories/cleaner_test.go @@ -0,0 +1,68 @@ +package repositories + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/ovh/cds/sdk" +) + +func Test_vacuumFileSystemCleanerFunc(t *testing.T) { + s, err := newTestService(t) + require.NoError(t, err) + s.Cfg.Basedir = t.TempDir() + + otherDao := dao{store: s.Cache, hostname: "other-instance"} + + tests := []struct { + name string + protect func(repoID string) + expectKept bool + }{ + { + name: "clone without last access is removed", + protect: nil, + expectKept: false, + }, + { + name: "clone recently used by this instance is kept", + protect: func(repoID string) { s.dao.saveLastAccess(repoID, time.Now().Add(time.Minute), 60) }, + expectKept: true, + }, + { + name: "last access from another instance does not protect the clone", + protect: func(repoID string) { otherDao.saveLastAccess(repoID, time.Now().Add(time.Minute), 60) }, + expectKept: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + repoID := sdk.UUID() + path := filepath.Join(s.Cfg.Basedir, repoID) + require.NoError(t, os.MkdirAll(path, os.FileMode(0700))) + t.Cleanup(func() { + _ = s.Cache.Delete(s.dao.lastAccessKey(repoID)) + _ = s.Cache.Delete(otherDao.lastAccessKey(repoID)) + }) + + if tt.protect != nil { + tt.protect(repoID) + } + + require.NoError(t, s.vacuumFileSystemCleanerFunc(context.TODO(), repoID)) + + _, err := os.Stat(path) + if tt.expectKept { + require.NoError(t, err, "clone directory must still exist") + } else { + require.True(t, os.IsNotExist(err), "clone directory must have been removed") + } + }) + } +} diff --git a/engine/repositories/dao.go b/engine/repositories/dao.go index 3a02de56aa..0073de579e 100644 --- a/engine/repositories/dao.go +++ b/engine/repositories/dao.go @@ -12,14 +12,25 @@ import ( ) var ( - rootKey = cache.Key("repositories", "operations") - processorKey = cache.Key("repositories", "processor") - locksKey = cache.Key("repositories", "locks") - lastAccessKey = cache.Key("repositories", "lastAccess") + rootKey = cache.Key("repositories", "operations") + processorKey = cache.Key("repositories", "processor") + locksKey = cache.Key("repositories", "locks") + lastAccessRootKey = cache.Key("repositories", "lastAccess") ) type dao struct { store cache.Store + // hostname scopes lastAccess keys to this instance: clones live on a local + // filesystem, so retention must reflect each instance's own usage. + hostname string +} + +func (d *dao) lastAccessKey(repoUUID string) string { + return cache.Key(lastAccessRootKey, d.hostname, repoUUID) +} + +func (d *dao) saveLastAccess(repoUUID string, expiration time.Time, ttlSeconds int) { + d.store.SetWithTTL(d.lastAccessKey(repoUUID), expiration, ttlSeconds) } func (d *dao) saveOperation(o *sdk.Operation) error { @@ -88,7 +99,7 @@ func (d *dao) unlock(ctx context.Context, uuid string) error { } func (d *dao) isExpired(ctx context.Context, uuid string) (time.Time, bool) { - k := cache.Key(lastAccessKey, uuid) + k := d.lastAccessKey(uuid) var v time.Time find, err := d.store.Get(k, &v) if err != nil { diff --git a/engine/repositories/dao_test.go b/engine/repositories/dao_test.go new file mode 100644 index 0000000000..5d78229293 --- /dev/null +++ b/engine/repositories/dao_test.go @@ -0,0 +1,32 @@ +package repositories + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/ovh/cds/sdk" +) + +func Test_daoLastAccessPerInstance(t *testing.T) { + s, err := newTestService(t) + require.NoError(t, err) + + daoA := dao{store: s.Cache, hostname: "instance-a"} + daoB := dao{store: s.Cache, hostname: "instance-b"} + + repoID := sdk.UUID() + t.Cleanup(func() { + _ = s.Cache.Delete(daoA.lastAccessKey(repoID)) + }) + + daoA.saveLastAccess(repoID, time.Now().Add(time.Minute), 60) + + _, expired := daoA.isExpired(context.TODO(), repoID) + require.False(t, expired, "repository must not be expired for the instance that accessed it") + + _, expired = daoB.isExpired(context.TODO(), repoID) + require.True(t, expired, "a last access from another instance must not protect this instance's clone") +} diff --git a/engine/repositories/processor.go b/engine/repositories/processor.go index 9e196d0e7c..c1e7c2cb6a 100644 --- a/engine/repositories/processor.go +++ b/engine/repositories/processor.go @@ -7,7 +7,6 @@ import ( "github.com/rockbears/log" - "github.com/ovh/cds/engine/cache" "github.com/ovh/cds/sdk" cdslog "github.com/ovh/cds/sdk/log" ) @@ -76,7 +75,7 @@ func (s *Service) do(ctx context.Context, op sdk.Operation) error { ttl := int(ttlDuration.Seconds()) ttlTime := time.Now().Add(ttlDuration) log.Info(ctx, "%s protected until %s", r.ID(), ttlTime.String()) - s.dao.store.SetWithTTL(cache.Key(lastAccessKey, r.ID()), ttlTime, ttl) + s.dao.saveLastAccess(r.ID(), ttlTime, ttl) }() switch { diff --git a/engine/repositories/repositories.go b/engine/repositories/repositories.go index f0312b45c0..05c306c64a 100644 --- a/engine/repositories/repositories.go +++ b/engine/repositories/repositories.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "net/http" + "os" "time" "github.com/gorilla/mux" @@ -114,9 +115,15 @@ func (s *Service) Serve(c context.Context) error { MaxHeaderBytes: 1 << 20, } + hostname, err := os.Hostname() + if err != nil { + return sdk.WrapError(err, "unable to get hostname") + } + //Set the dao s.dao = dao{ - store: s.Cache, + store: s.Cache, + hostname: hostname, } log.Info(ctx, "Initializing processor...") diff --git a/engine/repositories/repositories_test.go b/engine/repositories/repositories_test.go index 0ef58874eb..f4c52a10dc 100644 --- a/engine/repositories/repositories_test.go +++ b/engine/repositories/repositories_test.go @@ -72,7 +72,8 @@ func newTestService(t *testing.T) (*Service, error) { } service.dao = dao{ - store: service.Cache, + store: service.Cache, + hostname: "test-instance", } return service, nil