master

Documentation

Index

• Variables
• func ModifyShardAssignment(storageNodeIDs []models.NodeID, cfg *models.Database, ...) error
• func ShardAssignment(storageNodeIDs []models.NodeID, cfg *models.Database, fixedStartIndex int, ...) (*models.ShardAssignment, error)
• type ReplicaLeaderElector
• type StateMachineFactory
  • func NewStateMachineFactory(ctx context.Context, discoveryFactory discovery.Factory, stateMgr StateManager) *StateMachineFactory
  • func (f *StateMachineFactory) Start() (err error)
  • func (f *StateMachineFactory) Stop()
• type StateManager
  • func NewStateManager(ctx context.Context, masterRepo statepkg.Repository, ...) StateManager
• type StorageCluster

Variables

var StateMachinePaths = make(map[string]models.StateMachineInfo)

StateMachinePaths represents the paths which master state machine need watch.

Functions

func ModifyShardAssignment

func ModifyShardAssignment(storageNodeIDs []models.NodeID, cfg *models.Database, shardAssignment *models.ShardAssignment,
	fixedStartIndex int, startShardID models.ShardID) error

func ShardAssignment

func ShardAssignment(storageNodeIDs []models.NodeID, cfg *models.Database,
	fixedStartIndex int, startShardID models.ShardID) (*models.ShardAssignment, error)

ShardAssignment assigns replica list for storage storageCluster which database's each shard based on selected node list in storageCluster. There are 2 goals of replica assignment: 1. Spread the replicas evenly among storage nodes for currently storageCluster state. 2. For shards assigned to a particular storage node, their other replicas are spread over the other storage nodes.

TO achieve this goal, we: 1. Assign the first replica of each shard by round-robin, starting from a random position in the storage node list. 2. Assign the remaining replicas of each shard with an increasing shift.

Here is an example of assigning, (num. of nodes = 5, num of shards = 10, replica factor = 3) node-0 node-1 node-2 node-3 node-4 s0 s1 s2 s3 s4 (1st replica) s5 s6 s7 s8 s9 (1st replica) s4 s0 s1 s2 s3 (2st replica) s8 s9 s5 s6 s7 (2st replica) s3 s4 s0 s1 s2 (3st replica) s7 s8 s9 s5 s6 (3st replica)

Types

type ReplicaLeaderElector

type ReplicaLeaderElector interface {
	// ElectLeader elects the replica's leader based on shard assignment.
	ElectLeader(shardAssignment *models.ShardAssignment,
		liveNodes map[models.NodeID]models.StatefulNode,
		shardID models.ShardID,
	) (leader models.NodeID, err error)
}

ReplicaLeaderElector represents replica leader elector for shard.

type StateMachineFactory

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

StateMachineFactory represents master state machine maintainer.

func NewStateMachineFactory

func NewStateMachineFactory(ctx context.Context,
	discoveryFactory discovery.Factory,
	stateMgr StateManager,
) *StateMachineFactory

NewStateMachineFactory creates a StateMachineFactory instance.

func (*StateMachineFactory) Start

func (f *StateMachineFactory) Start() (err error)

Start starts all master related state machines.

func (*StateMachineFactory) Stop

func (f *StateMachineFactory) Stop()

Stop stops the broker's state machines.

type StateManager

type StateManager interface {
	discovery.StateMachineEventHandle

	// SetStateMachineFactory sets state machine factory.
	SetStateMachineFactory(stateMachineFct *StateMachineFactory)
	// GetStateMachineFactory returns state machine factory.
	GetStateMachineFactory() *StateMachineFactory
	// GetStorageCluster returns cluster controller for maintain the metadata of storage cluster.
	GetStorageCluster(name string) StorageCluster
	// GetDatabases returns the current databases.
	GetDatabases() []models.Database
	// GetStorages returns the current storage cluster list.
	GetStorages() []config.StorageCluster
	// GetShardAssignments returns the current shard assignment list.
	GetShardAssignments() []models.ShardAssignment
	// GetStorageStates returns current storage state list.
	GetStorageStates() []*models.StorageState
}

StateManager represents master state manager, state coordinator.

func NewStateManager

func NewStateManager(
	ctx context.Context,
	masterRepo statepkg.Repository,
	repoFactory statepkg.RepositoryFactory,
) StateManager

NewStateManager creates a StateManager instance.

type StorageCluster

type StorageCluster interface {
	// GetConfig returns storage configuration.
	GetConfig() *config.StorageCluster
	// Start starts the state machine for storage state change.
	Start() error
	// GetState returns the current state of storage cluster.
	GetState() *models.StorageState
	// GetLiveNodes returns the current live nodes of storage cluster.
	GetLiveNodes() ([]models.StatefulNode, error)
	// FlushDatabase submits the coordinator task for flushing memory database by name
	FlushDatabase(databaseName string) error
	// SaveDatabaseAssignment saves database assignment in storage state repo.
	SaveDatabaseAssignment(
		shardAssign *models.ShardAssignment,
		databaseOption *option.DatabaseOption,
	) error
	// SetDatabaseLimits sets the database's limits.
	SetDatabaseLimits(database string, limits []byte) error
	// DropDatabaseAssignment drops database assignment from storage state repo.
	DropDatabaseAssignment(databaseName string) error
	// GetRepo returns current storage cluster's state repo
	GetRepo() state.Repository
	// Close closes storage cluster controller
	Close()
}

StorageCluster represents storage cluster controller, 1) discovery active node list in storage cluster 2) save shard assignment 3) generate coordinator task