log

Documentation

Index

• Variables
• type Log
  • func Load(opts ...Option) (l *Log, err error)
  • func New(opts ...Option) (*Log, error)
  • func (l *Log) After(index uint64) ([]*pb.LogEntry, error)
  • func (l *Log) Append(entries ...*pb.LogEntry) error
  • func (l *Log) AsUpToDate(lastIndex, lastTerm uint64) bool
  • func (l *Log) Commit(index uint64) error
  • func (l *Log) CommitIndex() uint64
  • func (l *Log) CommitTerm() uint64
  • func (l *Log) Create(key, value []byte, term uint64) (*pb.LogEntry, error)
  • func (l *Log) Get(index uint64) (*pb.LogEntry, error)
  • func (l *Log) LastApplied() uint64
  • func (l *Log) LastCommit() *pb.LogEntry
  • func (l *Log) LastEntry() *pb.LogEntry
  • func (l *Log) LastTerm() uint64
  • func (l *Log) Length() uint64
  • func (l *Log) Meta() *pb.LogMeta
  • func (l *Log) Prev(index uint64) (*pb.LogEntry, error)
  • func (l *Log) Truncate(index, term uint64) error
• type Option
  • func WithStateMachine(sm StateMachine) Option
  • func WithSync(sync Sync) Option
• type Reader
• type StateMachine
• type Sync
• type Writer

Variables

var (
	ErrAppendEarlierTerm     = errors.New("cannot append entry in earlier term")
	ErrAppendSmallerIndex    = errors.New("cannot append entry with smaller index")
	ErrAppendSkipIndex       = errors.New("cannot skip index")
	ErrCommitInvalidIndex    = errors.New("cannot commit invalid index")
	ErrIndexAlreadyCommitted = errors.New("index already committed")
	ErrTruncInvalidIndex     = errors.New("cannot truncate invalid index")
	ErrTruncCommittedIndex   = errors.New("cannot truncate already committed index")
	ErrTruncTermMismatch     = errors.New("the first entry being truncated must match expected term")
	ErrSyncRequired          = errors.New("cannot load log from disk without a sync")
)

Types

type Log

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

Log implements the sequence of commands applied to the Raft state machine. This implementation uses an in-memory log that is backed by a disk based store for durability. The in-memory portion of the log contains all uncommitted entries and recently committed entries that need to be applied to the state machine. Once applied successfully, the in-memory log is truncated.

The log's primarily responsibility is to ensure that the sequence of commands is consistent, e.g. that entries are appended in a monotonically increasing time order as defined by the Raft leader's term.

Note that the log is not thread-safe and is not intended to be accessed from multiple go routines. Instead the log should be maintained by a single state machine that updates it sequentially when entries are committed.

TODO: right now the log stores everything in-memory; refactor to store partial in-mem log TODO: implement snapshotting functionality TODO: implement log validation to ensure the log is in a correct state

func Load

func Load(opts ...Option) (l *Log, err error)

Load a log from disk. This method creates a new log and reads entries and meta data from the sync reader. An error is returned if there is no WithSync() option.

func New

func New(opts ...Option) (*Log, error)

New creates an empty log with a null entry at index 0. It is a fresh log ready for any log operations that may be applied to it. Generally, external users will want to load the log from disk using a sync command.

func (*Log) After

func (l *Log) After(index uint64) ([]*pb.LogEntry, error)

After returns all entries after the specified index, inclusive

func (*Log) Append

func (l *Log) Append(entries ...*pb.LogEntry) error

Append one ore more entries and perform log invariant checks. If appending an entry creates a log inconsistency (out of order term or index), then an error is returned. A couple of important notes:

1. Append does not undo any successful appends even on error
2. Append will not compare entries that specify the same index

These notes mean that all entries being appended to this log should be consistent with each other as well as the end of the log, and that the log needs to be truncated in order to "update" or splice two logs together.

func (*Log) AsUpToDate

func (l *Log) AsUpToDate(lastIndex, lastTerm uint64) bool

AsUpToDate returns true if the remote log specified by the last index and last term are at least as up to date (or farther ahead) than the local log.

func (*Log) Commit

func (l *Log) Commit(index uint64) error

Commit all entries up to and including the specified index.

func (*Log) CommitIndex

func (l *Log) CommitIndex() uint64

CommitIndex returns the index of the last committed log entry.

func (*Log) CommitTerm

func (l *Log) CommitTerm() uint64

CommitTerm is a helper function to get the term of the entry at the commit index.

func (*Log) Create

func (l *Log) Create(key, value []byte, term uint64) (*pb.LogEntry, error)

Create an entry in the log and append it. This is essentially a helper method for quickly adding a command to the state machine consistent with the local log.

func (*Log) Get

func (l *Log) Get(index uint64) (*pb.LogEntry, error)

Get the entry at the specified index (whether or not it is committed). Returns an error if no entry exists at the index.

func (*Log) LastApplied

func (l *Log) LastApplied() uint64

LastApplied returns the index of the last applied log entry.

func (*Log) LastCommit

func (l *Log) LastCommit() *pb.LogEntry

LastCommit returns the log entry at the commit index.

func (*Log) LastEntry

func (l *Log) LastEntry() *pb.LogEntry

LastEntry returns the log entry at the last applied index.

func (*Log) LastTerm

func (l *Log) LastTerm() uint64

LastTerm is a helper function to get the term of the entry at the last applied index.

func (*Log) Length

func (l *Log) Length() uint64

Length returns the number of entries in the log

func (*Log) Meta

func (l *Log) Meta() *pb.LogMeta

Meta updates the state of the saved metadata on the log and returns a pointer to it. This means that the returned Meta is not safe for concurrent use and any writer that operates in another go routine should create a copy of it.

func (*Log) Prev

func (l *Log) Prev(index uint64) (*pb.LogEntry, error)

Prev returns the entry before the specified index (whether or not it is committed). Returns an error if no entry exists before.

func (*Log) Truncate

func (l *Log) Truncate(index, term uint64) error

Truncate the log to the given position, conditioned by term. This method returns an error if the log has been committed after the specified index, there is an epoch mismatch, or there is some other log operation error.

This method truncates everything after the given index, but keeps the entry at the specified index; e.g. truncate after.

type Option

type Option func(l *Log) error

func WithStateMachine

func WithStateMachine(sm StateMachine) Option

func WithSync

func WithSync(sync Sync) Option

type Reader

type Reader interface {
	// Read should return the entry at the specified index.
	Read(index uint64) (*pb.LogEntry, error)

	// ReadFrom should return all entries starting at the given index and following.
	ReadFrom(index uint64) ([]*pb.LogEntry, error)

	// ReadMeta should return the log metadata from disk.
	ReadMeta() (*pb.LogMeta, error)
}

type StateMachine

type StateMachine interface {
	CommitEntry(*pb.LogEntry) error
	DropEntry(*pb.LogEntry) error
}

type Sync

type Sync interface {
	Writer
	Reader
	io.Closer
}

type Writer

type Writer interface {
	// Write should append all of the log entries to disk.
	Write(...*pb.LogEntry) error

	// WriteMeta should sync the log metadata to disk.
	WriteMeta(*pb.LogMeta) error

	// Trunc should delete all entries starting with the given index and all entries
	// that follow. Note that this is different than the log.Truncate() semantics.
	Trunc(startIndex uint64) error
}