datanode

package
v2.4.0+incompatible Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 14, 2021 License: Apache-2.0 Imports: 37 Imported by: 1

Documentation

Index

Constants

View Source 
const (
	IntervalToUpdateReplica       = 600 // interval to update the replica
	IntervalToUpdatePartitionSize = 60  // interval to update the partition size
	NumOfFilesToRecoverInParallel = 10  // number of files to be recovered simultaneously
)
View Source 
const (
	ReadFlag  = 1
	WriteFlag = 2
)

cmd response

View Source 
const (
	ActionNotifyFollowerToRepair        = "ActionNotifyFollowerRepair"
	ActionStreamRead                    = "ActionStreamRead"
	ActionCreateExtent                  = "ActionCreateExtent:"
	ActionMarkDelete                    = "ActionMarkDelete:"
	ActionGetAllExtentWatermarks        = "ActionGetAllExtentWatermarks:"
	ActionWrite                         = "ActionWrite:"
	ActionRepair                        = "ActionRepair:"
	ActionDecommissionPartition         = "ActionDecommissionPartition"
	ActionAddDataPartitionRaftMember    = "ActionAddDataPartitionRaftMember"
	ActionRemoveDataPartitionRaftMember = "ActionRemoveDataPartitionRaftMember"
	ActionDataPartitionTryToLeader      = "ActionDataPartitionTryToLeader"

	ActionCreateDataPartition        = "ActionCreateDataPartition"
	ActionLoadDataPartition          = "ActionLoadDataPartition"
	ActionDeleteDataPartition        = "ActionDeleteDataPartition"
	ActionStreamReadTinyDeleteRecord = "ActionStreamReadTinyDeleteRecord"
	ActionSyncTinyDeleteRecord       = "ActionSyncTinyDeleteRecord"
	ActionStreamReadTinyExtentRepair = "ActionStreamReadTinyExtentRepair"
	ActionBatchMarkDelete            = "ActionBatchMarkDelete"
)

Action description

View Source 
const (
	RepairRead = true
	StreamRead = false
)
View Source 
const (
	EmptyResponse                      = 'E'
	TinyExtentRepairReadResponseArgLen = 17
	MaxSyncTinyDeleteBufferSize        = 2400000
	MaxFullSyncTinyDeleteTime          = 3600 * 24
	MinTinyExtentDeleteRecordSyncSize  = 4 * 1024 * 1024
)
View Source 
const (
	MetricPartitionIOName      = "dataPartitionIO"
	MetricPartitionIOBytesName = "dataPartitionIOBytes"
)
View Source 
const (
	DataPartitionPrefix           = "datapartition"
	DataPartitionMetadataFileName = "META"
	TempMetadataFileName          = ".meta"
	ApplyIndexFile                = "APPLY"
	TempApplyIndexFile            = ".apply"
	TimeLayout                    = "2006-01-02 15:04:05"
)
View Source 
const (
	RaftStatusStopped = 0
	RaftStatusRunning = 1
)
View Source 
const (
	DefaultZoneName         = proto.DefaultZoneName
	DefaultRaftDir          = "raft"
	DefaultRaftLogsToRetain = 10 // Count of raft logs per data partition
	DefaultDiskMaxErr       = 1
	DefaultDiskRetainMin    = 5 * util.GB  // GB
	DefaultDiskRetainMax    = 30 * util.GB // GB
)
View Source 
const (
	ConfigKeyLocalIP       = "localIP"       // string
	ConfigKeyPort          = "port"          // int
	ConfigKeyMasterAddr    = "masterAddr"    // array
	ConfigKeyZone          = "zoneName"      // string
	ConfigKeyDisks         = "disks"         // array
	ConfigKeyRaftDir       = "raftDir"       // string
	ConfigKeyRaftHeartbeat = "raftHeartbeat" // string
	ConfigKeyRaftReplica   = "raftReplica"   // string
	CfgTickInterval        = "tickInterval"  // int
	// smux Config
	ConfigKeyEnableSmuxClient  = "enableSmuxConnPool" //bool
	ConfigKeySmuxPortShift     = "smuxPortShift"      //int
	ConfigKeySmuxMaxConn       = "smuxMaxConn"        //int
	ConfigKeySmuxStreamPerConn = "smuxStreamPerConn"  //int
	ConfigKeySmuxMaxBuffer     = "smuxMaxBuffer"      //int
)
View Source 
const (
	BinaryMarshalMagicVersion = 0xFF
)
View Source 
const (
	BufferWrite = false
)
View Source 
const (
	DiskSectorSize = 512
)

Sector size

View Source 
const (
	DiskStatusFile = ".diskStatus"
)
View Source 
const ExpiredPartitionPrefix = "expired_"
View Source 
const (
	FinishLoadDataPartitionExtentHeader = 1
)

Status of load data partition extent header

View Source 
const (
	IsReleased = 1
)

Tiny extent has been put back to store

View Source 
const (
	MinAvaliTinyExtentCnt = 5
)
View Source 
const (
	MinTinyExtentsToRepair = 10 // minimum number of tiny extents to repair
)

Apply the raft log operation. Currently we only have the random write operation.

View Source 
const (
	ModuleName = "dataNode"
)
View Source 
const (
	NetworkProtocol = "tcp"
)

Network protocol

View Source 
const (
	RaftNotStarted = "RaftNotStarted"
)

Error code

View Source 
const (
	SyncTinyDeleteRecordFromLeaderOnEveryDisk = 5
)
View Source 
const (
	UpdateNodeInfoTicket = 1 * time.Minute
)

Variables

View Source 
var (
	MaxExtentRepairLimit = 20000
	MinExtentRepairLimit = 5
	CurExtentRepairLimit = MaxExtentRepairLimit
)
View Source 
var (
	ErrIncorrectStoreType          = errors.New("Incorrect store type")
	ErrNoSpaceToCreatePartition    = errors.New("No disk space to create a data partition")
	ErrNewSpaceManagerFailed       = errors.New("Creater new space manager failed")
	ErrGetMasterDatanodeInfoFailed = errors.New("Failed to get datanode info from master")

	LocalIP string

	MasterClient = masterSDK.NewMasterClient(nil, false)
)
View Source 
var (
	AutoRepairStatus = true
)
View Source 
var (
	// RegexpDataPartitionDir validates the directory name of a data partition.
	RegexpDataPartitionDir, _ = regexp.Compile("^datapartition_(\\d)+_(\\d)+$")
)

Functions

func DeleteLimiterWait 

func DeleteLimiterWait()

func GetIoMetricLabels 

func GetIoMetricLabels(partition *DataPartition, tp string) map[string]string

func IsDiskErr 

func IsDiskErr(errMsg string) bool

func MarshalRandWriteRaftLog added in v1.4.0 

func MarshalRandWriteRaftLog(opcode uint8, extentID uint64, offset, size int64, data []byte, crc uint32) (result []byte, err error)

func NewPacketToBroadcastMinAppliedID 

func NewPacketToBroadcastMinAppliedID(partitionID uint64, minAppliedID uint64) (p *repl.Packet)

NewPacketToBroadcastMinAppliedID returns a new packet to broadcast the min applied ID.

func NewPacketToGetAppliedID 

func NewPacketToGetAppliedID(partitionID uint64) (p *repl.Packet)

NewPacketToGetAppliedID returns a new packet to get the applied ID.

func NewPacketToGetMaxExtentIDAndPartitionSIze added in v1.4.0 

func NewPacketToGetMaxExtentIDAndPartitionSIze(partitionID uint64) (p *repl.Packet)

NewPacketToGetPartitionSize returns a new packet to get the partition size.

func NewPacketToGetPartitionSize 

func NewPacketToGetPartitionSize(partitionID uint64) (p *repl.Packet)

NewPacketToGetPartitionSize returns a new packet to get the partition size.

func UnmarshalOldVersionRaftLog added in v1.4.0 

func UnmarshalOldVersionRaftLog(raw []byte) (opItem *rndWrtOpItem, err error)

func UnmarshalOldVersionRandWriteOpItem added in v1.4.0 

func UnmarshalOldVersionRandWriteOpItem(raw []byte) (result *rndWrtOpItem, err error)

func UnmarshalRandWriteRaftLog added in v1.4.0 

func UnmarshalRandWriteRaftLog(raw []byte) (opItem *rndWrtOpItem, err error)

RandomWriteSubmit submits the proposal to raft.

Types

type DataNode 

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

DataNode defines the structure of a data node.

func NewServer 

func NewServer() *DataNode

func (*DataNode) OperatePacket 

func (s *DataNode) OperatePacket(p *repl.Packet, c net.Conn) (err error)

func (*DataNode) Post 

func (s *DataNode) Post(p *repl.Packet) error

func (*DataNode) Prepare 

func (s *DataNode) Prepare(p *repl.Packet) (err error)

func (*DataNode) Shutdown 

func (s *DataNode) Shutdown()

Shutdown shuts down the current data node.

func (*DataNode) Start 

func (s *DataNode) Start(cfg *config.Config) (err error)

func (*DataNode) Sync 

func (s *DataNode) Sync()

Sync keeps data node in sync.

type DataNodeInfo added in v1.4.0 

type DataNodeInfo struct {
	Addr                      string
	PersistenceDataPartitions []uint64
}

type DataNodeMetrics 

type DataNodeMetrics struct {
	MetricIOBytes *exporter.Counter
}

type DataPartition 

type DataPartition struct {
	DataPartitionCreateType int
	// contains filtered or unexported fields
}

func CreateDataPartition 

func CreateDataPartition(dpCfg *dataPartitionCfg, disk *Disk, request *proto.CreateDataPartitionRequest) (dp *DataPartition, err error)

func LoadDataPartition 

func LoadDataPartition(partitionDir string, disk *Disk) (dp *DataPartition, err error)

LoadDataPartition loads and returns a partition instance based on the specified directory. It reads the partition metadata file stored under the specified directory and creates the partition instance.

func (*DataPartition) Apply 

func (dp *DataPartition) Apply(command []byte, index uint64) (resp interface{}, err error)

Apply puts the data onto the disk.

func (*DataPartition) ApplyMemberChange 

func (dp *DataPartition) ApplyMemberChange(confChange *raftproto.ConfChange, index uint64) (resp interface{}, err error)

ApplyMemberChange supports adding new raft member or deleting an existing raft member. It does not support updating an existing member at this point.

func (*DataPartition) ApplyRandomWrite 

func (dp *DataPartition) ApplyRandomWrite(command []byte, raftApplyID uint64) (resp interface{}, err error)

ApplyRandomWrite random write apply

func (*DataPartition) ApplySnapshot 

func (dp *DataPartition) ApplySnapshot(peers []raftproto.Peer, iterator raftproto.SnapIterator) (err error)

ApplySnapshot asks the raft leader for the snapshot data to recover the contents on the local disk.

func (*DataPartition) Available 

func (dp *DataPartition) Available() int

Available returns the available space.

func (*DataPartition) CanRemoveRaftMember added in v1.4.0 

func (dp *DataPartition) CanRemoveRaftMember(peer proto.Peer) error

func (*DataPartition) ChangeRaftMember 

func (dp *DataPartition) ChangeRaftMember(changeType raftProto.ConfChangeType, peer raftProto.Peer, context []byte) (resp interface{}, err error)

ChangeRaftMember is a wrapper function of changing the raft member.

func (*DataPartition) CheckLeader 

func (dp *DataPartition) CheckLeader(request *repl.Packet, connect net.Conn) (err error)

CheckLeader checks if itself is the leader during read

func (*DataPartition) Del 

func (dp *DataPartition) Del(key interface{}) (interface{}, error)

Del deletes the raft log based on the given key. It is not needed for replicating data partition.

func (*DataPartition) Disk 

func (dp *DataPartition) Disk() *Disk

Disk returns the disk instance.

func (*DataPartition) DoExtentStoreRepair 

func (dp *DataPartition) DoExtentStoreRepair(repairTask *DataPartitionRepairTask)

DoExtentStoreRepair performs the repairs of the extent store. 1. when the extent size is smaller than the max size on the record, start to repair the missing part. 2. if the extent does not even exist, create the extent first, and then repair.

func (*DataPartition) DoRepair 

func (dp *DataPartition) DoRepair(repairTasks []*DataPartitionRepairTask)

DoRepair asks the leader to perform the repair tasks.

func (*DataPartition) ExtentStore 

func (dp *DataPartition) ExtentStore() *storage.ExtentStore

func (*DataPartition) ForceLoadHeader 

func (dp *DataPartition) ForceLoadHeader()

func (*DataPartition) ForceSetDataPartitionToFininshLoad 

func (dp *DataPartition) ForceSetDataPartitionToFininshLoad()

func (*DataPartition) ForceSetDataPartitionToLoadding 

func (dp *DataPartition) ForceSetDataPartitionToLoadding()

func (*DataPartition) ForceSetRaftRunning 

func (dp *DataPartition) ForceSetRaftRunning()

func (*DataPartition) Get 

func (dp *DataPartition) Get(key interface{}) (interface{}, error)

Get returns the raft log based on the given key. It is not needed for replicating data partition.

func (*DataPartition) GetAppliedID 

func (dp *DataPartition) GetAppliedID() (id uint64)

func (*DataPartition) GetExtentCount 

func (dp *DataPartition) GetExtentCount() int

func (*DataPartition) HandleFatalEvent 

func (dp *DataPartition) HandleFatalEvent(err *raft.FatalError)

HandleFatalEvent notifies the application when panic happens.

func (*DataPartition) HandleLeaderChange 

func (dp *DataPartition) HandleLeaderChange(leader uint64)

HandleLeaderChange notifies the application when the raft leader has changed.

func (*DataPartition) IsEquareCreateDataPartitionRequst 

func (dp *DataPartition) IsEquareCreateDataPartitionRequst(request *proto.CreateDataPartitionRequest) (err error)

func (*DataPartition) IsExsitReplica added in v1.4.0 

func (dp *DataPartition) IsExsitReplica(addr string) bool

func (*DataPartition) IsRaftLeader 

func (dp *DataPartition) IsRaftLeader() (addr string, ok bool)

IsRaftLeader tells if the given address belongs to the raft leader.

func (*DataPartition) IsRejectWrite added in v1.4.0 

func (dp *DataPartition) IsRejectWrite() bool

func (*DataPartition) LaunchRepair 

func (dp *DataPartition) LaunchRepair(extentType uint8)

LaunchRepair launches the repair of extents.

func (*DataPartition) Load 

func (dp *DataPartition) Load() (response *proto.LoadDataPartitionResponse)

func (*DataPartition) LoadAppliedID 

func (dp *DataPartition) LoadAppliedID() (err error)

LoadAppliedID loads the applied IDs to the memory.

func (*DataPartition) NotifyExtentRepair 

func (dp *DataPartition) NotifyExtentRepair(members []*DataPartitionRepairTask) (err error)

NotifyExtentRepair notifies the followers to repair.

func (*DataPartition) Path 

func (dp *DataPartition) Path() string

func (*DataPartition) PersistMetadata 

func (dp *DataPartition) PersistMetadata() (err error)

PersistMetadata persists the file metadata on the disk.

func (*DataPartition) Put 

func (dp *DataPartition) Put(key interface{}, val interface{}) (resp interface{}, err error)

Put submits the raft log to the raft store.

func (*DataPartition) RandomWriteSubmit 

func (dp *DataPartition) RandomWriteSubmit(pkg *repl.Packet) (err error)

RandomWriteSubmit submits the proposal to raft.

func (*DataPartition) ReloadSnapshot 

func (dp *DataPartition) ReloadSnapshot()

func (*DataPartition) Replicas 

func (dp *DataPartition) Replicas() []string

func (*DataPartition) SetMinAppliedID 

func (dp *DataPartition) SetMinAppliedID(id uint64)

func (*DataPartition) Size 

func (dp *DataPartition) Size() int

Size returns the partition size.

func (*DataPartition) SnapShot 

func (dp *DataPartition) SnapShot() (files []*proto.File)

Snapshot returns the snapshot of the data partition.

func (*DataPartition) Snapshot 

func (dp *DataPartition) Snapshot() (raftproto.Snapshot, error)

Snapshot persists the in-memory data (as a snapshot) to the disk. Note that the data in each data partition has already been saved on the disk. Therefore there is no need to take the snapshot in this case.

func (*DataPartition) StartRaft 

func (dp *DataPartition) StartRaft() (err error)

StartRaft start raft instance when data partition start or restore.

func (*DataPartition) StartRaftAfterRepair 

func (dp *DataPartition) StartRaftAfterRepair()

StartRaftAfterRepair starts the raft after repairing a partition. It can only happens after all the extent files are repaired by the leader. When the repair is finished, the local dp.partitionSize is same as the leader's dp.partitionSize. The repair task can be done in statusUpdateScheduler->LaunchRepair.

func (*DataPartition) StartRaftLoggingSchedule 

func (dp *DataPartition) StartRaftLoggingSchedule()

StartRaftLoggingSchedule starts the task schedule as follows: 1. write the raft applied id into disk. 2. collect the applied ids from raft members. 3. based on the minimum applied id to cutoff and delete the saved raft log in order to free the disk space.

func (*DataPartition) Status 

func (dp *DataPartition) Status() int

Status returns the partition status.

func (*DataPartition) Stop 

func (dp *DataPartition) Stop()

Stop close the store and the raft store.

func (*DataPartition) String 

func (dp *DataPartition) String() (m string)

String returns the string format of the data partition information.

func (*DataPartition) Used 

func (dp *DataPartition) Used() int

Used returns the used space.

type DataPartitionMetadata 

type DataPartitionMetadata struct {
	VolumeID                string
	PartitionID             uint64
	PartitionSize           int
	CreateTime              string
	Peers                   []proto.Peer
	Hosts                   []string
	DataPartitionCreateType int
	LastTruncateID          uint64
}

func (*DataPartitionMetadata) Validate 

func (md *DataPartitionMetadata) Validate() (err error)

type DataPartitionRepairTask 

type DataPartitionRepairTask struct {
	TaskType uint8

	ExtentsToBeCreated             []*storage.ExtentInfo
	ExtentsToBeRepaired            []*storage.ExtentInfo
	LeaderTinyDeleteRecordFileSize int64
	LeaderAddr                     string
	// contains filtered or unexported fields
}

DataPartitionRepairTask defines the reapir task for the data partition.

func NewDataPartitionRepairTask 

func NewDataPartitionRepairTask(extentFiles []*storage.ExtentInfo, tinyDeleteRecordFileSize int64, source, leaderAddr string) (task *DataPartitionRepairTask)

type Disk 

type Disk struct {
	sync.RWMutex
	Path        string
	ReadErrCnt  uint64 // number of read errors
	WriteErrCnt uint64 // number of write errors

	Total       uint64
	Used        uint64
	Available   uint64
	Unallocated uint64
	Allocated   uint64

	MaxErrCnt     int // maximum number of errors
	Status        int // disk status such as READONLY
	ReservedSpace uint64

	RejectWrite bool
	// contains filtered or unexported fields
}

Disk represents the structure of the disk

func NewDisk 

func NewDisk(path string, reservedSpace uint64, maxErrCnt int, space *SpaceManager) (d *Disk)

func (*Disk) AddSize added in v1.4.0 

func (d *Disk) AddSize(size uint64)

func (*Disk) AttachDataPartition 

func (d *Disk) AttachDataPartition(dp *DataPartition)

AttachDataPartition adds a data partition to the partition map.

func (*Disk) DataPartitionList 

func (d *Disk) DataPartitionList() (partitionIDs []uint64)

DataPartitionList returns a list of the data partitions

func (*Disk) DetachDataPartition 

func (d *Disk) DetachDataPartition(dp *DataPartition)

DetachDataPartition removes a data partition from the partition map.

func (*Disk) ForceExitRaftStore added in v1.1.1 

func (d *Disk) ForceExitRaftStore()

func (*Disk) GetDataPartition 

func (d *Disk) GetDataPartition(partitionID uint64) (partition *DataPartition)

GetDataPartition returns the data partition based on the given partition ID.

func (*Disk) PartitionCount 

func (d *Disk) PartitionCount() int

PartitionCount returns the number of partitions in the partition map.

func (*Disk) RestorePartition 

func (d *Disk) RestorePartition(visitor PartitionVisitor)

RestorePartition reads the files stored on the local disk and restores the data partitions.

type ItemIterator 

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

func NewItemIterator 

func NewItemIterator(applyID uint64) *ItemIterator

NewItemIterator creates a new item iterator.

func (*ItemIterator) ApplyIndex 

func (si *ItemIterator) ApplyIndex() uint64

ApplyIndex returns the appliedID

func (*ItemIterator) Close 

func (si *ItemIterator) Close()

Close Closes the iterator.

func (*ItemIterator) Next 

func (si *ItemIterator) Next() (data []byte, err error)

Next returns the next item in the iterator.

type PartitionVisitor 

type PartitionVisitor func(dp *DataPartition)

type RaftCmdItem 

type RaftCmdItem struct {
	Op uint32 `json:"op"`
	K  []byte `json:"k"`
	V  []byte `json:"v"`
}

type SpaceManager 

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

SpaceManager manages the disk space.

func NewSpaceManager 

func NewSpaceManager(dataNode *DataNode) *SpaceManager

NewSpaceManager creates a new space manager.

func (*SpaceManager) AttachPartition added in v1.4.0 

func (manager *SpaceManager) AttachPartition(dp *DataPartition)

func (*SpaceManager) CreatePartition 

func (manager *SpaceManager) CreatePartition(request *proto.CreateDataPartitionRequest) (dp *DataPartition, err error)

func (*SpaceManager) DeletePartition 

func (manager *SpaceManager) DeletePartition(dpID uint64)

DeletePartition deletes a partition based on the partition id.

func (*SpaceManager) DetachDataPartition added in v1.4.0 

func (manager *SpaceManager) DetachDataPartition(partitionID uint64)

DetachDataPartition removes a data partition from the partition map.

func (*SpaceManager) GetClusterID 

func (manager *SpaceManager) GetClusterID() (clusterID string)

func (*SpaceManager) GetDisk 

func (manager *SpaceManager) GetDisk(path string) (d *Disk, err error)

func (*SpaceManager) GetDisks 

func (manager *SpaceManager) GetDisks() (disks []*Disk)

func (*SpaceManager) GetNodeID 

func (manager *SpaceManager) GetNodeID() (nodeID uint64)

func (*SpaceManager) GetRaftStore 

func (manager *SpaceManager) GetRaftStore() (raftStore raftstore.RaftStore)

func (*SpaceManager) LoadDisk 

func (manager *SpaceManager) LoadDisk(path string, reservedSpace uint64, maxErrCnt int) (err error)

func (*SpaceManager) Partition 

func (manager *SpaceManager) Partition(partitionID uint64) (dp *DataPartition)

func (*SpaceManager) RangePartitions 

func (manager *SpaceManager) RangePartitions(f func(partition *DataPartition) bool)

func (*SpaceManager) SetClusterID 

func (manager *SpaceManager) SetClusterID(clusterID string)

func (*SpaceManager) SetNodeID 

func (manager *SpaceManager) SetNodeID(nodeID uint64)

func (*SpaceManager) SetRaftStore 

func (manager *SpaceManager) SetRaftStore(raftStore raftstore.RaftStore)

func (*SpaceManager) Stats 

func (manager *SpaceManager) Stats() *Stats

func (*SpaceManager) Stop 

func (manager *SpaceManager) Stop()

type Stats 

type Stats struct {
	Zone                               string
	ConnectionCnt                      int64
	ClusterID                          string
	TCPAddr                            string
	Start                              time.Time
	Total                              uint64
	Used                               uint64
	Available                          uint64 // available space
	TotalPartitionSize                 uint64 // dataPartitionCnt * dataPartitionSize
	RemainingCapacityToCreatePartition uint64
	CreatedPartitionCnt                uint64

	// the maximum capacity among all the nodes that can be used to create partition
	MaxCapacityToCreatePartition uint64

	sync.Mutex
	// contains filtered or unexported fields
}

Stats defines various metrics that will be collected during the execution.

func NewStats 

func NewStats(zone string) (s *Stats)

NewStats creates a new Stats.

func (*Stats) AddConnection 

func (s *Stats) AddConnection()

AddConnection adds a connection.

func (*Stats) GetConnectionCount 

func (s *Stats) GetConnectionCount() int64

GetConnectionCount gets the connection count.

func (*Stats) RemoveConnection 

func (s *Stats) RemoveConnection()

RemoveConnection removes a connection.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.