client

Documentation

Index

• Constants
• Variables
• type HashSelector
  • func (h *HashSelector) Next(binlog *pb.Binlog, retryTime int) *PumpStatus
  • func (h *HashSelector) Select(binlog *pb.Binlog) *PumpStatus
  • func (h *HashSelector) SetPumps(pumps []*PumpStatus)
• type PumpInfos
• type PumpSelector
  • func NewHashSelector() PumpSelector
  • func NewRangeSelector() PumpSelector
  • func NewScoreSelector() PumpSelector
  • func NewSelector(algorithm string) PumpSelector
• type PumpStatus
  • func NewPumpStatus(status *node.Status, security *tls.Config) *PumpStatus
• type PumpsClient
  • func NewPumpsClient(etcdURLs string, timeout time.Duration, securityOpt pd.SecurityOption) (*PumpsClient, error)
  • func (c *PumpsClient) Close()
  • func (c *PumpsClient) WriteBinlog(binlog *pb.Binlog) error
• type RangeSelector
  • func (r *RangeSelector) Next(binlog *pb.Binlog, retryTime int) *PumpStatus
  • func (r *RangeSelector) Select(binlog *pb.Binlog) *PumpStatus
  • func (r *RangeSelector) SetPumps(pumps []*PumpStatus)
• type ScoreSelector
  • func (s *ScoreSelector) Next(binlog *pb.Binlog, retryTime int) *PumpStatus
  • func (s *ScoreSelector) Select(binlog *pb.Binlog) *PumpStatus
  • func (s *ScoreSelector) SetPumps(pumps []*PumpStatus)

Constants

const (
	// DefaultEtcdTimeout is the default timeout config for etcd.
	DefaultEtcdTimeout = 5 * time.Second

	// DefaultRetryTime is the default time of retry.
	DefaultRetryTime = 20

	// DefaultBinlogWriteTimeout is the default max time binlog can use to write to pump.
	DefaultBinlogWriteTimeout = 15 * time.Second

	// CheckInterval is the default interval for check unavaliable pumps.
	CheckInterval = 30 * time.Second

	// RetryInterval is the default interval of retrying to write binlog.
	RetryInterval = 100 * time.Millisecond
)

const (
	// Range means range algorithm.
	Range = "range"

	// Hash means hash algorithm.
	Hash = "hash"

	// Score means choose pump by it's score.
	Score = "score"
)

Variables

var (
	// Logger is ...
	Logger = log.New()

	// ErrNoAvaliablePump means no avaliable pump to write binlog.
	ErrNoAvaliablePump = errors.New("no avaliable pump to write binlog")
)

Types

type HashSelector

type HashSelector struct {
	sync.RWMutex

	// TsMap saves the map of start_ts with pump when send prepare binlog.
	// And Commit binlog should send to the same pump.
	TsMap map[int64]*PumpStatus

	// PumpMap saves the map of pump's node id with pump.
	PumpMap map[string]*PumpStatus

	// the pumps to be selected.
	Pumps []*PumpStatus
}

HashSelector select a pump by hash.

func (*HashSelector) Next

func (h *HashSelector) Next(binlog *pb.Binlog, retryTime int) *PumpStatus

Next implement PumpSelector.Next. Only for Prewrite binlog.

func (*HashSelector) Select

func (h *HashSelector) Select(binlog *pb.Binlog) *PumpStatus

Select implement PumpSelector.Select.

func (*HashSelector) SetPumps

func (h *HashSelector) SetPumps(pumps []*PumpStatus)

SetPumps implement PumpSelector.SetPumps.

type PumpInfos

type PumpInfos struct {
	sync.RWMutex
	// Pumps saves the map of pump's nodeID and pump status.
	Pumps map[string]*PumpStatus

	// AvliablePumps saves the whole avaliable pumps' status.
	AvaliablePumps map[string]*PumpStatus

	// UnAvaliablePumps saves the unAvaliable pumps.
	// And only pump with Online state in this map need check is it avaliable.
	UnAvaliablePumps map[string]*PumpStatus
}

PumpInfos saves pumps' infomations in pumps client.

type PumpSelector

type PumpSelector interface {
	// SetPumps set pumps to be selected.
	SetPumps([]*PumpStatus)

	// Select returns a situable pump.
	Select(*pb.Binlog) *PumpStatus

	// returns the next pump.
	Next(*pb.Binlog, int) *PumpStatus
}

PumpSelector selects pump for sending binlog.

func NewHashSelector

func NewHashSelector() PumpSelector

NewHashSelector returns a new HashSelector.

func NewRangeSelector

func NewRangeSelector() PumpSelector

NewRangeSelector returns a new ScoreSelector.

func NewScoreSelector

func NewScoreSelector() PumpSelector

NewScoreSelector returns a new ScoreSelector.

func NewSelector

func NewSelector(algorithm string) PumpSelector

NewSelector returns a PumpSelector according to the algorithm.

type PumpStatus

type PumpStatus struct {
	/*
		Pump has these state:
		Online:
			only when pump's state is online that pumps client can write binlog to.
		Pausing:
			this pump is pausing, and can't provide write binlog service. And this state will turn into Paused when pump is quit.
		Paused:
			this pump is paused, and can't provide write binlog service.
		Closing:
			this pump is closing, and can't provide write binlog service. And this state will turn into Offline when pump is quit.
		Offline:
			this pump is offline, and can't provide write binlog service forever.
	*/
	node.Status

	// the pump is avaliable or not
	IsAvaliable bool

	// the client of this pump
	Client pb.PumpClient
	// contains filtered or unexported fields
}

PumpStatus saves pump's status.

func NewPumpStatus

func NewPumpStatus(status *node.Status, security *tls.Config) *PumpStatus

NewPumpStatus returns a new PumpStatus according to node's status.

type PumpsClient

type PumpsClient struct {

	// ClusterID is the cluster ID of this tidb cluster.
	ClusterID uint64

	// the registry of etcd.
	EtcdRegistry *node.EtcdRegistry

	// Pumps saves the pumps' information.
	Pumps *PumpInfos

	// Selector will select a suitable pump.
	Selector PumpSelector

	// the max retry time if write binlog failed.
	RetryTime int

	// BinlogWriteTimeout is the max time binlog can use to write to pump.
	BinlogWriteTimeout time.Duration

	// Security is the security config
	Security *tls.Config
	// contains filtered or unexported fields
}

PumpsClient is the client of pumps.

func NewPumpsClient

func NewPumpsClient(etcdURLs string, timeout time.Duration, securityOpt pd.SecurityOption) (*PumpsClient, error)

NewPumpsClient returns a PumpsClient.

func (*PumpsClient) Close

func (c *PumpsClient) Close()

Close closes the PumpsClient.

func (*PumpsClient) WriteBinlog

func (c *PumpsClient) WriteBinlog(binlog *pb.Binlog) error

WriteBinlog writes binlog to a situable pump.

type RangeSelector

type RangeSelector struct {
	sync.RWMutex

	// Offset saves the offset in Pumps.
	Offset int

	// TsMap saves the map of start_ts with pump when send prepare binlog.
	// And Commit binlog should send to the same pump.
	TsMap map[int64]*PumpStatus

	// PumpMap saves the map of pump's node id with pump.
	PumpMap map[string]*PumpStatus

	// the pumps to be selected.
	Pumps []*PumpStatus
}

RangeSelector select a pump by range.

func (*RangeSelector) Next

func (r *RangeSelector) Next(binlog *pb.Binlog, retryTime int) *PumpStatus

Next implement PumpSelector.Next. Only for Prewrite binlog.

func (*RangeSelector) Select

func (r *RangeSelector) Select(binlog *pb.Binlog) *PumpStatus

Select implement PumpSelector.Select.

func (*RangeSelector) SetPumps

func (r *RangeSelector) SetPumps(pumps []*PumpStatus)

SetPumps implement PumpSelector.SetPumps.

type ScoreSelector

type ScoreSelector struct{}

ScoreSelector select a pump by pump's score.

func (*ScoreSelector) Next

func (s *ScoreSelector) Next(binlog *pb.Binlog, retryTime int) *PumpStatus

Next implement PumpSelector.Next. Only for Prewrite binlog.

func (*ScoreSelector) Select

func (s *ScoreSelector) Select(binlog *pb.Binlog) *PumpStatus

Select implement PumpSelector.Select.

func (*ScoreSelector) SetPumps

func (s *ScoreSelector) SetPumps(pumps []*PumpStatus)

SetPumps implement PumpSelector.SetPumps.