Suppress error from resolving server addresses for blocks

Instead of returning the error, `FilterChunks` returns the full,
unfiltered list of chunks.

This makes sure that queries do not fail if no bloom gateway instances
are available.

Signed-off-by: Christian Haudum <[email protected]>

Author: chaudum

pkg/bloomgateway/client.go

@@ -116,11 +116,20 @@ type Client interface {
 	FilterChunks(ctx context.Context, tenant string, interval bloomshipper.Interval, blocks []blockWithSeries, plan plan.QueryPlan) ([]*logproto.GroupedChunkRefs, error)
 }
 
+// clientPool is a minimal interface that is satisfied by the JumpHashClientPool.
+// It does only expose functions that are used by the GatewayClient
+// and is required to mock the JumpHashClientPool in tests.
+type clientPool interface {
+	GetClientFor(string) (ringclient.PoolClient, error)
+	Addr(string) (string, error)
+	Stop()
+}
+
 type GatewayClient struct {
 	cfg         ClientConfig
 	logger      log.Logger
 	metrics     *clientMetrics
-	pool        *JumpHashClientPool
+	pool        clientPool
 	dnsProvider *discovery.DNS
 }
 
@@ -211,8 +220,15 @@ func (c *GatewayClient) FilterChunks(ctx context.Context, _ string, interval blo
 	servers := make([]addrWithGroups, 0, len(blocks))
 	for _, blockWithSeries := range blocks {
 		addr, err := c.pool.Addr(blockWithSeries.block.String())
+
+		// the client should return the full, unfiltered list of chunks instead of an error
 		if err != nil {
-			return nil, errors.Wrapf(err, "server address for block: %s", blockWithSeries.block)
+			level.Error(c.logger).Log("msg", "failed to resolve server address for block", "block", blockWithSeries.block, "err", err)
+			var series [][]*logproto.GroupedChunkRefs
+			for i := range blocks {
+				series = append(series, blocks[i].series)
+			}
+			return mergeSeries(series, nil)
 		}
 
 		if idx, found := pos[addr]; found {

pkg/bloomgateway/client_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/go-kit/log"
 	"github.com/grafana/dskit/flagext"
+	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/require"
@@ -16,16 +17,24 @@ import (
 	"github.com/grafana/loki/v3/pkg/storage/stores/shipper/bloomshipper"
 )
 
+type errorMockPool struct {
+	*JumpHashClientPool
+}
+
+func (p *errorMockPool) Addr(_ string) (string, error) {
+	return "", errors.New("no server found")
+}
+
 func TestBloomGatewayClient(t *testing.T) {
 	logger := log.NewNopLogger()
-	reg := prometheus.NewRegistry()
 
 	limits := newLimits()
 
 	cfg := ClientConfig{}
 	flagext.DefaultValues(&cfg)
 
 	t.Run("FilterChunks returns response", func(t *testing.T) {
+		reg := prometheus.NewRegistry()
 		c, err := NewClient(cfg, limits, reg, logger, nil, false)
 		require.NoError(t, err)
 		expr, err := syntax.ParseExpr(`{foo="bar"}`)
@@ -34,6 +43,33 @@ func TestBloomGatewayClient(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, 0, len(res))
 	})
+
+	t.Run("pool error is suppressed and returns full list of chunks", func(t *testing.T) {
+		reg := prometheus.NewRegistry()
+		c, err := NewClient(cfg, limits, reg, logger, nil, false)
+		require.NoError(t, err)
+		c.pool = &errorMockPool{}
+
+		expected := []*logproto.GroupedChunkRefs{
+			{Fingerprint: 0x00, Refs: []*logproto.ShortRef{shortRef(0, 1, 1)}},
+			{Fingerprint: 0x9f, Refs: []*logproto.ShortRef{shortRef(0, 1, 2)}},
+			{Fingerprint: 0xa0, Refs: []*logproto.ShortRef{shortRef(0, 1, 3)}},
+			{Fingerprint: 0xff, Refs: []*logproto.ShortRef{shortRef(0, 1, 4)}},
+		}
+
+		blocks := []blockWithSeries{
+			{block: mkBlockRef(0x00, 0x9f), series: expected[0:2]},
+			{block: mkBlockRef(0xa0, 0xff), series: expected[2:4]},
+		}
+		expr, err := syntax.ParseExpr(`{foo="bar"}`)
+		require.NoError(t, err)
+
+		res, err := c.FilterChunks(context.Background(), "tenant", bloomshipper.NewInterval(0, 0), blocks, plan.QueryPlan{AST: expr})
+		require.NoError(t, err)
+		require.Equal(t, 4, len(res))
+
+		require.Equal(t, expected, res)
+	})
 }
 
 func shortRef(f, t model.Time, c uint32) *logproto.ShortRef {
@@ -94,3 +130,6 @@ func TestGatewayClient_MergeChunkSets(t *testing.T) {
 	result := mergeChunkSets(inp1, inp2)
 	require.Equal(t, expected, result)
 }
+
+func TestGatewayClient_FilterChunks(t *testing.T) {
+}