Skip to content

Commit

Permalink
fix(blooms): Suppress error from resolving server addresses for blocks (
Browse files Browse the repository at this point in the history
#13385)

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 <christian.haudum@gmail.com>
  • Loading branch information
chaudum authored Jul 3, 2024
1 parent d451e23 commit 3ac2317
Show file tree
Hide file tree
Showing 2 changed files with 55 additions and 3 deletions.
20 changes: 18 additions & 2 deletions pkg/bloomgateway/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand Down Expand Up @@ -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 {
Expand Down
38 changes: 37 additions & 1 deletion pkg/bloomgateway/client_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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"}`)
Expand All @@ -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 {
Expand Down

0 comments on commit 3ac2317

Please sign in to comment.