Skip to content

Commit

Permalink
Add tsdb wal compression type zstd (cortexproject#6232)
Browse files Browse the repository at this point in the history
SungJin1212 authored Sep 25, 2024
1 parent f21e06d commit a54da24
Showing 7 changed files with 69 additions and 15 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -6,6 +6,7 @@
* [CHANGE] Enable Compactor and Alertmanager in target all. #6204
* [FEATURE] Ruler: Experimental: Add `ruler.frontend-address` to allow query to query frontends instead of ingesters. #6151
* [FEATURE] Ruler: Minimize chances of missed rule group evaluations that can occur due to OOM kills, bad underlying nodes, or due to an unhealthy ruler that appears in the ring as healthy. This feature is enabled via `-ruler.enable-ha-evaluation` flag. #6129
* [ENHANCEMENT] Ingester: Add `blocks-storage.tsdb.wal-compression-type` to support zstd wal compression type. #6232
* [ENHANCEMENT] Query Frontend: Add info field to query response. #6207
* [ENHANCEMENT] Query Frontend: Add peakSample in query stats response. #6188
* [ENHANCEMENT] Ruler: Add new ruler metric `cortex_ruler_rule_groups_in_store` that is the total rule groups per tenant in store, which can be used to compare with `cortex_prometheus_rule_group_rules` to count the number of rule groups that are not loaded by a ruler. #5869
8 changes: 7 additions & 1 deletion docs/blocks-storage/querier.md
Original file line number Diff line number Diff line change
@@ -1423,10 +1423,16 @@ blocks_storage:
# CLI flag: -blocks-storage.tsdb.stripe-size
[stripe_size: <int> | default = 16384]

# True to enable TSDB WAL compression.
# Deprecated (use blocks-storage.tsdb.wal-compression-type instead): True to
# enable TSDB WAL compression.
# CLI flag: -blocks-storage.tsdb.wal-compression-enabled
[wal_compression_enabled: <boolean> | default = false]

# TSDB WAL type. Supported values are: 'snappy', 'zstd' and '' (disable
# compression)
# CLI flag: -blocks-storage.tsdb.wal-compression-type
[wal_compression_type: <string> | default = ""]

# TSDB WAL segments files max size (bytes).
# CLI flag: -blocks-storage.tsdb.wal-segment-size-bytes
[wal_segment_size_bytes: <int> | default = 134217728]
8 changes: 7 additions & 1 deletion docs/blocks-storage/store-gateway.md
Original file line number Diff line number Diff line change
@@ -1538,10 +1538,16 @@ blocks_storage:
# CLI flag: -blocks-storage.tsdb.stripe-size
[stripe_size: <int> | default = 16384]

# True to enable TSDB WAL compression.
# Deprecated (use blocks-storage.tsdb.wal-compression-type instead): True to
# enable TSDB WAL compression.
# CLI flag: -blocks-storage.tsdb.wal-compression-enabled
[wal_compression_enabled: <boolean> | default = false]

# TSDB WAL type. Supported values are: 'snappy', 'zstd' and '' (disable
# compression)
# CLI flag: -blocks-storage.tsdb.wal-compression-type
[wal_compression_type: <string> | default = ""]

# TSDB WAL segments files max size (bytes).
# CLI flag: -blocks-storage.tsdb.wal-segment-size-bytes
[wal_segment_size_bytes: <int> | default = 134217728]
8 changes: 7 additions & 1 deletion docs/configuration/config-file-reference.md
Original file line number Diff line number Diff line change
@@ -1968,10 +1968,16 @@ tsdb:
# CLI flag: -blocks-storage.tsdb.stripe-size
[stripe_size: <int> | default = 16384]

# True to enable TSDB WAL compression.
# Deprecated (use blocks-storage.tsdb.wal-compression-type instead): True to
# enable TSDB WAL compression.
# CLI flag: -blocks-storage.tsdb.wal-compression-enabled
[wal_compression_enabled: <boolean> | default = false]

# TSDB WAL type. Supported values are: 'snappy', 'zstd' and '' (disable
# compression)
# CLI flag: -blocks-storage.tsdb.wal-compression-type
[wal_compression_type: <string> | default = ""]

# TSDB WAL segments files max size (bytes).
# CLI flag: -blocks-storage.tsdb.wal-segment-size-bytes
[wal_segment_size_bytes: <int> | default = 134217728]
7 changes: 4 additions & 3 deletions pkg/ingester/ingester.go
Original file line number Diff line number Diff line change
@@ -2172,11 +2172,12 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
enableExemplars = true
}
oooTimeWindow := i.limits.OutOfOrderTimeWindow(userID)

walCompressType := wlog.CompressionNone
// TODO(yeya24): expose zstd compression for WAL.
if i.cfg.BlocksStorageConfig.TSDB.WALCompressionEnabled {
walCompressType = wlog.CompressionSnappy
if i.cfg.BlocksStorageConfig.TSDB.WALCompressionType != "" {
walCompressType = wlog.CompressionType(i.cfg.BlocksStorageConfig.TSDB.WALCompressionType)
}

// Create a new user database
db, err := tsdb.Open(udir, userLogger, tsdbPromReg, &tsdb.Options{
RetentionDuration: i.cfg.BlocksStorageConfig.TSDB.Retention.Milliseconds(),
28 changes: 19 additions & 9 deletions pkg/storage/tsdb/config.go
Original file line number Diff line number Diff line change
@@ -42,14 +42,15 @@ const (

// Validation errors
var (
errInvalidShipConcurrency = errors.New("invalid TSDB ship concurrency")
errInvalidOpeningConcurrency = errors.New("invalid TSDB opening concurrency")
errInvalidCompactionInterval = errors.New("invalid TSDB compaction interval")
errInvalidCompactionConcurrency = errors.New("invalid TSDB compaction concurrency")
errInvalidWALSegmentSizeBytes = errors.New("invalid TSDB WAL segment size bytes")
errInvalidStripeSize = errors.New("invalid TSDB stripe size")
errInvalidOutOfOrderCapMax = errors.New("invalid TSDB OOO chunks capacity (in samples)")
errEmptyBlockranges = errors.New("empty block ranges for TSDB")
errInvalidShipConcurrency = errors.New("invalid TSDB ship concurrency")
errInvalidOpeningConcurrency = errors.New("invalid TSDB opening concurrency")
errInvalidCompactionInterval = errors.New("invalid TSDB compaction interval")
errInvalidCompactionConcurrency = errors.New("invalid TSDB compaction concurrency")
errInvalidWALSegmentSizeBytes = errors.New("invalid TSDB WAL segment size bytes")
errInvalidStripeSize = errors.New("invalid TSDB stripe size")
errInvalidOutOfOrderCapMax = errors.New("invalid TSDB OOO chunks capacity (in samples)")
errEmptyBlockranges = errors.New("empty block ranges for TSDB")
errUnSupportedWALCompressionType = errors.New("unsupported WAL compression type, valid types are (zstd, snappy and '')")

ErrInvalidBucketIndexBlockDiscoveryStrategy = errors.New("bucket index block discovery strategy can only be enabled when bucket index is enabled")
ErrBlockDiscoveryStrategy = errors.New("invalid block discovery strategy")
@@ -137,6 +138,7 @@ type TSDBConfig struct {
HeadChunksWriteBufferSize int `yaml:"head_chunks_write_buffer_size_bytes"`
StripeSize int `yaml:"stripe_size"`
WALCompressionEnabled bool `yaml:"wal_compression_enabled"`
WALCompressionType string `yaml:"wal_compression_type"`
WALSegmentSizeBytes int `yaml:"wal_segment_size_bytes"`
FlushBlocksOnShutdown bool `yaml:"flush_blocks_on_shutdown"`
CloseIdleTSDBTimeout time.Duration `yaml:"close_idle_tsdb_timeout"`
@@ -183,7 +185,8 @@ func (cfg *TSDBConfig) RegisterFlags(f *flag.FlagSet) {
f.DurationVar(&cfg.HeadCompactionIdleTimeout, "blocks-storage.tsdb.head-compaction-idle-timeout", 1*time.Hour, "If TSDB head is idle for this duration, it is compacted. Note that up to 25% jitter is added to the value to avoid ingesters compacting concurrently. 0 means disabled.")
f.IntVar(&cfg.HeadChunksWriteBufferSize, "blocks-storage.tsdb.head-chunks-write-buffer-size-bytes", chunks.DefaultWriteBufferSize, "The write buffer size used by the head chunks mapper. Lower values reduce memory utilisation on clusters with a large number of tenants at the cost of increased disk I/O operations.")
f.IntVar(&cfg.StripeSize, "blocks-storage.tsdb.stripe-size", 16384, "The number of shards of series to use in TSDB (must be a power of 2). Reducing this will decrease memory footprint, but can negatively impact performance.")
f.BoolVar(&cfg.WALCompressionEnabled, "blocks-storage.tsdb.wal-compression-enabled", false, "True to enable TSDB WAL compression.")
f.BoolVar(&cfg.WALCompressionEnabled, "blocks-storage.tsdb.wal-compression-enabled", false, "Deprecated (use blocks-storage.tsdb.wal-compression-type instead): True to enable TSDB WAL compression.")
f.StringVar(&cfg.WALCompressionType, "blocks-storage.tsdb.wal-compression-type", "", "TSDB WAL type. Supported values are: 'snappy', 'zstd' and '' (disable compression)")
f.IntVar(&cfg.WALSegmentSizeBytes, "blocks-storage.tsdb.wal-segment-size-bytes", wlog.DefaultSegmentSize, "TSDB WAL segments files max size (bytes).")
f.BoolVar(&cfg.FlushBlocksOnShutdown, "blocks-storage.tsdb.flush-blocks-on-shutdown", false, "True to flush blocks to storage on shutdown. If false, incomplete blocks will be reused after restart.")
f.DurationVar(&cfg.CloseIdleTSDBTimeout, "blocks-storage.tsdb.close-idle-tsdb-timeout", 0, "If TSDB has not received any data for this duration, and all blocks from TSDB have been shipped, TSDB is closed and deleted from local disk. If set to positive value, this value should be equal or higher than -querier.query-ingesters-within flag to make sure that TSDB is not closed prematurely, which could cause partial query results. 0 or negative value disables closing of idle TSDB.")
@@ -232,6 +235,13 @@ func (cfg *TSDBConfig) Validate() error {
return errInvalidOutOfOrderCapMax
}

switch cfg.WALCompressionType {
case "snappy", "zstd", "":
// valid
default:
return errUnSupportedWALCompressionType
}

return nil
}

24 changes: 24 additions & 0 deletions pkg/storage/tsdb/config_test.go
Original file line number Diff line number Diff line change
@@ -121,6 +121,30 @@ func TestConfig_Validate(t *testing.T) {
},
expectedErr: errInvalidOutOfOrderCapMax,
},
"should pass on valid wal compression type (snappy)": {
setup: func(cfg *BlocksStorageConfig) {
cfg.TSDB.WALCompressionType = "snappy"
},
expectedErr: nil,
},
"should pass on valid wal compression type (zstd)": {
setup: func(cfg *BlocksStorageConfig) {
cfg.TSDB.WALCompressionType = "zstd"
},
expectedErr: nil,
},
"should pass on valid wal compression type ('')": {
setup: func(cfg *BlocksStorageConfig) {
cfg.TSDB.WALCompressionType = ""
},
expectedErr: nil,
},
"should fail on invalid wal compression type": {
setup: func(cfg *BlocksStorageConfig) {
cfg.TSDB.WALCompressionType = "dummy"
},
expectedErr: errUnSupportedWALCompressionType,
},
}

for testName, testData := range tests {

0 comments on commit a54da24

Please sign in to comment.