checkpoint

Documentation

Index

• Variables
• type Agent
  • func NewAgentImpl(jobID string, pLogger *zap.Logger) Agent
• type AgentImpl
  • func (c *AgentImpl) Create(ctx context.Context, cfg *config.JobCfg) error
  • func (c *AgentImpl) FetchAllDoTables(ctx context.Context, cfg *config.JobCfg) (map[metadata.TargetTable][]metadata.SourceTable, error)
  • func (c *AgentImpl) FetchTableStmt(ctx context.Context, jobID string, cfg *config.JobCfg, ...) (string, error)
  • func (c *AgentImpl) IsFresh(ctx context.Context, workerType framework.WorkerType, task *metadata.Task) (bool, error)
  • func (c *AgentImpl) Remove(ctx context.Context, cfg *config.JobCfg) error
  • func (c *AgentImpl) UpgradeFuncs() []bootstrap.UpgradeFunc

Variables

var NewCheckpointAgent = NewAgentImpl

NewCheckpointAgent is a method to create a new checkpoint agent

Types

type Agent

type Agent interface {
	Create(ctx context.Context, cfg *config.JobCfg) error
	Remove(ctx context.Context, cfg *config.JobCfg) error
	IsFresh(ctx context.Context, workerType framework.WorkerType, task *metadata.Task) (bool, error)
	Upgrade(ctx context.Context, preVer semver.Version) error
	FetchAllDoTables(ctx context.Context, cfg *config.JobCfg) (map[metadata.TargetTable][]metadata.SourceTable, error)
	// FetchTableStmt fetch create table statement from checkpoint.
	FetchTableStmt(ctx context.Context, jobID string, cfg *config.JobCfg, sourceTable metadata.SourceTable) (string, error)
}

Agent defeins a checkpoint agent interface

func NewAgentImpl

func NewAgentImpl(jobID string, pLogger *zap.Logger) Agent

NewAgentImpl creates a new AgentImpl instance

type AgentImpl

type AgentImpl struct {
	*bootstrap.DefaultUpgrader
	// contains filtered or unexported fields
}

AgentImpl implements Agent

func (*AgentImpl) Create

func (c *AgentImpl) Create(ctx context.Context, cfg *config.JobCfg) error

Create implements Agent.Create Create will be called when the job is created/restarted/updated. We create checkpoint table in master rather than in workers, to avoid the annoying log of "table already exists", because one job only need to create one checkpoint table per unit. move these codes to tiflow later.

func (*AgentImpl) FetchAllDoTables

func (c *AgentImpl) FetchAllDoTables(ctx context.Context, cfg *config.JobCfg) (map[metadata.TargetTable][]metadata.SourceTable, error)

FetchAllDoTables returns all need to do tables after filtered and routed (fetches from upstream MySQL).

func (*AgentImpl) FetchTableStmt

func (c *AgentImpl) FetchTableStmt(ctx context.Context, jobID string, cfg *config.JobCfg, sourceTable metadata.SourceTable) (string, error)

FetchTableStmt fetch create table statement from checkpoint. TODO(https://github.com/pingcap/tiflow/issues/5334): save create table statement to checkpoint instead of table info.

func (*AgentImpl) IsFresh

func (c *AgentImpl) IsFresh(ctx context.Context, workerType framework.WorkerType, task *metadata.Task) (bool, error)

IsFresh implements Agent.IsFresh

func (*AgentImpl) Remove

func (c *AgentImpl) Remove(ctx context.Context, cfg *config.JobCfg) error

Remove implements Agent.Remove

func (*AgentImpl) UpgradeFuncs

func (c *AgentImpl) UpgradeFuncs() []bootstrap.UpgradeFunc

UpgradeFuncs implement the Upgrader interface.