raftstore

Documentation

Index

• Constants
• Variables
• func AddNode(manager NodeManager, nodeId uint64, addr string, heartbeat int, replicate int)
• func DeleteNode(manager NodeManager, nodeId uint64)
• type Config
• type NodeManager
• type NodeResolver
  • func NewNodeResolver() NodeResolver
• type Partition
• type PartitionConfig
• type PartitionFsm
• type PartitionStatus
• type PeerAddress
• type RaftStore
  • func NewRaftStore(cfg *Config) (mr RaftStore, err error)
• type RocksDBStore
  • func NewRocksDBStore(dir string) (store *RocksDBStore)
  • func (rs *RocksDBStore) BatchPut(cmdMap map[string][]byte) error
  • func (rs *RocksDBStore) Del(key interface{}) (result interface{}, err error)
  • func (rs *RocksDBStore) DeleteKeyAndPutIndex(key string, cmdMap map[string][]byte) error
  • func (rs *RocksDBStore) Get(key interface{}) (result interface{}, err error)
  • func (rs *RocksDBStore) Iterator(snapshot *gorocksdb.Snapshot) *gorocksdb.Iterator
  • func (rs *RocksDBStore) Open() error
  • func (rs *RocksDBStore) Put(key, value interface{}) (result interface{}, err error)
  • func (rs *RocksDBStore) ReleaseSnapshot(snapshot *gorocksdb.Snapshot)
  • func (rs *RocksDBStore) RocksDBSnapshot() *gorocksdb.Snapshot
• type Store

Constants

const (
	DefaultHeartbeatPort = 5901
	DefaultReplicatePort = 5902
	DefaultRetainLogs    = 20000
)

Constants for network port definition.

Variables

var (
	ErrNoSuchNode        = errors.New("no such node")
	ErrIllegalAddress    = errors.New("illegal address")
	ErrUnknownSocketType = errors.New("unknown socket type")
)

Error definitions.

var (
	ErrNotLeader = errors.New("not leader")
)

Error definitions for raft store partition.

Functions

func AddNode

func AddNode(manager NodeManager, nodeId uint64, addr string, heartbeat int, replicate int)

AddNode add node address into specified NodeManger if possible.

func DeleteNode

func DeleteNode(manager NodeManager, nodeId uint64)

DeleteNode delete node address data from specified NodeManager if possible.

Types

type Config

type Config struct {
	NodeID        uint64 // Identity of raft server instance.
	WalPath       string // Path of WAL(Write after Log)
	IpAddr        string // IP address of node
	HeartbeatPort int
	ReplicatePort int
	RetainLogs    uint64 // // RetainLogs controls how many logs we leave after truncate. The default value is 20000.
}

Config defined necessary configuration properties for raft store.

type NodeManager

type NodeManager interface {
	// AddNode used to add node address information.
	AddNode(nodeId uint64, addr string)

	// AddNode adds node address with specified port.
	AddNodeWithPort(nodeId uint64, addr string, heartbeat int, replicate int)

	// DeleteNode used to delete node address information
	// of specified node ID from NodeManager if possible.
	DeleteNode(nodeId uint64)
}

NodeManager defined necessary methods for node address management.

type NodeResolver

type NodeResolver interface {
	raft.SocketResolver
	NodeManager
}

NodeResolver defined necessary methods for both node address resolving and management. It extends from SocketResolver and NodeManager.

func NewNodeResolver

func NewNodeResolver() NodeResolver

NewNodeResolver returns a new NodeResolver instance for node address management and resolving.

type Partition

type Partition interface {
	// Submit submits command data to raft log.
	Submit(cmd []byte) (resp interface{}, err error)

	// ChaneMember submits member change event and information to raft log.
	ChangeMember(changeType proto.ConfChangeType, peer proto.Peer, context []byte) (resp interface{}, err error)

	// Stop removes this raft partition from raft server and make this partition shutdown.
	Stop() error

	// Delete stop and delete this partition forever.
	Delete() error

	// Status returns current raft status.
	Status() (status *PartitionStatus)

	// LeaderTerm returns current term of leader in raft group.
	LeaderTerm() (leaderId, term uint64)

	// IsLeader returns true if this node current is the leader in the raft group it belong to.
	IsLeader() bool

	// AppliedIndex returns current index value of applied raft log in this raft store partition.
	AppliedIndex() uint64

	// Truncate raft log
	Truncate(index uint64)
}

Partition wraps necessary methods for raft store partition operation. Partition is a shard for multi-raft in RaftSore. RaftStore based on multi-raft which managing multiple raft replication group at same time through single raft server instance and system resource.

type PartitionConfig

type PartitionConfig struct {
	ID      uint64
	Applied uint64
	Leader  uint64
	Term    uint64
	Peers   []PeerAddress
	SM      PartitionFsm
}

PartitionConfig defined necessary configuration properties for raft store partition.

type PartitionFsm

type PartitionFsm interface {
	raft.StateMachine
	Store
}

PartitionFsm wraps necessary methods include both FSM implementation and data storage operation for raft store partition. It extends from raft StateMachine and Store.

type PartitionStatus

type PartitionStatus = raft.Status

PartitionStatus is a type alias of raft.Status

type PeerAddress

type PeerAddress struct {
	proto.Peer
	Address       string
	HeartbeatPort int
	ReplicatePort int
}

type RaftStore

type RaftStore interface {
	CreatePartition(cfg *PartitionConfig) (Partition, error)
	Stop()
	RaftConfig() *raft.Config
	NodeManager
}

func NewRaftStore

func NewRaftStore(cfg *Config) (mr RaftStore, err error)

type RocksDBStore

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

func NewRocksDBStore

func NewRocksDBStore(dir string) (store *RocksDBStore)

func (*RocksDBStore) BatchPut

func (rs *RocksDBStore) BatchPut(cmdMap map[string][]byte) error

func (*RocksDBStore) Del

func (rs *RocksDBStore) Del(key interface{}) (result interface{}, err error)

func (*RocksDBStore) DeleteKeyAndPutIndex

func (rs *RocksDBStore) DeleteKeyAndPutIndex(key string, cmdMap map[string][]byte) error

func (*RocksDBStore) Get

func (rs *RocksDBStore) Get(key interface{}) (result interface{}, err error)

func (*RocksDBStore) Iterator

func (rs *RocksDBStore) Iterator(snapshot *gorocksdb.Snapshot) *gorocksdb.Iterator

func (*RocksDBStore) Open

func (rs *RocksDBStore) Open() error

func (*RocksDBStore) Put

func (rs *RocksDBStore) Put(key, value interface{}) (result interface{}, err error)

func (*RocksDBStore) ReleaseSnapshot

func (rs *RocksDBStore) ReleaseSnapshot(snapshot *gorocksdb.Snapshot)

func (*RocksDBStore) RocksDBSnapshot

func (rs *RocksDBStore) RocksDBSnapshot() *gorocksdb.Snapshot

type Store

type Store interface {
	Put(key, val interface{}) (interface{}, error)
	Get(key interface{}) (interface{}, error)
	Del(key interface{}) (interface{}, error)
}

Store is the interface defined the abstract and necessary methods for storage operation.