Function TestHead_mmapHeadChunks

tsdb/head_test.go:8756–9250

(t *testing.T)

}

func TestHead_mmapHeadChunks(t *testing.T) {
	// Interval such that one chunk's worth of samples (DefaultSamplesPerChunk) spans DefaultBlockDuration/4.
	interval := DefaultBlockDuration / (4 * DefaultSamplesPerChunk)
	// Enough samples to cross the nextAt boundary multiple times, producing 3 head chunks per series.
	const chunkCutIterations = 2*DefaultSamplesPerChunk + 10

	t.Run("basic", func(t *testing.T) {
		h, _ := newTestHead(t, DefaultBlockDuration, compression.None, false)
		require.NoError(t, h.Init(0))

		countReady := func() int {
			n := 0
			for i := range h.series.size {
				h.series.locks[i].RLock()
				for _, s := range h.series.series[i] {
					if s.headChunkCount.Load() >= 2 {
						n++
					}
				}
				h.series.locks[i].RUnlock()
			}
			return n
		}

		getCount := func(lbls labels.Labels) uint32 {
			s := h.series.getByHash(lbls.Hash(), lbls)
			require.NotNil(t, s, "series %s not found", lbls)
			return s.headChunkCount.Load()
		}

		lblsA := labels.FromStrings("__name__", "seriesA")
		lblsB := labels.FromStrings("__name__", "seriesB")
		lblsC := labels.FromStrings("__name__", "seriesC")

		ts := int64(0)

		// First chunk creation should set headChunkCount to 1.
		app := h.Appender(t.Context())
		_, err := app.Append(0, lblsA, ts, 1.0)
		require.NoError(t, err)
		require.NoError(t, app.Commit())
		ts += interval

		require.Equal(t, uint32(1), getCount(lblsA), "first chunk should set headChunkCount to 1")
		require.Equal(t, 0, countReady(), "series with only a first chunk should not be ready")

		// Appending enough samples to trigger chunk cuts should update headChunkCount.
		var refB, refC storage.SeriesRef
		app = h.Appender(t.Context())
		for range chunkCutIterations {
			var err error
			refB, err = app.Append(refB, lblsB, ts, float64(ts))
			require.NoError(t, err)
			refC, err = app.Append(refC, lblsC, ts, float64(ts))
			require.NoError(t, err)
			ts += interval
		}
		require.NoError(t, app.Commit())