queue

Documentation

Index

• Variables
• type FIFORequestQueue
  • func NewFIFORequestQueue(queue chan Request, userID string, queueLength *prometheus.GaugeVec) *FIFORequestQueue
• type Limits
• type MockLimits
  • func (l MockLimits) MaxOutstandingPerTenant(_ string) int
  • func (l MockLimits) MaxQueriersPerUser(_ string) float64
  • func (l MockLimits) QueryPriority(_ string) validation.QueryPriority
• type PriorityRequestQueue
  • func NewPriorityRequestQueue(queue *util.PriorityQueue, userID string, queueLength *prometheus.GaugeVec) *PriorityRequestQueue
• type Request
• type RequestQueue
  • func NewRequestQueue(forgetDelay time.Duration, queueLength *prometheus.GaugeVec, ...) *RequestQueue
  • func (q *RequestQueue) EnqueueRequest(userID string, req Request, maxQueriers float64, successFn func()) error
  • func (q *RequestQueue) GetConnectedQuerierWorkersMetric() float64
  • func (q *RequestQueue) GetNextRequestForQuerier(ctx context.Context, last UserIndex, querierID string) (Request, UserIndex, error)
  • func (q *RequestQueue) NotifyQuerierShutdown(querierID string)
  • func (q *RequestQueue) QuerierDisconnecting()
  • func (q *RequestQueue) RegisterQuerierConnection(querier string)
  • func (q *RequestQueue) UnregisterQuerierConnection(querier string)
• type UserIndex
  • func FirstUser() UserIndex
  • func (ui UserIndex) ReuseLastUser() UserIndex

Variables

var (
	ErrTooManyRequests = errors.New("too many outstanding requests")
	ErrStopped         = errors.New("queue is stopped")
)

Types

type FIFORequestQueue

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

func NewFIFORequestQueue

func NewFIFORequestQueue(queue chan Request, userID string, queueLength *prometheus.GaugeVec) *FIFORequestQueue

type Limits

type Limits interface {
	// MaxOutstandingPerTenant returns the limit to the maximum number
	// of outstanding requests per tenant per request queue.
	MaxOutstandingPerTenant(user string) int

	// QueryPriority returns query priority config for the tenant, including priority level,
	// their attributes, and how many reserved queriers each priority has.
	QueryPriority(user string) validation.QueryPriority
}

Limits needed for the Query Scheduler - interface used for decoupling.

type MockLimits

type MockLimits struct {
	MaxOutstanding        int
	MaxQueriersPerUserVal float64
	QueryPriorityVal      validation.QueryPriority
}

MockLimits implements the Limits interface. Used in tests only.

func (MockLimits) MaxOutstandingPerTenant

func (l MockLimits) MaxOutstandingPerTenant(_ string) int

func (MockLimits) MaxQueriersPerUser

func (l MockLimits) MaxQueriersPerUser(_ string) float64

func (MockLimits) QueryPriority

func (l MockLimits) QueryPriority(_ string) validation.QueryPriority

type PriorityRequestQueue

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

func NewPriorityRequestQueue

func NewPriorityRequestQueue(queue *util.PriorityQueue, userID string, queueLength *prometheus.GaugeVec) *PriorityRequestQueue

type Request

type Request interface {
	util.PriorityOp
}

Request stored into the queue.

type RequestQueue

type RequestQueue struct {
	services.Service
	// contains filtered or unexported fields
}

RequestQueue holds incoming requests in per-user queues. It also assigns each user specified number of queriers, and when querier asks for next request to handle (using GetNextRequestForQuerier), it returns requests in a fair fashion.

func NewRequestQueue

func NewRequestQueue(forgetDelay time.Duration, queueLength *prometheus.GaugeVec, discardedRequests *prometheus.CounterVec, limits Limits, registerer prometheus.Registerer) *RequestQueue

func (*RequestQueue) EnqueueRequest

func (q *RequestQueue) EnqueueRequest(userID string, req Request, maxQueriers float64, successFn func()) error

EnqueueRequest puts the request into the queue. MaxQueries is user-specific value that specifies how many queriers can this user use (zero or negative = all queriers). It is passed to each EnqueueRequest, because it can change between calls.

If request is successfully enqueued, successFn is called with the lock held, before any querier can receive the request.

func (*RequestQueue) GetConnectedQuerierWorkersMetric

func (q *RequestQueue) GetConnectedQuerierWorkersMetric() float64

func (*RequestQueue) GetNextRequestForQuerier

func (q *RequestQueue) GetNextRequestForQuerier(ctx context.Context, last UserIndex, querierID string) (Request, UserIndex, error)

GetNextRequestForQuerier find next user queue and takes the next request off of it. Will block if there are no requests. By passing user index from previous call of this method, querier guarantees that it iterates over all users fairly. If querier finds that request from the user is already expired, it can get a request for the same user by using UserIndex.ReuseLastUser.

func (*RequestQueue) NotifyQuerierShutdown

func (q *RequestQueue) NotifyQuerierShutdown(querierID string)

func (*RequestQueue) QuerierDisconnecting

func (q *RequestQueue) QuerierDisconnecting()

When querier is waiting for next request, this unblocks the method.

func (*RequestQueue) RegisterQuerierConnection

func (q *RequestQueue) RegisterQuerierConnection(querier string)

func (*RequestQueue) UnregisterQuerierConnection

func (q *RequestQueue) UnregisterQuerierConnection(querier string)

type UserIndex

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

UserIndex is opaque type that allows to resume iteration over users between successive calls of RequestQueue.GetNextRequestForQuerier method.

func FirstUser

func FirstUser() UserIndex

FirstUser returns UserIndex that starts iteration over user queues from the very first user.

func (UserIndex) ReuseLastUser

func (ui UserIndex) ReuseLastUser() UserIndex

Modify index to start iteration on the same user, for which last queue was returned.