mysql

Documentation

Index

• Constants
• Variables
• type BaseConn
  • func NewBaseConn(conn net.Conn) *BaseConn
  • func (c *BaseConn) Close() error
  • func (c *BaseConn) ReadPacket() ([]byte, error)
  • func (c *BaseConn) ReadPacketTo(w io.Writer) error
  • func (c *BaseConn) ResetSequence()
  • func (c *BaseConn) WriteEvent(data []byte, first bool) error
  • func (c *BaseConn) WritePacket(data []byte) error
• type BinlogServer
• type Conn
  • func NewConn(conn net.Conn, s BinlogServer, user string, password string) (*Conn, error)
  • func (c *Conn) Close()
  • func (c *Conn) Run()
• type EventBoundaryType
• type MasterInfo
  • func (m *MasterInfo) EncodeMasterInfo() ([]byte, error)
• type ReplicationState
• type Slave
  • func (s *Slave) Close()
• type Stats
• type TransactionBoundaryParser
  • func (p *TransactionBoundaryParser) GetEventBoundaryType(h *replication.EventHeader, eventBody []byte) (EventBoundaryType, error)
  • func (p *TransactionBoundaryParser) IsError() bool
  • func (p *TransactionBoundaryParser) IsInsideTransaction() bool
  • func (p *TransactionBoundaryParser) IsNotInsideTransaction() bool
  • func (p *TransactionBoundaryParser) Reset()
  • func (p *TransactionBoundaryParser) UpdateState(eventBoundaryType EventBoundaryType) error

Constants

const (
	//MaxPayloadLen is the max size of one packet in mysql
	MaxPayloadLen int = 1<<24 - 1
	//OKHeaderByte is is the flag for OK packet header
	OKHeaderByte = 1
)

const (
	//ServerID is the variable name of server id in mysql
	ServerID = "server_id"
	//ServerUUID is the variable name of server uuid in mysql
	ServerUUID = "server_uuid"
	//SlaveUUID is the variable name of slave uuid in mysql
	SlaveUUID = "slave_uuid"
	//MasterHeartbeatPeriod is the variable name of master_heartbeat_period in mysql
	MasterHeartbeatPeriod = "master_heartbeat_period"
	//GtidMode is the variable name of gtid_mode in mysql
	GtidMode = "gtid_mode"
	//GtidPurged is the variable name of gtid_purged in mysql
	GtidPurged = "gtid_purged"
	//VersionComment is the variable name of version_comment in mysql
	VersionComment = "version_comment"
	//BinlogFormat is the variable name of binlog_format in mysql
	BinlogFormat = "binlog_format"
	//BinlogChecksum is the variable name of binlog_checksum in mysql
	BinlogChecksum = "binlog_checksum"
	//MasterBinlogChecksum is the variable name of master_binlog_checksum in mysql
	MasterBinlogChecksum = "master_binlog_checksum"
	//SetNames using in the replication interaction process
	SetNames = "setnames"
	//UnixTimestamp is the variable name of unix_timestamp in mysql
	UnixTimestamp = "unix_timestamp"
	//Version is the variable name of version in mysql
	Version = "version"
)

const (
	// EventParserNone is set after DDL-3 or DML-4
	EventParserNone eventParserState = iota
	// EventParserGtid is set after DDL-1 or DML-1
	EventParserGtid
	// EventParserDDL is set after DDL-2
	EventParserDDL
	// EventParserDML is set after DML-2
	EventParserDML
	// EventParserError is set whenever the above pattern is not followed
	EventParserError
)

Variables

var (
	//ErrSQLNotSupport returns for sql is not support
	ErrSQLNotSupport = errors.New("mysql:sql not support")
	//ErrBadConn returns for connection was bad
	ErrBadConn = errors.New("mysql:connection was bad")
	//ErrLargePacket returns for the the packet is large than 16MB
	ErrLargePacket = errors.New("mysql:the packet is large than 16MB")
	//ErrChecksum returns for master_binlog_checksum is not crc32
	ErrChecksum = errors.New("mysql:master_binlog_checksum must be crc32")
	//ErrUpdateState returns for transaction boundary parser update state error
	ErrUpdateState = errors.New("mysql:transaction boundary parser update state error")
)

var MaxHeartbeatPeriod = time.Hour * 24 * 30 * 12 * 10

MaxHeartbeatPeriod is ten years

Types

type BaseConn

type BaseConn struct {
	net.Conn

	Sequence uint8
	// contains filtered or unexported fields
}

BaseConn is the base class to handle MySQL protocol.

func NewBaseConn

func NewBaseConn(conn net.Conn) *BaseConn

NewBaseConn implements create a BaseConn

func (*BaseConn) Close

func (c *BaseConn) Close() error

Close the BaseConn

func (*BaseConn) ReadPacket

func (c *BaseConn) ReadPacket() ([]byte, error)

ReadPacket implements read data from connection into buffer

func (*BaseConn) ReadPacketTo

func (c *BaseConn) ReadPacketTo(w io.Writer) error

ReadPacketTo implements read data from connection into writer

func (*BaseConn) ResetSequence

func (c *BaseConn) ResetSequence()

ResetSequence implements reset the sequence of BaseConn

func (*BaseConn) WriteEvent

func (c *BaseConn) WriteEvent(data []byte, first bool) error

WriteEvent implements encode data into a binlog event,then write by the connection. the syncer will divide the size large than 16MB packet WriteEvent don't need dividing packet

func (*BaseConn) WritePacket

func (c *BaseConn) WritePacket(data []byte) error

WritePacket implements encode data into a mysql packet, then write the packet by connection

type BinlogServer

type BinlogServer interface {
	RegisterSlave(slave *Slave) error
	UnregisterSlave(uuid string)
	GetSlaves() map[string]*Slave

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

	CheckGtidSet(flavor string, slaveExecutedGtidSet gomysql.GTIDSet) error
	GetMySQLDumpAt(slaveExecutedGtids *gomysql.MysqlGTIDSet) (uint64, error)
	GetFde(raftIndex uint64) ([]byte, error)
	GetNextBinlogFile(startRaftIndex uint64) (string, error)
	GetMasterInfo() (*MasterInfo, error)
	GetGtidSet(flavor string, key string) (gomysql.GTIDSet, error)

	LastBinlogFile() string
	LastFilePosition() uint32
}

BinlogServer implements binlog master functions used in replication

type Conn

type Conn struct {
	*BaseConn
	// contains filtered or unexported fields
}

Conn acts like a MySQL server connection, you can use MySQL client to communicate with it.

func NewConn

func NewConn(conn net.Conn, s BinlogServer, user string, password string) (*Conn, error)

NewConn create a Conn

func (*Conn) Close

func (c *Conn) Close()

Close the Conn

func (*Conn) Run

func (c *Conn) Run()

Run implements handle client request in Conn

type EventBoundaryType

type EventBoundaryType int8

EventBoundaryType is the type of binlog event boundary

const (
	//EventBoundaryTypeError represents error
	EventBoundaryTypeError EventBoundaryType = iota

	//EventBoundaryTypeGtid represents Gtid_log_event
	EventBoundaryTypeGtid

	//EventBoundaryTypeBeginTrx represents Query_log_event(BEGIN) or Query_log_event(XA START)
	EventBoundaryTypeBeginTrx

	//EventBoundaryTypeEndTrx represents Xid, Query_log_event(COMMIT), Query_log_event(ROLLBACK), XA_Prepare_log_event
	EventBoundaryTypeEndTrx

	//EventBoundaryTypeEndXaTrx represents Query_log_event(XA ROLLBACK)
	EventBoundaryTypeEndXaTrx

	//EventBoundaryTypePreStatement represents User_var, Intvar and Rand
	EventBoundaryTypePreStatement

	//EventBoundaryTypeStatement represents All other Query_log_events and all other DML events
	// (Rows, Load_data, etc.)
	EventBoundaryTypeStatement

	//EventBoundaryTypeIncident represents Incident
	EventBoundaryTypeIncident

	//EventBoundaryTypeIgnore represents All non DDL/DML events: Format_desc, Rotate,
	//Previous_gtids, Stop, etc.
	EventBoundaryTypeIgnore
)

type MasterInfo

type MasterInfo struct {
	ServerID   int32  `json:"server_id"`
	ServerUUID string `json:"server_uuid"`
	Version    string `json:"version"`
}

MasterInfo is the master information of syncer connected

func (*MasterInfo) EncodeMasterInfo

func (m *MasterInfo) EncodeMasterInfo() ([]byte, error)

EncodeMasterInfo implements encode MasterInfo into byte

type ReplicationState

type ReplicationState int8

ReplicationState is the slave state

const (
	//UNREGISTERED represents slave has not been register
	UNREGISTERED ReplicationState = iota
	//REGISTERED represents slave has been register
	REGISTERED
)

type Slave

type Slave struct {
	State ReplicationState `json:"replication_state"`

	ServerID int32  `json:"server_id"`
	UUID     string `json:"server_uuid"`
	HostName string `json:"host_name"`
	Port     int16  `json:"port"`
	Rank     uint32 `json:"-"`
	MasterID uint32 `json:"-"`

	User     string `json:"user"`
	Password string `json:"password"`

	ConnectTime time.Time     `json:"connect_time"`
	HeartBeat   time.Duration `json:"heartbeat"`

	Conn  *Conn `json:"-"`
	Stats `json:"stats"`
}

Slave replicating binlogs from kingbus

func (*Slave) Close

func (s *Slave) Close()

Close slave

type Stats

type Stats struct {
}

Stats is slave stats

type TransactionBoundaryParser

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

TransactionBoundaryParser represents a transaction boundary parser

func (*TransactionBoundaryParser) GetEventBoundaryType

func (p *TransactionBoundaryParser) GetEventBoundaryType(h *replication.EventHeader,
	eventBody []byte) (EventBoundaryType, error)

GetEventBoundaryType implements

func (*TransactionBoundaryParser) IsError

func (p *TransactionBoundaryParser) IsError() bool

IsError return true if current state is error

func (*TransactionBoundaryParser) IsInsideTransaction

func (p *TransactionBoundaryParser) IsInsideTransaction() bool

IsInsideTransaction return true if current state is in transaction

func (*TransactionBoundaryParser) IsNotInsideTransaction

func (p *TransactionBoundaryParser) IsNotInsideTransaction() bool

IsNotInsideTransaction return true if current state is not in transaction

func (*TransactionBoundaryParser) Reset

func (p *TransactionBoundaryParser) Reset()

Reset TransactionBoundaryParser

func (*TransactionBoundaryParser) UpdateState

func (p *TransactionBoundaryParser) UpdateState(eventBoundaryType EventBoundaryType) error

UpdateState TransactionBoundaryParser state