raft

Documentation

Index

• type Backend
  • func NewBackend(h Handle, l hclog.Logger) (*Backend, error)
  • func (b *Backend) Apply(buf []byte, idx uint64) any
  • func (b *Backend) DeleteCAS(ctx context.Context, id *pbresource.ID, version string) error
  • func (b *Backend) HandleConnection(conn net.Conn)
  • func (b *Backend) LeaderChanged()
  • func (b *Backend) List(ctx context.Context, consistency storage.ReadConsistency, ...) ([]*pbresource.Resource, error)
  • func (b *Backend) ListByOwner(_ context.Context, id *pbresource.ID) ([]*pbresource.Resource, error)
  • func (b *Backend) Read(ctx context.Context, consistency storage.ReadConsistency, id *pbresource.ID) (*pbresource.Resource, error)
  • func (b *Backend) Restore() (*Restoration, error)
  • func (b *Backend) Run(ctx context.Context)
  • func (b *Backend) Snapshot() (*Snapshot, error)
  • func (b *Backend) WatchList(_ context.Context, resType storage.UnversionedType, ...) (storage.Watch, error)
  • func (b *Backend) WriteCAS(ctx context.Context, res *pbresource.Resource) (*pbresource.Resource, error)
• type Handle
• type Restoration
  • func (r *Restoration) Abort()
  • func (r *Restoration) Apply(msg []byte) error
  • func (r *Restoration) Commit()
• type Snapshot
  • func (s *Snapshot) Next() ([]byte, error)

Types

type Backend

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

Backend is a Raft-backed storage backend implementation.

func NewBackend

func NewBackend(h Handle, l hclog.Logger) (*Backend, error)

NewBackend returns a storage backend that uses Raft for durable persistence and serves reads from an in-memory database. It's suitable for production use.

It's not an entirely clean abstraction because rather than owning the Raft subsystem directly, it has to integrate with the existing FSM and related machinery from before generic resources.

The given Handle will be used to apply logs and interrogate leadership state. In certain restricted circumstances, Handle may be nil, such as during tests that only exercise snapshot restoration, or when initializing a throwaway FSM during peers.json recovery - but calling any of the data access methods (read or write) will result in a panic.

With Raft, writes and strongly consistent reads must be done on the leader. Backend implements a gRPC server, which followers will use to transparently forward operations to the leader. To do so, they will obtain a connection using Handle.DialLeader. Connections are cached for re-use, so when there's a new leader, you must call LeaderChanged to refresh the connection. Leaders must accept connections and hand them off by calling Backend.HandleConnection. Backend's gRPC client and server *DO NOT* handle TLS themselves, as they are intended to communicate over Consul's multiplexed server port (which handles TLS).

For more information, see here: https://github.com/hashicorp/consul/tree/main/docs/resources#raft-storage-backend

You must call Run before using the backend.

func (*Backend) Apply

func (b *Backend) Apply(buf []byte, idx uint64) any

Apply is called by the FSM with the bytes of a Raft log entry, with Consul's envelope (i.e. type prefix and msgpack wrapper) stripped off.

func (*Backend) DeleteCAS

func (b *Backend) DeleteCAS(ctx context.Context, id *pbresource.ID, version string) error

DeleteCAS implements the storage.Backend interface.

func (*Backend) HandleConnection

func (b *Backend) HandleConnection(conn net.Conn)

HandleConnection should be called whenever a forwarding connection is opened.

func (*Backend) LeaderChanged

func (b *Backend) LeaderChanged()

LeaderChanged should be called whenever the current Raft leader changes, to drop and re-create the gRPC connection used for forwarding.

func (*Backend) List

func (b *Backend) List(ctx context.Context, consistency storage.ReadConsistency, resType storage.UnversionedType, tenancy *pbresource.Tenancy, namePrefix string) ([]*pbresource.Resource, error)

List implements the storage.Backend interface.

func (*Backend) ListByOwner

func (b *Backend) ListByOwner(_ context.Context, id *pbresource.ID) ([]*pbresource.Resource, error)

ListByOwner implements the storage.Backend interface.

func (*Backend) Read

func (b *Backend) Read(ctx context.Context, consistency storage.ReadConsistency, id *pbresource.ID) (*pbresource.Resource, error)

Read implements the storage.Backend interface.

func (*Backend) Restore

func (b *Backend) Restore() (*Restoration, error)

Restore starts the process of restoring a snapshot (i.e. from an on-disk backup, or to bootstrap from a leader).

Callers *must* call Abort or Commit when done, to free resources.

func (*Backend) Run

func (b *Backend) Run(ctx context.Context)

Run until the given context is canceled. This method blocks, so should be called in a goroutine.

func (*Backend) Snapshot

func (b *Backend) Snapshot() (*Snapshot, error)

Snapshot obtains a point-in-time snapshot of the backend's state, so that it can be written to disk as a backup or sent to bootstrap a follower.

func (*Backend) WatchList

func (b *Backend) WatchList(_ context.Context, resType storage.UnversionedType, tenancy *pbresource.Tenancy, namePrefix string) (storage.Watch, error)

WatchList implements the storage.Backend interface.

func (*Backend) WriteCAS

func (b *Backend) WriteCAS(ctx context.Context, res *pbresource.Resource) (*pbresource.Resource, error)

WriteCAS implements the storage.Backend interface.

type Handle

type Handle interface {
	// Apply the given log message.
	Apply(msg []byte) (any, error)

	// IsLeader determines if this server is the Raft leader (so can handle writes).
	IsLeader() bool

	// EnsureStrongConsistency checks the server is able to handle consistent reads by
	// verifying its leadership and checking the FSM has applied all queued writes.
	EnsureStrongConsistency(ctx context.Context) error

	// DialLeader dials a gRPC connection to the leader for forwarding.
	DialLeader() (*grpc.ClientConn, error)
}

Handle provides glue for interacting with the Raft subsystem via existing machinery on consul.Server.

type Restoration

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

Restoration is a handle that can be used to restore a snapshot.

func (*Restoration) Abort

func (r *Restoration) Abort()

Abort the restoration. It's safe to always call this in a defer statement because aborting a committed restoration is a no-op.

func (*Restoration) Apply

func (r *Restoration) Apply(msg []byte) error

Apply the given protobuf-encoded resource to the backend.

func (*Restoration) Commit

func (r *Restoration) Commit()

Commit the restoration.

type Snapshot

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

Snapshot is a point-in-time snapshot of a backend's state.

func (*Snapshot) Next

func (s *Snapshot) Next() ([]byte, error)

Next returns the next resource in the snapshot, protobuf encoded. nil bytes will be returned when the end of the snapshot has been reached.