Flusher target to flush WAL

Signed-off-by: Ganesh Vernekar <cs15btech11018@iith.ac.in>

Author: Ganesh Vernekar

CHANGELOG.md

@@ -8,7 +8,10 @@
   * `--store.min-chunk-age` has been removed
   * `--querier.query-store-after` has been added in it's place.
 * [FEATURE] Added user sub rings to distribute users to a subset of ingesters. #1947
-  * `--experimental.distributor.user-subring-size`
+* [FEATURE] Flusher target to flush the WAL.
+  * `-flusher.wal-dir` for the WAL directory to recover from.
+  * `-flusher.concurrent-flushes` for number of concurrent flushes.
+  * `-flusher.flush-op-timeout` is duration after which a flush should timeout.
 * [ENHANCEMENT] Experimental TSDB: Export TSDB Syncer metrics from Compactor component, they are prefixed with `cortex_compactor_`. #2023
 * [ENHANCEMENT] Experimental TSDB: Added dedicated flag `-experimental.tsdb.bucket-store.tenant-sync-concurrency` to configure the maximum number of concurrent tenants for which blocks are synched. #2026
 * [ENHANCEMENT] Experimental TSDB: Expose metrics for objstore operations (prefixed with `cortex_<component>_thanos_objstore_`, component being one of `ingester`, `querier` and `compactor`). #2027

cmd/cortex/main.go

@@ -7,6 +7,7 @@ import (
 	"math/rand"
 	"os"
 	"runtime"
+	"sync"
 	"time"
 
 	"github.com/go-kit/kit/log/level"
@@ -99,12 +100,26 @@ func main() {
 
 	level.Info(util.Logger).Log("msg", "Starting Cortex", "version", version.Info())
 
-	if err := t.Run(); err != nil {
-		level.Error(util.Logger).Log("msg", "error running Cortex", "err", err)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := t.Run(); err != nil {
+			level.Error(util.Logger).Log("msg", "error running Cortex", "err", err)
+		}
+	}()
+
+	if cfg.Target.IsJob() {
+		err = t.Stop()
+	}
+
+	wg.Wait()
+
+	if !cfg.Target.IsJob() {
+		err = t.Stop()
 	}
 
 	runtime.KeepAlive(ballast)
-	err = t.Stop()
 	util.CheckFatal("initializing cortex", err)
 }
 

docs/configuration/arguments.md

@@ -323,6 +323,17 @@ It also talks to a KVStore and has it's own copies of the same flags used by the
 - `-ingester.recover-from-wal`
    Set this to `true` to recover data from an existing WAL. The data is recovered even if WAL is disabled and this is set to `true`. The WAL dir needs to be set for this.
 
+#### Flusher
+
+- `-flusher.wal-dir`
+   Directory where the WAL data should be recovered from.
+
+- `-flusher.concurrent-flushes`
+   Number of concurrent flushes.
+
+- `-flusher.flush-op-timeout`
+   Duration after which a flush should timeout.
+
 ## Runtime Configuration file
 
 Cortex has a concept of "runtime config" file, which is simply a file that is reloaded while Cortex is running. It is used by some Cortex components to allow operator to change some aspects of Cortex configuration without restarting it. File is specified by using `-runtime-config.file=<filename>` flag and reload period (which defaults to 10 seconds) can be changed by `-runtime-config.reload-period=<duration>` flag. Previously this mechanism was only used by limits overrides, and flags were called `-limits.per-user-override-config=<filename>` and `-limits.per-user-override-period=10s` respectively. These are still used, if `-runtime-config.file=<filename>` is not specified.

docs/configuration/config-file-reference.md

@@ -53,6 +53,19 @@ Supported contents and default values of the config file:
 # The ingester_config configures the Cortex ingester.
 [ingester: <ingester_config>]
 
+flusher:
+  # Directory to read WAL from.
+  # CLI flag: -flusher.wal-dir
+  [wal_dir: <string> | default = "wal"]
+
+  # Number of concurrent goroutines flushing to dynamodb.
+  # CLI flag: -flusher.concurrent-flushes
+  [concurrent_flushes: <int> | default = 50]
+
+  # Timeout for individual flush operations.
+  # CLI flag: -flusher.flush-op-timeout
+  [flush_op_timeout: <duration> | default = 2m0s]
+
 # The storage_config configures where Cortex stores the data (chunks storage
 # engine).
 [storage: <storage_config>]

pkg/cortex/cortex.go

@@ -24,6 +24,7 @@ import (
 	config_client "github.com/cortexproject/cortex/pkg/configs/client"
 	"github.com/cortexproject/cortex/pkg/configs/db"
 	"github.com/cortexproject/cortex/pkg/distributor"
+	"github.com/cortexproject/cortex/pkg/flusher"
 	"github.com/cortexproject/cortex/pkg/ingester"
 	"github.com/cortexproject/cortex/pkg/ingester/client"
 	"github.com/cortexproject/cortex/pkg/querier"
@@ -66,6 +67,7 @@ type Config struct {
 	Querier        querier.Config           `yaml:"querier,omitempty"`
 	IngesterClient client.Config            `yaml:"ingester_client,omitempty"`
 	Ingester       ingester.Config          `yaml:"ingester,omitempty"`
+	Flusher        flusher.Config           `yaml:"flusher,omitempty"`
 	Storage        storage.Config           `yaml:"storage,omitempty"`
 	ChunkStore     chunk.StoreConfig        `yaml:"chunk_store,omitempty"`
 	Schema         chunk.SchemaConfig       `yaml:"schema,omitempty" doc:"hidden"` // Doc generation tool doesn't support it because part of the SchemaConfig doesn't support CLI flags (needs manual documentation)
@@ -101,6 +103,7 @@ func (c *Config) RegisterFlags(f *flag.FlagSet) {
 	c.Querier.RegisterFlags(f)
 	c.IngesterClient.RegisterFlags(f)
 	c.Ingester.RegisterFlags(f)
+	c.Flusher.RegisterFlags(f)
 	c.Storage.RegisterFlags(f)
 	c.ChunkStore.RegisterFlags(f)
 	c.Schema.RegisterFlags(f)
@@ -164,6 +167,7 @@ type Cortex struct {
 	overrides     *validation.Overrides
 	distributor   *distributor.Distributor
 	ingester      *ingester.Ingester
+	flusher       *flusher.Flusher
 	store         chunk.Store
 	worker        frontend.Worker
 	frontend      *frontend.Frontend

pkg/cortex/modules.go

@@ -25,6 +25,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/configs/api"
 	"github.com/cortexproject/cortex/pkg/configs/db"
 	"github.com/cortexproject/cortex/pkg/distributor"
+	"github.com/cortexproject/cortex/pkg/flusher"
 	"github.com/cortexproject/cortex/pkg/ingester"
 	"github.com/cortexproject/cortex/pkg/ingester/client"
 	"github.com/cortexproject/cortex/pkg/querier"
@@ -47,6 +48,7 @@ const (
 	Server
 	Distributor
 	Ingester
+	Flusher
 	Querier
 	QueryFrontend
 	Store
@@ -74,6 +76,8 @@ func (m moduleName) String() string {
 		return "store"
 	case Ingester:
 		return "ingester"
+	case Flusher:
+		return "flusher"
 	case Querier:
 		return "querier"
 	case QueryFrontend:
@@ -115,6 +119,9 @@ func (m *moduleName) Set(s string) error {
 	case "ingester":
 		*m = Ingester
 		return nil
+	case "flusher":
+		*m = Flusher
+		return nil
 	case "querier":
 		*m = Querier
 		return nil
@@ -156,6 +163,15 @@ func (m *moduleName) UnmarshalYAML(unmarshal func(interface{}) error) error {
 	return m.Set(s)
 }
 
+func (m moduleName) IsJob() bool {
+	switch m {
+	case Flusher:
+		return true
+	}
+
+	return false
+}
+
 func (t *Cortex) initServer(cfg *Config) (err error) {
 	t.server, err = server.New(cfg.Server)
 	return
@@ -311,6 +327,28 @@ func (t *Cortex) stopIngester() error {
 	return nil
 }
 
+func (t *Cortex) initFlusher(cfg *Config) (err error) {
+	// By the end of this call, the chunks should be recovered
+	// from the WAL and flushed.
+	t.flusher, err = flusher.New(
+		cfg.Flusher,
+		cfg.Ingester,
+		cfg.IngesterClient,
+		t.store,
+		prometheus.DefaultRegisterer,
+	)
+	if err != nil {
+		return
+	}
+
+	t.server.HTTP.Path("/ready").Handler(http.HandlerFunc(t.flusher.ReadinessHandler))
+	return
+}
+
+func (t *Cortex) stopFlusher() error {
+	return t.flusher.Close()
+}
+
 func (t *Cortex) initStore(cfg *Config) (err error) {
 	if cfg.Storage.Engine == storage.StorageEngineTSDB {
 		return nil
@@ -520,6 +558,12 @@ var modules = map[moduleName]module{
 		stop: (*Cortex).stopIngester,
 	},
 
+	Flusher: {
+		deps: []moduleName{Store, Server},
+		init: (*Cortex).initFlusher,
+		stop: (*Cortex).stopFlusher,
+	},
+
 	Querier: {
 		deps: []moduleName{Distributor, Store, Ring, Server},
 		init: (*Cortex).initQuerier,