scaling

Documentation

Overview

Package scaling provides the autoscaling capability for Numaflow.

A workqueue is implemented in this package to watch vertices in the cluster, calculate the desired replica number for each of them periodically, and patch the vertex spec.

Function StartWatching() and StopWatching() are also provided in the package, so that vertices can be added into and removed from the workqueue.

Index

• func KeyOfVertex(vertex dfv1.Vertex) string
• type Option
  • func WithBackPressureThreshold(n float64) Option
  • func WithTaskInterval(n int) Option
  • func WithWorkers(n int) Option
• type Scaler
  • func NewScaler(client client.Client, opts ...Option) *Scaler
  • func (s *Scaler) Contains(key string) bool
  • func (s *Scaler) Length() int
  • func (s *Scaler) Start(ctx context.Context) error
  • func (s *Scaler) StartWatching(key string)
  • func (s *Scaler) StopWatching(key string)

Functions

func KeyOfVertex

func KeyOfVertex(vertex dfv1.Vertex) string

KeyOfVertex returns the unique key of a vertex

Types

type Option

type Option func(*options)

func WithBackPressureThreshold

func WithBackPressureThreshold(n float64) Option

func WithTaskInterval

func WithTaskInterval(n int) Option

func WithWorkers

func WithWorkers(n int) Option

type Scaler

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

func NewScaler

func NewScaler(client client.Client, opts ...Option) *Scaler

NewScaler returns a Scaler instance.

func (*Scaler) Contains

func (s *Scaler) Contains(key string) bool

Contains returns if the Scaler contains the key.

func (*Scaler) Length

func (s *Scaler) Length() int

Length returns how many vertices are being watched for autoscaling

func (*Scaler) Start

func (s *Scaler) Start(ctx context.Context) error

Start function starts the autoscaling worker group. Each worker keeps picking up scaling tasks (which contains vertex keys) to calculate the desired replicas, and patch the vertex spec with the new replica number if needed.

func (*Scaler) StartWatching

func (s *Scaler) StartWatching(key string)

StartWatching put a key (namespace/name) into the Scaler

func (*Scaler) StopWatching

func (s *Scaler) StopWatching(key string)

StopWatching stops autoscaling on the key (namespace/name)