func (w *SegmentWAL) Sync() error {
	var head *segmentFile
	var err error
	func() {
		w.mtx.Lock()
		defer w.mtx.Unlock()
		if err = w.flush(); err != nil {
			return
		}
		head = w.head()
	}()
	if err != nil {
		return errors.Wrap(err, "flush buffer")
	}
	if head != nil {
		start := time.Now()
		err := fileutil.Fdatasync(head.File)
		w.metrics.fsyncDuration.Observe(time.Since(start).Seconds())
		return err
	}
	return nil
}
Sync flushes the changes to disk.
package tsdb
import (
	"bufio"
	"encoding/binary"
	"fmt"
	"hash"
	"hash/crc32"
	"io"
	"math"
	"os"
	"path/filepath"
	"sort"
	"sync"
	"time"
	"github.com/go-kit/kit/log"
	"github.com/go-kit/kit/log/level"
	"github.com/pkg/errors"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/prometheus/tsdb/encoding"
	"github.com/prometheus/prometheus/tsdb/fileutil"
	"github.com/prometheus/prometheus/tsdb/labels"
	"github.com/prometheus/prometheus/tsdb/record"
	"github.com/prometheus/prometheus/tsdb/tombstones"
	"github.com/prometheus/prometheus/tsdb/wal"
)
type WALEntryType uint8
const (
	WALMagic = uint32(0x43AF00EF)
	WALFormatDefault = byte(1)
)
const (
	WALEntrySymbols WALEntryType = 1
	WALEntrySeries  WALEntryType = 2
	WALEntrySamples WALEntryType = 3
	WALEntryDeletes WALEntryType = 4
)
type walMetrics struct {
	fsyncDuration prometheus.Summary
	corruptions   prometheus.Counter
}
func newWalMetrics(wal *SegmentWAL, r prometheus.Registerer) *walMetrics {
	m := &walMetrics{}
	m.fsyncDuration = prometheus.NewSummary(prometheus.SummaryOpts{
		Name:       "prometheus_tsdb_wal_fsync_duration_seconds",
		Help:       "Duration of WAL fsync.",
		Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
	})
	m.corruptions = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "prometheus_tsdb_wal_corruptions_total",
		Help: "Total number of WAL corruptions.",
	})
	if r != nil {
		r.MustRegister(
			m.fsyncDuration,
			m.corruptions,
		)
	}
	return m
}
type WAL interface {
	Reader() WALReader
	LogSeries([]record.RefSeries) error
	LogSamples([]record.RefSample) error
	LogDeletes([]tombstones.Stone) error
	Truncate(mint int64, keep func(uint64) bool) error
	Close() error
}
type WALReader interface {
	Read(
		seriesf func([]record.RefSeries),
		samplesf func([]record.RefSample),
		deletesf func([]tombstones.Stone),
	) error
}
type segmentFile struct {
	*os.File
	maxTime   int64   
	minSeries uint64  
}
func newSegmentFile(f *os.File) *segmentFile {
	return &segmentFile{
		File:      f,
		maxTime:   math.MinInt64,
		minSeries: math.MaxUint64,
	}
}
const (
	walSegmentSizeBytes = 256 * 1024 * 1024  
)
var castagnoliTable *crc32.Table
func init() {
	castagnoliTable = crc32.MakeTable(crc32.Castagnoli)
}
func newCRC32() hash.Hash32 {
	return crc32.New(castagnoliTable)
}
type SegmentWAL struct {
	mtx     sync.Mutex
	metrics *walMetrics
	dirFile *os.File
	files   []*segmentFile
	logger        log.Logger
	flushInterval time.Duration
	segmentSize   int64
	crc32 hash.Hash32
	cur   *bufio.Writer
	curN  int64
	stopc   chan struct{}
	donec   chan struct{}
	actorc  chan func() error  
	buffers sync.Pool
}
func OpenSegmentWAL(dir string, logger log.Logger, flushInterval time.Duration, r prometheus.Registerer) (*SegmentWAL, error) {
	if err := os.MkdirAll(dir, 0777); err != nil {
		return nil, err
	}
	df, err := fileutil.OpenDir(dir)
	if err != nil {
		return nil, err
	}
	if logger == nil {
		logger = log.NewNopLogger()
	}
	w := &SegmentWAL{
		dirFile:       df,
		logger:        logger,
		flushInterval: flushInterval,
		donec:         make(chan struct{}),
		stopc:         make(chan struct{}),
		actorc:        make(chan func() error, 1),
		segmentSize:   walSegmentSizeBytes,
		crc32:         newCRC32(),
	}
	w.metrics = newWalMetrics(w, r)
	fns, err := sequenceFiles(w.dirFile.Name())
	if err != nil {
		return nil, err
	}
	for i, fn := range fns {
		f, err := w.openSegmentFile(fn)
		if err == nil {
			w.files = append(w.files, newSegmentFile(f))
			continue
		}
		level.Warn(logger).Log("msg", "invalid segment file detected, truncating WAL", "err", err, "file", fn)
		for _, fn := range fns[i:] {
			if err := os.Remove(fn); err != nil {
				return w, errors.Wrap(err, "removing segment failed")
			}
		}
		break
	}
	go w.run(flushInterval)
	return w, nil
}
type repairingWALReader struct {
	wal *SegmentWAL
	r   WALReader
}
func (r *repairingWALReader) Read(
	seriesf func([]record.RefSeries),
	samplesf func([]record.RefSample),
	deletesf func([]tombstones.Stone),
) error {
	err := r.r.Read(seriesf, samplesf, deletesf)
	if err == nil {
		return nil
	}
	cerr, ok := errors.Cause(err).(walCorruptionErr)
	if !ok {
		return err
	}
	r.wal.metrics.corruptions.Inc()
	return r.wal.truncate(cerr.err, cerr.file, cerr.lastOffset)
}
func (w *SegmentWAL) truncate(err error, file int, lastOffset int64) error {
	level.Error(w.logger).Log("msg", "WAL corruption detected; truncating",
		"err", err, "file", w.files[file].Name(), "pos", lastOffset)
	for _, f := range w.files[file+1:] {
		if err := f.Close(); err != nil {
			return err
		}
		if err := os.Remove(f.Name()); err != nil {
			return err
		}
	}
	w.mtx.Lock()
	defer w.mtx.Unlock()
	w.files = w.files[:file+1]
	_, err = w.files[file].Seek(lastOffset, io.SeekStart)
	return err
}
func (w *SegmentWAL) Reader() WALReader {
	return &repairingWALReader{
		wal: w,
		r:   newWALReader(w.files, w.logger),
	}
}
func (w *SegmentWAL) getBuffer() *encoding.Encbuf {
	b := w.buffers.Get()
	if b == nil {
		return &encoding.Encbuf{B: make([]byte, 0, 64*1024)}
	}
	return b.(*encoding.Encbuf)
}
func (w *SegmentWAL) putBuffer(b *encoding.Encbuf) {
	b.Reset()
	w.buffers.Put(b)
}
func (w *SegmentWAL) Truncate(mint int64, keep func(uint64) bool) error {
	if len(w.files) < 2 {
		return nil
	}
	var candidates []*segmentFile
	for _, sf := range w.files[:len(w.files)-1] {
		if sf.maxTime >= mint {
			break
		}
		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)
	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()  
	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()  
		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
	}
	csf.File, err = w.openSegmentFile(candidates[0].Name())
	if err != nil {
		return err
	}
	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
}
func (w *SegmentWAL) LogSeries(series []record.RefSeries) error {
	buf := w.getBuffer()
	flag := w.encodeSeries(buf, series)
	w.mtx.Lock()
	defer w.mtx.Unlock()
	err := w.write(WALEntrySeries, flag, buf.Get())
	w.putBuffer(buf)
	if err != nil {
		return errors.Wrap(err, "log series")
	}
	tf := w.head()
	for _, s := range series {
		if tf.minSeries > s.Ref {
			tf.minSeries = s.Ref
		}
	}
	return nil
}
func (w *SegmentWAL) LogSamples(samples []record.RefSample) error {
	buf := w.getBuffer()
	flag := w.encodeSamples(buf, samples)
	w.mtx.Lock()
	defer w.mtx.Unlock()
	err := w.write(WALEntrySamples, flag, buf.Get())
	w.putBuffer(buf)
	if err != nil {
		return errors.Wrap(err, "log series")
	}
	tf := w.head()
	for _, s := range samples {
		if tf.maxTime < s.T {
			tf.maxTime = s.T
		}
	}
	return nil
}
func (w *SegmentWAL) LogDeletes(stones []tombstones.Stone) error {
	buf := w.getBuffer()
	flag := w.encodeDeletes(buf, stones)
	w.mtx.Lock()
	defer w.mtx.Unlock()
	err := w.write(WALEntryDeletes, flag, buf.Get())
	w.putBuffer(buf)
	if err != nil {
		return errors.Wrap(err, "log series")
	}
	tf := w.head()
	for _, s := range stones {
		for _, iv := range s.Intervals {
			if tf.maxTime < iv.Maxt {
				tf.maxTime = iv.Maxt
			}
		}
	}
	return nil
}
func (w *SegmentWAL) openSegmentFile(name string) (*os.File, error) {
	f, err := os.OpenFile(name, os.O_RDWR, 0666)
	if err != nil {
		return nil, err
	}
	metab := make([]byte, 8)
	hasError := true
	defer func() {
		if hasError {
			f.Close()
		}
	}()
	if n, err := f.Read(metab); err != nil {
		return nil, errors.Wrapf(err, "validate meta %q", f.Name())
	} else if n != 8 {
		return nil, errors.Errorf("invalid header size %d in %q", n, f.Name())
	}
	if m := binary.BigEndian.Uint32(metab[:4]); m != WALMagic {
		return nil, errors.Errorf("invalid magic header %x in %q", m, f.Name())
	}
	if metab[4] != WALFormatDefault {
		return nil, errors.Errorf("unknown WAL segment format %d in %q", metab[4], f.Name())
	}
	hasError = false
	return f, nil
}
func (w *SegmentWAL) createSegmentFile(name string) (*os.File, error) {
	f, err := os.Create(name)
	if err != nil {
		return nil, err
	}
	if err = fileutil.Preallocate(f, w.segmentSize, true); err != nil {
		return nil, err
	}
	metab := make([]byte, 8)
	binary.BigEndian.PutUint32(metab[:4], WALMagic)
	metab[4] = WALFormatDefault
	if _, err := f.Write(metab); err != nil {
		return nil, err
	}
	return f, err
}
func (w *SegmentWAL) cut() error {
	if hf := w.head(); hf != nil {
		if err := w.flush(); err != nil {
			return err
		}
		go func() {
			w.actorc <- func() error {
				off, err := hf.Seek(0, io.SeekCurrent)
				if err != nil {
					return errors.Wrapf(err, "finish old segment %s", hf.Name())
				}
				if err := hf.Truncate(off); err != nil {
					return errors.Wrapf(err, "finish old segment %s", hf.Name())
				}
				if err := hf.Sync(); err != nil {
					return errors.Wrapf(err, "finish old segment %s", hf.Name())
				}
				if err := hf.Close(); err != nil {
					return errors.Wrapf(err, "finish old segment %s", hf.Name())
				}
				return nil
			}
		}()
	}
	p, _, err := nextSequenceFile(w.dirFile.Name())
	if err != nil {
		return err
	}
	f, err := w.createSegmentFile(p)
	if err != nil {
		return err
	}
	go func() {
		w.actorc <- func() error {
			return errors.Wrap(w.dirFile.Sync(), "sync WAL directory")
		}
	}()
	w.files = append(w.files, newSegmentFile(f))
	w.cur = bufio.NewWriterSize(f, 8*1024*1024)
	w.curN = 8
	return nil
}
func (w *SegmentWAL) head() *segmentFile {
	if len(w.files) == 0 {
		return nil
	}
	return w.files[len(w.files)-1]
}