queue

Documentation

Index

• Constants
• func TimeToFirstByteTimeoutHandler(h http.Handler, dt time.Duration, msg string) http.Handler
• type Breaker
  • func NewBreaker(queueDepth, maxConcurrency, initialCapacity int32) *Breaker
  • func (b *Breaker) Maybe(thunk func()) bool
• type Channels
• type Measurement
• type Reporter
  • func NewStatsReporter(namespace string, config string, revision string, pod string) (*Reporter, error)
  • func (r *Reporter) Report(lameDuck bool, operationsPerSecond float64, averageConcurrentRequests float64) error
  • func (r *Reporter) UnregisterViews() error
• type ReqEvent
• type ReqEventType
• type Semaphore
  • func NewSemaphore(maxCapacity, initialCapacity int32) *Semaphore
  • func (s *Semaphore) Acquire()
  • func (s *Semaphore) AddCapacity(size int32)
  • func (s *Semaphore) ReduceCapacity(size int32) error
  • func (s *Semaphore) Release()
• type Stats
  • func NewStats(podName string, channels Channels, startedAt time.Time) *Stats

Constants

const (
	// RequestQueueQuitPath specifies the path to send quit request to
	// queue-proxy. This is used for preStop hook of queue-proxy. It:
	// - marks the service as not ready, so that requests will no longer
	//   be routed to it,
	// - adds a small delay, so that the container doesn't get killed at
	//   the same time the pod is marked for removal.
	RequestQueueQuitPath = "quitquitquit"

	// RequestQueueHealthPath specifies the path for health checks for
	// queue-proxy.
	RequestQueueHealthPath = "health"
)

const (
	// ReportingPeriod interval of time for reporting.
	ReportingPeriod = 10 * time.Second

	// OperationsPerSecondN
	OperationsPerSecondN = "operations_per_second"
	// AverageConcurrentRequestsN
	AverageConcurrentRequestsN = "average_concurrent_requests"
	// LameDuckN
	LameDuckN = "lame_duck"

	// OperationsPerSecondM number of operations per second.
	OperationsPerSecondM Measurement = iota
	// AverageConcurrentRequestsM average number of requests currently being handled by this pod.
	AverageConcurrentRequestsM
	// LameDuckM indicates this Pod has received a shutdown signal.
	LameDuckM
)

Functions

func TimeToFirstByteTimeoutHandler

func TimeToFirstByteTimeoutHandler(h http.Handler, dt time.Duration, msg string) http.Handler

TimeToFirstByteTimeoutHandler returns a Handler that runs h with the given time limit in which the first byte of the response must be written.

The new Handler calls h.ServeHTTP to handle each request, but if a call runs for longer than its time limit, the handler responds with a 503 Service Unavailable error and the given message in its body. (If msg is empty, a suitable default message will be sent.) After such a timeout, writes by h to its ResponseWriter will return ErrHandlerTimeout.

A panic from the underlying handler is propagated as-is to be able to make use of custom panic behavior by HTTP handlers. See https://golang.org/pkg/net/http/#Handler.

The implementation is largely inspired by http.TimeoutHandler.

Types

type Breaker

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

Breaker is a component that enforces a concurrency limit on the execution of a function. It also maintains a queue of function executions in excess of the concurrency limit. Function call attempts beyond the limit of the queue are failed immediately.

func NewBreaker

func NewBreaker(queueDepth, maxConcurrency, initialCapacity int32) *Breaker

NewBreaker creates a Breaker with the desired queue depth, concurrency limit and initial capacity

func (*Breaker) Maybe

func (b *Breaker) Maybe(thunk func()) bool

Maybe conditionally executes thunk based on the Breaker concurrency and queue parameters. If the concurrency limit and queue capacity are already consumed, Maybe returns immediately without calling thunk. If the thunk was executed, Maybe returns true, else false.

type Channels

type Channels struct {
	// Ticks with every request arrived/completed respectively
	ReqChan chan ReqEvent
	// Ticks with every stat report request
	ReportChan <-chan time.Time
	// Stat reporting channel
	StatChan chan *autoscaler.Stat
}

Channels is a structure for holding the channels for driving Stats. It's just to make the NewStats signature easier to read.

type Measurement

type Measurement int

Measurement type for reporting.

type Reporter

type Reporter struct {
	Initialized bool
	// contains filtered or unexported fields
}

Reporter structure representing a prometheus expoerter.

func NewStatsReporter

func NewStatsReporter(namespace string, config string, revision string, pod string) (*Reporter, error)

NewStatsReporter creates a reporter that collects and reports queue metrics

func (*Reporter) Report

func (r *Reporter) Report(lameDuck bool, operationsPerSecond float64, averageConcurrentRequests float64) error

Report captures request metrics

func (*Reporter) UnregisterViews

func (r *Reporter) UnregisterViews() error

UnregisterViews Unregister views

type ReqEvent

type ReqEvent struct {
	Time      time.Time
	EventType ReqEventType
}

ReqEvent represents either an incoming or closed request.

type ReqEventType

type ReqEventType int

ReqEventType denotes the type (incoming/closed) of a ReqEvent.

const (
	// ReqIn represents an incoming request
	ReqIn ReqEventType = iota
	// ReqOut represents a finished request
	ReqOut
)

type Semaphore

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

Semaphore is an implementation of a semaphore based on Go channels The number of available tokens is the number of elements in the buffered channel

func NewSemaphore

func NewSemaphore(maxCapacity, initialCapacity int32) *Semaphore

NewSemaphore creates a semaphore with the desired maximal and initial capacity

func (*Semaphore) Acquire

func (s *Semaphore) Acquire()

Acquire receives the token from the semaphore, potentially blocking

func (*Semaphore) AddCapacity

func (s *Semaphore) AddCapacity(size int32)

AddCapacity conditionally adds capacity to the semaphore If there are tokens that must be reduced, release them first Otherwise, add tokens to the queue

func (*Semaphore) ReduceCapacity

func (s *Semaphore) ReduceCapacity(size int32) error

ReduceCapacity removes tokens from the rotation It tries to acquire as many tokens as possible, if there are not enough tokens in the queue, it postpones the operation for the future by increasing the `reducers` counter

func (*Semaphore) Release

func (s *Semaphore) Release()

Release releases the token to the queue The operation is potentially blocking when the queue is full

type Stats

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

Stats is a structure for holding channels per pod.

func NewStats

func NewStats(podName string, channels Channels, startedAt time.Time) *Stats

NewStats instantiates a new instance of Stats.