fix(blooms): Use correct key to populate blockscache at startup (#13624)

The cache key for block directories in the blocks cache are the block's address without the directory prefix. This is how the directory is put to the LRU cache after downloading and extracting the block tarball.

This PR fixes the incorrect cache key used to populate the cache from disk on startup, which contained the file system directory prefix.

Since the cached item from startup is never used, it gets evicted first in case of a full cache, or due to its TTL. This causes also the underlying directory on disk to be deleted, which can however still be referenced from the correct cache key for that directory from a later download.
That caused the error `getting index reader: opening series file: open /path/to/block/series: no such file or directory` when trying to query the block, because the correct cache key was still present.

Signed-off-by: Christian Haudum <christian.haudum@gmail.com>
(cherry picked from commit 2624a4b)

pkg/storage/stores/shipper/bloomshipper/cache.go

@@ -100,9 +100,12 @@ func loadBlockDirectories(root string, logger log.Logger) (keys []string, values
 		}
 
 		if ok, clean := isBlockDir(path, logger); ok {
-			keys = append(keys, resolver.Block(ref).Addr())
+			// the cache key must not contain the directory prefix
+			// therefore we use the defaultKeyResolver to resolve the block's address
+			key := defaultKeyResolver{}.Block(ref).Addr()
+			keys = append(keys, key)
 			values = append(values, NewBlockDirectory(ref, path))
-			level.Debug(logger).Log("msg", "found block directory", "ref", ref, "path", path)
+			level.Debug(logger).Log("msg", "found block directory", "path", path, "key", key)
 		} else {
 			level.Warn(logger).Log("msg", "skip directory entry", "err", "not a block directory containing blooms and series", "path", path)
 			_ = clean(path)

pkg/storage/stores/shipper/bloomshipper/cache_test.go

@@ -99,7 +99,7 @@ func Test_LoadBlocksDirIntoCache(t *testing.T) {
 
 	require.Equal(t, 1, len(c.entries))
 
-	key := filepath.Join(wd, validDir) + ".tar.gz"
+	key := validDir + ".tar.gz" // cache key must not contain directory prefix
 	elem, found := c.entries[key]
 	require.True(t, found)
 	blockDir := elem.Value.(*Entry).Value