wal

Documentation

Index

• Constants
• func DeleteCheckpoints(dir string, maxIndex int) error
• func LastCheckpoint(dir string) (string, int, error)
• func NewLiveReaderMetrics(reg prometheus.Registerer) *liveReaderMetrics
• func NewSegmentBufReader(segs ...*Segment) *segmentBufReader
• func NewSegmentsRangeReader(sr ...SegmentRange) (io.ReadCloser, error)
• func NewSegmentsReader(dir string) (io.ReadCloser, error)
• func SegmentName(dir string, i int) string
• type CheckpointStats
  • func Checkpoint(w *WAL, from, to int, keep func(id uint64) bool, mint int64) (*CheckpointStats, error)
• type CorruptionErr
  • func (e *CorruptionErr) Error() string
• type LiveReader
  • func NewLiveReader(logger log.Logger, metrics *liveReaderMetrics, r io.Reader) *LiveReader
  • func (r *LiveReader) Err() error
  • func (r *LiveReader) Next() bool
  • func (r *LiveReader) Offset() int64
  • func (r *LiveReader) Record() []byte
• type Reader
  • func NewReader(r io.Reader) *Reader
  • func (r *Reader) Err() error
  • func (r *Reader) Next() bool
  • func (r *Reader) Offset() int64
  • func (r *Reader) Record() []byte
  • func (r *Reader) Segment() int
• type Segment
  • func CreateSegment(dir string, k int) (*Segment, error)
  • func OpenReadSegment(fn string) (*Segment, error)
  • func OpenWriteSegment(logger log.Logger, dir string, k int) (*Segment, error)
  • func (s *Segment) Dir() string
  • func (s *Segment) Index() int
• type SegmentRange
• type WAL
  • func New(logger log.Logger, reg prometheus.Registerer, dir string, compress bool) (*WAL, error)
  • func NewSize(logger log.Logger, reg prometheus.Registerer, dir string, segmentSize int, ...) (*WAL, error)
  • func Open(logger log.Logger, reg prometheus.Registerer, dir string) (*WAL, error)
  • func (w *WAL) Close() (err error)
  • func (w *WAL) CompressionEnabled() bool
  • func (w *WAL) Dir() string
  • func (w *WAL) Log(recs ...[]byte) error
  • func (w *WAL) NextSegment() error
  • func (w *WAL) Repair(origErr error) error
  • func (w *WAL) Segments() (first, last int, err error)
  • func (w *WAL) Truncate(i int) (err error)
• type Watcher
  • func NewWatcher(reg prometheus.Registerer, metrics *WatcherMetrics, logger log.Logger, ...) *Watcher
  • func (w *Watcher) Run() error
  • func (w *Watcher) Start()
  • func (w *Watcher) Stop()
• type WatcherMetrics
  • func NewWatcherMetrics(reg prometheus.Registerer) *WatcherMetrics
• type WriteTo

Constants

const (
	DefaultSegmentSize = 128 * 1024 * 1024 // 128 MB

)

Functions

func DeleteCheckpoints

func DeleteCheckpoints(dir string, maxIndex int) error

DeleteCheckpoints deletes all checkpoints in a directory below a given index.

func LastCheckpoint

func LastCheckpoint(dir string) (string, int, error)

LastCheckpoint returns the directory name and index of the most recent checkpoint. If dir does not contain any checkpoints, ErrNotFound is returned.

func NewLiveReaderMetrics

func NewLiveReaderMetrics(reg prometheus.Registerer) *liveReaderMetrics

NewLiveReaderMetrics instantiates, registers and returns metrics to be injected at LiveReader instantiation.

func NewSegmentBufReader

func NewSegmentBufReader(segs ...*Segment) *segmentBufReader

func NewSegmentsRangeReader

func NewSegmentsRangeReader(sr ...SegmentRange) (io.ReadCloser, error)

NewSegmentsRangeReader returns a new reader over the given WAL segment ranges. If first or last are -1, the range is open on the respective end.

func NewSegmentsReader

func NewSegmentsReader(dir string) (io.ReadCloser, error)

NewSegmentsReader returns a new reader over all segments in the directory.

func SegmentName

func SegmentName(dir string, i int) string

SegmentName builds a segment name for the directory.

Types

type CheckpointStats

type CheckpointStats struct {
	DroppedSeries     int
	DroppedSamples    int
	DroppedTombstones int
	TotalSeries       int // Processed series including dropped ones.
	TotalSamples      int // Processed samples including dropped ones.
	TotalTombstones   int // Processed tombstones including dropped ones.
}

CheckpointStats returns stats about a created checkpoint.

func Checkpoint

func Checkpoint(w *WAL, from, to int, keep func(id uint64) bool, mint int64) (*CheckpointStats, error)

Checkpoint creates a compacted checkpoint of segments in range [first, last] in the given WAL. It includes the most recent checkpoint if it exists. All series not satisfying keep and samples below mint are dropped.

The checkpoint is stored in a directory named checkpoint.N in the same segmented format as the original WAL itself. This makes it easy to read it through the WAL package and concatenate it with the original WAL.

type CorruptionErr

type CorruptionErr struct {
	Dir     string
	Segment int
	Offset  int64
	Err     error
}

CorruptionErr is an error that's returned when corruption is encountered.

func (*CorruptionErr) Error

func (e *CorruptionErr) Error() string

type LiveReader

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

LiveReader reads WAL records from an io.Reader. It allows reading of WALs that are still in the process of being written, and returns records as soon as they can be read.

func NewLiveReader

func NewLiveReader(logger log.Logger, metrics *liveReaderMetrics, r io.Reader) *LiveReader

NewLiveReader returns a new live reader.

func (*LiveReader) Err

func (r *LiveReader) Err() error

Err returns any errors encountered reading the WAL. io.EOFs are not terminal and Next can be tried again. Non-EOFs are terminal, and the reader should not be used again. It is up to the user to decide when to stop trying should io.EOF be returned.

func (*LiveReader) Next

func (r *LiveReader) Next() bool

Next returns true if Record() will contain a full record. If Next returns false, you should always checked the contents of Error(). Return false guarantees there are no more records if the segment is closed and not corrupt, otherwise if Err() == io.EOF you should try again when more data has been written.

func (*LiveReader) Offset

func (r *LiveReader) Offset() int64

Offset returns the number of bytes consumed from this segment.

func (*LiveReader) Record

func (r *LiveReader) Record() []byte

Record returns the current record. The returned byte slice is only valid until the next call to Next.

type Reader

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

Reader reads WAL records from an io.Reader.

func NewReader

func NewReader(r io.Reader) *Reader

NewReader returns a new reader.

func (*Reader) Err

func (r *Reader) Err() error

Err returns the last encountered error wrapped in a corruption error. If the reader does not allow to infer a segment index and offset, a total offset in the reader stream will be provided.

func (*Reader) Next

func (r *Reader) Next() bool

Next advances the reader to the next records and returns true if it exists. It must not be called again after it returned false.

func (*Reader) Offset

func (r *Reader) Offset() int64

Offset returns the current position of the segment being read.

func (*Reader) Record

func (r *Reader) Record() []byte

Record returns the current record. The returned byte slice is only valid until the next call to Next.

func (*Reader) Segment

func (r *Reader) Segment() int

Segment returns the current segment being read.

type Segment

type Segment struct {
	*os.File
	// contains filtered or unexported fields
}

Segment represents a segment file.

func CreateSegment

func CreateSegment(dir string, k int) (*Segment, error)

CreateSegment creates a new segment k in dir.

func OpenReadSegment

func OpenReadSegment(fn string) (*Segment, error)

OpenReadSegment opens the segment with the given filename.

func OpenWriteSegment

func OpenWriteSegment(logger log.Logger, dir string, k int) (*Segment, error)

OpenWriteSegment opens segment k in dir. The returned segment is ready for new appends.

func (*Segment) Dir

func (s *Segment) Dir() string

Dir returns the directory of the segment.

func (*Segment) Index

func (s *Segment) Index() int

Index returns the index of the segment.

type SegmentRange

type SegmentRange struct {
	Dir         string
	First, Last int
}

SegmentRange groups segments by the directory and the first and last index it includes.

type WAL

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

WAL is a write ahead log that stores records in segment files. It must be read from start to end once before logging new data. If an error occurs during read, the repair procedure must be called before it's safe to do further writes.

Segments are written to in pages of 32KB, with records possibly split across page boundaries. Records are never split across segments to allow full segments to be safely truncated. It also ensures that torn writes never corrupt records beyond the most recent segment.

func New

func New(logger log.Logger, reg prometheus.Registerer, dir string, compress bool) (*WAL, error)

New returns a new WAL over the given directory.

func NewSize

func NewSize(logger log.Logger, reg prometheus.Registerer, dir string, segmentSize int, compress bool) (*WAL, error)

NewSize returns a new WAL over the given directory. New segments are created with the specified size.

func Open

func Open(logger log.Logger, reg prometheus.Registerer, dir string) (*WAL, error)

Open an existing WAL.

func (*WAL) Close

func (w *WAL) Close() (err error)

Close flushes all writes and closes active segment.

func (*WAL) CompressionEnabled

func (w *WAL) CompressionEnabled() bool

CompressionEnabled returns if compression is enabled on this WAL.

func (*WAL) Dir

func (w *WAL) Dir() string

Dir returns the directory of the WAL.

func (*WAL) Log

func (w *WAL) Log(recs ...[]byte) error

Log writes the records into the log. Multiple records can be passed at once to reduce writes and increase throughput.

func (*WAL) NextSegment

func (w *WAL) NextSegment() error

NextSegment creates the next segment and closes the previous one.

func (*WAL) Repair

func (w *WAL) Repair(origErr error) error

Repair attempts to repair the WAL based on the error. It discards all data after the corruption.

func (*WAL) Segments

func (w *WAL) Segments() (first, last int, err error)

Segments returns the range [first, n] of currently existing segments. If no segments are found, first and n are -1.

func (*WAL) Truncate

func (w *WAL) Truncate(i int) (err error)

Truncate drops all segments before i.

type Watcher

type Watcher struct {
	StartTime int64

	// For testing, stop when we hit this segment.
	MaxSegment int
	// contains filtered or unexported fields
}

Watcher watches the TSDB WAL for a given WriteTo.

func NewWatcher

func NewWatcher(reg prometheus.Registerer, metrics *WatcherMetrics, logger log.Logger, name string, writer WriteTo, walDir string) *Watcher

NewWatcher creates a new WAL watcher for a given WriteTo.

func (*Watcher) Run

func (w *Watcher) Run() error

Run the watcher, which will tail the WAL until the quit channel is closed or an error case is hit.

func (*Watcher) Start

func (w *Watcher) Start()

Start the Watcher.

func (*Watcher) Stop

func (w *Watcher) Stop()

Stop the Watcher.

type WatcherMetrics

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

func NewWatcherMetrics

func NewWatcherMetrics(reg prometheus.Registerer) *WatcherMetrics

type WriteTo

type WriteTo interface {
	Append([]record.RefSample) bool
	StoreSeries([]record.RefSeries, int)
	SeriesReset(int)
}

WriteTo is an interface used by the Watcher to send the samples it's read from the WAL on to somewhere else.