Repository: prometheus/prometheus
File: tsdb/head.go (Go)

Code before:
	}
	// The lower two thirds of segments should contain mostly obsolete samples.
	// If we have less than two segments, it's not worth checkpointing yet.
	// With the default 2h blocks, this will keeping up to around 3h worth
	// of WAL segments.
	last = first + (last-first)*2/3
	if last <= first {
		return nil
	}

	keep := func(s record.RefSeries) bool {
		// Keep the record if the series exists in the head.
		if h.series.getByID(s.Ref) != nil {
			return true
		}

		// Keep the record if it was recently deleted.
		h.deletedMtx.Lock()
		keepUntil, ok := h.deleted[s.Ref]
		h.deletedMtx.Unlock()
		if ok {
			return keepUntil > last
		}

		// Keep the record if it is a duplicate of a series in the head.
		if !s.Labels.IsEmpty() && h.series.getByHash(s.Labels.Hash(), s.Labels) != nil {
			// If it is a duplicate, new samples will be written with the id of the series already in the head,
			// so keep this record only until the last current WAL segment.
			_, last, _ := wlog.Segments(h.wal.Dir())
			h.deletedMtx.Lock()
			h.deleted[s.Ref] = last
			h.deletedMtx.Unlock()
			return true
		}

		// TODO: Do we need to check if the series is a duplicate of a series in deleted?

		return false
	}
	h.metrics.checkpointCreationTotal.Inc()
	if _, err = wlog.Checkpoint(h.logger, h.wal, first, last, keep, mint); err != nil {
		h.metrics.checkpointCreationFail.Inc()
		var cerr *chunks.CorruptionErr
		if errors.As(err, &cerr) {
			h.metrics.walCorruptionsTotal.Inc()
		}
		return fmt.Errorf("create checkpoint: %w", err)
	}
	if err := h.wal.Truncate(last + 1); err != nil {
		// If truncating fails, we'll just try again at the next checkpoint.
		// Leftover segments will just be ignored in the future if there's a checkpoint

Review comment (suggestion):
I guess the `IsEmpty` is there for the case where we're checking metadata, but it isn't strictly necessary.

Assuming you keep it, perhaps more readable as `if s.Labels.IsEmpty() return false`.

Note to self: I expected there would be a `getByLabels` so you don't have to do the hash computation, but aparently not.

Code after:
	// Keep the record if the series has an expiry set.
	keepUntil, ok := h.getWALExpiry(id)
	return ok && keepUntil > last
}

// truncateWAL removes old data before mint from the WAL.
func (h *Head) truncateWAL(mint int64) error {
	h.chunkSnapshotMtx.Lock()
	defer h.chunkSnapshotMtx.Unlock()

	if h.wal == nil || mint <= h.lastWALTruncationTime.Load() {
		return nil
	}
	start := time.Now()
	h.lastWALTruncationTime.Store(mint)

	first, last, err := wlog.Segments(h.wal.Dir())
	if err != nil {
		return fmt.Errorf("get segment range: %w", err)
	}
	// Start a new segment, so low ingestion volume TSDB don't have more WAL than
	// needed.
	if _, err := h.wal.NextSegment(); err != nil {
		return fmt.Errorf("next segment: %w", err)
	}
	last-- // Never consider last segment for checkpoint.
	if last < 0 {
		return nil // no segments yet.
	}
	// The lower two thirds of segments should contain mostly obsolete samples.
	// If we have less than two segments, it's not worth checkpointing yet.
	// With the default 2h blocks, this will keeping up to around 3h worth
	// of WAL segments.
	last = first + (last-first)*2/3
	if last <= first {
		return nil
	}

	h.metrics.checkpointCreationTotal.Inc()
	if _, err = wlog.Checkpoint(h.logger, h.wal, first, last, h.keepSeriesInWALCheckpoint, mint); err != nil {
		h.metrics.checkpointCreationFail.Inc()
		var cerr *chunks.CorruptionErr
		if errors.As(err, &cerr) {
			h.metrics.walCorruptionsTotal.Inc()
		}
		return fmt.Errorf("create checkpoint: %w", err)
	}
	if err := h.wal.Truncate(last + 1); err != nil {
		// If truncating fails, we'll just try again at the next checkpoint.
		// Leftover segments will just be ignored in the future if there's a checkpoint
		// that supersedes them.