fix(blooms): bloomshipper no longer returns empty metas on fetch (#13130)

Traced some bloom compactor problems to this store interface returning empty `Metas` (with the exception of the reference key) on fetch when the object does not exist in storage. I originally found this due to a panic in the outdated meta removal code because empty `Metas` don't have any source TSDBs.

Fixes the impl to only return metas which existed which have their contents populated.

Author: owen-d

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

@@ -386,7 +386,7 @@ func (b *BloomClient) Stop() {
 }
 
 func (b *BloomClient) GetMetas(ctx context.Context, refs []MetaRef) ([]Meta, error) {
-	results := make([]Meta, len(refs))
+	results := make([]*Meta, len(refs))
 	err := concurrency.ForEachJob(ctx, len(refs), b.concurrency, func(ctx context.Context, idx int) error {
 		meta, err := b.GetMeta(ctx, refs[idx])
 		if err != nil {
@@ -395,11 +395,19 @@ func (b *BloomClient) GetMetas(ctx context.Context, refs []MetaRef) ([]Meta, err
 				return fmt.Errorf("failed to get meta file %s: %w", key, err)
 			}
 			level.Error(b.logger).Log("msg", "failed to get meta file", "ref", key, "err", err)
+			return nil
 		}
-		results[idx] = meta
+		results[idx] = &meta
 		return nil
 	})
-	return results, err
+
+	filtered := make([]Meta, 0, len(results))
+	for _, r := range results {
+		if r != nil {
+			filtered = append(filtered, *r)
+		}
+	}
+	return filtered, err
 }
 
 // GetMeta fetches the meta file for given MetaRef from object storage and

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

@@ -107,8 +107,8 @@ func TestBloomClient_GetMetas(t *testing.T) {
 		require.Equal(t, metas, []Meta{m1, m2})
 	})
 
-	t.Run("does not exist - yields empty meta", func(t *testing.T) {
-		ref := MetaRef{
+	t.Run("does not exist - skips empty meta", func(t *testing.T) {
+		notExist := MetaRef{
 			Ref: Ref{
 				TenantID:       "tenant",
 				TableName:      "table",
@@ -118,9 +118,9 @@ func TestBloomClient_GetMetas(t *testing.T) {
 				Checksum:       1234,
 			},
 		}
-		metas, err := c.GetMetas(ctx, []MetaRef{ref})
+		metas, err := c.GetMetas(ctx, []MetaRef{notExist, m1.MetaRef})
 		require.NoError(t, err)
-		require.Equal(t, metas, []Meta{{MetaRef: ref}})
+		require.Equal(t, metas, []Meta{m1})
 	})
 }