ghostferry

type AtomicBoolean int32

type BatchWriter struct {
	DB *sql.DB

	DatabaseRewrites map[string]string
	TableRewrites    map[string]string

	WriteRetries int
	// contains filtered or unexported fields
}

type BinlogDeleteEvent struct {
	*DMLEventBase
	// contains filtered or unexported fields
}

type BinlogInsertEvent struct {
	*DMLEventBase
	// contains filtered or unexported fields
}

type BinlogStreamer struct {
	Db           *sql.DB
	Config       *Config
	ErrorHandler ErrorHandler
	Filter       CopyFilter

	TableSchema TableSchemaCache
	// contains filtered or unexported fields
}

type BinlogUpdateEvent struct {
	*DMLEventBase
	// contains filtered or unexported fields
}

type BinlogWriter struct {
	DB               *sql.DB
	DatabaseRewrites map[string]string
	TableRewrites    map[string]string
	Throttler        Throttler

	BatchSize    int
	WriteRetries int

	ErrorHandler ErrorHandler
	// contains filtered or unexported fields
}

type ChecksumTableVerifier struct {
	Tables           []*schema.Table
	DatabaseRewrites map[string]string
	TableRewrites    map[string]string
	SourceDB         *sql.DB
	TargetDB         *sql.DB
	// contains filtered or unexported fields
}

type Config struct {
	// Source database connection configuration
	//
	// Required
	Source DatabaseConfig

	// Target database connection configuration
	//
	// Required
	Target DatabaseConfig

	// Map database name on the source database (key of the map) to a
	// different name on the target database (value of the associated key).
	// This allows one to move data and change the database name in the
	// process.
	//
	// Optional: defaults to empty map/no rewrites
	DatabaseRewrites map[string]string

	// Map the table name on the source dataabase to a different name on
	// the target database. See DatabaseRewrite.
	//
	// Optional: defaults to empty map/no rewrites
	TableRewrites map[string]string

	// The maximum number of retries for writes if the writes failed on
	// the target database.
	//
	// Optional: defaults to 5.
	DBWriteRetries int

	// Filter out the databases/tables when detecting the source databases
	// and tables.
	//
	// Required
	TableFilter TableFilter

	// Filter out unwanted data/events from being copied.
	//
	// Optional: defaults to nil/no filter.
	CopyFilter CopyFilter

	// The server id used by Ghostferry to connect to MySQL as a replication
	// slave. This id must be unique on the MySQL server. If 0 is specified,
	// a random id will be generated upon connecting to the MySQL server.
	//
	// Optional: defaults to an automatically generated one
	MyServerId uint32

	// The maximum number of binlog events to write at once. Note this is a
	// maximum: if there are not a lot of binlog events, they will be written
	// one at a time such the binlog streamer lag is as low as possible. This
	// batch size will only be hit if there is a log of binlog at the same time.
	//
	// Optional: defaults to 100
	BinlogEventBatchSize int

	// The batch size used to iterate the data during data copy. This batch size
	// is always used: if this is specified to be 100, 100 rows will be copied
	// per iteration.
	//
	// With the current implementation of Ghostferry, we need to lock the rows
	// we select. This means, the larger this number is, the longer we need to
	// hold this lock. On the flip side, the smaller this number is, the slower
	// the copy will likely be.
	//
	// Optional: defaults to 200
	DataIterationBatchSize uint64

	// The maximum number of retries for reads if the reads fail on the source
	// database.
	//
	// Optional: defaults to 5
	DBReadRetries int

	// This specify the number of concurrent goroutines, each iterating over
	// a single table.
	//
	// At this point in time, parallelize iteration within a single table. This
	// may be possible to add to the future.
	//
	// Optional: defaults to 4
	DataIterationConcurrency int

	// This specifies if Ghostferry will pause before cutover or not.
	//
	// Optional: defaults to false
	AutomaticCutover bool

	// Config for the ControlServer
	ServerBindAddr string
	WebBasedir     string
}

type ControlServer struct {
	F        *Ferry
	Verifier Verifier
	Addr     string
	Basedir  string
	// contains filtered or unexported fields
}

type CopyFilter interface {
	// BuildSelect is used to set up the query used for batch data copying,
	// allowing for restricting copying to a subset of data. Returning an error
	// here will cause the query to be retried, until the retry limit is
	// reached, at which point the ferry will be aborted. BuildSelect is passed
	// the columns to be selected, table being copied, the last primary key value
	// from the previous batch, and the batch size. Call DefaultBuildSelect to
	// generate the default query, which may be used as a starting point.
	BuildSelect([]string, *schema.Table, uint64, uint64) (sq.SelectBuilder, error)

	// ApplicableEvent is used to filter events for rows that have been
	// filtered in ConstrainSelect. ApplicableEvent should return true if the
	// event is for a row that would be selected by ConstrainSelect, and false
	// otherwise.
	// Returning an error here will cause the ferry to be aborted.
	ApplicableEvent(DMLEvent) (bool, error)
}

type CountMetric struct {
	MetricBase
	Value int64
}

type Cursor struct {
	CursorConfig

	Table         *schema.Table
	MaxPrimaryKey uint64
	RowLock       bool
	// contains filtered or unexported fields
}

type CursorConfig struct {
	DB        *sql.DB
	Throttler Throttler

	ColumnsToSelect []string
	BuildSelect     func([]string, *schema.Table, uint64, uint64) (squirrel.SelectBuilder, error)
	BatchSize       uint64
	ReadRetries     int
}

type DMLEvent interface {
	Database() string
	Table() string
	TableSchema() *schema.Table
	AsSQLString(target *schema.Table) (string, error)
	OldValues() RowData
	NewValues() RowData
	PK() (uint64, error)
}

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

type DataIterator struct {
	DB          *sql.DB
	Tables      []*schema.Table
	Concurrency int

	ErrorHandler ErrorHandler
	CursorConfig *CursorConfig

	CurrentState *DataIteratorState
	// contains filtered or unexported fields
}

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

type DatabaseConfig struct {
	Host      string
	Port      uint16
	User      string
	Pass      string
	Collation string
	Params    map[string]string

	TLS *TLSConfig
}

type ErrorHandler interface {
	Fatal(from string, err error)
}

type Ferry struct {
	*Config

	SourceDB *sql.DB
	TargetDB *sql.DB

	BinlogStreamer *BinlogStreamer
	BinlogWriter   *BinlogWriter

	DataIterator *DataIterator
	BatchWriter  *BatchWriter

	ErrorHandler ErrorHandler
	Throttler    Throttler

	Tables TableSchemaCache

	StartTime    time.Time
	DoneTime     time.Time
	OverallState string
	// contains filtered or unexported fields
}

type GaugeMetric struct {
	MetricBase
	Value float64
}

type IncompleteVerificationError struct{}

type IterativeVerifier struct {
	CursorConfig     *CursorConfig
	BinlogStreamer   *BinlogStreamer
	TableSchemaCache TableSchemaCache
	SourceDB         *sql.DB
	TargetDB         *sql.DB

	Tables           []*schema.Table
	IgnoredTables    []string
	DatabaseRewrites map[string]string
	TableRewrites    map[string]string
	Concurrency      int
	// contains filtered or unexported fields
}

type LagThrottler struct {
	ThrottlerBase
	PauserThrottler

	DB *sql.DB
	// contains filtered or unexported fields
}

type LagThrottlerConfig struct {
	Connection     DatabaseConfig
	MaxLag         int
	Query          string
	UpdateInterval string
}

type MetricBase struct {
	Key        string
	Tags       []MetricTag
	SampleRate float64
}

type MetricTag struct {
	Name  string
	Value string
}

type Metrics struct {
	Prefix      string
	DefaultTags []MetricTag
	Sink        chan interface{}
	// contains filtered or unexported fields
}

type PKPositionLog struct {
	Position uint64
	At       time.Time
}

type PanicErrorHandler struct {
	Ferry *Ferry
	// contains filtered or unexported fields
}

type PauserThrottler struct {
	ThrottlerBase
	// contains filtered or unexported fields
}

type ReverifyBatch struct {
	Pks   []uint64
	Table TableIdentifier
}

type ReverifyEntry struct {
	Pk    uint64
	Table *schema.Table
}

type ReverifyStore struct {
	MapStore map[TableIdentifier]map[uint64]struct{}

	BatchStore         []ReverifyBatch
	RowCount           uint64
	EmitLogPerRowCount uint64
	// contains filtered or unexported fields
}

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

type RowData []interface{}

type SqlDBWithFakeRollback struct {
	*sql.DB
}

type SqlPreparer interface {
	Prepare(string) (*sql.Stmt, error)
}

type SqlPreparerAndRollbacker interface {
	SqlPreparer
	Rollback() error
}

type Status struct {
	GhostferryVersion string

	SourceHostPort string
	TargetHostPort string

	OverallState      string
	StartTime         time.Time
	CurrentTime       time.Time
	TimeTaken         time.Duration
	ETA               time.Duration
	BinlogStreamerLag time.Duration
	PKsPerSecond      uint64

	AutomaticCutover            bool
	BinlogStreamerStopRequested bool
	LastSuccessfulBinlogPos     mysql.Position
	TargetBinlogPos             mysql.Position

	Throttled bool

	CompletedTableCount int
	TotalTableCount     int
	TableStatuses       []*TableStatus
	AllTableNames       []string
	AllDatabaseNames    []string

	VerifierSupport     bool
	VerifierAvailable   bool
	VerificationStarted bool
	VerificationDone    bool
	VerificationResult  VerificationResult
	VerificationErr     error
}

type TLSConfig struct {
	CertPath   string
	ServerName string
	// contains filtered or unexported fields
}

type TableFilter interface {
	ApplicableTables([]*schema.Table) ([]*schema.Table, error)
	ApplicableDatabases([]string) ([]string, error)
}

type TableIdentifier struct {
	SchemaName string
	TableName  string
}

type TableSchemaCache map[string]*schema.Table

type TableStatus struct {
	TableName        string
	PrimaryKeyName   string
	Status           string
	LastSuccessfulPK uint64
	TargetPK         uint64
}

type Throttler interface {
	Throttled() bool
	Disabled() bool
	SetDisabled(bool)
	SetPaused(bool)
	Run(context.Context) error
}

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

type TimerMetric struct {
	MetricBase
	Value time.Duration
}

type VerificationResult struct {
	DataCorrect bool
	Message     string
}

type VerificationResultAndStatus struct {
	VerificationResult

	StartTime time.Time
	DoneTime  time.Time
}

type Verifier interface {
	// Start the verifier in the background during the cutover phase.
	// Traditionally, this is called from within the ControlServer.
	//
	// This method maybe called multiple times and it's up to the verifier
	// to decide if it is possible to re-run the verification.
	StartInBackground() error

	// Wait for the verifier until it finishes verification after it was
	// started with the StartInBackground.
	//
	// A verification is "done" when it verified the dbs (either
	// correct or incorrect) OR when it experiences an error.
	Wait()

	// Returns the result and the status of the verification.
	// To check the status, call IsStarted() and IsDone() on
	// VerificationResultAndStatus.
	//
	// If the verification has been completed successfully (without errors) and
	// the data checks out to be "correct", the result will be
	// VerificationResult{true, ""}, with error = nil.
	// Otherwise, the result will be VerificationResult{false, "message"}, with
	// error = nil.
	//
	// If the verification is "done" but experienced an error during the check,
	// the result will be VerificationResult{}, with err = yourErr.
	Result() (VerificationResultAndStatus, error)
}

type WorkerPool struct {
	Concurrency int
	Process     func(int) (interface{}, error)
}