Loki: Allow configuring query_store_max_look_back_period when running a filesystem store and boltdb-shipper

docs/configuration/README.md

@@ -317,6 +317,13 @@ The `ingester_config` block configures Ingesters.
 # The maximum duration of a timeseries chunk in memory. If a timeseries runs for longer than this the current chunk will be flushed to the store and a new chunk created.
 [max_chunk_age: <duration> | default = 1h]
 
+# How far in the past an ingester is allowed to query the store for data.  
+# This is only useful for running multiple loki binaries with a shared ring with a `filesystem` store which is NOT shared between the binaries
+# When using any "shared" object store like S3 or GCS this value must always be left as 0
+# It is an error to configure this to a non-zero value when using any object store other than `filesystem`
+# Use a value of -1 to allow the ingester to query the store infinitely far back in time.
+[query_store_max_look_back_period: <duration> | default = 0]
+
 ```
 
 ### lifecycler_config

docs/operations/storage/boltdb-shipper.md

@@ -83,5 +83,21 @@ Frequency for checking updates can be configured with `resync_interval` config.
 To avoid keeping downloaded index files forever there is a ttl for them which defaults to 24 hours, which means if index files for a period are not used for 24 hours they would be removed from cache location.
 ttl can be configured using `cache_ttl` config.
 
+## Horizontal scaling of non-shared filesystem stores
+
+Using the boltdb-shipper also allows running the single binary Loki (or really just the ingesters) with a `filesystem` object store also using a shared hash ring.
+
+If you configure a shared ring via etcd/consul/memberlist you can run multiple instances of Loki on separate machines with separate filesystems,
+this now works because the ingesters are able to query the store directly.
+
+To enable this configuration in the ingester config you must set `query_store_max_look_back_period` according to how far back you want to store data, or use a value of -1 for infinite.
+
+At query time, any Loki instance can field the query, the instance will then use the ring to ask every other Loki instance for relevant data, 
+and because the ingesters can each query their store as far back as `query_store_max_look_back_period` allows, the correct data can be returned.
+
+Scaling up is as easy as adding more loki instances and letting them talk to the same ring.
+
+Scaling down is possible but manual, you would need to shutdown the loki instance and then physically copy the chunks directory and its index files in their entirety to another Loki instance.
+You cannot move them in partial it must be all, this other Loki instance will then find the boltdb index files and serve the chunks copied.
 
 

pkg/ingester/ingester.go

@@ -64,7 +64,7 @@ type Config struct {
 	ingesterClientFactory func(cfg client.Config, addr string) (client.HealthAndIngesterClient, error)
 
 	QueryStore                  bool          `yaml:"-"`
-	QueryStoreMaxLookBackPeriod time.Duration `yaml:"-"`
+	QueryStoreMaxLookBackPeriod time.Duration `yaml:"query_store_max_look_back_period"`
 }
 
 // RegisterFlags registers the flags.
@@ -84,6 +84,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.Float64Var(&cfg.SyncMinUtilization, "ingester.sync-min-utilization", 0, "Minimum utilization of chunk when doing synchronization.")
 	f.IntVar(&cfg.MaxReturnedErrors, "ingester.max-ignored-stream-errors", 10, "Maximum number of ignored stream errors to return. 0 to return all errors.")
 	f.DurationVar(&cfg.MaxChunkAge, "ingester.max-chunk-age", time.Hour, "Maximum chunk age before flushing.")
+	f.DurationVar(&cfg.QueryStoreMaxLookBackPeriod, "ingester.query-store-max-look-back-period", 0, "How far back should an ingester be allowed to query the store for data, for use only with boltdb-shipper index and filesystem object store. -1 for infinite.")
 }
 
 // Ingester builds chunks for incoming log streams.
@@ -414,7 +415,7 @@ func buildStoreRequest(cfg Config, req *logproto.QueryRequest) *logproto.QueryRe
 	}
 	start := req.Start
 	end := req.End
-	if cfg.QueryStoreMaxLookBackPeriod != 0 {
+	if cfg.QueryStoreMaxLookBackPeriod > 0 {
 		oldestStartTime := time.Now().Add(-cfg.QueryStoreMaxLookBackPeriod)
 		if oldestStartTime.After(req.Start) {
 			start = oldestStartTime

pkg/loki/modules.go

@@ -1,6 +1,7 @@
 package loki
 
 import (
+	"errors"
 	"fmt"
 	"net/http"
 	"os"
@@ -198,10 +199,14 @@ func (t *Loki) initIngester() (_ services.Service, err error) {
 	t.cfg.Ingester.LifecyclerConfig.ListenPort = t.cfg.Server.GRPCListenPort
 
 	// We want ingester to also query the store when using boltdb-shipper
-	if activeIndexType(t.cfg.SchemaConfig) == local.BoltDBShipperType {
+	pc := activePeriodConfig(t.cfg.SchemaConfig)
+	if pc.IndexType == local.BoltDBShipperType {
 		t.cfg.Ingester.QueryStore = true
-		// 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
-		t.cfg.Ingester.QueryStoreMaxLookBackPeriod = t.cfg.Ingester.MaxChunkAge + local.ShipperFileUploadInterval + (15 * time.Minute)
+		mlb, err := calculateMaxLookBack(pc, t.cfg.Ingester.QueryStoreMaxLookBackPeriod, t.cfg.Ingester.MaxChunkAge)
+		if err != nil {
+			return nil, err
+		}
+		t.cfg.Ingester.QueryStoreMaxLookBackPeriod = mlb
 	}
 
 	t.ingester, err = ingester.New(t.cfg.Ingester, t.cfg.IngesterClient, t.store, t.overrides)
@@ -256,7 +261,7 @@ func (t *Loki) initTableManager() (services.Service, error) {
 }
 
 func (t *Loki) initStore() (_ services.Service, err error) {
-	if activeIndexType(t.cfg.SchemaConfig) == local.BoltDBShipperType {
+	if activePeriodConfig(t.cfg.SchemaConfig).IndexType == local.BoltDBShipperType {
 		t.cfg.StorageConfig.BoltDBShipperConfig.IngesterName = t.cfg.Ingester.LifecyclerConfig.ID
 		switch t.cfg.Target {
 		case Ingester:
@@ -483,15 +488,35 @@ var modules = map[moduleName]module{
 	},
 }
 
-// activeIndexType type returns index type which would be applicable to logs that would be pushed starting now
+// activePeriodConfig type returns index type which would be applicable to logs that would be pushed starting now
 // Note: Another periodic config can be applicable in future which can change index type
-func activeIndexType(cfg chunk.SchemaConfig) string {
+func activePeriodConfig(cfg chunk.SchemaConfig) chunk.PeriodConfig {
 	now := model.Now()
 	i := sort.Search(len(cfg.Configs), func(i int) bool {
 		return cfg.Configs[i].From.Time > now
 	})
 	if i > 0 {
 		i--
 	}
-	return cfg.Configs[i].IndexType
+	return cfg.Configs[i]
+}
+
+func calculateMaxLookBack(pc chunk.PeriodConfig, maxLookBackConfig, maxChunkAge time.Duration) (time.Duration, error) {
+	if pc.ObjectType != local.FilesystemObjectStoreType && maxLookBackConfig.Milliseconds() != 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)
+
+	if maxLookBackConfig == 0 {
+		// If the QueryStoreMaxLookBackPeriod is still it's default value of 0, set it to the default calculated value.
+		return defaultMaxLookBack, nil
+	} else if maxLookBackConfig > 0 && maxLookBackConfig < defaultMaxLookBack {
+		// If the QueryStoreMaxLookBackPeriod is > 0 (-1 is allowed for infinite), make sure it's at least greater than the default or throw an error
+		return 0, fmt.Errorf("the configured query_store_max_look_back_period of '%v' is less than the calculated default of '%v' "+
+			"which is calculated based on the max_chunk_age + 15 minute boltdb-shipper interval + 15 min additional buffer.  Increase this value"+
+			"greater than the default or remove it from the configuration to use the default", maxLookBackConfig, defaultMaxLookBack)
+
+	}
+	return maxLookBackConfig, nil
 }

pkg/loki/modules_test.go

@@ -48,20 +48,20 @@ func TestActiveIndexType(t *testing.T) {
 		IndexType: "first",
 	}}
 
-	assert.Equal(t, "first", activeIndexType(cfg))
+	assert.Equal(t, cfg.Configs[0], activePeriodConfig(cfg))
 
 	// add a newer PeriodConfig in the past which should be considered
 	cfg.Configs = append(cfg.Configs, chunk.PeriodConfig{
 		From:      chunk.DayTime{Time: model.Now().Add(-12 * time.Hour)},
 		IndexType: "second",
 	})
-	assert.Equal(t, "second", activeIndexType(cfg))
+	assert.Equal(t, cfg.Configs[1], activePeriodConfig(cfg))
 
 	// add a newer PeriodConfig in the future which should not be considered
 	cfg.Configs = append(cfg.Configs, chunk.PeriodConfig{
 		From:      chunk.DayTime{Time: model.Now().Add(time.Hour)},
 		IndexType: "third",
 	})
-	assert.Equal(t, "second", activeIndexType(cfg))
+	assert.Equal(t, cfg.Configs[1], activePeriodConfig(cfg))
 
 }

pkg/storage/stores/local/shipper.go

@@ -34,6 +34,9 @@ const (
 	// BoltDBShipperType holds the index type for using boltdb with shipper which keeps flushing them to a shared storage
 	BoltDBShipperType = "boltdb-shipper"
 
+	// FilesystemObjectStoreType holds the periodic config type for the filesystem store
+	FilesystemObjectStoreType = "filesystem"
+
 	cacheCleanupInterval = 24 * time.Hour
 	storageKeyPrefix     = "index/"
 )
@@ -128,7 +131,7 @@ func NewShipper(cfg ShipperConfig, storageClient chunk.ObjectClient, boltDBGette
 // avoid uploading same files again with different name. If the filed does not exist we would create one with uploader name set to
 // ingester name and startup timestamp so that we randomise the name and do not override files from other ingesters.
 func (s *Shipper) getUploaderName() (string, error) {
-	uploader := fmt.Sprintf("%s-%d", s.cfg.IngesterName, time.Now().Unix())
+	uploader := fmt.Sprintf("%s-%d", s.cfg.IngesterName, time.Now().UnixNano())
 
 	uploaderFilePath := path.Join(s.cfg.ActiveIndexDirectory, "uploader", "name")
 	if err := chunk_util.EnsureDirectory(path.Dir(uploaderFilePath)); err != nil {