receive: allow enabling local tsdb compaction

Signed-off-by: Brett Jones <blockloop@users.noreply.github.com>

CHANGELOG.md

@@ -13,6 +13,7 @@ We use *breaking* word for marking changes that are not backward compatible (rel
 
 ### Fixed
 
+- [#1967](https://github.com/thanos-io/thanos/issues/1967) Receive: Allow enabling local TSDB compaction
 - [#1919](https://github.com/thanos-io/thanos/issues/1919) Compactor: Fixed potential data loss when uploading older blocks, or upload taking long time while compactor is
 running.
 - [#1937](https://github.com/thanos-io/thanos/pull/1937) Compactor: Improved synchronization of meta JSON files.

cmd/thanos/receive.go

@@ -71,7 +71,9 @@ func registerReceive(m map[string]setupFunc, app *kingpin.Application) {
 
 	replicationFactor := cmd.Flag("receive.replication-factor", "How many times to replicate incoming write requests.").Default("1").Uint64()
 
-	tsdbBlockDuration := modelDuration(cmd.Flag("tsdb.block-duration", "Duration for local TSDB blocks").Default("2h").Hidden())
+	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()
 
 	m[comp.String()] = func(g *run.Group, logger log.Logger, reg *prometheus.Registry, tracer opentracing.Tracer, _ bool) error {
 		lset, err := parseFlagLabels(*labelStrs)
@@ -128,7 +130,9 @@ func registerReceive(m map[string]setupFunc, app *kingpin.Application) {
 			*tenantHeader,
 			*replicaHeader,
 			*replicationFactor,
-			*tsdbBlockDuration,
+			*tsdbMinBlockDuration,
+			*tsdbMaxBlockDuration,
+			*ignoreBlockSize,
 			comp,
 		)
 	}
@@ -163,7 +167,9 @@ func runReceive(
 	tenantHeader string,
 	replicaHeader string,
 	replicationFactor uint64,
-	tsdbBlockDuration model.Duration,
+	tsdbMinBlockDuration model.Duration,
+	tsdbMaxBlockDuration model.Duration,
+	ignoreBlockSize bool,
 	comp component.SourceStoreAPI,
 ) error {
 	logger = log.With(logger, "component", "receive")
@@ -172,8 +178,8 @@ func runReceive(
 	tsdbCfg := &tsdb.Options{
 		RetentionDuration: retention,
 		NoLockfile:        true,
-		MinBlockDuration:  tsdbBlockDuration,
-		MaxBlockDuration:  tsdbBlockDuration,
+		MinBlockDuration:  tsdbMinBlockDuration,
+		MaxBlockDuration:  tsdbMaxBlockDuration,
 		WALCompression:    true,
 	}
 
@@ -207,6 +213,14 @@ func runReceive(
 	if len(confContentYaml) == 0 {
 		level.Info(logger).Log("msg", "No supported bucket was configured, uploads will be disabled")
 		upload = false
+
+		if tsdbMinBlockDuration != tsdbMaxBlockDuration {
+			if !ignoreBlockSize {
+				return errors.Errorf("found that TSDB Max time is %s and Min time is %s. "+
+					"Compaction needs to be disabled (tsdb.min-block-duration = tsdb.max-block-duration)", tsdbMaxBlockDuration, tsdbMinBlockDuration)
+			}
+			level.Warn(logger).Log("msg", "flag to ignore min/max block duration flags differing is being used. If the upload of a 2h block fails and a TSDB compaction happens that block may be missing from your Thanos bucket storage.")
+		}
 	}
 
 	// Start all components while we wait for TSDB to open but only load