bloom client records cache locality score

Signed-off-by: Owen Diehl <[email protected]>
grafana · Apr 3, 2024 · 24ef2cb · 24ef2cb
1 parent d39dc09
commit 24ef2cb
Show file tree

Hide file tree

Showing 2 changed files with 39 additions and 11 deletions.
diff --git a/pkg/bloomgateway/client.go b/pkg/bloomgateway/client.go
@@ -144,11 +144,12 @@ type Client interface {
 }
 
 type GatewayClient struct {
-	cfg    ClientConfig
-	limits Limits
-	logger log.Logger
-	pool   *ringclient.Pool
-	ring   ring.ReadRing
+	cfg     ClientConfig
+	limits  Limits
+	logger  log.Logger
+	metrics *clientMetrics
+	pool    *ringclient.Pool
+	ring    ring.ReadRing
 }
 
 func NewClient(
@@ -206,11 +207,12 @@ func NewClient(
 	}
 
 	return &GatewayClient{
-		cfg:    cfg,
-		logger: logger,
-		limits: limits,
-		pool:   clientpool.NewPool("bloom-gateway", cfg.PoolConfig, cfg.Ring, ringclient.PoolAddrFunc(poolFactory), logger, metricsNamespace),
-		ring:   readRing,
+		cfg:     cfg,
+		logger:  logger,
+		limits:  limits,
+		metrics: newClientMetrics(registerer),
+		pool:    clientpool.NewPool("bloom-gateway", cfg.PoolConfig, cfg.Ring, ringclient.PoolAddrFunc(poolFactory), logger, metricsNamespace),
+		ring:    readRing,
 	}, nil
 }
 
@@ -231,7 +233,7 @@ func shuffleAddrs(addrs []string) []string {
 
 // FilterChunkRefs implements Client
 func (c *GatewayClient) FilterChunks(ctx context.Context, tenant string, from, through model.Time, groups []*logproto.GroupedChunkRefs, plan plan.QueryPlan) ([]*logproto.GroupedChunkRefs, error) {
-	if !c.limits.BloomGatewayEnabled(tenant) {
+	if !c.limits.BloomGatewayEnabled(tenant) || len(groups) == 0 {
 		return groups, nil
 	}
 
@@ -248,6 +250,15 @@ func (c *GatewayClient) FilterChunks(ctx context.Context, tenant string, from, t
 	}
 	servers = partitionByReplicationSet(groups, servers)
 
+	// `% instances / % kesypace `. Ideally converges to 1,
+	// with theoretical max of `1 + 2/n_replicas` since the left and right keyspace bounds
+	// may lightly touch neighboring replicas.
+	firstFp, lastFp := groups[0].Fingerprint, groups[len(groups)-1].Fingerprint
+	pctKeyspace := float64(lastFp-firstFp) / float64(math.MaxUint64)
+	pctInstances := float64(len(servers)) / float64(len(rs.Instances))
+	cacheLocalityScore := pctInstances / pctKeyspace
+	c.metrics.cacheLocalityScore.Observe(cacheLocalityScore)
+
 	results := make([][]*logproto.GroupedChunkRefs, len(servers))
 	count := 0
 	err = concurrency.ForEachJob(ctx, len(servers), len(servers), func(ctx context.Context, i int) error {

diff --git a/pkg/bloomgateway/metrics.go b/pkg/bloomgateway/metrics.go
@@ -3,6 +3,7 @@ package bloomgateway
 import (
 	"time"
 
+	"github.com/grafana/loki/v3/pkg/util/constants"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 )
@@ -12,6 +13,22 @@ type metrics struct {
 	*serverMetrics
 }
 
+type clientMetrics struct {
+	cacheLocalityScore prometheus.Histogram
+}
+
+func newClientMetrics(registerer prometheus.Registerer) *clientMetrics {
+	return &clientMetrics{
+		cacheLocalityScore: promauto.With(registerer).NewHistogram(prometheus.HistogramOpts{
+			Namespace: constants.Loki,
+			Subsystem: "bloom_gateway_client",
+			Name:      "cache_locality_score",
+			Help:      "Cache locality score of the bloom filter, as measured by % of keyspace touched / % of bloom_gws required",
+			Buckets:   prometheus.ExponentialBucketsRange(1, 100, 8),
+		}),
+	}
+}
+
 type serverMetrics struct {
 	inflightRequests prometheus.Summary
 	requestedSeries  prometheus.Histogram