mysqlstream

package

v0.9.18 Latest Latest Go to latest Published: Feb 14, 2019 License: Apache-2.0 Imports: 29 Imported by: 2

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/moiot/gravity

Links

Open Source Insights

Documentation ¶

Index ¶

Variables
func GetCurrentPositionValue(cache position_store.PositionCacheInterface) (*utils.MySQLBinlogPosition, error)
func IsEventBelongsToMyself(event *replication.RowsEvent, pipelineName string) bool
func NewBarrierMsg(callback core.AfterMsgCommitFunc) *core.Msg
func NewDDLMsg(callback core.AfterMsgCommitFunc, dbName string, table string, ...) *core.Msg
func NewDeleteMsgs(host string, database string, table string, ts int64, ...) ([]*core.Msg, error)
func NewInsertMsgs(host string, database string, table string, ts int64, ...) ([]*core.Msg, error)
func NewUpdateMsgs(host string, database string, table string, ts int64, ...) ([]*core.Msg, error)
func NewXIDMsg(ts int64, callback core.AfterMsgCommitFunc, position utils.MySQLBinlogPosition) *core.Msg
func SetupInitialPosition(db *sql.DB, positionCache position_store.PositionCacheInterface, ...) error
func ToGoMySQLPosition(p utils.MySQLBinlogPosition) (gomysql.Position, gomysql.MysqlGTIDSet, error)
func UpdateCurrentPositionValue(cache position_store.PositionCacheInterface, ...) error
type BinlogEventSchemaFilterFunc
type BinlogTailer
- func NewBinlogTailer(pipelineName string, cfg *MySQLBinlogInputPluginConfig, gravityServerID uint32, ...) (*BinlogTailer, error)
type MySQLBinlogInputPluginConfig

Constants ¶

This section is empty.

Variables ¶

View Source

var (
	GravityTableInsertCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: "drc_v2",
		Subsystem: "gravity",
		Name:      "table_insert_rows_counter",
		Help:      "table insert rows counter",
	}, []string{metrics.PipelineTag, "db"})

	GravityTableUpdateCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: "drc_v2",
		Subsystem: "gravity",
		Name:      "table_update_rows_counter",
		Help:      "table update rows counter",
	}, []string{metrics.PipelineTag, "db"})

	GravityTableDeleteCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: "drc_v2",
		Subsystem: "gravity",
		Name:      "table_delete_rows_counter",
		Help:      "table delete rows counter",
	}, []string{metrics.PipelineTag, "db"})

	GravityTableRowsEventSize = prometheus.NewGaugeVec(prometheus.GaugeOpts{
		Namespace: "drc_v2",
		Subsystem: "gravity",
		Name:      "rows_event_size",
		Help:      "rows event size per txn",
	}, []string{metrics.PipelineTag})
)

View Source

var (
	// re-sync retry timeout
	RetryTimeout = 3 * time.Second

	BinlogProbeInterval = 3 * time.Second
)

Functions ¶

func GetCurrentPositionValue ¶ added in v0.9.17

func GetCurrentPositionValue(cache position_store.PositionCacheInterface) (*utils.MySQLBinlogPosition, error)

func IsEventBelongsToMyself ¶

func IsEventBelongsToMyself(event *replication.RowsEvent, pipelineName string) bool

func NewBarrierMsg ¶

func NewBarrierMsg(callback core.AfterMsgCommitFunc) *core.Msg

func NewDDLMsg ¶

func NewDDLMsg(
	callback core.AfterMsgCommitFunc,
	dbName string,
	table string,
	ast ast.StmtNode,
	ddlSQL string,
	ts int64,
	position utils.MySQLBinlogPosition) *core.Msg

func NewDeleteMsgs ¶

func NewDeleteMsgs(
	host string,
	database string,
	table string,
	ts int64,
	ev *replication.RowsEvent,
	tableDef *schema_store.Table) ([]*core.Msg, error)

func NewInsertMsgs ¶

func NewInsertMsgs(
	host string,
	database string,
	table string,
	ts int64,
	ev *replication.RowsEvent,
	tableDef *schema_store.Table) ([]*core.Msg, error)

func NewUpdateMsgs ¶

func NewUpdateMsgs(
	host string,
	database string,
	table string,
	ts int64,
	ev *replication.RowsEvent,
	tableDef *schema_store.Table) ([]*core.Msg, error)

func NewXIDMsg ¶

func NewXIDMsg(ts int64, callback core.AfterMsgCommitFunc, position utils.MySQLBinlogPosition) *core.Msg

func SetupInitialPosition ¶ added in v0.9.17

func SetupInitialPosition(db *sql.DB, positionCache position_store.PositionCacheInterface, startPositionSpec *utils.MySQLBinlogPosition) error

func ToGoMySQLPosition ¶ added in v0.9.17

func ToGoMySQLPosition(p utils.MySQLBinlogPosition) (gomysql.Position, gomysql.MysqlGTIDSet, error)

func UpdateCurrentPositionValue ¶ added in v0.9.17

func UpdateCurrentPositionValue(cache position_store.PositionCacheInterface, currentPosition *utils.MySQLBinlogPosition) error

Types ¶

type BinlogEventSchemaFilterFunc ¶

type BinlogEventSchemaFilterFunc func(schemaName string) bool

type BinlogTailer ¶

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

func NewBinlogTailer ¶

func NewBinlogTailer(
	pipelineName string,
	cfg *MySQLBinlogInputPluginConfig,
	gravityServerID uint32,
	positionCache position_store.PositionCacheInterface,
	sourceSchemaStore schema_store.SchemaStore,
	sourceDB *sql.DB,
	emitter core.Emitter,
	router core.Router,
	binlogChecker binlog_checker.BinlogChecker,
	binlogEventSchemaFilter BinlogEventSchemaFilterFunc,
) (*BinlogTailer, error)

NewBinlogTailer creats a new binlog tailer

func (*BinlogTailer) AfterMsgCommit ¶

func (tailer *BinlogTailer) AfterMsgCommit(msg *core.Msg) error

func (*BinlogTailer) AppendMsgTxnBuffer ¶

func (tailer *BinlogTailer) AppendMsgTxnBuffer(msg *core.Msg)

AppendMsgTxnBuffer adds basic job information to txn buffer

func (*BinlogTailer) ClearMsgTxnBuffer ¶

func (tailer *BinlogTailer) ClearMsgTxnBuffer()

func (*BinlogTailer) Close ¶

func (tailer *BinlogTailer) Close()

func (*BinlogTailer) FlushMsgTxnBuffer ¶

func (tailer *BinlogTailer) FlushMsgTxnBuffer()

FlushMsgTxnBuffer will flush job in txn buffer to queue. We will also filter out job that don't need to send out in this stage.

func (*BinlogTailer) Start ¶

func (tailer *BinlogTailer) Start() error

func (*BinlogTailer) Wait ¶

func (tailer *BinlogTailer) Wait()

type MySQLBinlogInputPluginConfig ¶

type MySQLBinlogInputPluginConfig struct {
	Source                  *utils.DBConfig            `mapstructure:"source" toml:"source" json:"source"`
	IgnoreBiDirectionalData bool                       `mapstructure:"ignore-bidirectional-data" toml:"ignore-bidirectional-data" json:"ignore-bidirectional-data"`
	StartPosition           *utils.MySQLBinlogPosition `mapstructure:"start-position" toml:"start-position" json:"start-position"`

	SourceProbeCfg *helper.SourceProbeCfg `mapstructure:"source-probe-config"json:"source-probe-config"`

	//
	// internal configurations that is not exposed to users
	//
	DisableBinlogChecker bool   `mapstructure:"-"json:"-"`
	DebugBinlog          bool   `mapstructure:"-"json:"-"`
	BinlogSyncerTimeout  string `mapstructure:"-"json:"-"`
}

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL