queue

Documentation

Overview

Package queue provides components for the queue-proxy binary.

Index

• Constants
• Variables
• func ConcurrencyStateHandler(logger *zap.SugaredLogger, h http.Handler, ...) http.HandlerFunc
• func ForwardedShimHandler(h http.Handler) http.Handler
• func GetRouteTagNameFromRequest(r *http.Request) string
• func NewAppRequestMetricsHandler(next http.Handler, b *Breaker, ns, service, config, rev, pod string) (http.Handler, error)
• func NewRequestMetricsHandler(next http.Handler, ns, service, config, rev, pod string) (http.Handler, error)
• func NewStatsHandler(proto http.Handler) http.Handler
• func ProxyHandler(breaker *Breaker, stats *netstats.RequestStats, tracingEnabled bool, ...) http.HandlerFunc
• type Breaker
  • func NewBreaker(params BreakerParams) *Breaker
  • func (b *Breaker) Capacity() int
  • func (b *Breaker) InFlight() int
  • func (b *Breaker) Maybe(ctx context.Context, thunk func()) error
  • func (b *Breaker) Reserve(ctx context.Context) (func(), bool)
  • func (b *Breaker) UpdateConcurrency(size int)
• type BreakerParams
• type ConcurrencyEndpoint
  • func NewConcurrencyEndpoint(e, m string) *ConcurrencyEndpoint
  • func (c *ConcurrencyEndpoint) Endpoint() string
  • func (c *ConcurrencyEndpoint) Pause(logger *zap.SugaredLogger)
  • func (c *ConcurrencyEndpoint) RefreshToken() error
  • func (c *ConcurrencyEndpoint) Request(action string) error
  • func (c *ConcurrencyEndpoint) Resume(logger *zap.SugaredLogger)
  • func (c *ConcurrencyEndpoint) Terminating(logger *zap.SugaredLogger)
• type ProtobufStatsReporter
  • func NewProtobufStatsReporter(pod string, reportingPeriod time.Duration) *ProtobufStatsReporter
  • func (r *ProtobufStatsReporter) Report(stats netstats.RequestStatsReport)
  • func (r *ProtobufStatsReporter) ServeHTTP(w http.ResponseWriter, _ *http.Request)

Constants

const (
	// Name is the name of the component.
	Name = "queue"

	// RequestQueueDrainPath specifies the path to wait until the proxy
	// server is shut down. Any subsequent calls to this endpoint after
	// the server has finished shutting down it will return immediately.
	// Main usage is to delay the termination of user-container until all
	// accepted requests have been processed.
	RequestQueueDrainPath = "/wait-for-drain"

	// CertDirectory is the name of the directory path where certificates are stored.
	CertDirectory = "/var/lib/knative/certs"

	// PodInfoDirectory is the name of the directory path where podinfo are stored.
	PodInfoDirectory = "/etc/podinfo"

	// TokenDirectory is the name of the directory path where tokens are stored.
	TokenDirectory = "/var/run/secrets/tokens"

	// ConcurrencyStateTokenFilename is the file name of the concurency state in TokenDirectory.
	ConcurrencyStateTokenFilename = "state-token"

	// PodInfoAnnotationsFilename is the file name of the annotations in PodInfoDirectory.
	PodInfoAnnotationsFilename = "annotations"
)

const MaxBreakerCapacity = math.MaxInt32

MaxBreakerCapacity is the largest valid value for the MaxConcurrency value of BreakerParams. This is limited by the maximum size of a chan struct{} in the current implementation.

Variables

var (
	// ErrRequestQueueFull indicates the breaker queue depth was exceeded.
	ErrRequestQueueFull = errors.New("pending request queue full")
)

Functions

func ConcurrencyStateHandler

func ConcurrencyStateHandler(logger *zap.SugaredLogger, h http.Handler, pause, resume func(*zap.SugaredLogger)) http.HandlerFunc

ConcurrencyStateHandler tracks the in flight requests for the pod. When the requests drop to zero, it runs the `pause` function, and when requests scale up from zero, it runs the `resume` function. If either of `pause` or `resume` are not passed, it runs the respective local function(s). The local functions are the expected behavior; the function parameters are enabled primarily for testing purposes.

func ForwardedShimHandler

func ForwardedShimHandler(h http.Handler) http.Handler

ForwardedShimHandler attempts to shim a `forwarded` HTTP header from the information available in the `x-forwarded-*` headers. When available, each node in the `x-forwarded-for` header is combined with the `x-forwarded-proto` and `x-forwarded-host` fields to construct a `forwarded` header. The `x-forwarded-by` header is ignored entirely, since it cannot be reliably combined with `x-forwarded-for`. No-op if a `forwarded` header is already present.

func GetRouteTagNameFromRequest

func GetRouteTagNameFromRequest(r *http.Request) string

GetRouteTagNameFromRequest extracts the value of the tag header from http.Request

func NewAppRequestMetricsHandler

func NewAppRequestMetricsHandler(next http.Handler, b *Breaker,
	ns, service, config, rev, pod string) (http.Handler, error)

NewAppRequestMetricsHandler creates an http.Handler that emits request metrics.

func NewRequestMetricsHandler

func NewRequestMetricsHandler(next http.Handler,
	ns, service, config, rev, pod string) (http.Handler, error)

NewRequestMetricsHandler creates an http.Handler that emits request metrics.

func NewStatsHandler

func NewStatsHandler(proto http.Handler) http.Handler

NewStatsHandler returns a new StatHandler.

func ProxyHandler

func ProxyHandler(breaker *Breaker, stats *netstats.RequestStats, tracingEnabled bool, next http.Handler) http.HandlerFunc

ProxyHandler sends requests to the `next` handler at a rate controlled by the passed `breaker`, while recording stats to `stats`.

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(params BreakerParams) *Breaker

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

func (*Breaker) Capacity

func (b *Breaker) Capacity() int

Capacity returns the number of allowed in-flight requests on this breaker.

func (*Breaker) InFlight

func (b *Breaker) InFlight() int

InFlight returns the number of requests currently in flight in this breaker.

func (*Breaker) Maybe

func (b *Breaker) Maybe(ctx context.Context, thunk func()) error

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 nil, else error.

func (*Breaker) Reserve

func (b *Breaker) Reserve(ctx context.Context) (func(), bool)

Reserve reserves an execution slot in the breaker, to permit richer semantics in the caller. The caller on success must execute the callback when done with work.

func (*Breaker) UpdateConcurrency

func (b *Breaker) UpdateConcurrency(size int)

UpdateConcurrency updates the maximum number of in-flight requests.

type BreakerParams

type BreakerParams struct {
	QueueDepth      int
	MaxConcurrency  int
	InitialCapacity int
}

BreakerParams defines the parameters of the breaker.

type ConcurrencyEndpoint

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

func NewConcurrencyEndpoint

func NewConcurrencyEndpoint(e, m string) *ConcurrencyEndpoint

func (*ConcurrencyEndpoint) Endpoint

func (c *ConcurrencyEndpoint) Endpoint() string

func (*ConcurrencyEndpoint) Pause

func (c *ConcurrencyEndpoint) Pause(logger *zap.SugaredLogger)

Pause freezes a container, retrying until either successful or a timeout is reached, at which point the container is killed

func (*ConcurrencyEndpoint) RefreshToken

func (c *ConcurrencyEndpoint) RefreshToken() error

func (*ConcurrencyEndpoint) Request

func (c *ConcurrencyEndpoint) Request(action string) error

func (*ConcurrencyEndpoint) Resume

func (c *ConcurrencyEndpoint) Resume(logger *zap.SugaredLogger)

Resume thaws a container, retrying until either successful or a timeout is reached, at which point the container is killed

func (*ConcurrencyEndpoint) Terminating

func (c *ConcurrencyEndpoint) Terminating(logger *zap.SugaredLogger)

type ProtobufStatsReporter

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

ProtobufStatsReporter structure represents a protobuf stats reporter.

func NewProtobufStatsReporter

func NewProtobufStatsReporter(pod string, reportingPeriod time.Duration) *ProtobufStatsReporter

NewProtobufStatsReporter creates a reporter that collects and reports queue metrics.

func (*ProtobufStatsReporter) Report

func (r *ProtobufStatsReporter) Report(stats netstats.RequestStatsReport)

Report captures request metrics.

func (*ProtobufStatsReporter) ServeHTTP

func (r *ProtobufStatsReporter) ServeHTTP(w http.ResponseWriter, _ *http.Request)

ServeHTTP serves the stats in protobuf format over HTTP.