func()

in tsdb/wal.go [294:419]
103 lines of code
25 McCabe index (conditional complexity)

func (w *SegmentWAL) Truncate(mint int64, keep func(uint64) bool) error {
	// The last segment is always active.
	if len(w.files) < 2 {
		return nil
	}
	var candidates []*segmentFile

	// All files have to be traversed as there could be two segments for a block
	// with first block having times (10000, 20000) and SECOND one having (0, 10000).
	for _, sf := range w.files[:len(w.files)-1] {
		if sf.maxTime >= mint {
			break
		}
		// Past WAL files are closed. We have to reopen them for another read.
		f, err := w.openSegmentFile(sf.Name())
		if err != nil {
			return errors.Wrap(err, "open old WAL segment for read")
		}
		candidates = append(candidates, &segmentFile{
			File:      f,
			minSeries: sf.minSeries,
			maxTime:   sf.maxTime,
		})
	}
	if len(candidates) == 0 {
		return nil
	}

	r := newWALReader(candidates, w.logger)

	// Create a new tmp file.
	f, err := w.createSegmentFile(filepath.Join(w.dirFile.Name(), "compact.tmp"))
	if err != nil {
		return errors.Wrap(err, "create compaction segment")
	}
	defer func() {
		if err := os.RemoveAll(f.Name()); err != nil {
			level.Error(w.logger).Log("msg", "remove tmp file", "err", err.Error())
		}
	}()

	var (
		csf          = newSegmentFile(f)
		crc32        = newCRC32()
		decSeries    = []record.RefSeries{}
		activeSeries = []record.RefSeries{}
	)

	for r.next() {
		rt, flag, byt := r.at()

		if rt != WALEntrySeries {
			continue
		}
		decSeries = decSeries[:0]
		activeSeries = activeSeries[:0]

		err := r.decodeSeries(flag, byt, &decSeries)
		if err != nil {
			return errors.Wrap(err, "decode samples while truncating")
		}
		for _, s := range decSeries {
			if keep(s.Ref) {
				activeSeries = append(activeSeries, s)
			}
		}

		buf := w.getBuffer()
		flag = w.encodeSeries(buf, activeSeries)

		_, err = w.writeTo(csf, crc32, WALEntrySeries, flag, buf.Get())
		w.putBuffer(buf)

		if err != nil {
			return errors.Wrap(err, "write to compaction segment")
		}
	}
	if r.Err() != nil {
		return errors.Wrap(r.Err(), "read candidate WAL files")
	}

	off, err := csf.Seek(0, io.SeekCurrent)
	if err != nil {
		return err
	}
	if err := csf.Truncate(off); err != nil {
		return err
	}
	if err := csf.Sync(); err != nil {
		return nil
	}
	if err := csf.Close(); err != nil {
		return nil
	}

	_ = candidates[0].Close() // need close before remove on platform windows
	if err := fileutil.Replace(csf.Name(), candidates[0].Name()); err != nil {
		return errors.Wrap(err, "rename compaction segment")
	}
	for _, f := range candidates[1:] {
		f.Close() // need close before remove on platform windows
		if err := os.RemoveAll(f.Name()); err != nil {
			return errors.Wrap(err, "delete WAL segment file")
		}
	}
	if err := w.dirFile.Sync(); err != nil {
		return err
	}

	// The file object of csf still holds the name before rename. Recreate it so
	// subsequent truncations do not look at a non-existent file name.
	csf.File, err = w.openSegmentFile(candidates[0].Name())
	if err != nil {
		return err
	}
	// We don't need it to be open.
	if err := csf.Close(); err != nil {
		return err
	}

	w.mtx.Lock()
	w.files = append([]*segmentFile{csf}, w.files[len(candidates):]...)
	w.mtx.Unlock()

	return nil
}