wal

Documentation

Overview

Package wal implements a segmented write-ahead log for crash-safe persistence of time-series data.

WAL design informed by Prometheus tsdb/wal. See /NOTICE.md.

Index

• Variables
• func EncodeRecord(dst []byte, typ RecordType, payload []byte) []byte
• func EncodeSamplesRecord(dst []byte, samples []RefSample) []byte
• func EncodeSeriesRecord(dst []byte, rec SeriesRecord) []byte
• func RecordSize(payloadLen int) int
• type Options
• type Reader
  • func NewReader(dir string) (*Reader, error)
  • func (r *Reader) Close() error
  • func (r *Reader) Err() error
  • func (r *Reader) Next() bool
  • func (r *Reader) Record() Record
• type Record
• type RecordType
  • func DecodeRecord(b []byte) (typ RecordType, payload []byte, consumed int, err error)
• type RefSample
  • func DecodeSamplesRecord(data []byte) ([]RefSample, error)
• type SeriesRecord
  • func DecodeSeriesRecord(data []byte) (SeriesRecord, error)
• type WAL
  • func Open(dir string, opts Options) (*WAL, error)
  • func (w *WAL) Close() error
  • func (w *WAL) LastSegment() int
  • func (w *WAL) LastSyncDuration() float64
  • func (w *WAL) Log(typ RecordType, payload []byte) error
  • func (w *WAL) LogSamples(samples []RefSample) error
  • func (w *WAL) LogSeries(recs []SeriesRecord) error
  • func (w *WAL) Replay() (*Reader, error)
  • func (w *WAL) Sync() error
  • func (w *WAL) Truncate(below int) error

Variables

var (
	ErrInvalidRecord = errors.New("wal: invalid record")
	ErrCorruptRecord = errors.New("wal: corrupt record (CRC mismatch)")
)

var ErrShortPayload = errors.New("wal: payload too short")

Functions

func EncodeRecord

func EncodeRecord(dst []byte, typ RecordType, payload []byte) []byte

EncodeRecord appends a framed record (type + len + payload + crc32c) to dst and returns the extended slice.

func EncodeSamplesRecord

func EncodeSamplesRecord(dst []byte, samples []RefSample) []byte

EncodeSamplesRecord appends the encoded samples record to dst.

nsamples(4) | for each: ref(8) t(8) v(8)

func EncodeSeriesRecord

func EncodeSeriesRecord(dst []byte, rec SeriesRecord) []byte

EncodeSeriesRecord appends the encoded series record to dst.

ref(8) | nlabels(4) | for each: namelen(2) name valuelen(2) value

func RecordSize

func RecordSize(payloadLen int) int

RecordSize returns the total on-disk size of a record with the given payload length.

Types

type Options

type Options struct {
	// SegmentMaxSize is the maximum size of a single segment file in bytes.
	// A new segment is created when the current one would exceed this.
	// Default: 128 MiB.
	SegmentMaxSize int

	// SyncInterval controls background fsync frequency.
	// Default (zero): 1s. Negative: sync on every Log call.
	SyncInterval time.Duration
}

Options configures WAL behavior.

type Reader

type Reader struct {
	// contains filtered or unexported fields
}

Reader scans WAL segments sequentially, validating CRC on each record. It follows the iterator pattern: Next() advances, Record() returns the current record, Err() returns any error after Next() returns false.

func NewReader

func NewReader(dir string) (*Reader, error)

NewReader creates a Reader over all segments in dir.

func (*Reader) Close

func (r *Reader) Close() error

Close releases any open file handle.

func (*Reader) Err

func (r *Reader) Err() error

Err returns the error encountered during reading, if any. A nil error after Next() returns false means all records were read cleanly.

func (*Reader) Next

func (r *Reader) Next() bool

Next advances to the next record. Returns false when no more records are available or an error is encountered. After Next returns false, call Err() to distinguish clean EOF from corruption.

func (*Reader) Record

func (r *Reader) Record() Record

Record returns the most recently read record.

type Record

type Record struct {
	Type RecordType
	Data []byte // raw payload; decode with DecodeSeriesRecord / DecodeSamplesRecord
}

Record is a decoded WAL record returned by the Reader.

type RecordType

type RecordType byte

RecordType identifies the kind of record stored in the WAL.

const (
	RecordSeries  RecordType = 1
	RecordSamples RecordType = 2
)

func DecodeRecord

func DecodeRecord(b []byte) (typ RecordType, payload []byte, consumed int, err error)

DecodeRecord parses a framed record from b. It returns the record type, the payload slice (a sub-slice of b), the total number of bytes consumed, and any error. On success, consumed == RecordSize(len(payload)).

type RefSample

type RefSample struct {
	Ref uint64
	T   int64
	V   float64
}

RefSample is a single sample keyed by series ref.

func DecodeSamplesRecord

func DecodeSamplesRecord(data []byte) ([]RefSample, error)

DecodeSamplesRecord decodes a samples payload.

type SeriesRecord

type SeriesRecord struct {
	Ref    uint64
	Labels []labels.Label
}

SeriesRecord is a WAL record that registers a new series.

func DecodeSeriesRecord

func DecodeSeriesRecord(data []byte) (SeriesRecord, error)

DecodeSeriesRecord decodes a series payload. The returned Labels hold copies of the strings (not sub-slices of data).

type WAL

type WAL struct {
	// contains filtered or unexported fields
}

WAL is a segmented write-ahead log.

func Open

func Open(dir string, opts Options) (*WAL, error)

Open opens or creates a WAL in dir. If segments already exist, it runs recovery (truncating at the first corrupt record) before returning.

func (*WAL) Close

func (w *WAL) Close() error

Close stops the background syncer, fsyncs, and closes the active segment.

func (*WAL) LastSegment

func (w *WAL) LastSegment() int

LastSegment returns the index of the current active segment.

func (*WAL) LastSyncDuration

func (w *WAL) LastSyncDuration() float64

LastSyncDuration returns the duration of the most recent fsync in seconds.

func (*WAL) Log

func (w *WAL) Log(typ RecordType, payload []byte) error

Log writes a framed record to the WAL. The payload is wrapped with the record envelope (type + length + CRC).

func (*WAL) LogSamples

func (w *WAL) LogSamples(samples []RefSample) error

LogSamples encodes and writes a samples record.

func (*WAL) LogSeries

func (w *WAL) LogSeries(recs []SeriesRecord) error

LogSeries encodes and writes a series record.

func (*WAL) Replay

func (w *WAL) Replay() (*Reader, error)

Replay returns a Reader over all WAL segments. The caller must Close the reader when done.

func (*WAL) Sync

func (w *WAL) Sync() error

Sync forces an fsync of the current segment.

func (*WAL) Truncate

func (w *WAL) Truncate(below int) error

Truncate deletes all segments with index less than below.