add wal compression option in rule and receive (#1933)

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

CHANGELOG.md

@@ -51,6 +51,7 @@ Compactor now properly handles partial block uploads for all operation like rete
 - [#1833](https://github.com/thanos-io/thanos/pull/1833) `--shipper.upload-compacted` flag has been promoted to non hidden, non experimental state. More info available [here](docs/quick-tutorial.md#uploading-old-metrics).
 - [#1867](https://github.com/thanos-io/thanos/pull/1867) Ruler: now sets a `Thanos/$version` `User-Agent` in requests
 - [#1887](https://github.com/thanos-io/thanos/pull/1887) Service discovery now deduplicates targets between different target groups
+- [#1933](https://github.com/thanos-io/thanos/pull/1933) Add a flag `--tsdb.wal-compression` to configure whether to enable tsdb wal compression in ruler and receiver.
 
 ## [v0.9.0](https://github.com/thanos-io/thanos/releases/tag/v0.9.0) - 2019.12.03
 

cmd/thanos/receive.go

@@ -13,7 +13,6 @@ import (
 	opentracing "github.com/opentracing/opentracing-go"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/storage/tsdb"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
@@ -73,6 +72,8 @@ func registerReceive(m map[string]setupFunc, app *kingpin.Application) {
 
 	tsdbBlockDuration := modelDuration(cmd.Flag("tsdb.block-duration", "Duration for local TSDB blocks").Default("2h").Hidden())
 
+	walCompression := cmd.Flag("tsdb.wal-compression", "Compress the tsdb WAL.").Default("true").Bool()
+
 	m[comp.String()] = func(g *run.Group, logger log.Logger, reg *prometheus.Registry, tracer opentracing.Tracer, _ bool) error {
 		lset, err := parseFlagLabels(*labelStrs)
 		if err != nil {
@@ -87,6 +88,14 @@ func registerReceive(m map[string]setupFunc, app *kingpin.Application) {
 			}
 		}
 
+		tsdbOpts := &tsdb.Options{
+			MinBlockDuration:  *tsdbBlockDuration,
+			MaxBlockDuration:  *tsdbBlockDuration,
+			RetentionDuration: *retention,
+			NoLockfile:        true,
+			WALCompression:    *walCompression,
+		}
+
 		// Local is empty, so try to generate a local endpoint
 		// based on the hostname and the listening port.
 		if *local == "" {
@@ -121,14 +130,13 @@ func registerReceive(m map[string]setupFunc, app *kingpin.Application) {
 			*rwClientServerName,
 			*dataDir,
 			objStoreConfig,
+			tsdbOpts,
 			lset,
-			*retention,
 			cw,
 			*local,
 			*tenantHeader,
 			*replicaHeader,
 			*replicationFactor,
-			*tsdbBlockDuration,
 			comp,
 		)
 	}
@@ -156,27 +164,18 @@ func runReceive(
 	rwClientServerName string,
 	dataDir string,
 	objStoreConfig *extflag.PathOrContent,
+	tsdbOpts *tsdb.Options,
 	lset labels.Labels,
-	retention model.Duration,
 	cw *receive.ConfigWatcher,
 	endpoint string,
 	tenantHeader string,
 	replicaHeader string,
 	replicationFactor uint64,
-	tsdbBlockDuration model.Duration,
 	comp component.SourceStoreAPI,
 ) error {
 	logger = log.With(logger, "component", "receive")
 	level.Warn(logger).Log("msg", "setting up receive; the Thanos receive component is EXPERIMENTAL, it may break significantly without notice")
 
-	tsdbCfg := &tsdb.Options{
-		RetentionDuration: retention,
-		NoLockfile:        true,
-		MinBlockDuration:  tsdbBlockDuration,
-		MaxBlockDuration:  tsdbBlockDuration,
-		WALCompression:    true,
-	}
-
 	localStorage := &tsdb.ReadyStorage{}
 	rwTLSConfig, err := tls.NewServerConfig(log.With(logger, "protocol", "HTTP"), rwServerCert, rwServerKey, rwServerClientCA)
 	if err != nil {
@@ -225,7 +224,7 @@ func runReceive(
 	{
 		// TSDB.
 		cancel := make(chan struct{})
-		startTimeMargin := int64(2 * time.Duration(tsdbCfg.MinBlockDuration).Seconds() * 1000)
+		startTimeMargin := int64(2 * time.Duration(tsdbOpts.MinBlockDuration).Seconds() * 1000)
 		g.Add(func() error {
 			defer close(dbReady)
 			defer close(uploadC)
@@ -237,7 +236,7 @@ func runReceive(
 				dataDir,
 				log.With(logger, "component", "tsdb"),
 				reg,
-				tsdbCfg,
+				tsdbOpts,
 			)
 
 			// Before quitting, ensure the WAL is flushed and the DB is closed.

cmd/thanos/rule.go

@@ -79,6 +79,8 @@ func registerRule(m map[string]setupFunc, app *kingpin.Application) {
 	tsdbRetention := modelDuration(cmd.Flag("tsdb.retention", "Block retention time on local disk.").
 		Default("48h"))
 
+	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.").
 		Strings()
 	alertmgrsTimeout := cmd.Flag("alertmanagers.send-timeout", "Timeout for sending alerts to Alertmanager").Default("10s").Duration()
@@ -126,6 +128,7 @@ func registerRule(m map[string]setupFunc, app *kingpin.Application) {
 			MaxBlockDuration:  *tsdbBlockDuration,
 			RetentionDuration: *tsdbRetention,
 			NoLockfile:        true,
+			WALCompression:    *walCompression,
 		}
 
 		lookupQueries := map[string]struct{}{}

docs/components/rule.md

@@ -281,6 +281,7 @@ 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.wal-compression     Compress the tsdb WAL.
       --alertmanagers.url=ALERTMANAGERS.URL ...
                                  Alertmanager replica URLs to push firing
                                  alerts. Ruler claims success if push to at