binlog

Documentation

Index

• Constants
• Variables
• type DNSTracker
  • func NewDNSTracker(host string, logger *log.Logger, changeCallback func()) *DNSTracker
  • func (t *DNSTracker) Start()
  • func (t *DNSTracker) Stop()
• type EventConsumer
  • func (c *EventConsumer) Close()
  • func (c *EventConsumer) Handle(event *replication.BinlogEvent) (err error)
  • func (c *EventConsumer) HandleError(err error)
• type GoMysqlLoggerAdapter
  • func (a *GoMysqlLoggerAdapter) Close() error
  • func (a *GoMysqlLoggerAdapter) Write(p []byte) (n int, err error)
• type MysqlBinlogInput
  • func NewMysqlBinlogInput() *MysqlBinlogInput
  • func (in *MysqlBinlogInput) Ack(msg *core.Message, err error)
  • func (in *MysqlBinlogInput) Configure(config core.StringMap) (err error)
  • func (in *MysqlBinlogInput) GetState() (state []byte, done bool)
  • func (in *MysqlBinlogInput) SetState(state []byte) (err error)
  • func (in *MysqlBinlogInput) Start() (err error)
  • func (in *MysqlBinlogInput) Stop()
• type MysqlBinlogInputConfig
• type ServerInfo
  • func LoadFromServer(host string, port uint16, user string, password string) (st *ServerInfo, err error)
  • func (s *ServerInfo) BinlogPosition() (core.MysqlBinlogPosition, error)
  • func (s *ServerInfo) String() string

Constants

const (
	// ReplicationModeFilepos is Mysql binlog file position replication
	ReplicationModeFilepos = "filepos"
	// ReplicationModeGtid is Mysql GTID replication
	ReplicationModeGtid = "gtid"
)

const (
	NoSwitch    = 0
	SwitchByIP  = 1 // master slave switch in a list of ip
	SwitchByDNS = 2 // switch by dns change
)

const (
	FakeMysqlSlaveHostName        = "pipe_s_binlog_extractor_server"
	DefaultGoMysqlRecvBufferSize  = 1024 * 1024 * 10
	DefaultRingBufferSize         = 1024 * 8
	DefaultBinlogParseConcurrency = 4
)

const DefaultDNSCheckInterval = 5000 // milliseconds

Variables

var TokenBegin = []byte("BEGIN")

Types

type DNSTracker

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

func NewDNSTracker

func NewDNSTracker(host string, logger *log.Logger, changeCallback func()) *DNSTracker

func (*DNSTracker) Start

func (t *DNSTracker) Start()

func (*DNSTracker) Stop

func (t *DNSTracker) Stop()

type EventConsumer

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

EventConsumer consumes binlog event from DisruptorBinlogSyncer.

func (*EventConsumer) Close

func (c *EventConsumer) Close()

func (*EventConsumer) Handle

func (c *EventConsumer) Handle(event *replication.BinlogEvent) (err error)

func (*EventConsumer) HandleError

func (c *EventConsumer) HandleError(err error)

type GoMysqlLoggerAdapter

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

GoMysqlLoggerAdapter is adapter to pass logs from go mysql logger to our logger.

func (*GoMysqlLoggerAdapter) Close

func (a *GoMysqlLoggerAdapter) Close() error

func (*GoMysqlLoggerAdapter) Write

func (a *GoMysqlLoggerAdapter) Write(p []byte) (n int, err error)

type MysqlBinlogInput

type MysqlBinlogInput struct {
	*core.BaseInput
	Config *MysqlBinlogInputConfig
	// contains filtered or unexported fields
}

MysqlBinlogInput extracts binlog events from mysql server.

func NewMysqlBinlogInput

func NewMysqlBinlogInput() *MysqlBinlogInput

func (*MysqlBinlogInput) Ack

func (in *MysqlBinlogInput) Ack(msg *core.Message, err error)

Ack won't be invoked concurrently which is guaranteed by pipeline, but it may happen at the same time with GetState so still need atomic operations.

func (*MysqlBinlogInput) Configure

func (in *MysqlBinlogInput) Configure(config core.StringMap) (err error)

func (*MysqlBinlogInput) GetState

func (in *MysqlBinlogInput) GetState() (state []byte, done bool)

func (*MysqlBinlogInput) SetState

func (in *MysqlBinlogInput) SetState(state []byte) (err error)

func (*MysqlBinlogInput) Start

func (in *MysqlBinlogInput) Start() (err error)

func (*MysqlBinlogInput) Stop

func (in *MysqlBinlogInput) Stop()

type MysqlBinlogInputConfig

type MysqlBinlogInputConfig struct {
	ID              string
	Address         string // the mysql server ip:port address
	BackupAddress   string // the backup server address, if connecting Address fails will try this
	User            string
	Password        string
	ReplicationMode string // gtid or filepos
	SyncDDL         bool   // whether generate events for ddl
	SyncTxInfo      bool   // whether generate events for transaction boundaries
	SwitchByDNS     bool   // if switch mysql master slave by changing dns
}

type ServerInfo

type ServerInfo struct {
	Host            string
	Port            uint16
	ServerID        uint32
	ServerUUID      string
	GtidEnabled     bool
	File            string
	Pos             uint32
	ExecutedGtidSet string
	PurgedGtidSet   string
}

func LoadFromServer

func LoadFromServer(host string, port uint16, user string, password string) (st *ServerInfo, err error)

func (*ServerInfo) BinlogPosition

func (s *ServerInfo) BinlogPosition() (core.MysqlBinlogPosition, error)

BinlogPosition returns server status's corresponding position.

func (*ServerInfo) String

func (s *ServerInfo) String() string