Add index gateway to the read path (#6385)

Author: trevorwhitney

pkg/loki/loki.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},
 	}

pkg/loki/modules.go

@@ -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
 	}

pkg/loki/modules_test.go

@@ -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
+}