Ingester: Added -blocks-storage.tsdb.head-chunks-write-queue-size allowing to configure the size of the in-memory queue used before flushing chunks to the disk . #5000

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

Author: t00350320

CHANGELOG.md

@@ -73,6 +73,7 @@
 * [FEATURE] QueryFrontend: Support vertical sharding for subqueries. #4955
 * [FEATURE] Querier: Added a new limit `-querier.max-fetched-data-bytes-per-query` allowing to limit the maximum size of all data in bytes that a query can fetch from each ingester and storage. #4854
 * [FEATURE] Added 2 flags `-alertmanager.alertmanager-client.grpc-compression` and `-querier.store-gateway-client.grpc-compression` to configure compression methods for grpc clients. #4889
+* [FEATURE] Ingester: Added `-blocks-storage.tsdb.head-chunks-write-queue-size` allowing to configure the size of the in-memory queue used before flushing chunks to the disk . #5000
 * [BUGFIX] Storage/Bucket: Enable AWS SDK for go authentication for s3 to fix IMDSv1 authentication. #4897
 * [BUGFIX] Memberlist: Add join with no retrying when starting service. #4804
 * [BUGFIX] Ruler: Fix /ruler/rule_groups returns YAML with extra fields. #4767

pkg/ingester/ingester.go

@@ -1846,6 +1846,7 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
 		EnableExemplarStorage:     enableExemplars,
 		IsolationDisabled:         true,
 		MaxExemplars:              int64(i.cfg.BlocksStorageConfig.TSDB.MaxExemplars),
+		HeadChunksWriteQueueSize:  i.cfg.BlocksStorageConfig.TSDB.HeadChunksWriteQueueSize,
 	}, nil)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to open TSDB: %s", udir)

pkg/ingester/metrics.go

@@ -336,25 +336,26 @@ type tsdbMetrics struct {
 	uploadFailures  *prometheus.Desc // sum(thanos_shipper_upload_failures_total)
 
 	// Metrics aggregated from TSDB.
-	tsdbCompactionsTotal         *prometheus.Desc
-	tsdbCompactionDuration       *prometheus.Desc
-	tsdbFsyncDuration            *prometheus.Desc
-	tsdbPageFlushes              *prometheus.Desc
-	tsdbPageCompletions          *prometheus.Desc
-	tsdbWALTruncateFail          *prometheus.Desc
-	tsdbWALTruncateTotal         *prometheus.Desc
-	tsdbWALTruncateDuration      *prometheus.Desc
-	tsdbWALCorruptionsTotal      *prometheus.Desc
-	tsdbWALWritesFailed          *prometheus.Desc
-	tsdbHeadTruncateFail         *prometheus.Desc
-	tsdbHeadTruncateTotal        *prometheus.Desc
-	tsdbHeadGcDuration           *prometheus.Desc
-	tsdbActiveAppenders          *prometheus.Desc
-	tsdbSeriesNotFound           *prometheus.Desc
-	tsdbChunks                   *prometheus.Desc
-	tsdbChunksCreatedTotal       *prometheus.Desc
-	tsdbChunksRemovedTotal       *prometheus.Desc
-	tsdbMmapChunkCorruptionTotal *prometheus.Desc
+	tsdbCompactionsTotal               *prometheus.Desc
+	tsdbCompactionDuration             *prometheus.Desc
+	tsdbFsyncDuration                  *prometheus.Desc
+	tsdbPageFlushes                    *prometheus.Desc
+	tsdbPageCompletions                *prometheus.Desc
+	tsdbWALTruncateFail                *prometheus.Desc
+	tsdbWALTruncateTotal               *prometheus.Desc
+	tsdbWALTruncateDuration            *prometheus.Desc
+	tsdbWALCorruptionsTotal            *prometheus.Desc
+	tsdbWALWritesFailed                *prometheus.Desc
+	tsdbHeadTruncateFail               *prometheus.Desc
+	tsdbHeadTruncateTotal              *prometheus.Desc
+	tsdbHeadGcDuration                 *prometheus.Desc
+	tsdbActiveAppenders                *prometheus.Desc
+	tsdbSeriesNotFound                 *prometheus.Desc
+	tsdbChunks                         *prometheus.Desc
+	tsdbChunksCreatedTotal             *prometheus.Desc
+	tsdbChunksRemovedTotal             *prometheus.Desc
+	tsdbMmapChunkCorruptionTotal       *prometheus.Desc
+	tsdbChunkwriteQueueOperationsTotal *prometheus.Desc
 
 	tsdbExemplarsTotal          *prometheus.Desc
 	tsdbExemplarsInStorage      *prometheus.Desc
@@ -478,6 +479,10 @@ func newTSDBMetrics(r prometheus.Registerer) *tsdbMetrics {
 			"cortex_ingester_tsdb_mmap_chunk_corruptions_total",
 			"Total number of memory-mapped TSDB chunk corruptions.",
 			nil, nil),
+		tsdbChunkwriteQueueOperationsTotal: prometheus.NewDesc(
+			"cortex_ingester_tsdb_chunk_write_queue_operations_total",
+			"Number of currently tsdb chunk write queues.",
+			[]string{"user", "operation"}, nil),
 		tsdbLoadedBlocks: prometheus.NewDesc(
 			"cortex_ingester_tsdb_blocks_loaded",
 			"Number of currently loaded data blocks",
@@ -579,6 +584,7 @@ func (sm *tsdbMetrics) Describe(out chan<- *prometheus.Desc) {
 	out <- sm.tsdbChunksCreatedTotal
 	out <- sm.tsdbChunksRemovedTotal
 	out <- sm.tsdbMmapChunkCorruptionTotal
+	out <- sm.tsdbChunkwriteQueueOperationsTotal
 	out <- sm.tsdbLoadedBlocks
 	out <- sm.tsdbSymbolTableSize
 	out <- sm.tsdbReloads
@@ -628,6 +634,7 @@ func (sm *tsdbMetrics) Collect(out chan<- prometheus.Metric) {
 	data.SendSumOfCountersPerUser(out, sm.tsdbChunksCreatedTotal, "prometheus_tsdb_head_chunks_created_total")
 	data.SendSumOfCountersPerUser(out, sm.tsdbChunksRemovedTotal, "prometheus_tsdb_head_chunks_removed_total")
 	data.SendSumOfCounters(out, sm.tsdbMmapChunkCorruptionTotal, "prometheus_tsdb_mmap_chunk_corruptions_total")
+	data.SendSumOfCountersPerUserWithLabels(out, sm.tsdbChunkwriteQueueOperationsTotal, "prometheus_tsdb_chunk_write_queue_operations_total", "operation")
 	data.SendSumOfGauges(out, sm.tsdbLoadedBlocks, "prometheus_tsdb_blocks_loaded")
 	data.SendSumOfGaugesPerUser(out, sm.tsdbSymbolTableSize, "prometheus_tsdb_symbol_table_size_bytes")
 	data.SendSumOfCounters(out, sm.tsdbReloads, "prometheus_tsdb_reloads_total")

pkg/storage/tsdb/config.go

@@ -136,6 +136,8 @@ type TSDBConfig struct {
 	WALSegmentSizeBytes       int           `yaml:"wal_segment_size_bytes"`
 	FlushBlocksOnShutdown     bool          `yaml:"flush_blocks_on_shutdown"`
 	CloseIdleTSDBTimeout      time.Duration `yaml:"close_idle_tsdb_timeout"`
+	//The size of the in-memory queue used before flushing chunks to the disk.
+	HeadChunksWriteQueueSize int `yaml:"head_chunks_write_queue_size"`
 
 	// MaxTSDBOpeningConcurrencyOnStartup limits the number of concurrently opening TSDB's during startup.
 	MaxTSDBOpeningConcurrencyOnStartup int `yaml:"max_tsdb_opening_concurrency_on_startup"`