binlog

Documentation

Index

• Variables
• type BinlogConnection
  • func NewBinlogConnection(cp dbconfigs.Connector) (*BinlogConnection, error)
  • func (bc *BinlogConnection) Close()
  • func (bc *BinlogConnection) StartBinlogDumpFromBinlogBeforeTimestamp(ctx context.Context, timestamp int64) (<-chan mysql.BinlogEvent, <-chan error, error)
  • func (bc *BinlogConnection) StartBinlogDumpFromCurrent(ctx context.Context) (replication.Position, <-chan mysql.BinlogEvent, <-chan error, error)
  • func (bc *BinlogConnection) StartBinlogDumpFromPosition(ctx context.Context, binlogFilename string, startPos replication.Position) (<-chan mysql.BinlogEvent, <-chan error, error)
• type EventStreamer
  • func NewEventStreamer(cp dbconfigs.Connector, se *schema.Engine, startPos replication.Position, ...) *EventStreamer
  • func (evs *EventStreamer) Stream(ctx context.Context) error
• type FullBinlogStatement
• type RegisterUpdateStreamServiceFunc
• type StreamList
  • func (sl *StreamList) Add(c context.CancelFunc) int
  • func (sl *StreamList) Delete(i int)
  • func (sl *StreamList) Init()
  • func (sl *StreamList) Stop()
• type Streamer
  • func NewStreamer(cp dbconfigs.Connector, se *schema.Engine, clientCharset *binlogdatapb.Charset, ...) *Streamer
  • func (bls *Streamer) Stream(ctx context.Context) (err error)
• type UpdateStream
• type UpdateStreamControl
• type UpdateStreamControlMock
  • func NewUpdateStreamControlMock() *UpdateStreamControlMock
  • func (m *UpdateStreamControlMock) Disable()
  • func (m *UpdateStreamControlMock) Enable()
  • func (m *UpdateStreamControlMock) InitDBConfig(*dbconfigs.DBConfigs)
  • func (m *UpdateStreamControlMock) IsEnabled() bool
  • func (m *UpdateStreamControlMock) RegisterService()
• type UpdateStreamImpl
  • func NewUpdateStream(ts *topo.Server, keyspace string, cell string, se *schema.Engine) *UpdateStreamImpl
  • func (updateStream *UpdateStreamImpl) Disable()
  • func (updateStream *UpdateStreamImpl) Enable()
  • func (updateStream *UpdateStreamImpl) HandlePanic(err *error)
  • func (updateStream *UpdateStreamImpl) InitDBConfig(dbcfgs *dbconfigs.DBConfigs)
  • func (updateStream *UpdateStreamImpl) IsEnabled() bool
  • func (updateStream *UpdateStreamImpl) RegisterService()
  • func (updateStream *UpdateStreamImpl) StreamKeyRange(ctx context.Context, position string, keyRange *topodatapb.KeyRange, ...) (err error)
  • func (updateStream *UpdateStreamImpl) StreamTables(ctx context.Context, position string, tables []string, ...) (err error)

Variables

var (

	// ErrClientEOF is returned by Streamer if the stream ended because the
	// consumer of the stream indicated it doesn't want any more events.
	ErrClientEOF = fmt.Errorf("binlog stream consumer ended the reply stream")
	// ErrServerEOF is returned by Streamer if the stream ended because the
	// connection to the mysqld server was lost, or the stream was terminated by
	// mysqld.
	ErrServerEOF = fmt.Errorf("binlog stream connection was closed by mysqld")
)

var (
	// ErrBinlogUnavailable is returned by this library when we
	// cannot find a suitable binlog to satisfy the request.
	ErrBinlogUnavailable = fmt.Errorf("cannot find relevant binlogs on this server")
)

var RegisterUpdateStreamServices []RegisterUpdateStreamServiceFunc

RegisterUpdateStreamServices is the list of all registration callbacks to invoke

Types

type BinlogConnection

type BinlogConnection struct {
	*mysql.Conn
	// contains filtered or unexported fields
}

BinlogConnection represents a connection to mysqld that pretends to be a replica connecting for replication. Each such connection must identify itself to mysqld with a server ID that is unique both among other BinlogConnections and among actual replicas in the topology.

func NewBinlogConnection

func NewBinlogConnection(cp dbconfigs.Connector) (*BinlogConnection, error)

NewBinlogConnection creates a new binlog connection to the mysqld instance.

func (*BinlogConnection) Close

func (bc *BinlogConnection) Close()

Close closes the binlog connection, which also signals an ongoing dump started with StartBinlogDump() to stop and close its BinlogEvent channel. The ID for the binlog connection is recycled back into the pool.

func (*BinlogConnection) StartBinlogDumpFromBinlogBeforeTimestamp

func (bc *BinlogConnection) StartBinlogDumpFromBinlogBeforeTimestamp(ctx context.Context, timestamp int64) (<-chan mysql.BinlogEvent, <-chan error, error)

StartBinlogDumpFromBinlogBeforeTimestamp requests a replication binlog dump from the source mysqld starting with a file that has timestamps smaller than the provided timestamp, and then sends binlog events to the provided channel.

The startup phase will list all the binary logs, and find the one that has events starting strictly before the provided timestamp. It will then start from there, and stream all events. It is the responsibility of the calling site to filter the events more.

MySQL 5.6+ note: we need to do it that way because of the way the GTIDSet works. In the previous two streaming functions, we pass in the full GTIDSet (that has the list of all transactions seen in the replication stream). In this case, we don't know it, all we have is the binlog file names. We depend on parsing the first PREVIOUS_GTIDS_EVENT event in the logs to get it. So we need the caller to parse that event, and it can't be skipped because its timestamp is lower. Then, for each subsequent event, the caller also needs to add the event GTID to its GTIDSet. Otherwise it won't be correct ever. So the caller really needs to build up its GTIDSet along the entire file, not just for events whose timestamp is in a given range.

The stream will continue in the background, waiting for new events if necessary, until the connection is closed, either by the source or by canceling the context.

Note the context is valid and used until eventChan is closed.

func (*BinlogConnection) StartBinlogDumpFromCurrent

func (bc *BinlogConnection) StartBinlogDumpFromCurrent(ctx context.Context) (replication.Position, <-chan mysql.BinlogEvent, <-chan error, error)

StartBinlogDumpFromCurrent requests a replication binlog dump from the current position.

func (*BinlogConnection) StartBinlogDumpFromPosition

func (bc *BinlogConnection) StartBinlogDumpFromPosition(ctx context.Context, binlogFilename string, startPos replication.Position) (<-chan mysql.BinlogEvent, <-chan error, error)

StartBinlogDumpFromPosition requests a replication binlog dump from the replication source mysqld (typically the primary server in the cluster) at the given Position and then sends binlog events to the provided channel. The stream will continue in the background, waiting for new events if necessary, until the connection is closed, either by the replication source or by canceling the context.

Note the context is valid and used until eventChan is closed.

type EventStreamer

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

EventStreamer is an adapter on top of a binlog Streamer that convert the events into StreamEvent objects.

func NewEventStreamer

func NewEventStreamer(cp dbconfigs.Connector, se *schema.Engine, startPos replication.Position, timestamp int64, sendEvent sendEventFunc) *EventStreamer

NewEventStreamer returns a new EventStreamer on top of a Streamer

func (*EventStreamer) Stream

func (evs *EventStreamer) Stream(ctx context.Context) error

Stream starts streaming updates

type FullBinlogStatement

type FullBinlogStatement struct {
	Statement  *binlogdatapb.BinlogTransaction_Statement
	Table      string
	KeyspaceID []byte
	PKNames    []*querypb.Field
	PKValues   []sqltypes.Value
}

FullBinlogStatement has all the information we can gather for an event. Some fields are only set if asked for, and if RBR is used. Otherwise we'll revert back to using the SQL comments, for SBR.

type RegisterUpdateStreamServiceFunc

type RegisterUpdateStreamServiceFunc func(UpdateStream)

RegisterUpdateStreamServiceFunc is the type to use for delayed registration of RPC servers until we have all the objects

type StreamList

type StreamList struct {
	sync.Mutex
	// contains filtered or unexported fields
}

StreamList is a map of context.CancelFunc to mass-interrupt ongoing calls.

func (*StreamList) Add

func (sl *StreamList) Add(c context.CancelFunc) int

Add adds a CancelFunc to the map.

func (*StreamList) Delete

func (sl *StreamList) Delete(i int)

Delete removes a CancelFunc from the list.

func (*StreamList) Init

func (sl *StreamList) Init()

Init must be called before using the list.

func (*StreamList) Stop

func (sl *StreamList) Stop()

Stop stops all the current streams.

type Streamer

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

Streamer streams binlog events from MySQL. A Streamer should only be used once. To start another stream, call NewStreamer() again.

func NewStreamer

func NewStreamer(cp dbconfigs.Connector, se *schema.Engine, clientCharset *binlogdatapb.Charset, startPos replication.Position, timestamp int64, sendTransaction sendTransactionFunc) *Streamer

NewStreamer creates a binlog Streamer.

dbname specifes the database to stream events for. mysqld is the local instance of mysqlctl.Mysqld. charset is the default character set on the BinlogPlayer side. startPos is the position to start streaming at. Incompatible with timestamp. timestamp is the timestamp to start streaming at. Incompatible with startPos. sendTransaction is called each time a transaction is committed or rolled back.

func (*Streamer) Stream

func (bls *Streamer) Stream(ctx context.Context) (err error)

Stream starts streaming binlog events using the settings from NewStreamer().

type UpdateStream

type UpdateStream interface {
	// StreamKeyRange streams events related to a KeyRange only
	StreamKeyRange(ctx context.Context, position string, keyRange *topodatapb.KeyRange, charset *binlogdatapb.Charset, callback func(trans *binlogdatapb.BinlogTransaction) error) error

	// StreamTables streams events related to a set of Tables only
	StreamTables(ctx context.Context, position string, tables []string, charset *binlogdatapb.Charset, callback func(trans *binlogdatapb.BinlogTransaction) error) error

	// HandlePanic should be called in a defer,
	// first thing in the RPC implementation.
	HandlePanic(*error)
}

UpdateStream is the interface for the binlog server

type UpdateStreamControl

type UpdateStreamControl interface {
	// InitDBConfigs is called after the db name is computed.
	InitDBConfig(*dbconfigs.DBConfigs)
	// RegisterService registers the UpdateStream service.
	RegisterService()
	// Enable will allow any new RPC calls
	Enable()
	// Disable will interrupt all current calls, and disallow any new call
	Disable()
	// IsEnabled returns true iff the service is enabled
	IsEnabled() bool
}

UpdateStreamControl is the interface an UpdateStream service implements to bring it up or down.

type UpdateStreamControlMock

type UpdateStreamControlMock struct {
	sync.Mutex
	// contains filtered or unexported fields
}

UpdateStreamControlMock is an implementation of UpdateStreamControl to be used in tests

func NewUpdateStreamControlMock

func NewUpdateStreamControlMock() *UpdateStreamControlMock

NewUpdateStreamControlMock creates a new UpdateStreamControlMock

func (*UpdateStreamControlMock) Disable

func (m *UpdateStreamControlMock) Disable()

Disable is part of UpdateStreamControl

func (*UpdateStreamControlMock) Enable

func (m *UpdateStreamControlMock) Enable()

Enable is part of UpdateStreamControl

func (*UpdateStreamControlMock) InitDBConfig

func (m *UpdateStreamControlMock) InitDBConfig(*dbconfigs.DBConfigs)

InitDBConfig is part of UpdateStreamControl

func (*UpdateStreamControlMock) IsEnabled

func (m *UpdateStreamControlMock) IsEnabled() bool

IsEnabled is part of UpdateStreamControl

func (*UpdateStreamControlMock) RegisterService

func (m *UpdateStreamControlMock) RegisterService()

RegisterService is part of UpdateStreamControl

type UpdateStreamImpl

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

UpdateStreamImpl is the real implementation of UpdateStream and UpdateStreamControl

func NewUpdateStream

func NewUpdateStream(ts *topo.Server, keyspace string, cell string, se *schema.Engine) *UpdateStreamImpl

NewUpdateStream returns a new UpdateStreamImpl object

func (*UpdateStreamImpl) Disable

func (updateStream *UpdateStreamImpl) Disable()

Disable will disallow any connection to the service

func (*UpdateStreamImpl) Enable

func (updateStream *UpdateStreamImpl) Enable()

Enable will allow connections to the service

func (*UpdateStreamImpl) HandlePanic

func (updateStream *UpdateStreamImpl) HandlePanic(err *error)

HandlePanic is part of the UpdateStream interface

func (*UpdateStreamImpl) InitDBConfig

func (updateStream *UpdateStreamImpl) InitDBConfig(dbcfgs *dbconfigs.DBConfigs)

InitDBConfig should be invoked after the db name is computed.

func (*UpdateStreamImpl) IsEnabled

func (updateStream *UpdateStreamImpl) IsEnabled() bool

IsEnabled returns true if UpdateStreamImpl is enabled

func (*UpdateStreamImpl) RegisterService

func (updateStream *UpdateStreamImpl) RegisterService()

RegisterService needs to be called to publish stats, and to start listening to clients. Only once instance can call this in a process.

func (*UpdateStreamImpl) StreamKeyRange

func (updateStream *UpdateStreamImpl) StreamKeyRange(ctx context.Context, position string, keyRange *topodatapb.KeyRange, charset *binlogdatapb.Charset, callback func(trans *binlogdatapb.BinlogTransaction) error) (err error)

StreamKeyRange is part of the UpdateStream interface

func (*UpdateStreamImpl) StreamTables

func (updateStream *UpdateStreamImpl) StreamTables(ctx context.Context, position string, tables []string, charset *binlogdatapb.Charset, callback func(trans *binlogdatapb.BinlogTransaction) error) (err error)

StreamTables is part of the UpdateStream interface