scheduler

Documentation

Index

• Constants
• Variables
• type Client
• type Metrics
  • func (s *Metrics) Equal(other *Metrics) bool
  • func (s *Metrics) TotalErrors() uint64
  • func (s *Metrics) TotalErrorsAtomic() uint64
  • func (s *Metrics) TotalPanics() uint64
  • func (s *Metrics) TotalPanicsAtomic() uint64
  • func (s *Metrics) TotalResources() uint64
  • func (s *Metrics) TotalResourcesAtomic() uint64
• type Option
  • func WithConcurrency(concurrency int) Option
  • func WithLogger(logger zerolog.Logger) Option
  • func WithMaxDepth(maxDepth uint64) Option
  • func WithStrategy(strategy Strategy) Option
• type Scheduler
  • func NewScheduler(opts ...Option) *Scheduler
  • func (s *Scheduler) Sync(ctx context.Context, client schema.ClientMeta, tables schema.Tables, ...) error
  • func (s *Scheduler) SyncAll(ctx context.Context, client schema.ClientMeta, tables schema.Tables) (message.SyncMessages, error)
• type Strategies
  • func (s Strategies) String() string
• type Strategy
  • func StrategyForName(s string) (Strategy, error)
  • func (Strategy) JSONSchema() *jsonschema.Schema
  • func (s *Strategy) MarshalJSON() ([]byte, error)
  • func (s *Strategy) String() string
  • func (s *Strategy) UnmarshalJSON(b []byte) error
  • func (s *Strategy) Validate() error
• type SyncOption
  • func WithSyncDeterministicCQID(deterministicCQID bool) SyncOption
• type TableClientMetrics
  • func (s *TableClientMetrics) Equal(other *TableClientMetrics) bool

Constants

const (
	DefaultConcurrency = 50000
	DefaultMaxDepth    = 4
)

Variables

var AllStrategies = Strategies{StrategyDFS, StrategyRoundRobin, StrategyShuffle}

var AllStrategyNames = [...]string{
	StrategyDFS:        "dfs",
	StrategyRoundRobin: "round-robin",
	StrategyShuffle:    "shuffle",
}

var ErrNoTables = errors.New("no tables specified for syncing, review `tables` and `skip_tables` in your config and specify at least one table to sync")

Types

type Client

type Client interface {
	ID() string
}

type Metrics

type Metrics struct {
	TableClient map[string]map[string]*TableClientMetrics
}

Metrics is deprecated as we move toward open telemetry for tracing and metrics

func (*Metrics) Equal

func (s *Metrics) Equal(other *Metrics) bool

Equal compares to stats. Mostly useful in testing

func (*Metrics) TotalErrors

func (s *Metrics) TotalErrors() uint64

func (*Metrics) TotalErrorsAtomic

func (s *Metrics) TotalErrorsAtomic() uint64

func (*Metrics) TotalPanics

func (s *Metrics) TotalPanics() uint64

func (*Metrics) TotalPanicsAtomic

func (s *Metrics) TotalPanicsAtomic() uint64

func (*Metrics) TotalResources

func (s *Metrics) TotalResources() uint64

func (*Metrics) TotalResourcesAtomic

func (s *Metrics) TotalResourcesAtomic() uint64

type Option

type Option func(*Scheduler)

func WithConcurrency

func WithConcurrency(concurrency int) Option

func WithLogger

func WithLogger(logger zerolog.Logger) Option

func WithMaxDepth

func WithMaxDepth(maxDepth uint64) Option

func WithStrategy

func WithStrategy(strategy Strategy) Option

type Scheduler

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

func NewScheduler

func NewScheduler(opts ...Option) *Scheduler

func (*Scheduler) Sync

func (s *Scheduler) Sync(ctx context.Context, client schema.ClientMeta, tables schema.Tables, res chan<- message.SyncMessage, opts ...SyncOption) error

func (*Scheduler) SyncAll

func (s *Scheduler) SyncAll(ctx context.Context, client schema.ClientMeta, tables schema.Tables) (message.SyncMessages, error)

SyncAll is mostly used for testing as it will sync all tables and can run out of memory in the real world. Should use Sync for production.

type Strategies

type Strategies []Strategy

func (Strategies) String

func (s Strategies) String() string

type Strategy

type Strategy int

const (
	StrategyDFS Strategy = iota
	StrategyRoundRobin
	StrategyShuffle
)

func StrategyForName

func StrategyForName(s string) (Strategy, error)

func (Strategy) JSONSchema

func (Strategy) JSONSchema() *jsonschema.Schema

func (*Strategy) MarshalJSON

func (s *Strategy) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler.

func (*Strategy) String

func (s *Strategy) String() string

func (*Strategy) UnmarshalJSON

func (s *Strategy) UnmarshalJSON(b []byte) error

UnmarshalJSON implements json.Unmarshaler.

func (*Strategy) Validate

func (s *Strategy) Validate() error

type SyncOption

type SyncOption func(*syncClient)

func WithSyncDeterministicCQID

func WithSyncDeterministicCQID(deterministicCQID bool) SyncOption

type TableClientMetrics

type TableClientMetrics struct {
	Resources uint64
	Errors    uint64
	Panics    uint64
	StartTime time.Time
	EndTime   time.Time
}

func (*TableClientMetrics) Equal

func (s *TableClientMetrics) Equal(other *TableClientMetrics) bool