drummer

README

About

Drummer is used for Dragonboat's monkey testing.

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.

Index

• Constants
• Variables
• func AddDrummerServer(ctx context.Context, client pb.DrummerClient, replicaID uint64, address string) (*pb.Empty, error)
• func EntityFailed(lastTick uint64, currentTick uint64) bool
• func GRPCError(err error) error
• func GetNodeHostCollection(ctx context.Context, client pb.DrummerClient) (*pb.NodeHostCollection, error)
• func GetShardCollection(ctx context.Context, client pb.DrummerClient) (*pb.ShardCollection, error)
• func GetShardConfig() pb.Config
• func GetShardConfigFromFile(fp string) pb.Config
• func GetShardStates(ctx context.Context, client pb.DrummerClient, shards []uint64) (*pb.ShardStates, error)
• func NewDB(shardID uint64, replicaID uint64) sm.IStateMachine
• func RemoveDrummerServer(ctx context.Context, client pb.DrummerClient, replicaID uint64) (*pb.Empty, error)
• func RunTicker(td time.Duration, tf TickerFunc, stopc1 <-chan struct{}, ...)
• func SubmitBootstrappped(ctx context.Context, client pb.DrummerClient) error
• func SubmitCreateDrummerChange(ctx context.Context, client pb.DrummerClient, shardID uint64, members []uint64, ...) error
• func SubmitRegions(ctx context.Context, client pb.DrummerClient, region pb.Regions) error
• func ToNodeHostSession(s *pb.Session) *client.Session
• func ToPBSession(s *client.Session) *pb.Session
• type DB
  • func (d *DB) Close() error
  • func (d *DB) GetHash() (uint64, error)
  • func (d *DB) Lookup(query interface{}) (interface{}, error)
  • func (d *DB) RecoverFromSnapshot(r io.Reader, files []sm.SnapshotFile, done <-chan struct{}) error
  • func (d *DB) SaveSnapshot(w io.Writer, fc sm.ISnapshotFileCollection, done <-chan struct{}) error
  • func (d *DB) Update(entry sm.Entry) (sm.Result, error)
• 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 *pb.Session) (*pb.SessionResponse, error)
  • func (api *NodehostAPI) GetSession(ctx context.Context, req *pb.SessionRequest) (*pb.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()
• type TickerFunc

Constants

const (
	// DBUpdated indicates DB has been successfully updated
	DBUpdated uint64 = 0
	// ShardExists means DB update has been rejected as the shard to
	// be created already exist.
	ShardExists 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")
)

var (

	// NodeHostInfoReportSecond defines how often should the NodeHost reports its
	// full details to drummer servers.
	NodeHostInfoReportSecond = settings.Soft.NodeHostInfoReportSecond
)

Functions

func AddDrummerServer

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

AddDrummerServer adds a new drummer node with specified replicaID and address to the Drummer shard.

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 GetNodeHostCollection

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

GetNodeHostCollection returns nodehosts known to the Drummer.

func GetShardCollection

func GetShardCollection(ctx context.Context,
	client pb.DrummerClient) (*pb.ShardCollection, error)

GetShardCollection returns known shards from the Drummer server.

func GetShardConfig

func GetShardConfig() pb.Config

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

func GetShardConfigFromFile

func GetShardConfigFromFile(fp string) pb.Config

GetShardConfigFromFile returns the DrummerConfig read from the specified configuration file.

func GetShardStates

func GetShardStates(ctx context.Context,
	client pb.DrummerClient, shards []uint64) (*pb.ShardStates, error)

GetShardStates returns shard states known to the Drummer server.

func NewDB

func NewDB(shardID uint64, replicaID uint64) sm.IStateMachine

NewDB creates a new DB instance.

func RemoveDrummerServer

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

RemoveDrummerServer removes the specified node from the Drummer shard.

func RunTicker

func RunTicker(td time.Duration,
	tf TickerFunc, stopc1 <-chan struct{}, stopc2 <-chan struct{})

RunTicker runs a ticker at the specified interval, the provided TickerFunc will be called after each tick. The ticker will be stopped when the TickerFunc return a true value or when any of the two specified stop channels is signalled.

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,
	shardID uint64, members []uint64, appName string) error

SubmitCreateDrummerChange submits Drummer change used for defining shards.

func SubmitRegions

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

SubmitRegions submits regions info to the Drummer server.

func ToNodeHostSession

func ToNodeHostSession(s *pb.Session) *client.Session

func ToPBSession

func ToPBSession(s *client.Session) *pb.Session

Types

type DB

type DB struct {
	ShardID        uint64 `json:"-"`
	ReplicaID      uint64 `json:"-"`
	Version        uint64
	Tick           uint64
	LaunchDeadline uint64
	Failed         bool
	Shards         map[uint64]*pb.Shard
	KVMap          map[string][]byte
	ShardImage     *multiShard
	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() error

Close closes the DB instance.

func (*DB) GetHash

func (d *DB) GetHash() (uint64, error)

GetHash returns the state machine hash.

func (*DB) Lookup

func (d *DB) Lookup(query interface{}) (interface{}, error)

Lookup performances local data lookup on the DB.

func (*DB) RecoverFromSnapshot

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

RecoverFromSnapshot recovers DB state from a snapshot.

func (*DB) SaveSnapshot

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

SaveSnapshot generates a snapshot of the DB

func (*DB) Update

func (d *DB) Update(entry sm.Entry) (sm.Result, error)

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 {
	pb.UnimplementedNodehostAPIServer
	// 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 *pb.Session) (*pb.SessionResponse, error)

CloseSession closes the specified client session instance.

func (*NodehostAPI) GetSession

func (api *NodehostAPI) GetSession(ctx context.Context,
	req *pb.SessionRequest) (*pb.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.

type TickerFunc

type TickerFunc func() bool

TickerFunc is type of the function that will be called by the RunTicker function after each tick. The returned boolean value indicates whether the ticker should stop.