Review fixes

Signed-off-by: Thor <[email protected]>

Author: Thor

CHANGELOG.md

@@ -16,6 +16,8 @@
 * [FEATURE] Added `global` ingestion rate limiter strategy. Deprecated `-distributor.limiter-reload-period` flag. #1766
 * [FEATURE] Added support for Microsoft Azure blob storage to be used for storing chunk data. #1913
 * [BUGFIX] Fixed unnecessary CAS operations done by the HA tracker when the jitter is enabled. #1861
+* [ENHANCEMENT] Experimental TSDB: Open existing TSDB on startup to prevent ingester from becoming ready before it can accept writes. #1917
+  * --experimental.tsdb.max-tsdb-opening-concurrency-on-startup
 
 ## 0.4.0 / 2019-12-02
 

pkg/ingester/ingester_v2.go

@@ -465,59 +465,57 @@ func (i *Ingester) closeAllTSDB() {
 // openExistingTSDB walks the user tsdb dir, and opens a tsdb for each user. This may start a WAL replay, so we limit the number of
 // concurrently opening TSDB.
 func (i *Ingester) openExistingTSDB(ctx context.Context) error {
+	level.Info(util.Logger).Log("msg", "opening existing TSDBs")
 	wg := &sync.WaitGroup{}
-	openGate := gate.New(i.cfg.TSDBConfig.MaxOpeningTSDBOnStartup)
+	openGate := gate.New(i.cfg.TSDBConfig.MaxTSDBOpeningConcurrencyOnStartup)
 
 	err := filepath.Walk(i.cfg.TSDBConfig.Dir, func(path string, info os.FileInfo, err error) error {
-		if path == i.cfg.TSDBConfig.Dir { // Nothing to do for root
+
+		// Skip root dir and all other files
+		if path == i.cfg.TSDBConfig.Dir || !info.IsDir() {
 			return nil
 		}
 
 		// Top level directories are assumed to be user TSDBs
-		if info.IsDir() {
+		userID := info.Name()
+		f, err := os.Open(path)
+		if err != nil {
+			level.Error(util.Logger).Log("msg", "unable to open user TSDB dir", "err", err, "user", userID, "path", path)
+			return filepath.SkipDir
+		}
+		defer f.Close()
 
-			userID := info.Name()
-			f, err := os.Open(path)
-			if err != nil {
-				level.Error(util.Logger).Log("msg", "unable to open user TSDB dir", "err", err, "user", userID)
+		// If the dir is empty skip it
+		if _, err := f.Readdirnames(1); err != nil {
+			if err != io.EOF {
+				level.Error(util.Logger).Log("msg", "unable to read TSDB dir", "err", err, "user", userID, "path", path)
 				return filepath.SkipDir
 			}
-			defer f.Close()
 
-			// If the dir is empty skip it
-			if _, err := f.Readdirnames(1); err != nil {
-				if err != io.EOF {
-					level.Error(util.Logger).Log("msg", "unable to read TSDB dir", "err", err, "user", userID)
-					return filepath.SkipDir
-				}
+			// Empty dir
+			return filepath.SkipDir
+		}
 
-				// Empty dir
-				return filepath.SkipDir
-			}
+		// Limit the number of TSDB's opening concurrently. Start blocks until there's a free spot available or the context is cancelled.
+		if err := openGate.Start(ctx); err != nil {
+			return err
+		}
 
-			// Limit the number of TSDB's opening concurrently
-			if err := openGate.Start(ctx); err != nil {
-				return err
+		wg.Add(1)
+		go func(userID string) {
+			defer wg.Done()
+			defer openGate.Done()
+			_, err := i.getOrCreateTSDB(userID, true) // force create the TSDB due to the lifecycler not having started yet
+			if err != nil {
+				level.Error(util.Logger).Log("msg", "unable to open user TSDB", "err", err, "user", userID)
 			}
+		}(userID)
 
-			wg.Add(1)
-			go func(userID string) {
-				defer wg.Done()
-				defer openGate.Done()
-				_, err := i.getOrCreateTSDB(userID, true) // force create the TSDB due to the lifecycler not having started yet
-				if err != nil {
-					level.Error(util.Logger).Log("msg", "unable to open user TSDB", "err", err, "user", userID)
-				}
-			}(userID)
-
-			return filepath.SkipDir // Don't descend into directories
-		}
-
-		// Skip all other files
-		return nil
+		return filepath.SkipDir // Don't descend into directories
 	})
 
 	// Wait for all opening routines to finish
 	wg.Wait()
+	level.Info(util.Logger).Log("msg", "completed opening existing TSDBs")
 	return err
 }

pkg/storage/tsdb/config.go

@@ -32,8 +32,8 @@ type Config struct {
 	Backend      string            `yaml:"backend"`
 	BucketStore  BucketStoreConfig `yaml:"bucket_store"`
 
-	// MaxOpeningTSDBOnStartup limits the number of concurrently opening TSDB's during startup
-	MaxOpeningTSDBOnStartup int `yaml:"max_opening_tsdb_on_startup"`
+	// MaxTSDBOpeningConcurrencyOnStartup limits the number of concurrently opening TSDB's during startup
+	MaxTSDBOpeningConcurrencyOnStartup int `yaml:"max_tsdb_opening_concurrency_on_startup"`
 
 	// Backends
 	S3  s3.Config  `yaml:"s3"`
@@ -85,7 +85,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.DurationVar(&cfg.Retention, "experimental.tsdb.retention-period", 6*time.Hour, "TSDB block retention")
 	f.DurationVar(&cfg.ShipInterval, "experimental.tsdb.ship-interval", 30*time.Second, "the frequency at which tsdb blocks are scanned for shipping")
 	f.StringVar(&cfg.Backend, "experimental.tsdb.backend", "s3", "TSDB storage backend to use")
-	f.IntVar(&cfg.MaxOpeningTSDBOnStartup, "experimental.tsdb.max-opening-tsdb-on-startup", 10, "limit the number of concurrently opening TSDB's on startup")
+	f.IntVar(&cfg.MaxTSDBOpeningConcurrencyOnStartup, "experimental.tsdb.max-tsdb-opening-concurrency-on-startup", 10, "limit the number of concurrently opening TSDB's on startup")
 }
 
 // Validate the config