receive, rule: Lock TSDB directories

CHANGELOG.md

@@ -29,6 +29,7 @@ We use *breaking* word for marking changes that are not backward compatible (rel
 ### Changed
 
 - [#2893](https://github.com/thanos-io/thanos/pull/2893) Store: Rename metric `thanos_bucket_store_cached_postings_compression_time_seconds` to `thanos_bucket_store_cached_postings_compression_time_seconds_total`.
+- [#2915](https://github.com/thanos-io/thanos/pull/2915) Receive,Ruler: Enable TSDB directory locking by default. Add a new flag (`--tsdb.no-lockfile`) to override behavior.
 
 ## [v0.14.0](https://github.com/thanos-io/thanos/releases/tag/v0.14.0) - 2020.07.10
 

cmd/thanos/receive.go

@@ -86,10 +86,10 @@ func registerReceive(m map[string]setupFunc, app *kingpin.Application) {
 
 	tsdbMinBlockDuration := modelDuration(cmd.Flag("tsdb.min-block-duration", "Min duration for local TSDB blocks").Default("2h").Hidden())
 	tsdbMaxBlockDuration := modelDuration(cmd.Flag("tsdb.max-block-duration", "Max duration for local TSDB blocks").Default("2h").Hidden())
-	ignoreBlockSize := cmd.Flag("shipper.ignore-unequal-block-size", "If true receive will not require min and max block size flags to be set to the same value. Only use this if you want to keep long retention and compaction enabled, as in the worst case it can result in ~2h data loss for your Thanos bucket storage.").Default("false").Hidden().Bool()
-
 	walCompression := cmd.Flag("tsdb.wal-compression", "Compress the tsdb WAL.").Default("true").Bool()
+	noLockFile := cmd.Flag("tsdb.no-lockfile", "Do not create lockfile in TSDB data directory. In any case, the lockfiles will be deleted on next startup.").Default("false").Bool()
 
+	ignoreBlockSize := cmd.Flag("shipper.ignore-unequal-block-size", "If true receive will not require min and max block size flags to be set to the same value. Only use this if you want to keep long retention and compaction enabled, as in the worst case it can result in ~2h data loss for your Thanos bucket storage.").Default("false").Hidden().Bool()
 	allowOutOfOrderUpload := cmd.Flag("shipper.allow-out-of-order-uploads",
 		"If true, shipper will skip failed block uploads in the given iteration and retry later. This means that some newer blocks might be uploaded sooner than older blocks."+
 			"This can trigger compaction without those blocks and as a result will create an overlap situation. Set it to true if you have vertical compaction enabled and wish to upload blocks as soon as possible without caring"+
@@ -114,7 +114,7 @@ func registerReceive(m map[string]setupFunc, app *kingpin.Application) {
 			MinBlockDuration:  int64(time.Duration(*tsdbMinBlockDuration) / time.Millisecond),
 			MaxBlockDuration:  int64(time.Duration(*tsdbMaxBlockDuration) / time.Millisecond),
 			RetentionDuration: int64(time.Duration(*retention) / time.Millisecond),
-			NoLockfile:        true,
+			NoLockfile:        *noLockFile,
 			WALCompression:    *walCompression,
 		}
 
@@ -304,6 +304,11 @@ func runReceive(
 			Help: "Number of Multi DB completed reloads with flush and potential upload due to hashring changes",
 		})
 
+		level.Debug(logger).Log("msg", "removing storage lock files if any")
+		if err := dbs.RemoveLockFilesIfAny(); err != nil {
+			return errors.Wrap(err, "remove storage lock files")
+		}
+
 		// TSDBs reload logic, listening on hashring changes.
 		cancel := make(chan struct{})
 		g.Add(func() error {

cmd/thanos/rule.go

@@ -8,6 +8,7 @@ import (
 	"math/rand"
 	"net/http"
 	"net/url"
+	"os"
 	"path/filepath"
 	"strconv"
 	"strings"
@@ -25,9 +26,11 @@ import (
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/rules"
-	tsdb "github.com/prometheus/prometheus/tsdb"
+	"github.com/prometheus/prometheus/tsdb"
 	tsdberrors "github.com/prometheus/prometheus/tsdb/errors"
 	"github.com/prometheus/prometheus/util/strutil"
+	"gopkg.in/alecthomas/kingpin.v2"
+
 	"github.com/thanos-io/thanos/pkg/alert"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/component"
@@ -51,7 +54,6 @@ import (
 	"github.com/thanos-io/thanos/pkg/tls"
 	"github.com/thanos-io/thanos/pkg/tracing"
 	"github.com/thanos-io/thanos/pkg/ui"
-	"gopkg.in/alecthomas/kingpin.v2"
 )
 
 // registerRule registers a rule command.
@@ -77,7 +79,7 @@ func registerRule(m map[string]setupFunc, app *kingpin.Application) {
 		Default("2h"))
 	tsdbRetention := modelDuration(cmd.Flag("tsdb.retention", "Block retention time on local disk.").
 		Default("48h"))
-
+	noLockFile := cmd.Flag("tsdb.no-lockfile", "Do not create lockfile in TSDB data directory. In any case, the lockfiles will be deleted on next startup.").Default("false").Bool()
 	walCompression := cmd.Flag("tsdb.wal-compression", "Compress the tsdb WAL.").Default("true").Bool()
 
 	alertmgrs := cmd.Flag("alertmanagers.url", "Alertmanager replica URLs to push firing alerts. Ruler claims success if push to at least one alertmanager from discovered succeeds. The scheme should not be empty e.g `http` might be used. The scheme may be prefixed with 'dns+' or 'dnssrv+' to detect Alertmanager IPs through respective DNS lookups. The port defaults to 9093 or the SRV record's value. The URL path is used as a prefix for the regular Alertmanager API path.").
@@ -134,7 +136,7 @@ func registerRule(m map[string]setupFunc, app *kingpin.Application) {
 			MinBlockDuration:  int64(time.Duration(*tsdbBlockDuration) / time.Millisecond),
 			MaxBlockDuration:  int64(time.Duration(*tsdbBlockDuration) / time.Millisecond),
 			RetentionDuration: int64(time.Duration(*tsdbRetention) / time.Millisecond),
-			NoLockfile:        true,
+			NoLockfile:        *noLockFile,
 			WALCompression:    *walCompression,
 		}
 
@@ -350,6 +352,12 @@ func runRule(
 	if err != nil {
 		return errors.Wrap(err, "open TSDB")
 	}
+
+	level.Debug(logger).Log("msg", "removing storage lock file if any")
+	if err := removeLockfileIfAny(logger, dataDir); err != nil {
+		return errors.Wrap(err, "remove storage lock files")
+	}
+
 	{
 		done := make(chan struct{})
 		g.Add(func() error {
@@ -642,6 +650,21 @@ func runRule(
 	return nil
 }
 
+func removeLockfileIfAny(logger log.Logger, dataDir string) error {
+	absdir, err := filepath.Abs(dataDir)
+	if err != nil {
+		return err
+	}
+	if err := os.Remove(filepath.Join(absdir, "lock")); err != nil {
+		if os.IsNotExist(err) {
+			return nil
+		}
+		return err
+	}
+	level.Info(logger).Log("msg", "a leftover lockfile found and removed")
+	return nil
+}
+
 func parseFlagLabels(s []string) (labels.Labels, error) {
 	var lset labels.Labels
 	for _, l := range s {

docs/components/receive.md

@@ -159,5 +159,8 @@ Flags:
                                  How many times to replicate incoming write
                                  requests.
       --tsdb.wal-compression     Compress the tsdb WAL.
+      --tsdb.no-lockfile         Do not create lockfile in TSDB data directory.
+                                 In any case, the lockfiles will be deleted on
+                                 next startup.
 
 ```

docs/components/rule.md

@@ -282,6 +282,9 @@ Flags:
       --eval-interval=30s        The default evaluation interval to use.
       --tsdb.block-duration=2h   Block duration for TSDB block.
       --tsdb.retention=48h       Block retention time on local disk.
+      --tsdb.no-lockfile         Do not create lockfile in TSDB data directory.
+                                 In any case, the lockfiles will be deleted on
+                                 next startup.
       --tsdb.wal-compression     Compress the tsdb WAL.
       --alertmanagers.url=ALERTMANAGERS.URL ...
                                  Alertmanager replica URLs to push firing

pkg/receive/multitsdb.go

@@ -8,6 +8,7 @@ import (
 	"io/ioutil"
 	"os"
 	"path"
+	"path/filepath"
 	"sync"
 
 	"github.com/go-kit/kit/log"
@@ -213,6 +214,32 @@ func (t *MultiTSDB) Sync(ctx context.Context) error {
 	return merr.Err()
 }
 
+func (t *MultiTSDB) RemoveLockFilesIfAny() error {
+	fis, err := ioutil.ReadDir(t.dataDir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil
+		}
+		return err
+	}
+
+	merr := terrors.MultiError{}
+	for _, fi := range fis {
+		if !fi.IsDir() {
+			continue
+		}
+		if err := os.Remove(filepath.Join(t.defaultTenantDataDir(fi.Name()), "lock")); err != nil {
+			if os.IsNotExist(err) {
+				continue
+			}
+			merr.Add(err)
+			continue
+		}
+		level.Info(t.logger).Log("msg", "a leftover lockfile found and removed", "tenant", fi.Name())
+	}
+	return merr.Err()
+}
+
 func (t *MultiTSDB) TSDBStores() map[string]storepb.StoreServer {
 	t.mtx.RLock()
 	defer t.mtx.RUnlock()
@@ -230,7 +257,7 @@ func (t *MultiTSDB) TSDBStores() map[string]storepb.StoreServer {
 func (t *MultiTSDB) startTSDB(logger log.Logger, tenantID string, tenant *tenant) error {
 	reg := prometheus.WrapRegistererWith(prometheus.Labels{"tenant": tenantID}, t.reg)
 	lbls := append(t.labels, labels.Label{Name: t.tenantLabelName, Value: tenantID})
-	dataDir := path.Join(t.dataDir, tenantID)
+	dataDir := t.defaultTenantDataDir(tenantID)
 
 	level.Info(logger).Log("msg", "opening TSDB")
 	opts := *t.tsdbOpts
@@ -263,6 +290,10 @@ func (t *MultiTSDB) startTSDB(logger log.Logger, tenantID string, tenant *tenant
 	return nil
 }
 
+func (t *MultiTSDB) defaultTenantDataDir(tenantID string) string {
+	return path.Join(t.dataDir, tenantID)
+}
+
 func (t *MultiTSDB) getOrLoadTenant(tenantID string, blockingStart bool) (*tenant, error) {
 	// Fast path, as creating tenants is a very rare operation.
 	t.mtx.RLock()