raft

Documentation

Index

• func NewRaftEventStream[V encoding.BinaryMarshaler, VPtr Unmarshallable[V], E RaftStreamEvent[V, VPtr], ...](ctx context.Context, cluster *Cluster, shardID uint64) eventstream.EventView[V, E]
• type Cluster
  • func New(cfg *RaftConfig) *Cluster
  • func (c *Cluster) AddMember(ctx context.Context, shardID uint64, replicaID uint64, address string) error
  • func (c *Cluster) Join(ctx context.Context) error
  • func (c *Cluster) Start(ctx context.Context) error
  • func (c *Cluster) Stop()
• type Event
• type RaftConfig
• type RaftEventView
  • func (s *RaftEventView[V, VPrt, E]) Publish(ctx context.Context, event E) error
  • func (s *RaftEventView[V, VPrt, E]) View(ctx context.Context) (V, error)
• type RaftStreamEvent
• type ShardHandle
  • func AddShard[Q any, R any, E Event, EPtr Unmarshallable[E]](ctx context.Context, to *Cluster, shardID uint64, ...) *ShardHandle[E, Q, R]
  • func (s *ShardHandle[E, Q, R]) Propose(ctx context.Context, msg E) error
  • func (s *ShardHandle[E, Q, R]) Query(ctx context.Context, query Q) (R, error)
• type StateMachine
• type UnitQuery
• type Unmarshallable

Functions

func NewRaftEventStream

func NewRaftEventStream[
	V encoding.BinaryMarshaler,
	VPtr Unmarshallable[V],
	E RaftStreamEvent[V, VPtr],
	EPtr Unmarshallable[E],
](ctx context.Context, cluster *Cluster, shardID uint64) eventstream.EventView[V, E]

Types

type Cluster

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

Cluster of dragonboat nodes.

func New

func New(cfg *RaftConfig) *Cluster

New creates a new cluster.

func (*Cluster) AddMember

func (c *Cluster) AddMember(ctx context.Context, shardID uint64, replicaID uint64, address string) error

AddMember to the cluster. This needs to be called on an existing running cluster member, before the new member is started.

func (*Cluster) Join

func (c *Cluster) Join(ctx context.Context) error

Join the cluster as a new member. Blocks until the cluster instance is ready.

func (*Cluster) Start

func (c *Cluster) Start(ctx context.Context) error

Start the cluster. Blocks until the cluster instance is ready.

func (*Cluster) Stop

func (c *Cluster) Stop()

type Event

type Event interface {
	encoding.BinaryMarshaler
}

Event to update the state machine. These are stored in the Raft log.

type RaftConfig

type RaftConfig struct {
	InitialMembers    []string      `help:"Initial members" required:""`
	ReplicaID         uint64        `help:"Node ID" required:""`
	DataDir           string        `help:"Data directory" required:""`
	RaftAddress       string        `help:"Address to advertise to other nodes" required:""`
	ListenAddress     string        `help:"Address to listen for incoming traffic. If empty, RaftAddress will be used."`
	ShardReadyTimeout time.Duration `help:"Timeout for shard to be ready" default:"5s"`
	// Raft configuration
	RTT                time.Duration `help:"Estimated average round trip time between nodes" default:"200ms"`
	ElectionRTT        uint64        `help:"Election RTT as a multiple of RTT" default:"10"`
	HeartbeatRTT       uint64        `help:"Heartbeat RTT as a multiple of RTT" default:"1"`
	SnapshotEntries    uint64        `help:"Snapshot entries" default:"10"`
	CompactionOverhead uint64        `help:"Compaction overhead" default:"100"`
}

type RaftEventView

type RaftEventView[V encoding.BinaryMarshaler, VPrt Unmarshallable[V], E RaftStreamEvent[V, VPrt]] struct {
	// contains filtered or unexported fields
}

func (*RaftEventView[V, VPrt, E]) Publish

func (s *RaftEventView[V, VPrt, E]) Publish(ctx context.Context, event E) error

func (*RaftEventView[V, VPrt, E]) View

func (s *RaftEventView[V, VPrt, E]) View(ctx context.Context) (V, error)

type RaftStreamEvent

type RaftStreamEvent[View encoding.BinaryMarshaler, VPtr Unmarshallable[View]] interface {
	encoding.BinaryMarshaler
	eventstream.Event[View]
}

type ShardHandle

type ShardHandle[E Event, Q any, R any] struct {
	// contains filtered or unexported fields
}

ShardHandle is a handle to a shard in the cluster. It is the interface to update and query the state of a shard.

E is the event type. Q is the query type. R is the query response type.

func AddShard

func AddShard[Q any, R any, E Event, EPtr Unmarshallable[E]](
	ctx context.Context,
	to *Cluster,
	shardID uint64,
	sm StateMachine[Q, R, E, EPtr],
) *ShardHandle[E, Q, R]

AddShard adds a shard to the cluster. This can be only called before the cluster is started.

func (*ShardHandle[E, Q, R]) Propose

func (s *ShardHandle[E, Q, R]) Propose(ctx context.Context, msg E) error

Propose an event to the shard.

func (*ShardHandle[E, Q, R]) Query

func (s *ShardHandle[E, Q, R]) Query(ctx context.Context, query Q) (R, error)

Query the state of the shard.

type StateMachine

type StateMachine[Q any, R any, E Event, EPtr Unmarshallable[E]] interface {
	// Query the state of the state machine.
	Lookup(key Q) (R, error)
	// Update the state of the state machine.
	Update(msg E) error
	// Save the state of the state machine to a snapshot.
	Save(writer io.Writer) error
	// Recover the state of the state machine from a snapshot.
	Recover(reader io.Reader) error
	// Close the state machine.
	Close() error
}

StateMachine is a typed interface to dragonboat's statemachine.IStateMachine. It is used to implement the state machine for a single shard.

Q is the query type. R is the query response type. E is the event type.

type UnitQuery

type UnitQuery struct{}

type Unmarshallable

type Unmarshallable[T any] interface {
	*T
	encoding.BinaryUnmarshaler
}

Unmarshallable is a type that can be unmarshalled from a binary representation.