rangedloop

Documentation

Index

• Variables
• func CreateUUIDBoundaries(nRanges uint32) ([]uuid.UUID, error)
• func MakeUUIDWithTopBits(topBits uint32) (uuid.UUID, error)
• type Config
• type LiveCountObserver
  • func NewLiveCountObserver(metabase *metabase.DB, suspiciousProcessedRatio float64, ...) *LiveCountObserver
  • func (o *LiveCountObserver) Finish(ctx context.Context) (err error)
  • func (o *LiveCountObserver) Fork(ctx context.Context) (Partial, error)
  • func (o *LiveCountObserver) Join(ctx context.Context, partial Partial) error
  • func (o *LiveCountObserver) Process(ctx context.Context, segments []Segment) error
  • func (o *LiveCountObserver) Start(ctx context.Context, startTime time.Time) (err error)
  • func (o *LiveCountObserver) Stats(cb func(key monkit.SeriesKey, field string, val float64))
• type MetabaseRangeSplitter
  • func NewMetabaseRangeSplitter(db *metabase.DB, asOfSystemInterval time.Duration, batchSize int) *MetabaseRangeSplitter
  • func (provider *MetabaseRangeSplitter) CreateRanges(nRanges int, batchSize int) ([]SegmentProvider, error)
• type MetabaseSegmentProvider
  • func (provider *MetabaseSegmentProvider) Iterate(ctx context.Context, fn func([]Segment) error) error
  • func (provider *MetabaseSegmentProvider) Range() UUIDRange
• type Observer
• type ObserverDuration
• type Partial
• type RangeSplitter
• type Segment
  • func (s *Segment) Expired(now time.Time) bool
  • func (s Segment) Inline() bool
  • func (s Segment) PieceSize() int64
• type SegmentProvider
• type SequenceObserver
  • func NewSequenceObserver(observers ...Observer) *SequenceObserver
  • func (o *SequenceObserver) Finish(ctx context.Context) (err error)
  • func (o *SequenceObserver) Fork(ctx context.Context) (Partial, error)
  • func (o *SequenceObserver) Join(ctx context.Context, partial Partial) error
  • func (o *SequenceObserver) Start(ctx context.Context, startTime time.Time) (err error)
• type Service
  • func NewService(log *zap.Logger, config Config, provider RangeSplitter, observers []Observer) *Service
  • func (service *Service) Close() error
  • func (service *Service) Run(ctx context.Context) (err error)
  • func (service *Service) RunOnce(ctx context.Context) (observerDurations []ObserverDuration, err error)
• type UUIDRange
  • func CreateUUIDRanges(nRanges uint32) ([]UUIDRange, error)

Variables

var (

	// Error is a standard error class for this component.
	Error = errs.Class("ranged loop")
)

Functions

func CreateUUIDBoundaries

func CreateUUIDBoundaries(nRanges uint32) ([]uuid.UUID, error)

CreateUUIDBoundaries splits up the entire 128-bit UUID range into equal parts.

func MakeUUIDWithTopBits

func MakeUUIDWithTopBits(topBits uint32) (uuid.UUID, error)

MakeUUIDWithTopBits creates a zeroed UUID with the top 32 bits set from the input. Technically the result is not a UUID since it doesn't have the version and variant bits set.

Types

type Config

type Config struct {
	Parallelism        int           `help:"how many chunks of segments to process in parallel" default:"2"`
	BatchSize          int           `help:"how many items to query in a batch" default:"2500"`
	AsOfSystemInterval time.Duration `help:"as of system interval" releaseDefault:"-5m" devDefault:"-1us" testDefault:"-1us"`
	Interval           time.Duration `help:"how often to run the loop" releaseDefault:"2h" devDefault:"10s" testDefault:"0"`

	SuspiciousProcessedRatio float64 `help:"ratio where to consider processed count as supicious" default:"0.03"`
}

Config contains configurable values for the shared loop.

type LiveCountObserver

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

LiveCountObserver reports a count of segments during loop execution. This can be used to report the rate and progress of the loop. TODO we may need better name for this type.

func NewLiveCountObserver

func NewLiveCountObserver(metabase *metabase.DB, suspiciousProcessedRatio float64, asOfSystemInterval time.Duration) *LiveCountObserver

NewLiveCountObserver . To avoid pollution, make sure to only use one instance of this observer. Also make sure to only add it to instances of the loop which are actually doing something.

func (*LiveCountObserver) Finish

func (o *LiveCountObserver) Finish(ctx context.Context) (err error)

Finish gets segments count after range execution and verifies them against processed segments and segments in table before loop execution.

func (*LiveCountObserver) Fork

func (o *LiveCountObserver) Fork(ctx context.Context) (Partial, error)

Fork returns a shared instance so we have a view of all loop ranges.

func (*LiveCountObserver) Join

func (o *LiveCountObserver) Join(ctx context.Context, partial Partial) error

Join does nothing because the instance is shared across ranges.

func (*LiveCountObserver) Process

func (o *LiveCountObserver) Process(ctx context.Context, segments []Segment) error

Process increments the counter.

func (*LiveCountObserver) Start

func (o *LiveCountObserver) Start(ctx context.Context, startTime time.Time) (err error)

Start resets the count at start of the ranged segment loop and gets statistis about segments table.

func (*LiveCountObserver) Stats

func (o *LiveCountObserver) Stats(cb func(key monkit.SeriesKey, field string, val float64))

Stats implements monkit.StatSource to report the number of segments.

type MetabaseRangeSplitter

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

MetabaseRangeSplitter implements RangeSplitter.

func NewMetabaseRangeSplitter

func NewMetabaseRangeSplitter(db *metabase.DB, asOfSystemInterval time.Duration, batchSize int) *MetabaseRangeSplitter

NewMetabaseRangeSplitter creates the segment provider.

func (*MetabaseRangeSplitter) CreateRanges

func (provider *MetabaseRangeSplitter) CreateRanges(nRanges int, batchSize int) ([]SegmentProvider, error)

CreateRanges splits the segment table into chunks.

type MetabaseSegmentProvider

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

MetabaseSegmentProvider implements SegmentProvider.

func (*MetabaseSegmentProvider) Iterate

func (provider *MetabaseSegmentProvider) Iterate(ctx context.Context, fn func([]Segment) error) error

Iterate loops over a part of the segment table.

func (*MetabaseSegmentProvider) Range

func (provider *MetabaseSegmentProvider) Range() UUIDRange

Range returns range which is processed by this provider.

type Observer

type Observer interface {
	// Start is called at the beginning of each segment loop.
	Start(context.Context, time.Time) error

	// Fork creates a Partial to process a chunk of all the segments. It is
	// called after Start. It is not called concurrently.
	Fork(context.Context) (Partial, error)

	// Join is called for each partial returned by Fork. This gives the
	// opportunity to merge the output like in a reduce step. It will be called
	// before Finish. It is not called concurrently.
	Join(context.Context, Partial) error

	// Finish is called after all segments are processed by all observers.
	Finish(context.Context) error
}

Observer subscribes to the parallel segment loop. It is intended that a naïve implementation is threadsafe.

type ObserverDuration

type ObserverDuration struct {
	Observer Observer
	// Duration is set to -1 when the observer has errored out
	// so someone watching metrics can tell that something went wrong.
	Duration time.Duration
}

ObserverDuration reports back on how long it took the observer to process all the segments.

type Partial

type Partial interface {
	// Process is called repeatedly with batches of segments.
	// It is not called concurrently on the same instance.
	Process(context.Context, []Segment) error
}

Partial processes a part of the total range of segments.

type RangeSplitter

type RangeSplitter interface {
	CreateRanges(nRanges int, batchSize int) ([]SegmentProvider, error)
}

RangeSplitter splits a source of segments into ranges, so that multiple segments can be processed concurrently. It usually abstracts over a database. It is a subcomponent of the ranged segment loop.

type Segment

type Segment metabase.LoopSegmentEntry

Segment contains information about segment metadata which will be received by observers.

func (*Segment) Expired

func (s *Segment) Expired(now time.Time) bool

Expired checks if segment expired relative to now.

func (Segment) Inline

func (s Segment) Inline() bool

Inline returns true if segment is inline.

func (Segment) PieceSize

func (s Segment) PieceSize() int64

PieceSize returns calculated piece size for segment.

type SegmentProvider

type SegmentProvider interface {
	Range() UUIDRange
	Iterate(ctx context.Context, fn func([]Segment) error) error
}

SegmentProvider iterates through a range of segments.

type SequenceObserver

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

SequenceObserver provides ability to run observers from the list sequentially through next loop iterations. TODO find better name.

func NewSequenceObserver

func NewSequenceObserver(observers ...Observer) *SequenceObserver

NewSequenceObserver creates new SequenceObserver instance.

func (*SequenceObserver) Finish

func (o *SequenceObserver) Finish(ctx context.Context) (err error)

Finish passes Finish operation to current observer.

func (*SequenceObserver) Fork

func (o *SequenceObserver) Fork(ctx context.Context) (Partial, error)

Fork passes Fork operation to current observer.

func (*SequenceObserver) Join

func (o *SequenceObserver) Join(ctx context.Context, partial Partial) error

Join passes Join operation to current observer.

func (*SequenceObserver) Start

func (o *SequenceObserver) Start(ctx context.Context, startTime time.Time) (err error)

Start passes Start operation to current observer.

type Service

type Service struct {
	Loop *sync2.Cycle
	// contains filtered or unexported fields
}

Service iterates through all segments and calls the attached observers for every segment

architecture: Service

func NewService

func NewService(log *zap.Logger, config Config, provider RangeSplitter, observers []Observer) *Service

NewService creates a new instance of the ranged loop service.

func (*Service) Close

func (service *Service) Close() error

Close stops the ranged loop.

func (*Service) Run

func (service *Service) Run(ctx context.Context) (err error)

Run starts the looping service.

func (*Service) RunOnce

func (service *Service) RunOnce(ctx context.Context) (observerDurations []ObserverDuration, err error)

RunOnce goes through one time and sends information to observers.

type UUIDRange

type UUIDRange struct {
	Start *uuid.UUID
	End   *uuid.UUID
}

UUIDRange describes a range of UUID values. Start and End can be open-ended.

func CreateUUIDRanges

func CreateUUIDRanges(nRanges uint32) ([]UUIDRange, error)

CreateUUIDRanges splits up the entire 128-bit UUID range into equal parts.