server

package
v0.0.0-...-13d8ce4 Latest Latest
Warning

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

 Go to latest Published: Apr 24, 2019 License: Apache-2.0 Imports: 37 Imported by: 3

Documentation

Index

Constants

View Source 
const (

	// HealthInterval is the minimum time the cluster should be healthy
	// before accepting add member requests.
	HealthInterval = 5 * time.Second
	//DialTimeout is the timeout of dial
	DialTimeout = 5 * time.Second
	//ProposeMaxRetryCount is the max count of propose retry times
	ProposeMaxRetryCount = 1000
)

Variables

View Source 
var (
	//ErrStopped return for kingbus server stopped
	ErrStopped = errors.New("kingbus: server stopped")
	//ErrStarted return for kingbus server started
	ErrStarted = errors.New("kingbus: server already started")
	//ErrCanceled return for request be cancelled
	ErrCanceled = errors.New("kingbus: request cancelled")
	//ErrTimeout return for request timed out
	ErrTimeout = errors.New("kingbus: request timed out")
	//ErrTimeoutDueToLeaderFail return for request timed out,possibly due to previous leader failure
	ErrTimeoutDueToLeaderFail = errors.New("kingbus: request timed out, possibly due to previous leader failure")
	//ErrTimeoutDueToConnectionLost return for request timed out,possibly due to connection lost
	ErrTimeoutDueToConnectionLost = errors.New("kingbus: request timed out, possibly due to connection lost")
	//ErrNotEnoughStartedMembers return for re-configuration failed due to not enough started members
	ErrNotEnoughStartedMembers = errors.New("kingbus: re-configuration failed due to not enough started members")
	//ErrRequestTooLarge return for request is too large
	ErrRequestTooLarge = errors.New("kingbus: request is too large")
	//ErrNoSpace return for no space
	ErrNoSpace = errors.New("kingbus: no space")
	//ErrTooManyRequests return for too many requests
	ErrTooManyRequests = errors.New("kingbus: too many requests")
	//ErrUnhealthy return for unhealthy cluster
	ErrUnhealthy = errors.New("kingbus: unhealthy cluster")
	//ErrKeyNotFound return for key not found
	ErrKeyNotFound = errors.New("kingbus: key not found")
	//ErrCorrupt return for corrupt cluster
	ErrCorrupt = errors.New("kingbus: corrupt cluster")
	//ErrUnsupport return for apply unsupport event type
	ErrUnsupport = errors.New("apply unsupport event type")
	//ErrArgs return for args are not available
	ErrArgs = errors.New("binlog_server:args are not available")
	//ErrNotContain return for slave gtidset not contain in master
	ErrNotContain = errors.New("binlog_server:slave gtidset not contain in master")
	//ErrUUIDIsNull return for the uuid of server is null
	ErrUUIDIsNull = errors.New("binlog_server:the uuid of server is null")
)

Functions

This section is empty.

Types

type BinlogProgress 

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

BinlogProgress is the progress of receiving binlog

func (*BinlogProgress) CurrentGtidStr 

func (s *BinlogProgress) CurrentGtidStr() string

CurrentGtidStr get the current gtid

func (*BinlogProgress) ExecutedGtidSetClone 

func (s *BinlogProgress) ExecutedGtidSetClone() gomysql.GTIDSet

ExecutedGtidSetClone geth the executed gtid set clone, deep copy

func (*BinlogProgress) ExecutedGtidSetStr 

func (s *BinlogProgress) ExecutedGtidSetStr() string

ExecutedGtidSetStr get the executed gtid set

func (*BinlogProgress) LastBinlogFile 

func (s *BinlogProgress) LastBinlogFile() string

LastBinlogFile get the last file of binlog event

func (*BinlogProgress) LastFilePosition 

func (s *BinlogProgress) LastFilePosition() uint32

LastFilePosition get the last file position of binlog event

type BinlogServer 

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

BinlogServer is a binlog server,send binlog event to slave. The generic process: 1.authentication SHOW GLOBAL VARIABLES LIKE 'BINLOG_CHECKSUM' SET @master_binlog_checksum='NONE' SET @master_heartbeat_period=%d 2.COM_REGISTER_SLAVE 3.semi-sync: SHOW VARIABLES LIKE 'rpl_semi_sync_master_enabled'; SET @rpl_semi_sync_slave = 1 4.COM_BINLOG_DUMP_GTID

func NewBinlogServer 

func NewBinlogServer(cfg *config.BinlogServerConfig, ki KingbusInfo, store storage.Storage, broadcast *utils.Broadcast) (*BinlogServer, error)

NewBinlogServer create a binlog server

func (*BinlogServer) CheckGtidSet 

func (s *BinlogServer) CheckGtidSet(flavor string, slaveExecutedGtidSet gomysql.GTIDSet) error

CheckGtidSet the illegal of the slave executed gtid

func (*BinlogServer) DumpBinlogAt 

func (s *BinlogServer) DumpBinlogAt(ctx context.Context,
	startRaftIndex uint64, slaveGtids *gomysql.MysqlGTIDSet,
	eventC chan<- *storagepb.BinlogEvent, errorC chan<- error) error

DumpBinlogAt implements dump binlog event by slave executed gtid set

func (*BinlogServer) GetFde 

func (s *BinlogServer) GetFde(preGtidEventIndex uint64) ([]byte, error)

GetFde get the FORMAT_DESCRIPTION_EVENT by previous gtids log event raft index

func (*BinlogServer) GetGtidSet 

func (s *BinlogServer) GetGtidSet(flavor string, key string) (gomysql.GTIDSet, error)

GetGtidSet get gtid set

func (*BinlogServer) GetMasterInfo 

func (s *BinlogServer) GetMasterInfo() (*mysql.MasterInfo, error)

GetMasterInfo get the master information connected by syncer

func (*BinlogServer) GetMySQLDumpAt 

func (s *BinlogServer) GetMySQLDumpAt(slaveExecutedGtids *gomysql.MysqlGTIDSet) (uint64, error)

GetMySQLDumpAt return raft index, dump binlog event from this position TODO if gtids is empty set, and previous_gtids also is empty, ok!

func (*BinlogServer) GetNextBinlogFile 

func (s *BinlogServer) GetNextBinlogFile(startRaftIndex uint64) (string, error)

GetNextBinlogFile get next binlog file by raft index

func (*BinlogServer) GetSlaves 

func (s *BinlogServer) GetSlaves() map[string]*mysql.Slave

GetSlaves get all slaves connected the binlog server

func (*BinlogServer) LastBinlogFile 

func (s *BinlogServer) LastBinlogFile() string

LastBinlogFile return the last binlog file

func (*BinlogServer) LastFilePosition 

func (s *BinlogServer) LastFilePosition() uint32

LastFilePosition return the last binlog file position

func (*BinlogServer) RegisterSlave 

func (s *BinlogServer) RegisterSlave(slave *mysql.Slave) error

RegisterSlave implements register slave into binlog server

func (*BinlogServer) Start 

func (s *BinlogServer) Start()

Start implements binlog server start

func (*BinlogServer) Stop 

func (s *BinlogServer) Stop()

Stop implements binlog server stop

func (*BinlogServer) UnregisterSlave 

func (s *BinlogServer) UnregisterSlave(uuid string)

UnregisterSlave unregister slave by uuid

type KingbusInfo 

type KingbusInfo interface {
	AppliedIndex() uint64
	LastBinlogFile() string
	LastFilePosition() uint32
	ExecutedGtidSetStr() string
}

KingbusInfo get the kingbus server information

type KingbusServer 

type KingbusServer struct {
	Cfg *config.KingbusServerConfig

	//lock for leadElectedTime
	Mu sync.RWMutex
	// contains filtered or unexported fields
}

KingbusServer is a instance run all sub servers

func NewKingbusServer 

func NewKingbusServer(cfg *config.KingbusServerConfig) (*KingbusServer, error)

NewKingbusServer create a kingbus server

func (*KingbusServer) AddMember 

func (s *KingbusServer) AddMember(ctx context.Context, memb membership.Member) ([]*membership.Member, error)

AddMember member into raft cluster, only executed in lead node

func (*KingbusServer) AppliedIndex 

func (s *KingbusServer) AppliedIndex() uint64

AppliedIndex get applied index

func (*KingbusServer) CommittedIndex 

func (s *KingbusServer) CommittedIndex() uint64

CommittedIndex get committed index

func (*KingbusServer) CurrentGtidStr 

func (s *KingbusServer) CurrentGtidStr() string

CurrentGtidStr return current gtid

func (*KingbusServer) ExecutedGtidSetStr 

func (s *KingbusServer) ExecutedGtidSetStr() string

ExecutedGtidSetStr return executed gtid

func (*KingbusServer) GetIP 

func (s *KingbusServer) GetIP() string

GetIP get ip of kingbus server

func (*KingbusServer) GetServerStatus 

func (s *KingbusServer) GetServerStatus(svrType config.SubServerType) interface{}

GetServerStatus get the sub server status

func (*KingbusServer) ID 

func (s *KingbusServer) ID() types.ID

ID get kingbus server id

func (*KingbusServer) IsBinlogServerStarted 

func (s *KingbusServer) IsBinlogServerStarted() bool

IsBinlogServerStarted return if binlog server started

func (*KingbusServer) IsIDRemoved 

func (s *KingbusServer) IsIDRemoved(id uint64) bool

IsIDRemoved return if the kingbus has been removed

func (*KingbusServer) IsLeader 

func (s *KingbusServer) IsLeader() bool

IsLeader return the node is lead

func (*KingbusServer) IsSyncerStarted 

func (s *KingbusServer) IsSyncerStarted() bool

IsSyncerStarted return if syncer started

func (*KingbusServer) LastBinlogFile 

func (s *KingbusServer) LastBinlogFile() string

LastBinlogFile return last binlog file

func (*KingbusServer) LastFilePosition 

func (s *KingbusServer) LastFilePosition() uint32

LastFilePosition return last binlog file position

func (*KingbusServer) Leader 

func (s *KingbusServer) Leader() types.ID

Leader get raft cluster leader

func (*KingbusServer) Process 

func (s *KingbusServer) Process(ctx context.Context, m raftpb.Message) error

Process takes a raft message and applies it to the server's raft state machine, respecting any timeout of the given context.

func (*KingbusServer) Propose 

func (s *KingbusServer) Propose(data []byte) error

Propose data,only execute in lead node,follower node will be forbidden

func (*KingbusServer) ProposeWithRetry 

func (s *KingbusServer) ProposeWithRetry(ctx context.Context, data []byte)

ProposeWithRetry implements propose data with retry

func (*KingbusServer) RemoveMember 

func (s *KingbusServer) RemoveMember(ctx context.Context, id uint64) ([]*membership.Member, error)

RemoveMember member from raft cluster, only executed in lead node

func (*KingbusServer) ReportSnapshot 

func (s *KingbusServer) ReportSnapshot(id uint64, status etcdraft.SnapshotStatus)

ReportSnapshot reports snapshot sent status to the raft state machine, and clears the used snapshot from the snapshot store.

func (*KingbusServer) ReportUnreachable 

func (s *KingbusServer) ReportUnreachable(id uint64)

ReportUnreachable report unreachable

func (*KingbusServer) Run 

func (s *KingbusServer) Run()

Run kingbus server

func (*KingbusServer) StartProposeBinlog 

func (s *KingbusServer) StartProposeBinlog(ctx context.Context)

StartProposeBinlog start propose binlog event

func (*KingbusServer) StartServer 

func (s *KingbusServer) StartServer(svrType config.SubServerType, args interface{}) error

StartServer start sub servers:syncer server or binlog master server

func (*KingbusServer) Stop 

func (s *KingbusServer) Stop()

Stop kingbus server

func (*KingbusServer) StopServer 

func (s *KingbusServer) StopServer(svrType config.SubServerType)

StopServer stop sub server

func (*KingbusServer) SyncAdminURL 

func (s *KingbusServer) SyncAdminURL()

SyncAdminURL sync the admin url between raft cluster since only propose data in lead node, other followers should wait apply the lead admin url,then send http api request to lead admin url.

func (*KingbusServer) Term 

func (s *KingbusServer) Term() uint64

Term get raft term

func (*KingbusServer) UpdateMember 

func (s *KingbusServer) UpdateMember(ctx context.Context, memb membership.Member) ([]*membership.Member, error)

UpdateMember member in raft cluster, only executed in lead node

type PrometheusServer 

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

PrometheusServer provides a container with config parameters for the Prometheus Exporter

func NewPrometheusServer 

func NewPrometheusServer(addr string, r metrics.Registry,
	promRegistry prometheus.Registerer, FlushInterval time.Duration) *PrometheusServer

NewPrometheusServer returns a Provider that produces Prometheus metrics. Namespace and subsystem are applied to all produced metrics.

func (*PrometheusServer) Run 

func (c *PrometheusServer) Run()

Run prometheus server

func (*PrometheusServer) Stop 

func (c *PrometheusServer) Stop()

Stop prometheus server

type Syncer 

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

Syncer is a mock mysql slave, and receive binlog from master and propose event into raft cluster

func NewSyncer 

func NewSyncer(cfg *config.SyncerConfig, store storage.Storage) (*Syncer, error)

NewSyncer create syncer in lead node, and init its gtid set

func (*Syncer) BinlogEventC 

func (s *Syncer) BinlogEventC() chan *storagepb.BinlogEvent

BinlogEventC get binlogEvent channel

func (*Syncer) DataC 

func (s *Syncer) DataC() chan []byte

DataC get data channel

func (*Syncer) Execute 

func (s *Syncer) Execute(cmd string, args ...interface{}) (rr *gomysql.Result, err error)

Execute a SQL in Master

func (*Syncer) GetMasterInfo 

func (s *Syncer) GetMasterInfo() error

GetMasterInfo get master info

func (*Syncer) Start 

func (s *Syncer) Start(gset gomysql.GTIDSet) error

Start syncer

func (*Syncer) Stop 

func (s *Syncer) Stop()

Stop syncer

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.