drummer

README

About Drummer

Drummer is a component used in Dragonboat's tests. It has been moved into github.com/lni/dragonboat/internal in the coming release, your application is not suppose to import drummer.

Documentation

Overview

Package drummer implements a reference Master server and client. They can manage and monitor large number of NodeHosts distributed across the network. Raft nodes running on failed NodeHost instances can be detected and replaced automatically by Drummer servers, thus bring in high availability to the system.

Note that both Drummer servers and its IMasterClient client provided in this package are optional.

Index

• Constants
• Variables
• func AddDrummerServer(ctx context.Context, client pb.DrummerClient, nodeID uint64, address string) (*pb.Empty, error)
• func EntityFailed(lastTick uint64, currentTick uint64) bool
• func GRPCError(err error) error
• func GetClusterCollection(ctx context.Context, client pb.DrummerClient) (*pb.ClusterCollection, error)
• func GetClusterConfig() pb.Config
• func GetClusterConfigFromFile(fp string) pb.Config
• func GetClusterStates(ctx context.Context, client pb.DrummerClient, clusters []uint64) (*pb.ClusterStates, error)
• func GetNodeHostCollection(ctx context.Context, client pb.DrummerClient) (*pb.NodeHostCollection, error)
• func NewDB(clusterID uint64, nodeID uint64) statemachine.IStateMachine
• func RemoveDrummerServer(ctx context.Context, client pb.DrummerClient, nodeID uint64) (*pb.Empty, error)
• func SubmitBootstrappped(ctx context.Context, client pb.DrummerClient) error
• func SubmitCreateDrummerChange(ctx context.Context, client pb.DrummerClient, clusterID uint64, ...) error
• func SubmitRegions(ctx context.Context, client pb.DrummerClient, region pb.Regions) error
• type DB
  • func (d *DB) Close()
  • func (d *DB) GetHash() uint64
  • func (d *DB) Lookup(key []byte) []byte
  • func (d *DB) RecoverFromSnapshot(r io.Reader, files []statemachine.SnapshotFile, done <-chan struct{}) error
  • func (d *DB) SaveSnapshot(w io.Writer, fc statemachine.ISnapshotFileCollection, done <-chan struct{}) (uint64, error)
  • func (d *DB) Update(data []byte) uint64
• type Drummer
  • func NewDrummer(nh *dragonboat.NodeHost, grpcHost string) *Drummer
  • func (d *Drummer) Start()
  • func (d *Drummer) Stop()
• type NodehostAPI
  • func NewNodehostAPI(address string, nh *dragonboat.NodeHost) *NodehostAPI
  • func (api *NodehostAPI) CloseSession(ctx context.Context, cs *client.Session) (*pb.SessionResponse, error)
  • func (api *NodehostAPI) GetSession(ctx context.Context, req *pb.SessionRequest) (*client.Session, error)
  • func (api *NodehostAPI) Propose(ctx context.Context, req *pb.RaftProposal) (*pb.RaftResponse, error)
  • func (api *NodehostAPI) Read(ctx context.Context, req *pb.RaftReadIndex) (*pb.RaftResponse, error)
  • func (api *NodehostAPI) Stop()

Constants

const (
	// DBUpdated indicates DB has been successfully updated
	DBUpdated uint64 = 0
	// ClusterExists means DB update has been rejected as the cluster to
	// be created already exist.
	ClusterExists uint64 = 1
	// DBBootstrapped means DB update has been rejected as the
	// DB has been bootstrapped.
	DBBootstrapped uint64 = 2
)

const (
	// DBKVUpdated means the KV update has been successfully completed
	DBKVUpdated uint64 = 0
	// DBKVFinalized indicates that the KV update is rejected as there is
	// already a finalized record in DB with the specified key
	DBKVFinalized uint64 = 1
	// DBKVRejected indicates that the KV update is rejected
	DBKVRejected uint64 = 2
)

Variables

var (
	// ErrDBKVUpdateRejected indicates that the update for KV has been
	// rejected.
	ErrDBKVUpdateRejected = errors.New("DB KV update rejected")
)

var (
	// ErrInvalidRequest indicates the request can not be fulfilled as it is
	// regarded as invalid.
	ErrInvalidRequest = errors.New("invalid drummer request")
)

Functions

func AddDrummerServer

func AddDrummerServer(ctx context.Context, client pb.DrummerClient,
	nodeID uint64, address string) (*pb.Empty, error)

AddDrummerServer adds a new drummer node with specified nodeID and address to the Drummer cluster.

func EntityFailed

func EntityFailed(lastTick uint64, currentTick uint64) bool

EntityFailed returns whether the timeline indicates that the entity has failed.

func GRPCError

func GRPCError(err error) error

GRPCError converts errors defined in package multiraft to gRPC errors

func GetClusterCollection

func GetClusterCollection(ctx context.Context,
	client pb.DrummerClient) (*pb.ClusterCollection, error)

GetClusterCollection returns known clusters from the Drummer server.

func GetClusterConfig

func GetClusterConfig() pb.Config

GetClusterConfig returns the configuration used by Raft clusters managed by the drummer server.

func GetClusterConfigFromFile

func GetClusterConfigFromFile(fp string) pb.Config

GetClusterConfigFromFile returns the DrummerConfig read from the specified configuration file.

func GetClusterStates

func GetClusterStates(ctx context.Context,
	client pb.DrummerClient, clusters []uint64) (*pb.ClusterStates, error)

GetClusterStates returns cluster states known to the Drummer server.

func GetNodeHostCollection

func GetNodeHostCollection(ctx context.Context,
	client pb.DrummerClient) (*pb.NodeHostCollection, error)

GetNodeHostCollection returns nodehosts known to the Drummer.

func NewDB

func NewDB(clusterID uint64, nodeID uint64) statemachine.IStateMachine

NewDB creates a new DB instance.

func RemoveDrummerServer

func RemoveDrummerServer(ctx context.Context, client pb.DrummerClient,
	nodeID uint64) (*pb.Empty, error)

RemoveDrummerServer removes the specified node from the Drummer cluster.

func SubmitBootstrappped

func SubmitBootstrappped(ctx context.Context,
	client pb.DrummerClient) error

SubmitBootstrappped sets the bootstrapped flag on Drummer server.

func SubmitCreateDrummerChange

func SubmitCreateDrummerChange(ctx context.Context, client pb.DrummerClient,
	clusterID uint64, members []uint64, appName string) error

SubmitCreateDrummerChange submits Drummer change used for defining clusters.

func SubmitRegions

func SubmitRegions(ctx context.Context,
	client pb.DrummerClient, region pb.Regions) error

SubmitRegions submits regions info to the Drummer server.

Types

type DB

type DB struct {
	ClusterID      uint64 `json:"-"`
	NodeID         uint64 `json:"-"`
	Version        uint64
	Tick           uint64
	LaunchDeadline uint64
	Failed         bool
	Clusters       map[uint64]*pb.Cluster
	KVMap          map[string][]byte
	ClusterImage   *multiCluster
	NodeHostImage  *multiNodeHost
	NodeHostInfo   map[string]pb.NodeHostInfo
	Requests       map[string][]pb.NodeHostRequest
	Outgoing       map[string][]pb.NodeHostRequest
}

DB is the struct used to maintain the raft-backed Drummer DB

func (*DB) Close

func (d *DB) Close()

Close closes the DB instance.

func (*DB) GetHash

func (d *DB) GetHash() uint64

GetHash returns the state machine hash.

func (*DB) Lookup

func (d *DB) Lookup(key []byte) []byte

Lookup performances local data lookup on the DB.

func (*DB) RecoverFromSnapshot

func (d *DB) RecoverFromSnapshot(r io.Reader, files []statemachine.SnapshotFile,
	done <-chan struct{}) error

RecoverFromSnapshot recovers DB state from a snapshot.

func (*DB) SaveSnapshot

func (d *DB) SaveSnapshot(w io.Writer, fc statemachine.ISnapshotFileCollection,
	done <-chan struct{}) (uint64, error)

SaveSnapshot generates a snapshot of the DB

func (*DB) Update

func (d *DB) Update(data []byte) uint64

Update updates the DB instance.

type Drummer

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

Drummer is the main struct used by the Drummer server

func NewDrummer

func NewDrummer(nh *dragonboat.NodeHost, grpcHost string) *Drummer

NewDrummer creates a new Drummer instance.

func (*Drummer) Start

func (d *Drummer) Start()

Start starts the Drummer server.

func (*Drummer) Stop

func (d *Drummer) Stop()

Stop gracefully stops the Drummer server.

type NodehostAPI

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

NodehostAPI implements the grpc server used for making raft IO requests.

func NewNodehostAPI

func NewNodehostAPI(address string, nh *dragonboat.NodeHost) *NodehostAPI

NewNodehostAPI creates a new NodehostAPI server instance.

func (*NodehostAPI) CloseSession

func (api *NodehostAPI) CloseSession(ctx context.Context,
	cs *client.Session) (*pb.SessionResponse, error)

CloseSession closes the specified client session instance.

func (*NodehostAPI) GetSession

func (api *NodehostAPI) GetSession(ctx context.Context,
	req *pb.SessionRequest) (*client.Session, error)

GetSession gets a new client session instance.

func (*NodehostAPI) Propose

func (api *NodehostAPI) Propose(ctx context.Context,
	req *pb.RaftProposal) (*pb.RaftResponse, error)

Propose makes a propose.

func (*NodehostAPI) Read

func (api *NodehostAPI) Read(ctx context.Context,
	req *pb.RaftReadIndex) (*pb.RaftResponse, error)

Read makes a linearizable read operation.

func (*NodehostAPI) Stop

func (api *NodehostAPI) Stop()

Stop stops the NodehostAPI instance.