admission2

README

Admission Package

Overview

The admission package provides a BoundedQueue object. This object implements a semaphore for limiting the number of bytes admitted into a collector pipeline. Additionally, the BoundedQueue limits the number of bytes allowed to block on a call to Acquire(pending int64).

There are two error conditions generated within this code:

• rejecting request, too much pending data: When the limit on waiting bytes its reached, this will be returned to limit the total amount waiting.
• rejecting request, request is too large: When an individual request exceeds the configured limit, this will be returned without acquiring or waiting.

The BoundedQueue implements LIFO semantics. See this article explaining why it is preferred to FIFO semantics.

Usage

Create a new BoundedQueue by calling bq := admission.NewBoundedQueue(maxLimitBytes, maxLimitWaiting)

Within the component call bq.Acquire(ctx, requestSize) which will:

1. succeed immediately if there is enough available memory,
2. fail immediately if there are too many waiters, or
3. block until context cancelation or enough bytes becomes available.

When the resources have been acquired successfully, a closure is returned that, when called, will release the semaphore. When the semaphore is released, pending waiters that can be satisfied will acquire the resource and become unblocked.

Documentation

Index

• Variables
• type BoundedQueue
  • func (bq *BoundedQueue) Acquire(ctx context.Context, pending uint64) (ReleaseFunc, error)
• type N
  • func (n *N) Chan() <-chan struct{}
  • func (n *N) HasBeenNotified() bool
  • func (n *N) Notify()
  • func (n *N) WaitForNotification()
• type Queue
  • func NewBoundedQueue(id component.ID, ts component.TelemetrySettings, ...) (Queue, error)
  • func NewUnboundedQueue() Queue
• type ReleaseFunc

Variables

var (
	ErrTooMuchWaiting  = status.Error(grpccodes.ResourceExhausted, "rejecting request, too much pending data")
	ErrRequestTooLarge = status.Errorf(grpccodes.InvalidArgument, "rejecting request, request is too large")
)

Types

type BoundedQueue

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

BoundedQueue is a LIFO-oriented admission-controlled Queue.

func (*BoundedQueue) Acquire

func (bq *BoundedQueue) Acquire(ctx context.Context, pending uint64) (ReleaseFunc, error)

Acquire implements Queue.

type N

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

notification.N is a minimal Go version of absl::Notification:

https://github.com/abseil/abseil-cpp/blob/master/absl/synchronization/notification.h

Use New() to construct a notification object (the zero value is not usable).

func (*N) Chan

func (n *N) Chan() <-chan struct{}

Chan allows a caller to wait for the notification as part of a select statement. Outside of a select statement, prefer writing WaitForNotification().

func (*N) HasBeenNotified

func (n *N) HasBeenNotified() bool

func (*N) Notify

func (n *N) Notify()

func (*N) WaitForNotification

func (n *N) WaitForNotification()

type Queue

type Queue interface {
	// Acquire asks the controller to admit the caller.
	//
	// The weight parameter specifies how large of an admission to make.
	// This might be used on the bytes of request (for example) to differentiate
	// between large and small requests.
	//
	// Admit will return when one of the following events occurs:
	//
	//   (1) admission is allowed, or
	//   (2) the provided ctx becomes canceled, or
	//   (3) there are so many existing waiters that the
	//       controller decides to reject this caller without
	//       admitting it.
	//
	// In case (1), the return value will be a non-nil
	// ReleaseFunc. The caller must invoke it after it is finished
	// with the resource being guarded by the admission
	// controller.
	//
	// In case (2), the return value will be a Cancelled or
	// DeadlineExceeded error.
	//
	// In case (3), the return value will be a ResourceExhausted
	// error.
	Acquire(ctx context.Context, weight uint64) (ReleaseFunc, error)
}

Queue is a weighted admission queue interface.

func NewBoundedQueue

func NewBoundedQueue(id component.ID, ts component.TelemetrySettings, maxLimitAdmit, maxLimitWait uint64) (Queue, error)

NewBoundedQueue returns a LIFO-oriented Queue implementation which admits `maxLimitAdmit` bytes concurrently and allows up to `maxLimitWait` bytes to wait for admission.

func NewUnboundedQueue

func NewUnboundedQueue() Queue

NewUnboundedQueue returns a no-op implementation of the Queue interface.

type ReleaseFunc

type ReleaseFunc func()

ReleaseFunc is returned by Acquire when the Acquire() was admitted.