-
Notifications
You must be signed in to change notification settings - Fork 3.5k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add basic structure of bloom gateways (#10782)
### Summary
This pull requests adds the basic structure for the new bloom gateway component.
- Adds new `bloom-gateway` target that runs with multiple instances joined by a ring
- Adds a querier and client component on the index gateway to filter chunk refs
- Adds the gRPC protobuf definitions for commication between index gateways and bloom gateways
- Adds a store component used on the bloom gateways to query binary bloom files
```
Querier Query Frontend
| |
................................... service boundary
| |
+----+------+
|
indexgateway.Gateway**
|
bloomgateway.BloomQuerier
|
bloomgateway.GatewayClient
|
logproto.BloomGatewayClient
|
................................... service boundary
|
bloomgateway.Gateway
|
bloomshipper.Store
|
bloomshipper.Shipper
|
bloomshipper.BloomFileClient**
|
ObjectClient**
|
................................... service boundary
|
object storage
** not part of this PR
```
This PR still contains a lot of TODOs and possibilities for optimisations, which will be addressed in subsequent pull requests.
Signed-off-by: Christian Haudum <[email protected]>- Loading branch information
Showing
35 changed files
with
4,000 additions
and
214 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,162 @@ | ||
| /* | ||
| Bloom Gateway package | ||
| The bloom gateway is a component that can be run as a standalone microserivce | ||
| target and provides capabilities for filtering ChunkRefs based on a given list | ||
| of line filter expressions. | ||
| Querier Query Frontend | ||
| | | | ||
| ................................... service boundary | ||
| | | | ||
| +----+------+ | ||
| | | ||
| indexgateway.Gateway | ||
| | | ||
| bloomgateway.BloomQuerier | ||
| | | ||
| bloomgateway.GatewayClient | ||
| | | ||
| logproto.BloomGatewayClient | ||
| | | ||
| ................................... service boundary | ||
| | | ||
| bloomgateway.Gateway | ||
| | | ||
| bloomshipper.Store | ||
| | | ||
| bloomshipper.Shipper | ||
| | | ||
| bloomshipper.BloomFileClient | ||
| | | ||
| ObjectClient | ||
| | | ||
| ................................... service boundary | ||
| | | ||
| object storage | ||
| */ | ||
| package bloomgateway | ||
|
|
||
| import ( | ||
| "context" | ||
| "sort" | ||
|
|
||
| "github.com/go-kit/log" | ||
| "github.com/grafana/dskit/services" | ||
| "github.com/grafana/dskit/tenant" | ||
| "github.com/pkg/errors" | ||
| "github.com/prometheus/client_golang/prometheus" | ||
|
|
||
| "github.com/grafana/loki/pkg/logproto" | ||
| "github.com/grafana/loki/pkg/storage" | ||
| "github.com/grafana/loki/pkg/storage/config" | ||
| "github.com/grafana/loki/pkg/storage/stores/shipper/bloomshipper" | ||
| ) | ||
|
|
||
| var errGatewayUnhealthy = errors.New("bloom-gateway is unhealthy in the ring") | ||
| var errInvalidTenant = errors.New("invalid tenant in chunk refs") | ||
|
|
||
| type metrics struct{} | ||
|
|
||
| func newMetrics(r prometheus.Registerer) *metrics { | ||
| return &metrics{} | ||
| } | ||
|
|
||
| type Gateway struct { | ||
| services.Service | ||
|
|
||
| cfg Config | ||
| logger log.Logger | ||
| metrics *metrics | ||
|
|
||
| bloomStore bloomshipper.Store | ||
|
|
||
| sharding ShardingStrategy | ||
| } | ||
|
|
||
| // New returns a new instance of the Bloom Gateway. | ||
| func New(cfg Config, schemaCfg config.SchemaConfig, storageCfg storage.Config, shardingStrategy ShardingStrategy, cm storage.ClientMetrics, logger log.Logger, reg prometheus.Registerer) (*Gateway, error) { | ||
| g := &Gateway{ | ||
| cfg: cfg, | ||
| logger: logger, | ||
| metrics: newMetrics(reg), | ||
| sharding: shardingStrategy, | ||
| } | ||
|
|
||
| client, err := bloomshipper.NewBloomClient(schemaCfg.Configs, storageCfg, cm) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| bloomShipper, err := bloomshipper.NewShipper(client, storageCfg.BloomShipperConfig, logger) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| bloomStore, err := bloomshipper.NewBloomStore(bloomShipper) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| g.bloomStore = bloomStore | ||
| g.Service = services.NewIdleService(g.starting, g.stopping) | ||
|
|
||
| return g, nil | ||
| } | ||
|
|
||
| func (g *Gateway) starting(ctx context.Context) error { | ||
| return nil | ||
| } | ||
|
|
||
| func (g *Gateway) stopping(_ error) error { | ||
| g.bloomStore.Stop() | ||
| return nil | ||
| } | ||
|
|
||
| // FilterChunkRefs implements BloomGatewayServer | ||
| func (g *Gateway) FilterChunkRefs(ctx context.Context, req *logproto.FilterChunkRefRequest) (*logproto.FilterChunkRefResponse, error) { | ||
| tenantID, err := tenant.TenantID(ctx) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| for _, ref := range req.Refs { | ||
| if ref.UserID != tenantID { | ||
| return nil, errors.Wrapf(errInvalidTenant, "expected chunk refs from tenant %s, got tenant %s", tenantID, ref.UserID) | ||
| } | ||
| } | ||
|
|
||
| // Sort ChunkRefs by fingerprint in ascending order | ||
| sort.Slice(req.Refs, func(i, j int) bool { | ||
| return req.Refs[i].Fingerprint < req.Refs[j].Fingerprint | ||
| }) | ||
|
|
||
| chunkRefs := req.Refs | ||
|
|
||
| // Only query bloom filters if filters are present | ||
| if len(req.Filters) > 0 { | ||
| chunkRefs, err = g.bloomStore.FilterChunkRefs(ctx, tenantID, req.From.Time(), req.Through.Time(), req.Refs, req.Filters...) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| } | ||
|
|
||
| // TODO(chaudum): Re-use buffers for response. | ||
| resp := make([]*logproto.GroupedChunkRefs, 0) | ||
| for idx, chunkRef := range chunkRefs { | ||
| fp := chunkRef.Fingerprint | ||
| shortRef := &logproto.ShortRef{From: chunkRef.From, Through: chunkRef.Through, Checksum: chunkRef.Checksum} | ||
| if idx == 0 || fp > resp[len(resp)-1].Fingerprint { | ||
| r := &logproto.GroupedChunkRefs{ | ||
| Fingerprint: fp, | ||
| Tenant: tenantID, | ||
| Refs: []*logproto.ShortRef{shortRef}, | ||
| } | ||
| resp = append(resp, r) | ||
| continue | ||
| } | ||
| resp[len(resp)-1].Refs = append(resp[len(resp)-1].Refs, shortRef) | ||
| } | ||
|
|
||
| return &logproto.FilterChunkRefResponse{ChunkRefs: resp}, nil | ||
| } |
Oops, something went wrong.