Method Init

tsdb/head.go:694–953

Init loads data from the write ahead log and prepares the head for writes. It should be called before using an appender so that it limits the ingested samples to the head min valid time.

(minValidTime int64)

// It should be called before using an appender so that it
// limits the ingested samples to the head min valid time.
func (h *Head) Init(minValidTime int64) error {
	h.minValidTime.Store(minValidTime)
	defer h.resetWLReplayResources()
	defer func() {
		h.postings.EnsureOrder(h.opts.WALReplayConcurrency)
	}()
	defer h.gc() // After loading the wal remove the obsolete data from the head.
	defer func() {
		// Loading of m-mapped chunks and snapshot can make the mint of the Head
		// to go below minValidTime.
		if h.MinTime() < h.minValidTime.Load() {
			h.minTime.Store(h.minValidTime.Load())
		}
	}()

	h.logger.Info("Replaying on-disk memory mappable chunks if any")
	start := time.Now()

	snapIdx, snapOffset := -1, 0
	refSeries := make(map[chunks.HeadSeriesRef]*memSeries)

	snapshotLoaded := false
	var chunkSnapshotLoadDuration time.Duration
	if h.opts.EnableMemorySnapshotOnShutdown {
		h.logger.Info("Chunk snapshot is enabled, replaying from the snapshot")
		// If there are any WAL files, there should be at least one WAL file with an index that is current or newer
		// than the snapshot index. If the WAL index is behind the snapshot index somehow, the snapshot is assumed
		// to be outdated.
		loadSnapshot := true
		if h.wal != nil {
			_, endAt, err := wlog.Segments(h.wal.Dir())
			if err != nil {
				return fmt.Errorf("finding WAL segments: %w", err)
			}

			_, idx, _, err := LastChunkSnapshot(h.opts.ChunkDirRoot)
			if err != nil && !errors.Is(err, record.ErrNotFound) {
				h.logger.Error("Could not find last snapshot", "err", err)
			}

			if err == nil && endAt < idx {
				loadSnapshot = false
				h.logger.Warn("Last WAL file is behind snapshot, removing snapshots")
				if err := DeleteChunkSnapshots(h.opts.ChunkDirRoot, math.MaxInt, math.MaxInt); err != nil {
					h.logger.Error("Error while deleting snapshot directories", "err", err)
				}
			}
		}
		if loadSnapshot {
			var err error
			snapIdx, snapOffset, refSeries, err = h.loadChunkSnapshot()
			if err == nil {
				snapshotLoaded = true
				chunkSnapshotLoadDuration = time.Since(start)
				h.logger.Info("Chunk snapshot loading time", "duration", chunkSnapshotLoadDuration.String())
			} else {
				snapIdx, snapOffset = -1, 0
				refSeries = make(map[chunks.HeadSeriesRef]*memSeries)