Ingest samples older than 1h for block store

docs/blocks-storage/querier.md

@@ -481,6 +481,14 @@ blocks_storage:
     # CLI flag: -experimental.blocks-storage.tsdb.flush-blocks-on-shutdown
     [flush_blocks_on_shutdown: <boolean> | default = false]
 
+    # Local directory to store backfill TSDBs in the ingesters.
+    # CLI flag: -experimental.blocks-storage.tsdb.backfill-dir
+    [backfill_dir: <string> | default = "backfill_tsdb"]
+
+    # Maximum accepted sample age by backfilling. 0 disables it.
+    # CLI flag: -experimental.blocks-storage.tsdb.backfill-max-age
+    [backfill_limit: <duration> | default = 0s]
+
     # limit the number of concurrently opening TSDB's on startup
     # CLI flag: -experimental.blocks-storage.tsdb.max-tsdb-opening-concurrency-on-startup
     [max_tsdb_opening_concurrency_on_startup: <int> | default = 10]

docs/blocks-storage/store-gateway.md

@@ -508,6 +508,14 @@ blocks_storage:
     # CLI flag: -experimental.blocks-storage.tsdb.flush-blocks-on-shutdown
     [flush_blocks_on_shutdown: <boolean> | default = false]
 
+    # Local directory to store backfill TSDBs in the ingesters.
+    # CLI flag: -experimental.blocks-storage.tsdb.backfill-dir
+    [backfill_dir: <string> | default = "backfill_tsdb"]
+
+    # Maximum accepted sample age by backfilling. 0 disables it.
+    # CLI flag: -experimental.blocks-storage.tsdb.backfill-max-age
+    [backfill_limit: <duration> | default = 0s]
+
     # limit the number of concurrently opening TSDB's on startup
     # CLI flag: -experimental.blocks-storage.tsdb.max-tsdb-opening-concurrency-on-startup
     [max_tsdb_opening_concurrency_on_startup: <int> | default = 10]

docs/configuration/config-file-reference.md

@@ -3251,6 +3251,14 @@ tsdb:
   # CLI flag: -experimental.blocks-storage.tsdb.flush-blocks-on-shutdown
   [flush_blocks_on_shutdown: <boolean> | default = false]
 
+  # Local directory to store backfill TSDBs in the ingesters.
+  # CLI flag: -experimental.blocks-storage.tsdb.backfill-dir
+  [backfill_dir: <string> | default = "backfill_tsdb"]
+
+  # Maximum accepted sample age by backfilling. 0 disables it.
+  # CLI flag: -experimental.blocks-storage.tsdb.backfill-max-age
+  [backfill_limit: <duration> | default = 0s]
+
   # limit the number of concurrently opening TSDB's on startup
   # CLI flag: -experimental.blocks-storage.tsdb.max-tsdb-opening-concurrency-on-startup
   [max_tsdb_opening_concurrency_on_startup: <int> | default = 10]

pkg/ingester/ingester_v2.go

@@ -2,8 +2,10 @@ package ingester
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"math"
 	"net/http"
 	"os"
@@ -12,6 +14,7 @@ import (
 	"time"
 
 	"github.com/go-kit/kit/log/level"
+	"github.com/oklog/ulid"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -20,6 +23,7 @@ import (
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/storage"
 	"github.com/prometheus/prometheus/tsdb"
+	tsdb_errors "github.com/prometheus/prometheus/tsdb/errors"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/objstore"
 	"github.com/thanos-io/thanos/pkg/shipper"
@@ -119,10 +123,42 @@ func (u *userTSDB) setLastUpdate(t time.Time) {
 	u.lastUpdate.Store(t.Unix())
 }
 
+func (u *userTSDB) getShippedBlocksULID() ([]ulid.ULID, error) {
+	b, err := ioutil.ReadFile(filepath.Join(u.Dir(), shipper.MetaFilename))
+	if err != nil {
+		return nil, errors.Wrap(err, "read shipper meta file")
+	}
+	var shipperMeta shipper.Meta
+	if err := json.Unmarshal(b, &shipperMeta); err != nil {
+		return nil, errors.Wrap(err, "unmarshal shipper meta file to json")
+	}
+	return shipperMeta.Uploaded, nil
+}
+
+func (u *userTSDB) getUnshippedBlocksULID() (unshipped []ulid.ULID, err error) {
+	shippedBlocks, err := u.getShippedBlocksULID()
+	if err != nil {
+		return nil, errors.Wrap(err, "get shipped blocks")
+	}
+
+Outer:
+	for _, b := range u.Blocks() {
+		for _, uid := range shippedBlocks {
+			if uid == b.Meta().ULID {
+				continue Outer
+			}
+		}
+		unshipped = append(unshipped, b.Meta().ULID)
+	}
+
+	return unshipped, nil
+}
+
 // TSDBState holds data structures used by the TSDB storage engine
 type TSDBState struct {
-	dbs    map[string]*userTSDB // tsdb sharded by userID
-	bucket objstore.Bucket
+	dbs         map[string]*userTSDB // tsdb sharded by userID
+	backfillDBs *backfillTSDBs
+	bucket      objstore.Bucket
 
 	// Value used by shipper as external label.
 	shipperIngesterID string
@@ -157,6 +193,7 @@ func newTSDBState(bucketClient objstore.Bucket, registerer prometheus.Registerer
 		tsdbMetrics:         newTSDBMetrics(registerer),
 		forceCompactTrigger: make(chan chan<- struct{}),
 		shipTrigger:         make(chan chan<- struct{}),
+		backfillDBs:         newBackfillTSDBs(),
 
 		compactionsTriggered: promauto.With(registerer).NewCounter(prometheus.CounterOpts{
 			Name: "cortex_ingester_tsdb_compactions_triggered_total",
@@ -274,6 +311,11 @@ func (i *Ingester) startingV2(ctx context.Context) error {
 		return errors.Wrap(err, "opening existing TSDBs")
 	}
 
+	// Scan and open backfill TSDB's that already exist on disk.
+	if err := i.openExistingBackfillTSDB(context.Background()); err != nil {
+		return errors.Wrap(err, "opening existing backfill TSDBs")
+	}
+
 	// Important: we want to keep lifecycler running until we ask it to stop, so we need to give it independent context
 	if err := i.lifecycler.StartAsync(context.Background()); err != nil {
 		return errors.Wrap(err, "failed to start lifecycler")
@@ -420,6 +462,7 @@ func (i *Ingester) v2Push(ctx context.Context, req *client.WriteRequest) (*clien
 
 	// Walk the samples, appending them to the users database
 	app := db.Appender()
+	var backfillApp *backfillAppender
 	for _, ts := range req.Timeseries {
 		// Check if we already have a cached reference for this series. Be aware
 		// that even if we have a reference it's not guaranteed to be still valid.
@@ -460,13 +503,27 @@ func (i *Ingester) v2Push(ctx context.Context, req *client.WriteRequest) (*clien
 				}
 			}
 
+			cause := errors.Cause(err)
+			if cause == storage.ErrOutOfBounds &&
+				i.cfg.BlocksStorageConfig.TSDB.BackfillLimit > 0 &&
+				s.TimestampMs > db.Head().MaxTime()-i.cfg.BlocksStorageConfig.TSDB.BackfillLimit.Milliseconds() {
+				if backfillApp == nil {
+					backfillApp = i.newBackfillAppender(userID)
+				}
+				err := backfillApp.add(ts.Labels, s)
+				if err == nil {
+					succeededSamplesCount++
+					continue
+				}
+				cause = errors.Cause(err)
+			}
+
 			failedSamplesCount++
 
 			// Check if the error is a soft error we can proceed on. If so, we keep track
 			// of it, so that we can return it back to the distributor, which will return a
 			// 400 error to the client. The client (Prometheus) will not retry on 400, and
 			// we actually ingested all samples which haven't failed.
-			cause := errors.Cause(err)
 			if cause == storage.ErrOutOfBounds || cause == storage.ErrOutOfOrderSample || cause == storage.ErrDuplicateSampleForTimestamp {
 				if firstPartialErr == nil {
 					firstPartialErr = errors.Wrapf(err, "series=%s, timestamp=%v", client.FromLabelAdaptersToLabels(ts.Labels).String(), model.Time(s.TimestampMs).Time().UTC().Format(time.RFC3339Nano))
@@ -495,8 +552,13 @@ func (i *Ingester) v2Push(ctx context.Context, req *client.WriteRequest) (*clien
 			}
 
 			// The error looks an issue on our side, so we should rollback
-			if rollbackErr := app.Rollback(); rollbackErr != nil {
-				level.Warn(util.Logger).Log("msg", "failed to rollback on error", "user", userID, "err", rollbackErr)
+			var merr tsdb_errors.MultiError
+			merr.Add(errors.Wrap(app.Rollback(), "main appender"))
+			if backfillApp != nil {
+				merr.Add(errors.Wrap(backfillApp.rollback(), "backfill appender"))
+			}
+			if merr.Err() != nil {
+				level.Warn(util.Logger).Log("msg", "failed to rollback on error", "user", userID, "err", merr.Err())
 			}
 
 			return nil, wrapWithUser(err, userID)
@@ -507,8 +569,13 @@ func (i *Ingester) v2Push(ctx context.Context, req *client.WriteRequest) (*clien
 	i.TSDBState.appenderAddDuration.Observe(time.Since(startAppend).Seconds())
 
 	startCommit := time.Now()
-	if err := app.Commit(); err != nil {
-		return nil, wrapWithUser(err, userID)
+	var merr tsdb_errors.MultiError
+	merr.Add(errors.Wrap(app.Commit(), "main appender"))
+	if backfillApp != nil {
+		merr.Add(errors.Wrap(backfillApp.commit(), "backfill appender"))
+	}
+	if merr.Err() != nil {
+		return nil, wrapWithUser(merr.Err(), userID)
 	}
 	i.TSDBState.appenderCommitDuration.Observe(time.Since(startCommit).Seconds())
 
@@ -571,6 +638,16 @@ func (i *Ingester) v2Query(ctx context.Context, req *client.QueryRequest) (*clie
 		return nil, ss.Err()
 	}
 
+	backfillSSs, err := i.backfillSelect(ctx, userID, int64(from), int64(through), matchers)
+	if err != nil {
+		return nil, err
+	}
+	if len(backfillSSs) > 0 {
+		// TODO(codesome): If any TSDB in backfill buckets were overlapping
+		// with main TSDB, then use tsdb.NewMergedVerticalSeriesSet
+		ss = tsdb.NewMergedSeriesSet(append(backfillSSs, ss))
+	}
+
 	numSamples := 0
 
 	result := &client.QueryResponse{}
@@ -802,6 +879,16 @@ func (i *Ingester) v2QueryStream(req *client.QueryRequest, stream client.Ingeste
 		return ss.Err()
 	}
 
+	backfillSSs, err := i.backfillSelect(ctx, userID, int64(from), int64(through), matchers)
+	if err != nil {
+		return err
+	}
+	if len(backfillSSs) > 0 {
+		// TODO(codesome): If any TSDB in backfill buckets were overlapping
+		// with main TSDB, then use tsdb.NewMergedVerticalSeriesSet
+		ss = tsdb.NewMergedSeriesSet(append(backfillSSs, ss))
+	}
+
 	timeseries := make([]client.TimeSeries, 0, queryStreamBatchSize)
 	batchSize := 0
 	numSamples := 0
@@ -921,11 +1008,28 @@ func (i *Ingester) getOrCreateTSDB(userID string, force bool) (*userTSDB, error)
 // createTSDB creates a TSDB for a given userID, and returns the created db.
 func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
 	tsdbPromReg := prometheus.NewRegistry()
-	udir := i.cfg.BlocksStorageConfig.TSDB.BlocksDir(userID)
-	userLogger := util.WithUserID(userID, util.Logger)
 
 	blockRanges := i.cfg.BlocksStorageConfig.TSDB.BlockRanges.ToMilliseconds()
+	userDB, err := i.createNewTSDB(
+		userID,
+		i.cfg.BlocksStorageConfig.TSDB.BlocksDir(userID),
+		blockRanges[0],
+		blockRanges[len(blockRanges)-1],
+		tsdbPromReg,
+	)
+	if err == nil {
+		// We set the limiter here because we don't want to limit
+		// series during WAL replay.
+		userDB.limiter = i.limiter
+		i.TSDBState.tsdbMetrics.setRegistryForUser(userID, tsdbPromReg)
+	}
+
+	return userDB, err
+}
 
+// createNewTSDB creates a TSDB for a given userID and data directory.
+func (i *Ingester) createNewTSDB(userID, dbDir string, minBlockDuration, maxBlockDuration int64, reg *prometheus.Registry) (*userTSDB, error) {
+	userLogger := util.WithUserID(userID, util.Logger)
 	userDB := &userTSDB{
 		userID:              userID,
 		refCache:            cortex_tsdb.NewRefCache(),
@@ -936,17 +1040,17 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
 	}
 
 	// Create a new user database
-	db, err := tsdb.Open(udir, userLogger, tsdbPromReg, &tsdb.Options{
+	db, err := tsdb.Open(dbDir, userLogger, reg, &tsdb.Options{
 		RetentionDuration:       i.cfg.BlocksStorageConfig.TSDB.Retention.Milliseconds(),
-		MinBlockDuration:        blockRanges[0],
-		MaxBlockDuration:        blockRanges[len(blockRanges)-1],
+		MinBlockDuration:        minBlockDuration,
+		MaxBlockDuration:        maxBlockDuration,
 		NoLockfile:              true,
 		StripeSize:              i.cfg.BlocksStorageConfig.TSDB.StripeSize,
 		WALCompression:          i.cfg.BlocksStorageConfig.TSDB.WALCompressionEnabled,
 		SeriesLifecycleCallback: userDB,
 	})
 	if err != nil {
-		return nil, errors.Wrapf(err, "failed to open TSDB: %s", udir)
+		return nil, errors.Wrapf(err, "failed to open TSDB: %s", dbDir)
 	}
 	db.DisableCompactions() // we will compact on our own schedule
 
@@ -956,13 +1060,10 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
 	level.Info(userLogger).Log("msg", "Running compaction after WAL replay")
 	err = db.Compact()
 	if err != nil {
-		return nil, errors.Wrapf(err, "failed to compact TSDB: %s", udir)
+		return nil, errors.Wrapf(err, "failed to compact TSDB: %s", dbDir)
 	}
 
 	userDB.DB = db
-	// We set the limiter here because we don't want to limit
-	// series during WAL replay.
-	userDB.limiter = i.limiter
 	userDB.setLastUpdate(time.Now()) // After WAL replay.
 
 	// Thanos shipper requires at least 1 external label to be set. For this reason,
@@ -982,16 +1083,15 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
 	if i.cfg.BlocksStorageConfig.TSDB.ShipInterval > 0 {
 		userDB.shipper = shipper.New(
 			userLogger,
-			tsdbPromReg,
-			udir,
+			reg,
+			dbDir,
 			cortex_tsdb.NewUserBucketClient(userID, i.TSDBState.bucket),
 			func() labels.Labels { return l },
 			metadata.ReceiveSource,
 			true, // Allow out of order uploads. It's fine in Cortex's context.
 		)
 	}
 
-	i.TSDBState.tsdbMetrics.setRegistryForUser(userID, tsdbPromReg)
 	return userDB, nil
 }
 
@@ -1023,8 +1123,15 @@ func (i *Ingester) closeAllTSDB() {
 		}(userDB)
 	}
 
-	// Wait until all Close() completed
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		i.closeAllBackfillTSDBs()
+	}()
+
 	i.userStatesMtx.Unlock()
+
+	// Wait until all Close() completed
 	wg.Wait()
 }
 
@@ -1177,9 +1284,17 @@ func (i *Ingester) compactionLoop(ctx context.Context) error {
 		select {
 		case <-ticker.C:
 			i.compactBlocks(ctx, false)
+			if err := i.compactOldBackfillTSDBsAndShip(i.cfg.BlocksStorageConfig.TSDB.BackfillLimit.Milliseconds()); err != nil {
+				level.Warn(util.Logger).Log("msg", "failed to compact and ship old backfill TSDBs", "err", err)
+			}
+
+			if err := i.closeOldBackfillTSDBsAndDelete(i.cfg.MaxChunkAge.Milliseconds()); err != nil {
+				level.Warn(util.Logger).Log("msg", "failed to close and delete old backfill TSDBs", "err", err)
+			}
 
 		case ch := <-i.TSDBState.forceCompactTrigger:
 			i.compactBlocks(ctx, true)
+			i.compactAllBackfillTSDBs(ctx)
 
 			// Notify back.
 			select {
@@ -1289,8 +1404,10 @@ func (i *Ingester) v2LifecyclerFlush() {
 	ctx := context.Background()
 
 	i.compactBlocks(ctx, true)
+	i.compactAllBackfillTSDBs(ctx)
 	if i.cfg.BlocksStorageConfig.TSDB.ShipInterval > 0 {
 		i.shipBlocks(ctx)
+		i.shipAllBackfillTSDBs(ctx)
 	}
 
 	level.Info(util.Logger).Log("msg", "finished flushing and shipping TSDB blocks")