operations

Documentation

Index

• Constants
• Variables
• func RequestCanceledCheck(err error) error
• type BackfillOperation
  • func NewBackfillOperation(ctx context.Context, plan config.OperationPlan, cancelFunc context.CancelFunc) (*BackfillOperation, error)
  • func (o *BackfillOperation) Checkpoint() string
  • func (o *BackfillOperation) Preflights(_ *dynamodb.DescribeTableOutput, _ *dynamodb.DescribeTableOutput) error
  • func (o *BackfillOperation) Rate() string
  • func (o *BackfillOperation) Run() error
  • func (o *BackfillOperation) Status() string
  • func (o *BackfillOperation) UpdateConsumedCapacity(capacities []*dynamodb.ConsumedCapacity)
• type BackfillRecord
• type DescribeOperation
  • func NewDescribeOperation(ctx context.Context, plan config.OperationPlan, cancelFunc context.CancelFunc) (*DescribeOperation, error)
  • func (o *DescribeOperation) ApproximateItemCount() string
  • func (o *DescribeOperation) ApproximateTableSize() string
  • func (o *DescribeOperation) Start()
  • func (o *DescribeOperation) Status() string
  • func (o *DescribeOperation) Stop()
• type ErrorCollator
  • func (c *ErrorCollator) Register(f func() error)
  • func (c *ErrorCollator) Run() error
• type LatencyLock
  • func (l *LatencyLock) Status() string
  • func (l *LatencyLock) Update(lastCheck time.Time)
• type Operation
• type OperationPhase
• type Operator
  • func NewOperator(ctx context.Context, plan config.OperationPlan, cancelFunc context.CancelFunc) (*Operator, error)
  • func (o *Operator) Checkpoint() string
  • func (o *Operator) Preflights() error
  • func (o *Operator) Run() error
  • func (o *Operator) Status() *status.Status
• type Phase
  • func (p *Phase) Complete() bool
  • func (p *Phase) Error() error
  • func (p *Phase) Errored() bool
  • func (p *Phase) Finish() error
  • func (p *Phase) Running() bool
  • func (p *Phase) Start() error
  • func (p *Phase) Status() string
  • func (p *Phase) StatusCode() OperationPhase
• type RateTracker
  • func NewRateTracker(rateType string, tickFreq time.Duration) *RateTracker
  • func (t *RateTracker) ApproximateCount() string
  • func (t *RateTracker) Count() int64
  • func (t *RateTracker) Duration() time.Duration
  • func (t *RateTracker) Increment(by int64)
  • func (t *RateTracker) RatePerSecond() string
  • func (t *RateTracker) Start()
  • func (t *RateTracker) Stop()
• type StreamOperation
  • func NewStreamOperation(ctx context.Context, plan config.OperationPlan, cancelFunc context.CancelFunc) (*StreamOperation, error)
  • func (o *StreamOperation) Checkpoint() string
  • func (o *StreamOperation) Preflights(in *dynamodb.DescribeTableOutput, _ *dynamodb.DescribeTableOutput) error
  • func (o *StreamOperation) Rate() string
  • func (o *StreamOperation) Run() error
  • func (o *StreamOperation) Status() string

Constants

const (
	NotStartedPhase operatorPhase = iota
	BackfillPhase
	StreamPhase
	NoopPhase
	CompletedPhase
)

Variables

var ErrOperationFailed = errors.New("Operation failed")

Functions

func RequestCanceledCheck

func RequestCanceledCheck(err error) error

Types

type BackfillOperation

type BackfillOperation struct {
	OperationPlan config.OperationPlan
	// contains filtered or unexported fields
}

func NewBackfillOperation

func NewBackfillOperation(ctx context.Context, plan config.OperationPlan, cancelFunc context.CancelFunc) (*BackfillOperation, error)

func (*BackfillOperation) Checkpoint

func (o *BackfillOperation) Checkpoint() string

Checkpoint prints a logging statement summarizing the current state. Meant for periodic update requests.

func (*BackfillOperation) Preflights

func (o *BackfillOperation) Preflights(_ *dynamodb.DescribeTableOutput, _ *dynamodb.DescribeTableOutput) error

func (*BackfillOperation) Rate

func (o *BackfillOperation) Rate() string

func (*BackfillOperation) Run

func (o *BackfillOperation) Run() error

func (*BackfillOperation) Status

func (o *BackfillOperation) Status() string

func (*BackfillOperation) UpdateConsumedCapacity

func (o *BackfillOperation) UpdateConsumedCapacity(capacities []*dynamodb.ConsumedCapacity)

type BackfillRecord

type BackfillRecord map[string]*dynamodb.AttributeValue

type DescribeOperation

type DescribeOperation struct {
	OperationPlan config.OperationPlan
	// contains filtered or unexported fields
}

func NewDescribeOperation

func NewDescribeOperation(ctx context.Context, plan config.OperationPlan, cancelFunc context.CancelFunc) (*DescribeOperation, error)

func (*DescribeOperation) ApproximateItemCount

func (o *DescribeOperation) ApproximateItemCount() string

func (*DescribeOperation) ApproximateTableSize

func (o *DescribeOperation) ApproximateTableSize() string

func (*DescribeOperation) Start

func (o *DescribeOperation) Start()

func (*DescribeOperation) Status

func (o *DescribeOperation) Status() string

func (*DescribeOperation) Stop

func (o *DescribeOperation) Stop()

type ErrorCollator

type ErrorCollator struct {
	Funcs  []func() error
	Cancel func()
}

func (*ErrorCollator) Register

func (c *ErrorCollator) Register(f func() error)

func (*ErrorCollator) Run

func (c *ErrorCollator) Run() error

type LatencyLock

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

func (*LatencyLock) Status

func (l *LatencyLock) Status() string

func (*LatencyLock) Update

func (l *LatencyLock) Update(lastCheck time.Time)

type Operation

type Operation interface {
	Checkpoint() string
	Preflights(*dynamodb.DescribeTableOutput, *dynamodb.DescribeTableOutput) error
	Rate() string
	Run() error
	Status() string
}

type OperationPhase

type OperationPhase int

const (
	Initialized OperationPhase = iota
	Started
	Finished
	Errored
)

type Operator

type Operator struct {
	OperationPlan config.OperationPlan
	// contains filtered or unexported fields
}

func NewOperator

func NewOperator(ctx context.Context, plan config.OperationPlan, cancelFunc context.CancelFunc) (*Operator, error)

func (*Operator) Checkpoint

func (o *Operator) Checkpoint() string

func (*Operator) Preflights

func (o *Operator) Preflights() error

func (*Operator) Run

func (o *Operator) Run() error

func (*Operator) Status

func (o *Operator) Status() *status.Status

type Phase

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

Phase a representation of the status of an operational phase

func (*Phase) Complete

func (p *Phase) Complete() bool

func (*Phase) Error

func (p *Phase) Error() error

Error mark the phase as errored

func (*Phase) Errored

func (p *Phase) Errored() bool

func (*Phase) Finish

func (p *Phase) Finish() error

Finish mark the phase as finished

func (*Phase) Running

func (p *Phase) Running() bool

func (*Phase) Start

func (p *Phase) Start() error

Start mark the phase as started

func (*Phase) Status

func (p *Phase) Status() string

Status return the string representation of the phase status

func (*Phase) StatusCode

func (p *Phase) StatusCode() OperationPhase

StatusCode returns the representative integer of the status 0 -> Initialized 1 -> Started 2 -> Finished 3 -> Errored

type RateTracker

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

RateTracker is a structure to hold and manage a rate tracking loop, counters and timers

func NewRateTracker

func NewRateTracker(rateType string, tickFreq time.Duration) *RateTracker

Return a new RateTracker

func (*RateTracker) ApproximateCount

func (t *RateTracker) ApproximateCount() string

ApproximateCount the approximate count of records we've written during the lifetime of the struct

func (*RateTracker) Count

func (t *RateTracker) Count() int64

Count the absolute count of records we've written during the lifetime of the struct

func (*RateTracker) Duration

func (t *RateTracker) Duration() time.Duration

Duration the duration since we started

func (*RateTracker) Increment

func (t *RateTracker) Increment(by int64)

Increment takes a value to add to the current counter

func (*RateTracker) RatePerSecond

func (t *RateTracker) RatePerSecond() string

RatePerSecond returns a pretty formatted description of the rate from the last completed window

func (*RateTracker) Start

func (t *RateTracker) Start()

Start start the rate tracker

func (*RateTracker) Stop

func (t *RateTracker) Stop()

Stop stop the rate tracker

type StreamOperation

type StreamOperation struct {
	OperationPlan config.OperationPlan
	// contains filtered or unexported fields
}

func NewStreamOperation

func NewStreamOperation(ctx context.Context, plan config.OperationPlan, cancelFunc context.CancelFunc) (*StreamOperation, error)

func (*StreamOperation) Checkpoint

func (o *StreamOperation) Checkpoint() string

Checkpoint is a periodic status output meant for historical tracking. This will be called when an update is desired.

func (*StreamOperation) Preflights

func (o *StreamOperation) Preflights(in *dynamodb.DescribeTableOutput, _ *dynamodb.DescribeTableOutput) error

func (*StreamOperation) Rate

func (o *StreamOperation) Rate() string

func (*StreamOperation) Run

func (o *StreamOperation) Run() error

func (*StreamOperation) Status

func (o *StreamOperation) Status() string