binlog

package
v0.19.3 Latest Latest
Warning

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

 Go to latest Published: Apr 4, 2024 License: Apache-2.0 Imports: 29 Imported by: 3

Documentation

Index

Constants

This section is empty.

Variables

View Source 
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")
)
View Source 
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")
)
View Source 
var RegisterUpdateStreamServices []RegisterUpdateStreamServiceFunc

RegisterUpdateStreamServices is the list of all registration callbacks to invoke

Functions

This section is empty.

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 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 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, parser *sqlparser.Parser) *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

Source Files

View all Source files

Directories

Path Synopsis
Package binlogplayer contains the code that plays a vreplication stream on a client database.
 Package binlogplayer contains the code that plays a vreplication stream on a client database.
Package eventtoken includes utility methods for event token handling.
 Package eventtoken includes utility methods for event token handling.
Package grpcbinlogstreamer contains the gRPC implementation of the binlog streamer server component.
 Package grpcbinlogstreamer contains the gRPC implementation of the binlog streamer server component.

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.