sharding

Documentation

Index

• func InitializeMetrics(prefix string, config *Config) error
• func SetGlobalMetrics(prefix string, metricsChan chan interface{})
• func StopAndFlushMetrics()
• type Config
• type HTTPCallback
  • func (h *HTTPCallback) Post(client *http.Client) error
• type JoinTable
• type ShardedCopyFilter
  • func (f *ShardedCopyFilter) ApplicableEvent(event ghostferry.DMLEvent) (bool, error)
  • func (f *ShardedCopyFilter) BuildSelect(columns []string, table *schema.Table, lastPk, batchSize uint64) (sq.SelectBuilder, error)
• type ShardedTableFilter
  • func (s *ShardedTableFilter) ApplicableDatabases(dbs []string) ([]string, error)
  • func (s *ShardedTableFilter) ApplicableTables(tables []*schema.Table) (applicable []*schema.Table, err error)
• type ShardingErrorHandler
  • func (this *ShardingErrorHandler) Fatal(from string, err error)
• type ShardingFerry
  • func NewFerry(config *Config) (*ShardingFerry, error)
  • func (r *ShardingFerry) Initialize() error
  • func (r *ShardingFerry) Run()
  • func (r *ShardingFerry) Start() error

Functions

func InitializeMetrics

func InitializeMetrics(prefix string, config *Config) error

func SetGlobalMetrics

func SetGlobalMetrics(prefix string, metricsChan chan interface{})

func StopAndFlushMetrics

func StopAndFlushMetrics()

Types

type Config

type Config struct {
	*ghostferry.Config

	ShardingKey   string
	ShardingValue int64
	SourceDB      string
	TargetDB      string

	StatsDAddress string
	CutoverLock   HTTPCallback
	CutoverUnlock HTTPCallback
	ErrorCallback HTTPCallback

	JoinedTables              map[string][]JoinTable
	IgnoredTables             []string
	IgnoredVerificationTables []string
	PrimaryKeyTables          []string

	VerifierIterationConcurrency int

	Throttle *ghostferry.LagThrottlerConfig
}

type HTTPCallback

type HTTPCallback struct {
	URI     string
	Payload string
}

func (*HTTPCallback) Post

func (h *HTTPCallback) Post(client *http.Client) error

type JoinTable

type JoinTable struct {
	TableName, JoinColumn string
}

type ShardedCopyFilter

type ShardedCopyFilter struct {
	ShardingKey      string
	ShardingValue    interface{}
	JoinedTables     map[string][]JoinTable
	PrimaryKeyTables map[string]struct{}
	// contains filtered or unexported fields
}

func (*ShardedCopyFilter) ApplicableEvent

func (f *ShardedCopyFilter) ApplicableEvent(event ghostferry.DMLEvent) (bool, error)

func (*ShardedCopyFilter) BuildSelect

func (f *ShardedCopyFilter) BuildSelect(columns []string, table *schema.Table, lastPk, batchSize uint64) (sq.SelectBuilder, error)

type ShardedTableFilter

type ShardedTableFilter struct {
	SourceShard      string
	ShardingKey      string
	JoinedTables     map[string][]JoinTable
	IgnoredTables    []*regexp.Regexp
	PrimaryKeyTables map[string]struct{}
}

func (*ShardedTableFilter) ApplicableDatabases

func (s *ShardedTableFilter) ApplicableDatabases(dbs []string) ([]string, error)

func (*ShardedTableFilter) ApplicableTables

func (s *ShardedTableFilter) ApplicableTables(tables []*schema.Table) (applicable []*schema.Table, err error)

type ShardingErrorHandler

type ShardingErrorHandler struct {
	ghostferry.ErrorHandler
	ErrorCallback HTTPCallback
	Logger        *logrus.Entry
}

func (*ShardingErrorHandler) Fatal

func (this *ShardingErrorHandler) Fatal(from string, err error)

type ShardingFerry

type ShardingFerry struct {
	Ferry *ghostferry.Ferry
	// contains filtered or unexported fields
}

func NewFerry

func NewFerry(config *Config) (*ShardingFerry, error)

func (*ShardingFerry) Initialize

func (r *ShardingFerry) Initialize() error

func (*ShardingFerry) Run

func (r *ShardingFerry) Run()

func (*ShardingFerry) Start

func (r *ShardingFerry) Start() error