binlogplayer

Documentation

Overview

Package binlogplayer contains the code that plays a filtered replication stream on a client database. It usually runs inside the destination master vttablet process.

Index

• Variables
• func CreateBlpCheckpoint() []string
• func PopulateBlpCheckpoint(index uint32, pos myproto.ReplicationPosition, timeUpdated int64, flags string) string
• func QueryBlpCheckpoint(index uint32) string
• func ReadStartPosition(dbClient VtClient, uid uint32) (*proto.BlpPosition, string, error)
• func RegisterBinlogPlayerClientFactory(name string, factory BinlogPlayerClientFactory)
• func UpdateBlpCheckpoint(uid uint32, pos myproto.ReplicationPosition, timeUpdated int64, ...) string
• type BinlogPlayer
  • func NewBinlogPlayerKeyRange(dbClient VtClient, addr string, keyspaceIdType key.KeyspaceIdType, ...) *BinlogPlayer
  • func NewBinlogPlayerTables(dbClient VtClient, addr string, tables []string, ...) *BinlogPlayer
  • func (blp *BinlogPlayer) ApplyBinlogEvents(interrupted chan struct{}) error
• type BinlogPlayerClient
• type BinlogPlayerClientFactory
• type BinlogPlayerResponse
• type BinlogPlayerStats
  • func NewBinlogPlayerStats() *BinlogPlayerStats
  • func (bps *BinlogPlayerStats) GetLastPosition() myproto.ReplicationPosition
  • func (bps *BinlogPlayerStats) SetLastPosition(pos myproto.ReplicationPosition)
• type DBClient
  • func NewDbClient(params *mysql.ConnectionParams) *DBClient
  • func (dc *DBClient) Begin() error
  • func (dc *DBClient) Close()
  • func (dc *DBClient) Commit() error
  • func (dc *DBClient) Connect() error
  • func (dc *DBClient) ExecuteFetch(query string, maxrows int, wantfields bool) (*mproto.QueryResult, error)
  • func (dc *DBClient) Rollback() error
• type DummyVtClient
  • func NewDummyVtClient() *DummyVtClient
  • func (dc DummyVtClient) Begin() error
  • func (dc DummyVtClient) Close()
  • func (dc DummyVtClient) Commit() error
  • func (dc DummyVtClient) Connect() error
  • func (dc DummyVtClient) ExecuteFetch(query string, maxrows int, wantfields bool) (qr *mproto.QueryResult, err error)
  • func (dc DummyVtClient) Rollback() error
• type VtClient

Variables

var (
	// we will log anything that's higher than that
	SLOW_QUERY_THRESHOLD = time.Duration(100 * time.Millisecond)

	// keys for the stats map
	BLPL_QUERY       = "Query"
	BLPL_TRANSACTION = "Transaction"

	// flags for the blp_checkpoint table. The database entry is just
	// a join(",") of these flags.
	BLP_FLAG_DONT_START = "DontStart"
)

Functions

func CreateBlpCheckpoint

func CreateBlpCheckpoint() []string

CreateBlpCheckpoint returns the statements required to create the _vt.blp_checkpoint table

func PopulateBlpCheckpoint

func PopulateBlpCheckpoint(index uint32, pos myproto.ReplicationPosition, timeUpdated int64, flags string) string

PopulateBlpCheckpoint returns a statement to populate the first value into the _vt.blp_checkpoint table.

func QueryBlpCheckpoint

func QueryBlpCheckpoint(index uint32) string

QueryBlpCheckpoint returns a statement to query the gtid and flags for a given shard from the _vt.blp_checkpoint table.

func ReadStartPosition

func ReadStartPosition(dbClient VtClient, uid uint32) (*proto.BlpPosition, string, error)

ReadStartPosition will return the current start position and the flags for the provided binlog player.

func RegisterBinlogPlayerClientFactory

func RegisterBinlogPlayerClientFactory(name string, factory BinlogPlayerClientFactory)

func UpdateBlpCheckpoint

func UpdateBlpCheckpoint(uid uint32, pos myproto.ReplicationPosition, timeUpdated int64, txTimestamp int64) string

UpdateBlpCheckpoint returns a statement to update a value in the _vt.blp_checkpoint table.

Types

type BinlogPlayer

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

BinlogPlayer is handling reading a stream of updates from BinlogServer

func NewBinlogPlayerKeyRange

func NewBinlogPlayerKeyRange(dbClient VtClient, addr string, keyspaceIdType key.KeyspaceIdType, keyRange key.KeyRange, startPosition *proto.BlpPosition, stopPosition myproto.ReplicationPosition, blplStats *BinlogPlayerStats) *BinlogPlayer

NewBinlogPlayerKeyRange returns a new BinlogPlayer pointing at the server replicating the provided keyrange, starting at the startPosition, and updating _vt.blp_checkpoint with uid=startPosition.Uid. If !stopPosition.IsZero(), it will stop when reaching that position.

func NewBinlogPlayerTables

func NewBinlogPlayerTables(dbClient VtClient, addr string, tables []string, startPosition *proto.BlpPosition, stopPosition myproto.ReplicationPosition, blplStats *BinlogPlayerStats) *BinlogPlayer

NewBinlogPlayerTables returns a new BinlogPlayer pointing at the server replicating the provided tables, starting at the startPosition, and updating _vt.blp_checkpoint with uid=startPosition.Uid. If !stopPosition.IsZero(), it will stop when reaching that position.

func (*BinlogPlayer) ApplyBinlogEvents

func (blp *BinlogPlayer) ApplyBinlogEvents(interrupted chan struct{}) error

ApplyBinlogEvents makes a gob rpc request to BinlogServer and processes the events. It will return nil if 'interrupted' was closed, or if we reached the stopping point. It will return io.EOF if the server stops sending us updates. It may return any other error it encounters.

type BinlogPlayerClient

type BinlogPlayerClient interface {
	// Dial a server
	Dial(addr string, connTimeout time.Duration) error

	// Close the connection
	Close()

	// Ask the server to stream binlog updates
	ServeUpdateStream(*proto.UpdateStreamRequest, chan *proto.StreamEvent) BinlogPlayerResponse

	// Ask the server to stream updates related to the provided tables
	StreamTables(*proto.TablesRequest, chan *proto.BinlogTransaction) BinlogPlayerResponse

	// Ask the server to stream updates related to thee provided keyrange
	StreamKeyRange(*proto.KeyRangeRequest, chan *proto.BinlogTransaction) BinlogPlayerResponse
}

BinlogPlayerClient is the interface all clients must satisfy

type BinlogPlayerClientFactory

type BinlogPlayerClientFactory func() BinlogPlayerClient

type BinlogPlayerResponse

type BinlogPlayerResponse interface {
	Error() error
}

BinlogPlayerResponse is the return value for streaming events

type BinlogPlayerStats

type BinlogPlayerStats struct {
	// Stats about the player, keys used are BLPL_QUERY and BLPL_TRANSACTION
	Timings *stats.Timings
	Rates   *stats.Rates

	SecondsBehindMaster sync2.AtomicInt64
	// contains filtered or unexported fields
}

BinlogPlayerStats is the internal stats of a player. It is a different structure that is passed in so stats can be collected over the life of multiple individual players.

func NewBinlogPlayerStats

func NewBinlogPlayerStats() *BinlogPlayerStats

NewBinlogPlayerStats creates a new BinlogPlayerStats structure

func (*BinlogPlayerStats) GetLastPosition

func (bps *BinlogPlayerStats) GetLastPosition() myproto.ReplicationPosition

func (*BinlogPlayerStats) SetLastPosition

func (bps *BinlogPlayerStats) SetLastPosition(pos myproto.ReplicationPosition)

type DBClient

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

DBClient is a real VtClient backed by a mysql connection

func NewDbClient

func NewDbClient(params *mysql.ConnectionParams) *DBClient

func (*DBClient) Begin

func (dc *DBClient) Begin() error

func (*DBClient) Close

func (dc *DBClient) Close()

func (*DBClient) Commit

func (dc *DBClient) Commit() error

func (*DBClient) Connect

func (dc *DBClient) Connect() error

func (*DBClient) ExecuteFetch

func (dc *DBClient) ExecuteFetch(query string, maxrows int, wantfields bool) (*mproto.QueryResult, error)

func (*DBClient) Rollback

func (dc *DBClient) Rollback() error

type DummyVtClient

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

DummyVtClient is a VtClient that writes to a writer instead of executing anything

func NewDummyVtClient

func NewDummyVtClient() *DummyVtClient

func (DummyVtClient) Begin

func (dc DummyVtClient) Begin() error

func (DummyVtClient) Close

func (dc DummyVtClient) Close()

func (DummyVtClient) Commit

func (dc DummyVtClient) Commit() error

func (DummyVtClient) Connect

func (dc DummyVtClient) Connect() error

func (DummyVtClient) ExecuteFetch

func (dc DummyVtClient) ExecuteFetch(query string, maxrows int, wantfields bool) (qr *mproto.QueryResult, err error)

func (DummyVtClient) Rollback

func (dc DummyVtClient) Rollback() error

type VtClient

type VtClient interface {
	Connect() error
	Begin() error
	Commit() error
	Rollback() error
	Close()
	ExecuteFetch(query string, maxrows int, wantfields bool) (qr *mproto.QueryResult, err error)
}

VtClient is a high level interface to the database