diff --git a/pkg/ingester/retention.go b/pkg/ingester/retention.go index fd2d6f22d8..b5e14843f6 100644 --- a/pkg/ingester/retention.go +++ b/pkg/ingester/retention.go @@ -361,17 +361,18 @@ func (bm *realFSBlockManager) GetTenantIDs(ctx context.Context) ([]string, error return nil, ctx.Err() } - tenantDirs, err := fs.ReadDir(bm.FS, bm.Root) + dirs, err := fs.ReadDir(bm.FS, bm.Root) if err != nil { return nil, err } tenantIDs := make([]string, 0) - for _, tenantDir := range tenantDirs { - if !tenantDir.IsDir() { + for _, dir := range dirs { + if !bm.isTenantDir(bm.Root, dir) { continue } - tenantIDs = append(tenantIDs, tenantDir.Name()) + + tenantIDs = append(tenantIDs, dir.Name()) } return tenantIDs, nil } @@ -446,3 +447,28 @@ func (bm *realFSBlockManager) DeleteBlock(ctx context.Context, block *tenantBloc return nil }) } + +// isTenantDir checks if a directory is a tenant directory. +func (bm *realFSBlockManager) isTenantDir(path string, entry fs.DirEntry) bool { + if !entry.IsDir() { + return false + } + + subEntries, err := bm.FS.ReadDir(filepath.Join(path, entry.Name())) + if err != nil { + return false + } + + foundLocalDir := false + for _, subEntry := range subEntries { + if !subEntry.IsDir() { + continue + } + + if subEntry.Name() == phlareDBLocalPath { + foundLocalDir = true + break + } + } + return foundLocalDir +} diff --git a/pkg/ingester/retention_test.go b/pkg/ingester/retention_test.go index 8ba8cd0c6a..80a88e4318 100644 --- a/pkg/ingester/retention_test.go +++ b/pkg/ingester/retention_test.go @@ -19,6 +19,7 @@ import ( "github.com/spf13/afero" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" + "golang.org/x/exp/slices" "github.com/grafana/pyroscope/pkg/phlaredb" "github.com/grafana/pyroscope/pkg/phlaredb/shipper" @@ -427,11 +428,16 @@ func TestFSBlockManager(t *testing.T) { fs.markBlocksShippedForTenant(t, tenantID, uploadedBlockIDs...) } + // Create a lost+found directory. + fs.createDirectories(t, "lost+found") + t.Run("GetTenantIDs", func(t *testing.T) { bm := newFSBlockManager(root, e, fs) tenantIDs, err := bm.GetTenantIDs(context.Background()) require.NoError(t, err) require.Equal(t, []string{"1218", "anonymous"}, tenantIDs) + // Explicitly check lost+found isn't in tenant id list. + require.NotContains(t, tenantIDs, "lost+found") }) t.Run("GetBlocksForTenant", func(t *testing.T) { @@ -461,6 +467,55 @@ func TestFSBlockManager(t *testing.T) { }) } +func TestFSBlockManager_isTenantDir(t *testing.T) { + const root = "/data" + dirPaths := []string{ + // Skip, not tenant ids + "lost+found", + ".DS_Store", + + // Skip, no local dir + "1234/head/01HKWWF79V1STKXBNYW7WCMDGM", + "1234/head/01HKWWF8939QM6E7BS69X0RASG", + + // Tenant dirs + "anonymous/local/01HKWWF3CTFC5EJN6JJ96TY4W9", + "anonymous/local/01HKWWF4C298KVTEEQ3RW6TVHZ", + "1218/local/01HKWWF5BB2DJVDP0DTMT9MDMN", + "1218/local/01HKWWF6AKVZDCWQB12MHWG7FN", + "9876/local", + } + filePaths := []string{ + // Skip all files + "somefile.txt", + } + + fs := &mockFS{ + Fs: afero.NewMemMapFs(), + Root: root, + } + fs.createDirectories(t, dirPaths...) + fs.createFiles(t, filePaths...) + + gotTenantIDs := []string{} + entries, err := fs.ReadDir(fs.Root) + require.NoError(t, err) + + bm := &realFSBlockManager{ + Root: fs.Root, + FS: fs, + } + for _, entry := range entries { + if bm.isTenantDir(fs.Root, entry) { + gotTenantIDs = append(gotTenantIDs, entry.Name()) + } + } + slices.Sort(gotTenantIDs) + + wantTenantIDs := []string{"1218", "9876", "anonymous"} + require.Equal(t, wantTenantIDs, gotTenantIDs) +} + func TestSortBlocks(t *testing.T) { createAnonymousBlock := func(t *testing.T, blockID string, uploaded bool) *tenantBlock { t.Helper() @@ -591,6 +646,30 @@ func (mfs *mockFS) markBlocksShippedForTenant(t *testing.T, tenantID string, blo } } +func (mfs *mockFS) createDirectories(t *testing.T, paths ...string) { + t.Helper() + for _, path := range paths { + path = filepath.Join(mfs.Root, path) + err := mfs.MkdirAll(path, 0755) + if err != nil { + t.Fatalf("failed to create directory: %s: %v", path, err) + return + } + } +} + +func (mfs *mockFS) createFiles(t *testing.T, paths ...string) { + t.Helper() + for _, path := range paths { + path = filepath.Join(mfs.Root, path) + _, err := mfs.Create(path) + if err != nil { + t.Fatalf("failed to create file: %s: %v", path, err) + return + } + } +} + type mockBlockManager struct { mock.Mock }