Merge pull request #10452 from prometheus/release-2.34

Merge Release 2.34 into main

CHANGELOG.md

@@ -1,7 +1,9 @@
-## 2.34.0-rc.0 / 2022-02-24
+## 2.34.0 / 2022-03-15
 
 * [CHANGE] UI: Classic UI removed. #10208
 * [CHANGE] Tracing: Migrate from Jaeger to OpenTelemetry based tracing. #9724, #10203, #10276
+* [ENHANCEMENT] TSDB: Disable the chunk write queue by default and allow configuration with the experimental flag `--storage.tsdb.head-chunks-write-queue-size`. #10425
+* [ENHANCEMENT] HTTP SD: Add a failure counter. #10372
 * [ENHANCEMENT] Azure SD: Set Prometheus User-Agent on requests. #10209
 * [ENHANCEMENT] Uyuni SD: Reduce the number of logins to Uyuni. #10072
 * [ENHANCEMENT] Scrape: Log when an invalid media type is encountered during a scrape. #10186
@@ -13,6 +15,8 @@
 * [BUGFIX] PromQL: Properly return an error from histogram_quantile when metrics have the same labelset. #10140 
 * [BUGFIX] UI: Fix bug that sets the range input to the resolution. #10227
 * [BUGFIX] TSDB: Fix a query panic when `memory-snapshot-on-shutdown` is enabled. #10348
+* [BUGFIX] Parser: Specify type in metadata parser errors. #10269
+* [BUGFIX] Scrape: Fix label limit changes not applying. #10370
 
 ## 2.33.5 / 2022-03-08
 

VERSION

@@ -1 +1 @@
-2.34.0-rc.0
+2.34.0

cmd/prometheus/main.go

@@ -305,6 +305,9 @@ func main() {
 	serverOnlyFlag(a, "storage.tsdb.wal-compression", "Compress the tsdb WAL.").
 		Hidden().Default("true").BoolVar(&cfg.tsdb.WALCompression)
 
+	serverOnlyFlag(a, "storage.tsdb.head-chunks-write-queue-size", "Size of the queue through which head chunks are written to the disk to be m-mapped, 0 disables the queue completely. Experimental.").
+		Default("0").IntVar(&cfg.tsdb.HeadChunksWriteQueueSize)
+
 	agentOnlyFlag(a, "storage.agent.path", "Base path for metrics storage.").
 		Default("data-agent/").StringVar(&cfg.agentStoragePath)
 
@@ -1484,6 +1487,7 @@ type tsdbOptions struct {
 	NoLockfile                     bool
 	AllowOverlappingBlocks         bool
 	WALCompression                 bool
+	HeadChunksWriteQueueSize       int
 	StripeSize                     int
 	MinBlockDuration               model.Duration
 	MaxBlockDuration               model.Duration
@@ -1501,6 +1505,7 @@ func (opts tsdbOptions) ToTSDBOptions() tsdb.Options {
 		NoLockfile:                     opts.NoLockfile,
 		AllowOverlappingBlocks:         opts.AllowOverlappingBlocks,
 		WALCompression:                 opts.WALCompression,
+		HeadChunksWriteQueueSize:       opts.HeadChunksWriteQueueSize,
 		StripeSize:                     opts.StripeSize,
 		MinBlockDuration:               int64(time.Duration(opts.MinBlockDuration) / time.Millisecond),
 		MaxBlockDuration:               int64(time.Duration(opts.MaxBlockDuration) / time.Millisecond),

go.mod

@@ -29,7 +29,7 @@ require (
 	github.com/golang/snappy v0.0.4
 	github.com/google/pprof v0.0.0-20220218203455-0368bd9e19a7
 	github.com/gophercloud/gophercloud v0.24.0
-	github.com/grafana/regexp v0.0.0-20220202152315-e74e38789280
+	github.com/grafana/regexp v0.0.0-20220304095617-2e8d9baf4ac2
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0
 	github.com/hashicorp/consul/api v1.12.0
 	github.com/hashicorp/go-hclog v0.12.2 // indirect

go.sum

@@ -737,8 +737,8 @@ github.com/gorilla/mux v1.7.3/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2z
 github.com/gorilla/websocket v0.0.0-20170926233335-4201258b820c/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
-github.com/grafana/regexp v0.0.0-20220202152315-e74e38789280 h1:MOND6wXrwVXEzmL2bZ+Jcbgycwt1LD5q6NQbqz/Nlic=
+github.com/grafana/regexp v0.0.0-20220304095617-2e8d9baf4ac2 h1:uirlL/j72L93RhV4+mkWhjv0cov2I0MIgPOG9rMDr1k=
-github.com/grafana/regexp v0.0.0-20220202152315-e74e38789280/go.mod h1:M5qHK+eWfAv8VR/265dIuEpL3fNfeC21tXXp9itM24A=
+github.com/grafana/regexp v0.0.0-20220304095617-2e8d9baf4ac2/go.mod h1:M5qHK+eWfAv8VR/265dIuEpL3fNfeC21tXXp9itM24A=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.1-0.20190118093823-f849b5445de4/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=

tsdb/chunks/head_chunks.go

@@ -69,7 +69,8 @@ const (
 	// DefaultWriteBufferSize is the default write buffer size.
 	DefaultWriteBufferSize = 4 * 1024 * 1024 // 4 MiB.
 	// DefaultWriteQueueSize is the default size of the in-memory queue used before flushing chunks to the disk.
-	DefaultWriteQueueSize = 1000
+	// A value of 0 completely disables this feature.
+	DefaultWriteQueueSize = 0
 )
 
 // ChunkDiskMapperRef represents the location of a head chunk on disk.
@@ -249,7 +250,10 @@ func NewChunkDiskMapper(reg prometheus.Registerer, dir string, pool chunkenc.Poo
 		crc32:           newCRC32(),
 		chunkBuffer:     newChunkBuffer(),
 	}
-	m.writeQueue = newChunkWriteQueue(reg, writeQueueSize, m.writeChunk)
+
+	if writeQueueSize > 0 {
+		m.writeQueue = newChunkWriteQueue(reg, writeQueueSize, m.writeChunk)
+	}
 
 	if m.pool == nil {
 		m.pool = chunkenc.NewPool()
@@ -375,18 +379,33 @@ func repairLastChunkFile(files map[int]string) (_ map[int]string, returnErr erro
 // WriteChunk writes the chunk to the disk.
 // The returned chunk ref is the reference from where the chunk encoding starts for the chunk.
 func (cdm *ChunkDiskMapper) WriteChunk(seriesRef HeadSeriesRef, mint, maxt int64, chk chunkenc.Chunk, callback func(err error)) (chkRef ChunkDiskMapperRef) {
-	var err error
+	// cdm.evtlPosMtx must be held to serialize the calls to cdm.evtlPos.getNextChunkRef() and the writing of the chunk (either with or without queue).
-	defer func() {
-		if err != nil && callback != nil {
-			callback(err)
-		}
-	}()
-
-	// cdm.evtlPosMtx must be held to serialize the calls to .getNextChunkRef() and .addJob().
 	cdm.evtlPosMtx.Lock()
 	defer cdm.evtlPosMtx.Unlock()
-
 	ref, cutFile := cdm.evtlPos.getNextChunkRef(chk)
+
+	if cdm.writeQueue != nil {
+		return cdm.writeChunkViaQueue(ref, cutFile, seriesRef, mint, maxt, chk, callback)
+	}
+
+	err := cdm.writeChunk(seriesRef, mint, maxt, chk, ref, cutFile)
+	if callback != nil {
+		callback(err)
+	}
+
+	return ref
+}
+
+func (cdm *ChunkDiskMapper) writeChunkViaQueue(ref ChunkDiskMapperRef, cutFile bool, seriesRef HeadSeriesRef, mint, maxt int64, chk chunkenc.Chunk, callback func(err error)) (chkRef ChunkDiskMapperRef) {
+	var err error
+	if callback != nil {
+		defer func() {
+			if err != nil {
+				callback(err)
+			}
+		}()
+	}
+
 	err = cdm.writeQueue.addJob(chunkWriteJob{
 		cutFile:   cutFile,
 		seriesRef: seriesRef,
@@ -473,6 +492,10 @@ func (cdm *ChunkDiskMapper) CutNewFile() {
 }
 
 func (cdm *ChunkDiskMapper) IsQueueEmpty() bool {
+	if cdm.writeQueue == nil {
+		return true
+	}
+
 	return cdm.writeQueue.queueIsEmpty()
 }
 
@@ -602,9 +625,11 @@ func (cdm *ChunkDiskMapper) Chunk(ref ChunkDiskMapperRef) (chunkenc.Chunk, error
 		return nil, ErrChunkDiskMapperClosed
 	}
 
-	chunk := cdm.writeQueue.get(ref)
+	if cdm.writeQueue != nil {
-	if chunk != nil {
+		chunk := cdm.writeQueue.get(ref)
-		return chunk, nil
+		if chunk != nil {
+			return chunk, nil
+		}
 	}
 
 	sgmIndex, chkStart := ref.Unpack()
@@ -962,7 +987,9 @@ func (cdm *ChunkDiskMapper) Close() error {
 	cdm.evtlPosMtx.Lock()
 	defer cdm.evtlPosMtx.Unlock()
 
-	cdm.writeQueue.stop()
+	if cdm.writeQueue != nil {
+		cdm.writeQueue.stop()
+	}
 
 	// 'WriteChunk' locks writePathMtx first and then readPathMtx for cutting head chunk file.
 	// The lock order should not be reversed here else it can cause deadlocks.

tsdb/chunks/head_chunks_test.go

@@ -27,6 +27,22 @@ import (
 	"github.com/prometheus/prometheus/tsdb/chunkenc"
 )
 
+var writeQueueSize int
+
+func TestMain(m *testing.M) {
+	// Run all tests with the chunk write queue disabled.
+	writeQueueSize = 0
+	exitVal := m.Run()
+	if exitVal != 0 {
+		os.Exit(exitVal)
+	}
+
+	// Re-run all tests with the chunk write queue size of 1e6.
+	writeQueueSize = 1000000
+	exitVal = m.Run()
+	os.Exit(exitVal)
+}
+
 func TestChunkDiskMapper_WriteChunk_Chunk_IterateChunks(t *testing.T) {
 	hrw := createChunkDiskMapper(t, "")
 	defer func() {
@@ -453,7 +469,7 @@ func createChunkDiskMapper(t *testing.T, dir string) *ChunkDiskMapper {
 		dir = t.TempDir()
 	}
 
-	hrw, err := NewChunkDiskMapper(nil, dir, chunkenc.NewPool(), DefaultWriteBufferSize, DefaultWriteQueueSize)
+	hrw, err := NewChunkDiskMapper(nil, dir, chunkenc.NewPool(), DefaultWriteBufferSize, writeQueueSize)
 	require.NoError(t, err)
 	require.False(t, hrw.fileMaxtSet)
 	require.NoError(t, hrw.IterateAllChunks(func(_ HeadSeriesRef, _ ChunkDiskMapperRef, _, _ int64, _ uint16) error { return nil }))