tsdb: expose stripe size option to reduce tsdb memory footprint (#2185)

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

Author: thorfour

CHANGELOG.md

@@ -35,6 +35,7 @@
   * `-config.expand-env`
 * [FEATURE] Add /config HTTP endpoint which exposes the current Cortex configuration as YAML. #2165
 * [FEATURE] Allow Prometheus remote write directly to ingesters. #1491
+* [FEATURE] Add flag `-experimental.tsdb.stripe-size` to expose TSDB stripe size option. #2185
 * [ENHANCEMENT] Add `status` label to `cortex_alertmanager_configs` metric to gauge the number of valid and invalid configs. #2125
 * [ENHANCEMENT] Cassandra Authentication: added the `custom_authenticators` config option that allows users to authenticate with cassandra clusters using password authenticators that are not approved by default in [gocql](https://github.com/gocql/gocql/blob/81b8263d9fe526782a588ef94d3fa5c6148e5d67/conn.go#L27) #2093
 * [ENHANCEMENT] Experimental TSDB: Export TSDB Syncer metrics from Compactor component, they are prefixed with `cortex_compactor_`. #2023

pkg/ingester/ingester_v2.go

@@ -609,6 +609,7 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
 		RetentionDuration: uint64(i.cfg.TSDBConfig.Retention / time.Millisecond),
 		BlockRanges:       i.cfg.TSDBConfig.BlockRanges.ToMilliseconds(),
 		NoLockfile:        true,
+		StripeSize:        i.cfg.TSDBConfig.StripeSize,
 	})
 	if err != nil {
 		return nil, err

pkg/storage/tsdb/config.go

@@ -34,6 +34,7 @@ var (
 	errInvalidShipConcurrency       = errors.New("invalid TSDB ship concurrency")
 	errInvalidCompactionInterval    = errors.New("invalid TSDB compaction interval")
 	errInvalidCompactionConcurrency = errors.New("invalid TSDB compaction concurrency")
+	errInvalidStripeSize            = errors.New("invalid TSDB stripe size")
 )
 
 // Config holds the config information for TSDB storage
@@ -47,6 +48,7 @@ type Config struct {
 	BucketStore               BucketStoreConfig `yaml:"bucket_store"`
 	HeadCompactionInterval    time.Duration     `yaml:"head_compaction_interval"`
 	HeadCompactionConcurrency int               `yaml:"head_compaction_concurrency"`
+	StripeSize                int               `yaml:"stripe_size"`
 
 	// MaxTSDBOpeningConcurrencyOnStartup limits the number of concurrently opening TSDB's during startup
 	MaxTSDBOpeningConcurrencyOnStartup int `yaml:"max_tsdb_opening_concurrency_on_startup"`
@@ -114,6 +116,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.IntVar(&cfg.MaxTSDBOpeningConcurrencyOnStartup, "experimental.tsdb.max-tsdb-opening-concurrency-on-startup", 10, "limit the number of concurrently opening TSDB's on startup")
 	f.DurationVar(&cfg.HeadCompactionInterval, "experimental.tsdb.head-compaction-interval", 1*time.Minute, "How frequently does Cortex try to compact TSDB head. Block is only created if data covers smallest block range. Must be greater than 0 and max 5 minutes.")
 	f.IntVar(&cfg.HeadCompactionConcurrency, "experimental.tsdb.head-compaction-concurrency", 5, "Maximum number of tenants concurrently compacting TSDB head into a new block")
+	f.IntVar(&cfg.StripeSize, "experimental.tsdb.stripe-size", 16384, "Power of 2 to use for the number of shards of series to use in TSDB. Reducing this will decrease memory footprint, but can negatively impact performance.")
 }
 
 // Validate the config
@@ -134,6 +137,10 @@ func (cfg *Config) Validate() error {
 		return errInvalidCompactionConcurrency
 	}
 
+	if cfg.StripeSize <= 1 || (cfg.StripeSize&(cfg.StripeSize-1)) != 0 { // ensure stripe size is a positive power of 2
+		return errInvalidStripeSize
+	}
+
 	return nil
 }
 

pkg/storage/tsdb/config_test.go

@@ -20,6 +20,7 @@ func TestConfig_Validate(t *testing.T) {
 				Backend:                   "s3",
 				HeadCompactionInterval:    1 * time.Minute,
 				HeadCompactionConcurrency: 5,
+				StripeSize:                2,
 			},
 			expectedErr: nil,
 		},
@@ -28,12 +29,14 @@ func TestConfig_Validate(t *testing.T) {
 				Backend:                   "gcs",
 				HeadCompactionInterval:    1 * time.Minute,
 				HeadCompactionConcurrency: 5,
+				StripeSize:                2,
 			},
 			expectedErr: nil,
 		},
 		"should fail on unknown backend": {
 			config: Config{
-				Backend: "unknown",
+				Backend:    "unknown",
+				StripeSize: 2,
 			},
 			expectedErr: errUnsupportedBackend,
 		},
@@ -42,6 +45,7 @@ func TestConfig_Validate(t *testing.T) {
 				Backend:         "s3",
 				ShipInterval:    time.Minute,
 				ShipConcurrency: 0,
+				StripeSize:      2,
 			},
 			expectedErr: errInvalidShipConcurrency,
 		},
@@ -52,20 +56,23 @@ func TestConfig_Validate(t *testing.T) {
 				ShipConcurrency:           0,
 				HeadCompactionInterval:    1 * time.Minute,
 				HeadCompactionConcurrency: 5,
+				StripeSize:                2,
 			},
 			expectedErr: nil,
 		},
 		"should fail on invalid compaction interval": {
 			config: Config{
 				Backend:                "s3",
 				HeadCompactionInterval: 0 * time.Minute,
+				StripeSize:             2,
 			},
 			expectedErr: errInvalidCompactionInterval,
 		},
 		"should fail on too high compaction interval": {
 			config: Config{
 				Backend:                "s3",
 				HeadCompactionInterval: 10 * time.Minute,
+				StripeSize:             2,
 			},
 			expectedErr: errInvalidCompactionInterval,
 		},
@@ -74,6 +81,7 @@ func TestConfig_Validate(t *testing.T) {
 				Backend:                   "s3",
 				HeadCompactionInterval:    time.Minute,
 				HeadCompactionConcurrency: 0,
+				StripeSize:                2,
 			},
 			expectedErr: errInvalidCompactionConcurrency,
 		},
@@ -82,6 +90,43 @@ func TestConfig_Validate(t *testing.T) {
 				Backend:                   "s3",
 				HeadCompactionInterval:    time.Minute,
 				HeadCompactionConcurrency: 10,
+				StripeSize:                2,
+			},
+			expectedErr: nil,
+		},
+		"should fail on negative stripe size": {
+			config: Config{
+				Backend:                   "s3",
+				HeadCompactionInterval:    1 * time.Minute,
+				HeadCompactionConcurrency: 5,
+				StripeSize:                -2,
+			},
+			expectedErr: errInvalidStripeSize,
+		},
+		"should fail on stripe size 0": {
+			config: Config{
+				Backend:                   "s3",
+				HeadCompactionInterval:    1 * time.Minute,
+				HeadCompactionConcurrency: 5,
+				StripeSize:                0,
+			},
+			expectedErr: errInvalidStripeSize,
+		},
+		"should fail on stripe size 1": {
+			config: Config{
+				Backend:                   "s3",
+				HeadCompactionInterval:    1 * time.Minute,
+				HeadCompactionConcurrency: 5,
+				StripeSize:                1,
+			},
+			expectedErr: errInvalidStripeSize,
+		},
+		"should pass on stripe size": {
+			config: Config{
+				Backend:                   "s3",
+				HeadCompactionInterval:    1 * time.Minute,
+				HeadCompactionConcurrency: 5,
+				StripeSize:                1 << 14,
 			},
 			expectedErr: nil,
 		},