Add index gateway to the read path (#6385)

pull/6412/head
Trevor Whitney 3 years ago committed by GitHub
parent 9ff621825a
commit 01fe534751
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 2
      pkg/loki/loki.go
  2. 6
      pkg/loki/modules.go
  3. 139
      pkg/loki/modules_test.go

@ -511,7 +511,7 @@ func (t *Loki) setupModuleManager() error {
IngesterQuerier: {Ring},
IndexGatewayRing: {RuntimeConfig, Server, MemberlistKV},
All: {QueryScheduler, QueryFrontend, Querier, Ingester, Distributor, Ruler, Compactor},
Read: {QueryScheduler, QueryFrontend, Querier, Ruler, Compactor},
Read: {QueryScheduler, QueryFrontend, Querier, Ruler, Compactor, IndexGateway},
Write: {Ingester, Distributor},
MemberlistKV: {Server},
}

@ -902,6 +902,12 @@ func (t *Loki) initIndexGateway() (services.Service, error) {
}
func (t *Loki) initIndexGatewayRing() (_ services.Service, err error) {
// IndexGateway runs by default on read target, and should always assume
// ring mode when run in this way.
if t.isModuleActive(Read) {
t.Cfg.IndexGateway.Mode = indexgateway.RingMode
}
if t.Cfg.IndexGateway.Mode != indexgateway.RingMode {
return
}

@ -7,6 +7,7 @@ import (
"github.com/grafana/dskit/flagext"
"github.com/prometheus/common/model"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/prometheus/client_golang/prometheus"
@ -16,6 +17,8 @@ import (
"github.com/grafana/loki/pkg/storage/config"
"github.com/grafana/loki/pkg/storage/stores/indexshipper"
"github.com/grafana/loki/pkg/storage/stores/shipper"
"github.com/grafana/loki/pkg/storage/stores/shipper/indexgateway"
)
func Test_calculateMaxLookBack(t *testing.T) {
@ -144,44 +147,8 @@ func TestMultiKVSetup(t *testing.T) {
t.Run(target, func(t *testing.T) {
prepareGlobalMetricsRegistry(t)
cfg := Config{}
cfg.SchemaConfig = config.SchemaConfig{
Configs: []config.PeriodConfig{
{
IndexType: config.StorageTypeInMemory,
ObjectType: config.StorageTypeFileSystem,
RowShards: 16,
Schema: "v11",
From: config.DayTime{
Time: model.Now(),
},
},
},
}
flagext.DefaultValues(&cfg)
// Set to 0 to find any free port.
cfg.Server.HTTPListenPort = 0
cfg.Server.GRPCListenPort = 0
cfg.Target = []string{target}
// Must be set, otherwise MultiKV config provider will not be set.
cfg := minimalWorkingConfig(t, dir, target)
cfg.RuntimeConfig.LoadPath = filepath.Join(dir, "config.yaml")
// This would be overwritten by the default values setting.
cfg.StorageConfig = storage.Config{
FSConfig: local.FSConfig{Directory: dir},
BoltDBShipperConfig: shipper.Config{
Config: indexshipper.Config{
SharedStoreType: config.StorageTypeFileSystem,
ActiveIndexDirectory: dir,
CacheLocation: dir,
Mode: indexshipper.ModeWriteOnly,
},
},
}
cfg.Ruler.Config.StoreConfig.Type = config.StorageTypeLocal
cfg.Ruler.Config.StoreConfig.Local.Directory = dir
c, err := New(cfg)
require.NoError(t, err)
@ -193,3 +160,101 @@ func TestMultiKVSetup(t *testing.T) {
})
}
}
func TestIndexGatewayRingMode_when_TargetIsRead(t *testing.T) {
dir := t.TempDir()
t.Run("IndexGateway always set to ring mode when running as part of read target", func(t *testing.T) {
cfg := minimalWorkingConfig(t, dir, Read)
c, err := New(cfg)
require.NoError(t, err)
services, err := c.ModuleManager.InitModuleServices(Read)
defer func() {
for _, service := range services {
service.StopAsync()
}
}()
require.NoError(t, err)
assert.Equal(t, c.Cfg.IndexGateway.Mode, indexgateway.RingMode)
})
t.Run("When IndexGateway is running independent of Read target", func(t *testing.T) {
t.Run("IndexGateway respects configured simple mode", func(t *testing.T) {
cfg := minimalWorkingConfig(t, dir, IndexGatewayRing)
cfg.IndexGateway.Mode = indexgateway.SimpleMode
c, err := New(cfg)
require.NoError(t, err)
services, err := c.ModuleManager.InitModuleServices(IndexGateway)
defer func() {
for _, service := range services {
service.StopAsync()
}
}()
require.NoError(t, err)
assert.Equal(t, c.Cfg.IndexGateway.Mode, indexgateway.SimpleMode)
})
t.Run("IndexGateway respects configured ring mode", func(t *testing.T) {
cfg := minimalWorkingConfig(t, dir, IndexGatewayRing)
cfg.IndexGateway.Mode = indexgateway.RingMode
c, err := New(cfg)
require.NoError(t, err)
services, err := c.ModuleManager.InitModuleServices(IndexGateway)
defer func() {
for _, service := range services {
service.StopAsync()
}
}()
require.NoError(t, err)
assert.Equal(t, c.Cfg.IndexGateway.Mode, indexgateway.RingMode)
})
})
}
func minimalWorkingConfig(t *testing.T, dir, target string) Config {
prepareGlobalMetricsRegistry(t)
cfg := Config{}
cfg.SchemaConfig = config.SchemaConfig{
Configs: []config.PeriodConfig{
{
IndexType: config.StorageTypeInMemory,
ObjectType: config.StorageTypeFileSystem,
RowShards: 16,
Schema: "v11",
From: config.DayTime{
Time: model.Now(),
},
},
},
}
flagext.DefaultValues(&cfg)
// Set to 0 to find any free port.
cfg.Server.HTTPListenPort = 0
cfg.Server.GRPCListenPort = 0
cfg.Target = []string{target}
// This would be overwritten by the default values setting.
cfg.StorageConfig = storage.Config{
FSConfig: local.FSConfig{Directory: dir},
BoltDBShipperConfig: shipper.Config{
Config: indexshipper.Config{
SharedStoreType: config.StorageTypeFileSystem,
ActiveIndexDirectory: dir,
CacheLocation: dir,
Mode: indexshipper.ModeWriteOnly,
},
},
}
cfg.Ruler.Config.StoreConfig.Type = config.StorageTypeLocal
cfg.Ruler.Config.StoreConfig.Local.Directory = dir
return cfg
}

Loading…
Cancel
Save