flowcontrol

Documentation

Index

• Constants
• func RequestDelegated(ctx context.Context)
• func WatchInitialized(ctx context.Context)
• func WithInitializationSignal(ctx context.Context, signal InitializationSignal) context.Context
• type ForgetWatchFunc
• type InitializationSignal
  • func NewInitializationSignal() InitializationSignal
• type Interface
  • func New(informerFactory kubeinformers.SharedInformerFactory, ...) Interface
  • func NewTestable(config TestableConfig) Interface
• type MaxSeatsTracker
  • func NewMaxSeatsTracker() MaxSeatsTracker
• type RequestDigest
  • func (rd RequestDigest) GoString() string
• type StartFunction
• type TestableConfig
• type WatchTracker
  • func NewWatchTracker() WatchTracker

Constants

const ConfigConsumerAsFieldManager = "api-priority-and-fairness-config-consumer-v1"

ConfigConsumerAsFieldManager is how the config consuminng controller appears in an ObjectMeta ManagedFieldsEntry.Manager

const MinTarget = 0.001

Functions

func RequestDelegated

func RequestDelegated(ctx context.Context)

RequestDelegated informs the priority and fairness dispatcher that a given request has been delegated to an aggregated API server. No-op when priority and fairness is disabled.

func WatchInitialized

func WatchInitialized(ctx context.Context)

WatchInitialized sends a signal to priority and fairness dispatcher that a given watch request has already been initialized.

func WithInitializationSignal

func WithInitializationSignal(ctx context.Context, signal InitializationSignal) context.Context

WithInitializationSignal creates a copy of parent context with priority and fairness initialization signal value.

Types

type ForgetWatchFunc

type ForgetWatchFunc func()

ForgetWatchFunc is a function that should be called to forget the previously registered watch from the watch tracker.

type InitializationSignal

type InitializationSignal interface {
	// Signal notifies the dispatcher about finished initialization.
	Signal()
	// Wait waits for the initialization signal.
	Wait()
}

InitializationSignal is an interface that allows sending and handling initialization signals.

func NewInitializationSignal

func NewInitializationSignal() InitializationSignal

type Interface

type Interface interface {
	// Handle takes care of queuing and dispatching a request
	// characterized by the given digest.  The given `noteFn` will be
	// invoked with the results of request classification.
	// The given `workEstimator` is called, if at all, after noteFn.
	// `workEstimator` will be invoked only when the request
	//  is classified as non 'exempt'.
	// 'workEstimator', when invoked, must return the
	// work parameters for the request.
	// If the request is queued then `queueNoteFn` will be called twice,
	// first with `true` and then with `false`; otherwise
	// `queueNoteFn` will not be called at all.  If Handle decides
	// that the request should be executed then `execute()` will be
	// invoked once to execute the request; otherwise `execute()` will
	// not be invoked.
	// Handle() should never return while execute() is running, even if
	// ctx is cancelled or times out.
	Handle(ctx context.Context,
		requestDigest RequestDigest,
		noteFn func(fs *flowcontrol.FlowSchema, pl *flowcontrol.PriorityLevelConfiguration, flowDistinguisher string),
		workEstimator func() fcrequest.WorkEstimate,
		queueNoteFn fq.QueueNoteFn,
		execFn func(),
	)

	// Run monitors config objects from the main apiservers and causes
	// any needed changes to local behavior.  This method ceases
	// activity and returns after the given channel is closed.
	Run(stopCh <-chan struct{}) error

	// Install installs debugging endpoints to the web-server.
	Install(c *mux.PathRecorderMux)

	// WatchTracker provides the WatchTracker interface.
	WatchTracker

	// MaxSeatsTracker is invoked from the work estimator to track max seats
	// that can be occupied by a request for a priority level.
	MaxSeatsTracker
}

Interface defines how the API Priority and Fairness filter interacts with the underlying system.

func New

func New(
	informerFactory kubeinformers.SharedInformerFactory,
	flowcontrolClient flowcontrolclient.FlowcontrolV1beta3Interface,
	serverConcurrencyLimit int,
	requestWaitLimit time.Duration,
) Interface

New creates a new instance to implement API priority and fairness

func NewTestable

func NewTestable(config TestableConfig) Interface

NewTestable is extra flexible to facilitate testing

type MaxSeatsTracker

type MaxSeatsTracker interface {
	// GetMaxSeats returns the maximum seats a request should occupy for a given priority level.
	GetMaxSeats(priorityLevelName string) uint64

	// SetMaxSeats configures max seats for a priority level.
	SetMaxSeats(priorityLevelName string, maxSeats uint64)

	// ForgetPriorityLevel removes max seats tracking for a priority level.
	ForgetPriorityLevel(priorityLevelName string)
}

MaxSeatsTracker is used to track max seats allocatable per priority level from the work estimator

func NewMaxSeatsTracker

func NewMaxSeatsTracker() MaxSeatsTracker

type RequestDigest

type RequestDigest struct {
	RequestInfo *request.RequestInfo
	User        user.Info
}

RequestDigest holds necessary info from request for flow-control

func (RequestDigest) GoString

func (rd RequestDigest) GoString() string

GoString produces a golang source expression of the value.

type StartFunction

type StartFunction func(ctx context.Context, hashValue uint64) (execute bool, afterExecution func())

StartFunction begins the process of handling a request. If the request gets queued then this function uses the given hashValue as the source of entropy as it shuffle-shards the request into a queue. The descr1 and descr2 values play no role in the logic but appear in log messages. This method does not return until the queuing, if any, for this request is done. If `execute` is false then `afterExecution` is irrelevant and the request should be rejected. Otherwise the request should be executed and `afterExecution` must be called exactly once.

type TestableConfig

type TestableConfig struct {
	// Name of the controller
	Name string

	// Clock to use in timing deliberate delays
	Clock clock.PassiveClock

	// AsFieldManager is the string to use in the metadata for
	// server-side apply.  Normally this is
	// `ConfigConsumerAsFieldManager`.  This is exposed as a parameter
	// so that a test of competing controllers can supply different
	// values.
	AsFieldManager string

	// FoundToDangling maps the boolean indicating whether a
	// FlowSchema's referenced PLC exists to the boolean indicating
	// that FlowSchema's status should indicate a dangling reference.
	// This is a parameter so that we can write tests of what happens
	// when servers disagree on that bit of Status.
	FoundToDangling func(bool) bool

	// InformerFactory to use in building the controller
	InformerFactory kubeinformers.SharedInformerFactory

	// FlowcontrolClient to use for manipulating config objects
	FlowcontrolClient flowcontrolclient.FlowcontrolV1beta3Interface

	// ServerConcurrencyLimit for the controller to enforce
	ServerConcurrencyLimit int

	// RequestWaitLimit configured on the server
	RequestWaitLimit time.Duration

	// GaugeVec for metrics about requests, broken down by phase and priority_level
	ReqsGaugeVec metrics.RatioedGaugeVec

	// RatioedGaugePairVec for metrics about seats occupied by all phases of execution
	ExecSeatsGaugeVec metrics.RatioedGaugeVec

	// QueueSetFactory for the queuing implementation
	QueueSetFactory fq.QueueSetFactory
}

TestableConfig carries the parameters to an implementation that is testable

type WatchTracker

type WatchTracker interface {
	// RegisterWatch reqisters a watch based on the provided http.Request
	// in the tracker. It returns the function that should be called
	// to forget the watcher once it is finished.
	RegisterWatch(r *http.Request) ForgetWatchFunc

	// GetInterestedWatchCount returns the number of watches that are
	// potentially interested in a request with a given RequestInfo
	// for the purpose of estimating cost of that request.
	GetInterestedWatchCount(requestInfo *request.RequestInfo) int
}

WatchTracker is an interface that allows tracking the number of watches in the system for the purpose of estimating the cost of incoming mutating requests.

func NewWatchTracker

func NewWatchTracker() WatchTracker