binlogplayer

package
v0.15.1 Latest Latest
Warning

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

 Go to latest Published: Nov 29, 2022 License: Apache-2.0 Imports: 28 Imported by: 4

Documentation

Overview

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

Index

Constants

This section is empty.

Variables

View Source 
var (
	// SlowQueryThreshold will cause we logging anything that's higher than it.
	SlowQueryThreshold = time.Duration(100 * time.Millisecond)

	// BlplQuery is the key for the stats map.
	BlplQuery = "Query"
	// BlplTransaction is the key for the stats map.
	BlplTransaction = "Transaction"

	// VReplicationInit is for the Init state.
	VReplicationInit = "Init"
	// VReplicationCopying is for the Copying state.
	VReplicationCopying = "Copying"
	// BlpRunning is for the Running state.
	BlpRunning = "Running"
	// BlpStopped is for the Stopped state.
	BlpStopped = "Stopped"
	// BlpError is for the Error state.
	BlpError = "Error"
)
View Source 
var AlterVReplicationTable = []string{
	"ALTER TABLE _vt.vreplication ADD COLUMN db_name VARBINARY(255) NOT NULL",
	"ALTER TABLE _vt.vreplication MODIFY source MEDIUMBLOB NOT NULL",
	"ALTER TABLE _vt.vreplication ADD KEY workflow_idx (workflow(64))",
	"ALTER TABLE _vt.vreplication ADD COLUMN rows_copied BIGINT(20) NOT NULL DEFAULT 0",
	"ALTER TABLE _vt.vreplication ADD COLUMN tags VARBINARY(1024) NOT NULL DEFAULT ''",

	"ALTER TABLE _vt.vreplication ADD COLUMN time_heartbeat BIGINT(20) NOT NULL DEFAULT 0",
	"ALTER TABLE _vt.vreplication ADD COLUMN workflow_type int NOT NULL DEFAULT 0",
	"ALTER TABLE _vt.vreplication ADD COLUMN time_throttled BIGINT NOT NULL DEFAULT 0",
	"ALTER TABLE _vt.vreplication ADD COLUMN component_throttled VARCHAR(255) NOT NULL DEFAULT ''",
	"ALTER TABLE _vt.vreplication ADD COLUMN workflow_sub_type int NOT NULL DEFAULT 0",
}

AlterVReplicationTable adds new columns to vreplication table

View Source 
var WithDDLInitialQueries = []string{
	"SELECT db_name FROM _vt.vreplication LIMIT 0",
	"SELECT rows_copied FROM _vt.vreplication LIMIT 0",
	"SELECT time_heartbeat FROM _vt.vreplication LIMIT 0",
	withddl.QueryToTriggerWithDDL,
}

WithDDLInitialQueries contains the queries that:

  • are to be expected by the mock db client during tests, or
  • trigger some of the above _vt.vreplication schema changes to take effect when the binlogplayer starts up

todo: cleanup here. QueryToTriggerWithDDL will be enough to ensure vreplication schema gets created/altered correctly. 

So do that explicitly and move queries required into the mock code.

Functions

func CreateVReplication 

func CreateVReplication(workflow string, source *binlogdatapb.BinlogSource, position string, maxTPS, maxReplicationLag, timeUpdated int64, dbName string,
	workflowType binlogdatapb.VReplicationWorkflowType, workflowSubType binlogdatapb.VReplicationWorkflowSubType) string

CreateVReplication returns a statement to populate the first value into the _vt.vreplication table.

func CreateVReplicationState 

func CreateVReplicationState(workflow string, source *binlogdatapb.BinlogSource, position, state string, dbName string,
	workflowType binlogdatapb.VReplicationWorkflowType, workflowSubType binlogdatapb.VReplicationWorkflowSubType) string

CreateVReplicationState returns a statement to create a stopped vreplication.

func CreateVReplicationTable 

func CreateVReplicationTable() []string

CreateVReplicationTable returns the statements required to create the _vt.vreplication table. id: is an auto-increment column that identifies the stream. workflow: documents the creator/manager of the stream. Example: 'MoveTables'. source: contains a string proto representation of binlogpb.BinlogSource. pos: initially, a start position, and is updated to the current position by the binlog player. stop_pos: optional column that specifies the stop position. max_tps: max transactions per second. max_replication_lag: if replication lag exceeds this amount writing is throttled accordingly. cell: optional column that overrides the current cell to replicate from. tablet_types: optional column that overrides the tablet types to look to replicate from. time_update: last time an event was applied. transaction_timestamp: timestamp of the transaction (from the primary). state: Running, Error or Stopped. message: Reason for current state.

func DecodePosition added in v0.11.0 

func DecodePosition(gtid string) (mysql.Position, error)

DecodePosition attempts to uncompress the passed value first and if it fails tries to decode it as a valid GTID

func DeleteVReplication 

func DeleteVReplication(uid uint32) string

DeleteVReplication returns a statement to delete the replication.

func GenerateUpdateHeartbeat added in v0.13.0 

func GenerateUpdateHeartbeat(uid uint32, timeUpdated int64) (string, error)

GenerateUpdateHeartbeat returns a statement to record the latest heartbeat in the _vt.vreplication table.

func GenerateUpdatePos 

func GenerateUpdatePos(uid uint32, pos mysql.Position, timeUpdated int64, txTimestamp int64, rowsCopied int64, compress bool) string

GenerateUpdatePos returns a statement to record the latest processed gtid in the _vt.vreplication table.

func GenerateUpdateRowsCopied added in v0.11.0 

func GenerateUpdateRowsCopied(uid uint32, rowsCopied int64) string

GenerateUpdateRowsCopied returns a statement to update the rows_copied value in the _vt.vreplication table.

func GenerateUpdateTimeThrottled added in v0.14.2 

func GenerateUpdateTimeThrottled(uid uint32, timeThrottledUnix int64, componentThrottled string) (string, error)

GenerateUpdateTimeThrottled returns a statement to record the latest throttle time in the _vt.vreplication table.

func LimitString added in v0.11.0 

func LimitString(s string, limit int) string

LimitString truncates string to specified size

func LogError added in v0.11.0 

func LogError(msg string, err error)

LogError logs a message after truncating it to avoid spamming logs

func MessageTruncate 

func MessageTruncate(msg string) string

MessageTruncate truncates the message string to a safe length.

func MysqlUncompress added in v0.11.0 

func MysqlUncompress(input string) []byte

MysqlUncompress will uncompress a binary string in the format stored by mysql's compress() function The first four bytes represent the size of the original string passed to compress() Remaining part is the compressed string using zlib, which we uncompress here using golang's zlib library

func ReadVReplicationPos 

func ReadVReplicationPos(index uint32) string

ReadVReplicationPos returns a statement to query the gtid for a given stream from the _vt.vreplication table.

func ReadVReplicationStatus 

func ReadVReplicationStatus(index uint32) string

ReadVReplicationStatus returns a statement to query the status fields for a given stream from the _vt.vreplication table.

func RegisterClientFactory 

func RegisterClientFactory(name string, factory ClientFactory)

RegisterClientFactory adds a new factory. Call during init().

func SetProtocol added in v0.15.0 

func SetProtocol(name string, protocol string) (reset func())

SetProtocol is a helper function to set the binlogplayer --binlog_player_protocol flag value for tests. If successful, it returns a function that, when called, returns the flag to its previous value.

Note that because this variable is bound to a flag, the effects of this function are global, not scoped to the calling test-case. Therefore, it should not be used in conjunction with t.Parallel.

func StartVReplication 

func StartVReplication(uid uint32) string

StartVReplication returns a statement to start the replication.

func StartVReplicationUntil 

func StartVReplicationUntil(uid uint32, pos string) string

StartVReplicationUntil returns a statement to start the replication with a stop position.

func StopVReplication 

func StopVReplication(uid uint32, message string) string

StopVReplication returns a statement to stop the replication.

Types

type BinlogPlayer 

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

BinlogPlayer is for reading a stream of updates from BinlogServer.

func NewBinlogPlayerKeyRange 

func NewBinlogPlayerKeyRange(dbClient DBClient, tablet *topodatapb.Tablet, keyRange *topodatapb.KeyRange, uid uint32, blplStats *Stats) *BinlogPlayer

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

func NewBinlogPlayerTables 

func NewBinlogPlayerTables(dbClient DBClient, tablet *topodatapb.Tablet, tables []string, uid uint32, blplStats *Stats) *BinlogPlayer

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

func (*BinlogPlayer) ApplyBinlogEvents 

func (blp *BinlogPlayer) ApplyBinlogEvents(ctx context.Context) error

ApplyBinlogEvents makes an RPC request to BinlogServer and processes the events. It returns nil if the provided context was canceled, or if we reached the stopping point. If an error is encountered, it updates the vreplication state to "Error". If a stop position was specified, and reached, the state is updated to "Stopped".

type BinlogTransactionStream 

type BinlogTransactionStream interface {
	// Recv returns the next BinlogTransaction, or an error if the RPC was
	// interrupted.
	Recv() (*binlogdatapb.BinlogTransaction, error)
}

BinlogTransactionStream is the interface of the object returned by StreamTables and StreamKeyRange

type Client 

type Client interface {
	// Dial a server
	Dial(tablet *topodatapb.Tablet) error

	// Close the connection
	Close()

	// Ask the server to stream updates related to the provided tables.
	// Should return context.Canceled if the context is canceled.
	StreamTables(ctx context.Context, position string, tables []string, charset *binlogdatapb.Charset) (BinlogTransactionStream, error)

	// Ask the server to stream updates related to the provided keyrange.
	// Should return context.Canceled if the context is canceled.
	StreamKeyRange(ctx context.Context, position string, keyRange *topodatapb.KeyRange, charset *binlogdatapb.Charset) (BinlogTransactionStream, error)
}

Client is the interface all clients must satisfy

type ClientFactory 

type ClientFactory func() Client

ClientFactory is the factory method to create a Client

type DBClient 

type DBClient interface {
	DBName() string
	Connect() error
	Begin() error
	Commit() error
	Rollback() error
	Close()
	ExecuteFetch(query string, maxrows int) (qr *sqltypes.Result, err error)
}

DBClient is a high level interface to the database.

func NewDBClient 

func NewDBClient(params dbconfigs.Connector) DBClient

NewDBClient creates a DBClient instance

func NewFakeDBClient 

func NewFakeDBClient() DBClient

NewFakeDBClient returns a fake DBClient. Its functions return preset responses to requests.

type MockDBClient 

type MockDBClient struct {
	UName string
	// contains filtered or unexported fields
}

MockDBClient mocks a DBClient. It must be configured to expect requests in a specific order.

func NewMockDBClient 

func NewMockDBClient(t *testing.T) *MockDBClient

NewMockDBClient returns a new DBClientMock with the default "Filtered" UName.

func NewMockDbaClient added in v0.11.0 

func NewMockDbaClient(t *testing.T) *MockDBClient

NewMockDbaClient returns a new DBClientMock with the default "Dba" UName.

func (*MockDBClient) Begin 

func (dc *MockDBClient) Begin() error

Begin is part of the DBClient interface

func (*MockDBClient) Close 

func (dc *MockDBClient) Close()

Close is part of the DBClient interface

func (*MockDBClient) Commit 

func (dc *MockDBClient) Commit() error

Commit is part of the DBClient interface

func (*MockDBClient) Connect 

func (dc *MockDBClient) Connect() error

Connect is part of the DBClient interface

func (*MockDBClient) DBName 

func (dc *MockDBClient) DBName() string

DBName is part of the DBClient interface

func (*MockDBClient) ExecuteFetch 

func (dc *MockDBClient) ExecuteFetch(query string, maxrows int) (qr *sqltypes.Result, err error)

ExecuteFetch is part of the DBClient interface

func (*MockDBClient) ExpectRequest 

func (dc *MockDBClient) ExpectRequest(query string, result *sqltypes.Result, err error)

ExpectRequest adds an expected result to the mock. This function should not be called conncurrently with other commands.

func (*MockDBClient) ExpectRequestRE 

func (dc *MockDBClient) ExpectRequestRE(queryRE string, result *sqltypes.Result, err error)

ExpectRequestRE adds an expected result to the mock. queryRE is a regular expression. This function should not be called conncurrently with other commands.

func (*MockDBClient) Rollback 

func (dc *MockDBClient) Rollback() error

Rollback is part of the DBClient interface

func (*MockDBClient) Wait 

func (dc *MockDBClient) Wait()

Wait waits for all expected requests to be executed. dc.t.Fatalf is executed on 1 second timeout. Wait should not be called concurrently with ExpectRequest.

type Stats 

type Stats struct {
	// Stats about the player, keys used are BlplQuery and BlplTransaction
	Timings *stats.Timings
	Rates   *stats.Rates

	ReplicationLagSeconds sync2.AtomicInt64
	History               *history.History

	State sync2.AtomicString

	PhaseTimings   *stats.Timings
	QueryTimings   *stats.Timings
	QueryCount     *stats.CountersWithSingleLabel
	CopyRowCount   *stats.Counter
	CopyLoopCount  *stats.Counter
	ErrorCounts    *stats.CountersWithMultiLabels
	NoopQueryCount *stats.CountersWithSingleLabel

	VReplicationLags     *stats.Timings
	VReplicationLagRates *stats.Rates
	// contains filtered or unexported fields
}

Stats 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 NewStats 

func NewStats() *Stats

NewStats creates a new Stats structure.

func (*Stats) Heartbeat added in v0.9.0 

func (bps *Stats) Heartbeat() int64

Heartbeat gets the time the last heartbeat from vstreamer was seen

func (*Stats) LastPosition 

func (bps *Stats) LastPosition() mysql.Position

LastPosition gets the last replication position.

func (*Stats) MessageHistory 

func (bps *Stats) MessageHistory() []string

MessageHistory gets all the messages, we store 3 at a time

func (*Stats) RecordHeartbeat added in v0.9.0 

func (bps *Stats) RecordHeartbeat(tm int64)

RecordHeartbeat updates the time the last heartbeat from vstreamer was seen

func (*Stats) SetLastPosition 

func (bps *Stats) SetLastPosition(pos mysql.Position)

SetLastPosition sets the last replication position.

type StatsHistoryRecord 

type StatsHistoryRecord struct {
	Time    time.Time
	Message string
}

StatsHistoryRecord is used to store a Message with timestamp

func (*StatsHistoryRecord) IsDuplicate 

func (r *StatsHistoryRecord) IsDuplicate(other any) bool

IsDuplicate implements history.Deduplicable

type VRSettings 

type VRSettings struct {
	StartPos          mysql.Position
	StopPos           mysql.Position
	MaxTPS            int64
	MaxReplicationLag int64
	State             string
	WorkflowType      int32
	WorkflowSubType   int32
	WorkflowName      string
}

VRSettings contains the settings of a vreplication table.

func ReadVRSettings 

func ReadVRSettings(dbClient DBClient, uid uint32) (VRSettings, error)

ReadVRSettings retrieves the throttler settings for vreplication from the checkpoint table.

Source Files

View all Source files

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.