create smaller unique files from boltdb shipper and other code refactorings

pkg/loki/loki.go

@@ -140,7 +140,7 @@ func New(cfg Config) (*Loki, error) {
 	if err := loki.setupModuleManager(); err != nil {
 		return nil, err
 	}
-	storage.RegisterCustomIndexClients(cfg.StorageConfig, prometheus.DefaultRegisterer)
+	storage.RegisterCustomIndexClients(&loki.cfg.StorageConfig, prometheus.DefaultRegisterer)
 
 	return loki, nil
 }

pkg/loki/modules.go

@@ -37,7 +37,7 @@ import (
 	"github.com/grafana/loki/pkg/querier"
 	"github.com/grafana/loki/pkg/querier/queryrange"
 	loki_storage "github.com/grafana/loki/pkg/storage"
-	"github.com/grafana/loki/pkg/storage/stores/local"
+	"github.com/grafana/loki/pkg/storage/stores/shipper"
 	serverutil "github.com/grafana/loki/pkg/util/server"
 	"github.com/grafana/loki/pkg/util/validation"
 )
@@ -182,7 +182,7 @@ func (t *Loki) initIngester() (_ services.Service, err error) {
 
 	// We want ingester to also query the store when using boltdb-shipper
 	pc := t.cfg.SchemaConfig.Configs[activePeriodConfig(t.cfg.SchemaConfig)]
-	if pc.IndexType == local.BoltDBShipperType {
+	if pc.IndexType == shipper.BoltDBShipperType {
 		t.cfg.Ingester.QueryStore = true
 		mlb, err := calculateMaxLookBack(pc, t.cfg.Ingester.QueryStoreMaxLookBackPeriod, t.cfg.Ingester.MaxChunkAge)
 		if err != nil {
@@ -243,17 +243,17 @@ func (t *Loki) initTableManager() (services.Service, error) {
 }
 
 func (t *Loki) initStore() (_ services.Service, err error) {
-	if t.cfg.SchemaConfig.Configs[activePeriodConfig(t.cfg.SchemaConfig)].IndexType == local.BoltDBShipperType {
+	if t.cfg.SchemaConfig.Configs[activePeriodConfig(t.cfg.SchemaConfig)].IndexType == shipper.BoltDBShipperType {
 		t.cfg.StorageConfig.BoltDBShipperConfig.IngesterName = t.cfg.Ingester.LifecyclerConfig.ID
 		switch t.cfg.Target {
 		case Ingester:
 			// We do not want ingester to unnecessarily keep downloading files
-			t.cfg.StorageConfig.BoltDBShipperConfig.Mode = local.ShipperModeWriteOnly
+			t.cfg.StorageConfig.BoltDBShipperConfig.Mode = shipper.ModeWriteOnly
 		case Querier:
 			// We do not want query to do any updates to index
-			t.cfg.StorageConfig.BoltDBShipperConfig.Mode = local.ShipperModeReadOnly
+			t.cfg.StorageConfig.BoltDBShipperConfig.Mode = shipper.ModeReadOnly
 		default:
-			t.cfg.StorageConfig.BoltDBShipperConfig.Mode = local.ShipperModeReadWrite
+			t.cfg.StorageConfig.BoltDBShipperConfig.Mode = shipper.ModeReadWrite
 		}
 	}
 
@@ -370,20 +370,20 @@ func activePeriodConfig(cfg chunk.SchemaConfig) int {
 // usingBoltdbShipper check whether current or the next index type is boltdb-shipper, returns true if yes.
 func usingBoltdbShipper(cfg chunk.SchemaConfig) bool {
 	activePCIndex := activePeriodConfig(cfg)
-	if cfg.Configs[activePCIndex].IndexType == local.BoltDBShipperType ||
-		(len(cfg.Configs)-1 > activePCIndex && cfg.Configs[activePCIndex+1].IndexType == local.BoltDBShipperType) {
+	if cfg.Configs[activePCIndex].IndexType == shipper.BoltDBShipperType ||
+		(len(cfg.Configs)-1 > activePCIndex && cfg.Configs[activePCIndex+1].IndexType == shipper.BoltDBShipperType) {
 		return true
 	}
 
 	return false
 }
 
 func calculateMaxLookBack(pc chunk.PeriodConfig, maxLookBackConfig, maxChunkAge time.Duration) (time.Duration, error) {
-	if pc.ObjectType != local.FilesystemObjectStoreType && maxLookBackConfig.Nanoseconds() != 0 {
+	if pc.ObjectType != shipper.FilesystemObjectStoreType && maxLookBackConfig.Nanoseconds() != 0 {
 		return 0, errors.New("it is an error to specify a non zero `query_store_max_look_back_period` value when using any object store other than `filesystem`")
 	}
 	// When using shipper, limit max look back for query to MaxChunkAge + upload interval by shipper + 15 mins to query only data whose index is not pushed yet
-	defaultMaxLookBack := maxChunkAge + local.ShipperFileUploadInterval + (15 * time.Minute)
+	defaultMaxLookBack := maxChunkAge + shipper.UploadInterval + (15 * time.Minute)
 
 	if maxLookBackConfig == 0 {
 		// If the QueryStoreMaxLookBackPeriod is still it's default value of 0, set it to the default calculated value.

pkg/storage/store.go

@@ -18,15 +18,15 @@ import (
 	"github.com/grafana/loki/pkg/logproto"
 	"github.com/grafana/loki/pkg/logql"
 	"github.com/grafana/loki/pkg/logql/stats"
-	"github.com/grafana/loki/pkg/storage/stores/local"
+	"github.com/grafana/loki/pkg/storage/stores/shipper"
 	"github.com/grafana/loki/pkg/util"
 )
 
 // Config is the loki storage configuration
 type Config struct {
 	storage.Config      `yaml:",inline"`
-	MaxChunkBatchSize   int                 `yaml:"max_chunk_batch_size"`
-	BoltDBShipperConfig local.ShipperConfig `yaml:"boltdb_shipper"`
+	MaxChunkBatchSize   int            `yaml:"max_chunk_batch_size"`
+	BoltDBShipperConfig shipper.Config `yaml:"boltdb_shipper"`
 }
 
 // RegisterFlags adds the flags required to configure this flag set.
@@ -64,7 +64,7 @@ func NewStore(cfg Config, storeCfg chunk.StoreConfig, schemaCfg chunk.SchemaConf
 // NewTableClient creates a TableClient for managing tables for index/chunk store.
 // ToDo: Add support in Cortex for registering custom table client like index client.
 func NewTableClient(name string, cfg Config) (chunk.TableClient, error) {
-	if name == local.BoltDBShipperType {
+	if name == shipper.BoltDBShipperType {
 		name = "boltdb"
 		cfg.FSConfig = cortex_local.FSConfig{Directory: cfg.BoltDBShipperConfig.ActiveIndexDirectory}
 	}
@@ -287,13 +287,13 @@ func filterChunksByTime(from, through model.Time, chunks []chunk.Chunk) []chunk.
 	return filtered
 }
 
-func RegisterCustomIndexClients(cfg Config, registerer prometheus.Registerer) {
+func RegisterCustomIndexClients(cfg *Config, registerer prometheus.Registerer) {
 	// BoltDB Shipper is supposed to be run as a singleton.
 	// This could also be done in NewBoltDBIndexClientWithShipper factory method but we are doing it here because that method is used
 	// in tests for creating multiple instances of it at a time.
 	var boltDBIndexClientWithShipper chunk.IndexClient
 
-	storage.RegisterIndexStore(local.BoltDBShipperType, func() (chunk.IndexClient, error) {
+	storage.RegisterIndexStore(shipper.BoltDBShipperType, func() (chunk.IndexClient, error) {
 		if boltDBIndexClientWithShipper != nil {
 			return boltDBIndexClientWithShipper, nil
 		}
@@ -303,9 +303,7 @@ func RegisterCustomIndexClients(cfg Config, registerer prometheus.Registerer) {
 			return nil, err
 		}
 
-		boltDBIndexClientWithShipper, err = local.NewBoltDBIndexClientWithShipper(
-			cortex_local.BoltDBConfig{Directory: cfg.BoltDBShipperConfig.ActiveIndexDirectory},
-			objectClient, cfg.BoltDBShipperConfig, registerer)
+		boltDBIndexClientWithShipper, err = shipper.NewShipper(cfg.BoltDBShipperConfig, objectClient, registerer)
 
 		return boltDBIndexClientWithShipper, err
 	}, func() (client chunk.TableClient, e error) {
@@ -314,6 +312,6 @@ func RegisterCustomIndexClients(cfg Config, registerer prometheus.Registerer) {
 			return nil, err
 		}
 
-		return local.NewBoltDBShipperTableClient(objectClient), nil
+		return shipper.NewBoltDBShipperTableClient(objectClient), nil
 	})
 }

pkg/storage/store_test.go

@@ -28,7 +28,7 @@ import (
 	"github.com/grafana/loki/pkg/logproto"
 	"github.com/grafana/loki/pkg/logql"
 	"github.com/grafana/loki/pkg/logql/marshal"
-	"github.com/grafana/loki/pkg/storage/stores/local"
+	"github.com/grafana/loki/pkg/storage/stores/shipper"
 	"github.com/grafana/loki/pkg/util/validation"
 )
 
@@ -731,7 +731,7 @@ func TestStore_MultipleBoltDBShippersInConfig(t *testing.T) {
 	require.NoError(t, err)
 
 	// config for BoltDB Shipper
-	boltdbShipperConfig := local.ShipperConfig{}
+	boltdbShipperConfig := shipper.Config{}
 	flagext.DefaultValues(&boltdbShipperConfig)
 	boltdbShipperConfig.ActiveIndexDirectory = path.Join(tempDir, "index")
 	boltdbShipperConfig.SharedStoreType = "filesystem"
@@ -748,7 +748,7 @@ func TestStore_MultipleBoltDBShippersInConfig(t *testing.T) {
 		BoltDBShipperConfig: boltdbShipperConfig,
 	}
 
-	RegisterCustomIndexClients(config, nil)
+	RegisterCustomIndexClients(&config, nil)
 
 	store, err := NewStore(config, chunk.StoreConfig{}, chunk.SchemaConfig{
 		Configs: []chunk.PeriodConfig{