Skip to content

Commit 8978ecf

Browse files
salvacortssalvacorts
authored
feat: Boilerplate for new bloom build planner and worker components. (#12989)
1 parent 88e545f commit 8978ecf

File tree

10 files changed

+284
-1
lines changed

10 files changed

+284
-1
lines changed

‎docs/sources/shared/configuration.md

+10
Original file line numberDiff line numberDiff line change
@@ -326,6 +326,16 @@ pattern_ingester:
326326
# merging them as bloom blocks.
327327
[bloom_compactor: <bloom_compactor>]
328328

329+
bloom_build:
330+
# Flag to enable or disable the usage of the bloom-planner and bloom-builder
331+
# components.
332+
# CLI flag: -bloom-build.enabled
333+
[enabled: <boolean> | default = false]
334+
335+
planner:
336+
337+
builder:
338+
329339
# Experimental: The bloom_gateway block configures the Loki bloom gateway
330340
# server, responsible for serving queries for filtering chunks based on filter
331341
# expressions.

‎pkg/bloombuild/builder/builder.go

+50
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
package builder
2+
3+
import (
4+
"context"
5+
6+
"github.com/go-kit/log"
7+
"github.com/grafana/dskit/services"
8+
"github.com/prometheus/client_golang/prometheus"
9+
10+
utillog "github.com/grafana/loki/v3/pkg/util/log"
11+
)
12+
13+
type Worker struct {
14+
services.Service
15+
16+
cfg Config
17+
metrics *Metrics
18+
logger log.Logger
19+
}
20+
21+
func New(
22+
cfg Config,
23+
logger log.Logger,
24+
r prometheus.Registerer,
25+
) (*Worker, error) {
26+
utillog.WarnExperimentalUse("Bloom Builder", logger)
27+
28+
w := &Worker{
29+
cfg: cfg,
30+
metrics: NewMetrics(r),
31+
logger: logger,
32+
}
33+
34+
w.Service = services.NewBasicService(w.starting, w.running, w.stopping)
35+
return w, nil
36+
}
37+
38+
func (w *Worker) starting(_ context.Context) (err error) {
39+
w.metrics.running.Set(1)
40+
return err
41+
}
42+
43+
func (w *Worker) stopping(_ error) error {
44+
w.metrics.running.Set(0)
45+
return nil
46+
}
47+
48+
func (w *Worker) running(_ context.Context) error {
49+
return nil
50+
}

‎pkg/bloombuild/builder/config.go

+21
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
package builder
2+
3+
import "flag"
4+
5+
// Config configures the bloom-builder component.
6+
type Config struct {
7+
// TODO: Add config
8+
}
9+
10+
// RegisterFlagsWithPrefix registers flags for the bloom-planner configuration.
11+
func (cfg *Config) RegisterFlagsWithPrefix(_ string, _ *flag.FlagSet) {
12+
// TODO: Register flags with flagsPrefix
13+
}
14+
15+
func (cfg *Config) Validate() error {
16+
return nil
17+
}
18+
19+
type Limits interface {
20+
// TODO: Add limits
21+
}

‎pkg/bloombuild/builder/metrics.go

+26
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
package builder
2+
3+
import (
4+
"github.com/prometheus/client_golang/prometheus"
5+
"github.com/prometheus/client_golang/prometheus/promauto"
6+
)
7+
8+
const (
9+
metricsNamespace = "loki"
10+
metricsSubsystem = "bloombuilder"
11+
)
12+
13+
type Metrics struct {
14+
running prometheus.Gauge
15+
}
16+
17+
func NewMetrics(r prometheus.Registerer) *Metrics {
18+
return &Metrics{
19+
running: promauto.With(r).NewGauge(prometheus.GaugeOpts{
20+
Namespace: metricsNamespace,
21+
Subsystem: metricsSubsystem,
22+
Name: "running",
23+
Help: "Value will be 1 if the bloom builder is currently running on this instance",
24+
}),
25+
}
26+
}

‎pkg/bloombuild/config.go

+40
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
package bloombuild
2+
3+
import (
4+
"flag"
5+
"fmt"
6+
7+
"github.com/grafana/loki/v3/pkg/bloombuild/builder"
8+
"github.com/grafana/loki/v3/pkg/bloombuild/planner"
9+
)
10+
11+
// Config configures the bloom-planner component.
12+
type Config struct {
13+
Enabled bool `yaml:"enabled"`
14+
15+
Planner planner.Config `yaml:"planner"`
16+
Builder builder.Config `yaml:"builder"`
17+
}
18+
19+
// RegisterFlags registers flags for the bloom building configuration.
20+
func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
21+
f.BoolVar(&cfg.Enabled, "bloom-build.enabled", false, "Flag to enable or disable the usage of the bloom-planner and bloom-builder components.")
22+
cfg.Planner.RegisterFlagsWithPrefix("bloom-build.planner", f)
23+
cfg.Builder.RegisterFlagsWithPrefix("bloom-build.builder", f)
24+
}
25+
26+
func (cfg *Config) Validate() error {
27+
if !cfg.Enabled {
28+
return nil
29+
}
30+
31+
if err := cfg.Planner.Validate(); err != nil {
32+
return fmt.Errorf("invalid bloom planner configuration: %w", err)
33+
}
34+
35+
if err := cfg.Builder.Validate(); err != nil {
36+
return fmt.Errorf("invalid bloom builder configuration: %w", err)
37+
}
38+
39+
return nil
40+
}

‎pkg/bloombuild/planner/config.go

+21
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
package planner
2+
3+
import "flag"
4+
5+
// Config configures the bloom-planner component.
6+
type Config struct {
7+
// TODO: Add config
8+
}
9+
10+
// RegisterFlagsWithPrefix registers flags for the bloom-planner configuration.
11+
func (cfg *Config) RegisterFlagsWithPrefix(_ string, _ *flag.FlagSet) {
12+
// TODO: Register flags with flagsPrefix
13+
}
14+
15+
func (cfg *Config) Validate() error {
16+
return nil
17+
}
18+
19+
type Limits interface {
20+
// TODO: Add limits
21+
}

‎pkg/bloombuild/planner/metrics.go

+26
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
package planner
2+
3+
import (
4+
"github.com/prometheus/client_golang/prometheus"
5+
"github.com/prometheus/client_golang/prometheus/promauto"
6+
)
7+
8+
const (
9+
metricsNamespace = "loki"
10+
metricsSubsystem = "bloomplanner"
11+
)
12+
13+
type Metrics struct {
14+
running prometheus.Gauge
15+
}
16+
17+
func NewMetrics(r prometheus.Registerer) *Metrics {
18+
return &Metrics{
19+
running: promauto.With(r).NewGauge(prometheus.GaugeOpts{
20+
Namespace: metricsNamespace,
21+
Subsystem: metricsSubsystem,
22+
Name: "running",
23+
Help: "Value will be 1 if bloom planner is currently running on this instance",
24+
}),
25+
}
26+
}

‎pkg/bloombuild/planner/planner.go

+50
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
package planner
2+
3+
import (
4+
"context"
5+
6+
"github.com/go-kit/log"
7+
"github.com/grafana/dskit/services"
8+
"github.com/prometheus/client_golang/prometheus"
9+
10+
utillog "github.com/grafana/loki/v3/pkg/util/log"
11+
)
12+
13+
type Planner struct {
14+
services.Service
15+
16+
cfg Config
17+
metrics *Metrics
18+
logger log.Logger
19+
}
20+
21+
func New(
22+
cfg Config,
23+
logger log.Logger,
24+
r prometheus.Registerer,
25+
) (*Planner, error) {
26+
utillog.WarnExperimentalUse("Bloom Planner", logger)
27+
28+
p := &Planner{
29+
cfg: cfg,
30+
metrics: NewMetrics(r),
31+
logger: logger,
32+
}
33+
34+
p.Service = services.NewBasicService(p.starting, p.running, p.stopping)
35+
return p, nil
36+
}
37+
38+
func (p *Planner) starting(_ context.Context) (err error) {
39+
p.metrics.running.Set(1)
40+
return err
41+
}
42+
43+
func (p *Planner) stopping(_ error) error {
44+
p.metrics.running.Set(0)
45+
return nil
46+
}
47+
48+
func (p *Planner) running(_ context.Context) error {
49+
return nil
50+
}

‎pkg/loki/loki.go

+7
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ import (
3131
"google.golang.org/grpc/health/grpc_health_v1"
3232

3333
"github.com/grafana/loki/v3/pkg/analytics"
34+
"github.com/grafana/loki/v3/pkg/bloombuild"
3435
"github.com/grafana/loki/v3/pkg/bloomcompactor"
3536
"github.com/grafana/loki/v3/pkg/bloomgateway"
3637
"github.com/grafana/loki/v3/pkg/compactor"
@@ -90,6 +91,7 @@ type Config struct {
9091
Pattern pattern.Config `yaml:"pattern_ingester,omitempty"`
9192
IndexGateway indexgateway.Config `yaml:"index_gateway"`
9293
BloomCompactor bloomcompactor.Config `yaml:"bloom_compactor,omitempty" category:"experimental"`
94+
BloomBuild bloombuild.Config `yaml:"bloom_build,omitempty" category:"experimental"`
9395
BloomGateway bloomgateway.Config `yaml:"bloom_gateway,omitempty" category:"experimental"`
9496
StorageConfig storage.Config `yaml:"storage_config,omitempty"`
9597
ChunkStoreConfig config.ChunkStoreConfig `yaml:"chunk_store_config,omitempty"`
@@ -173,6 +175,7 @@ func (c *Config) RegisterFlags(f *flag.FlagSet) {
173175
c.Tracing.RegisterFlags(f)
174176
c.CompactorConfig.RegisterFlags(f)
175177
c.BloomCompactor.RegisterFlags(f)
178+
c.BloomBuild.RegisterFlags(f)
176179
c.QueryScheduler.RegisterFlags(f)
177180
c.Analytics.RegisterFlags(f)
178181
c.OperationalConfig.RegisterFlags(f)
@@ -649,6 +652,8 @@ func (t *Loki) setupModuleManager() error {
649652
mm.RegisterModule(BloomStore, t.initBloomStore)
650653
mm.RegisterModule(BloomCompactor, t.initBloomCompactor)
651654
mm.RegisterModule(BloomCompactorRing, t.initBloomCompactorRing, modules.UserInvisibleModule)
655+
mm.RegisterModule(BloomPlanner, t.initBloomPlanner)
656+
mm.RegisterModule(BloomBuilder, t.initBloomBuilder)
652657
mm.RegisterModule(IndexGateway, t.initIndexGateway)
653658
mm.RegisterModule(IndexGatewayRing, t.initIndexGatewayRing, modules.UserInvisibleModule)
654659
mm.RegisterModule(IndexGatewayInterceptors, t.initIndexGatewayInterceptors, modules.UserInvisibleModule)
@@ -686,6 +691,8 @@ func (t *Loki) setupModuleManager() error {
686691
IndexGateway: {Server, Store, BloomStore, IndexGatewayRing, IndexGatewayInterceptors, Analytics},
687692
BloomGateway: {Server, BloomStore, Analytics},
688693
BloomCompactor: {Server, BloomStore, BloomCompactorRing, Analytics, Store},
694+
BloomPlanner: {Server, BloomStore, Analytics, Store},
695+
BloomBuilder: {Server, BloomStore, Analytics, Store},
689696
PatternIngester: {Server, MemberlistKV, Analytics},
690697
PatternRingClient: {Server, MemberlistKV, Analytics},
691698
IngesterQuerier: {Ring},

‎pkg/loki/modules.go

+33-1
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,8 @@ import (
3838
"github.com/grafana/loki/v3/pkg/storage/types"
3939

4040
"github.com/grafana/loki/v3/pkg/analytics"
41+
"github.com/grafana/loki/v3/pkg/bloombuild/builder"
42+
"github.com/grafana/loki/v3/pkg/bloombuild/planner"
4143
"github.com/grafana/loki/v3/pkg/bloomgateway"
4244
"github.com/grafana/loki/v3/pkg/compactor"
4345
compactorclient "github.com/grafana/loki/v3/pkg/compactor/client"
@@ -122,6 +124,8 @@ const (
122124
QuerySchedulerRing string = "query-scheduler-ring"
123125
BloomCompactor string = "bloom-compactor"
124126
BloomCompactorRing string = "bloom-compactor-ring"
127+
BloomPlanner string = "bloom-planner"
128+
BloomBuilder string = "bloom-builder"
125129
BloomStore string = "bloom-store"
126130
All string = "all"
127131
Read string = "read"
@@ -803,7 +807,7 @@ func (t *Loki) updateConfigForShipperStore() {
803807
t.Cfg.StorageConfig.TSDBShipperConfig.Mode = indexshipper.ModeWriteOnly
804808
t.Cfg.StorageConfig.TSDBShipperConfig.IngesterDBRetainPeriod = shipperQuerierIndexUpdateDelay(t.Cfg.StorageConfig.IndexCacheValidity, t.Cfg.StorageConfig.TSDBShipperConfig.ResyncInterval)
805809

806-
case t.Cfg.isTarget(Querier), t.Cfg.isTarget(Ruler), t.Cfg.isTarget(Read), t.Cfg.isTarget(Backend), t.isModuleActive(IndexGateway), t.Cfg.isTarget(BloomCompactor):
810+
case t.Cfg.isTarget(Querier), t.Cfg.isTarget(Ruler), t.Cfg.isTarget(Read), t.Cfg.isTarget(Backend), t.isModuleActive(IndexGateway), t.Cfg.isTarget(BloomCompactor), t.Cfg.isTarget(BloomPlanner), t.Cfg.isTarget(BloomBuilder):
807811
// We do not want query to do any updates to index
808812
t.Cfg.StorageConfig.BoltDBShipperConfig.Mode = indexshipper.ModeReadOnly
809813
t.Cfg.StorageConfig.TSDBShipperConfig.Mode = indexshipper.ModeReadOnly
@@ -1553,6 +1557,34 @@ func (t *Loki) initBloomCompactorRing() (services.Service, error) {
15531557
return t.bloomCompactorRingManager, nil
15541558
}
15551559

1560+
func (t *Loki) initBloomPlanner() (services.Service, error) {
1561+
if !t.Cfg.BloomBuild.Enabled {
1562+
return nil, nil
1563+
}
1564+
1565+
logger := log.With(util_log.Logger, "component", "bloom-planner")
1566+
1567+
return planner.New(
1568+
t.Cfg.BloomBuild.Planner,
1569+
logger,
1570+
prometheus.DefaultRegisterer,
1571+
)
1572+
}
1573+
1574+
func (t *Loki) initBloomBuilder() (services.Service, error) {
1575+
if !t.Cfg.BloomBuild.Enabled {
1576+
return nil, nil
1577+
}
1578+
1579+
logger := log.With(util_log.Logger, "component", "bloom-worker")
1580+
1581+
return builder.New(
1582+
t.Cfg.BloomBuild.Builder,
1583+
logger,
1584+
prometheus.DefaultRegisterer,
1585+
)
1586+
}
1587+
15561588
func (t *Loki) initQueryScheduler() (services.Service, error) {
15571589
s, err := scheduler.NewScheduler(t.Cfg.QueryScheduler, t.Overrides, util_log.Logger, t.querySchedulerRingManager, prometheus.DefaultRegisterer, t.Cfg.MetricsNamespace)
15581590
if err != nil {

0 commit comments

Comments
 (0)