tree

Documentation

Index

• func CurrentQuerier(qa *TenantQuerierQueuingAlgorithm) string
• func DeleteNode(startingNode *Node, pathFromNode QueuePath) bool
• func GetOrAddNode(path QueuePath, tree Tree) error
• func TenantQueueCount(tree *MultiAlgorithmTreeQueue) int
• type DequeueArgs
• type MultiAlgorithmTreeQueue
  • func NewTree(queuingAlgorithms ...QueuingAlgorithm) (*MultiAlgorithmTreeQueue, error)
  • func (t *MultiAlgorithmTreeQueue) Dequeue(dequeueArgs *DequeueArgs) (QueuePath, any)
  • func (t *MultiAlgorithmTreeQueue) EnqueueBackByPath(path QueuePath, v any) error
  • func (t *MultiAlgorithmTreeQueue) EnqueueFrontByPath(path QueuePath, v any) error
  • func (t *MultiAlgorithmTreeQueue) GetNode(path QueuePath) *Node
  • func (t *MultiAlgorithmTreeQueue) IsEmpty() bool
  • func (t *MultiAlgorithmTreeQueue) ItemCount() int
• type Node
  • func RootNode(tree *MultiAlgorithmTreeQueue) *Node
  • func (n *Node) IsEmpty() bool
  • func (n *Node) ItemCount() int
  • func (n *Node) Name() string
• type QuerierID
• type QuerierWorkerQueuePriorityAlgo
  • func NewQuerierWorkerQueuePriorityAlgo() *QuerierWorkerQueuePriorityAlgo
• type QueueIndex
• type QueuePath
• type QueuingAlgorithm
• type RoundRobinState
  • func NewRoundRobinState() *RoundRobinState
• type TenantID
• type TenantQuerierQueuingAlgorithm
  • func NewTenantQuerierQueuingAlgorithm() *TenantQuerierQueuingAlgorithm
  • func (qa *TenantQuerierQueuingAlgorithm) AddTenant(tenantID string) int
  • func (qa *TenantQuerierQueuingAlgorithm) QueriersForTenant(tenantID string) map[QuerierID]struct{}
  • func (qa *TenantQuerierQueuingAlgorithm) SetQueriersForTenant(tenantID string, querierSet map[QuerierID]struct{})
  • func (qa *TenantQuerierQueuingAlgorithm) TenantIDOrder() []string
  • func (qa *TenantQuerierQueuingAlgorithm) TenantOrderIndex() int
  • func (qa *TenantQuerierQueuingAlgorithm) TotalQueueSizeForTenant(tenantID string) int
• type Tree

Functions

func CurrentQuerier

func CurrentQuerier(qa *TenantQuerierQueuingAlgorithm) string

CurrentQuerier is a test utility for reading the currently-set querier.

func DeleteNode

func DeleteNode(startingNode *Node, pathFromNode QueuePath) bool

DeleteNode is a test utility

func GetOrAddNode

func GetOrAddNode(path QueuePath, tree Tree) error

func TenantQueueCount

func TenantQueueCount(tree *MultiAlgorithmTreeQueue) int

TenantQueueCount is a test utility which returns the number of distinct tenants with populated queues.

Types

type DequeueArgs

type DequeueArgs struct {
	QuerierID       string
	WorkerID        int
	LastTenantIndex int
}

type MultiAlgorithmTreeQueue

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

MultiAlgorithmTreeQueue holds metadata and a pointer to the root node of a hierarchical queue implementation. The root Node maintains a localQueue and an arbitrary number of child nodes (which themselves may have local queues and children). Each Node in MultiAlgorithmTreeQueue uses a QueuingAlgorithm (determined by node depth) to determine dequeue order of that Node's subtree.

Each queuing dimension is modeled as a node in the tree, internally reachable through a QueuePath.

The QueuePath is an ordered array of strings describing the path from the tree root to a Node. In addition to child Nodes, each Node contains a local queue (FIFO) of items.

When dequeuing from a given node, a Node will use its QueuingAlgorithm to choose either itself or a child node to dequeue from recursively (i.e., a child Node will use its own QueuingAlgorithm to determine how to proceed). MultiAlgorithmTreeQueue will not dequeue from two different Nodes at the same depth consecutively, unless the previously-checked Node was empty down to the leaf node.

func NewTree

func NewTree(queuingAlgorithms ...QueuingAlgorithm) (*MultiAlgorithmTreeQueue, error)

func (*MultiAlgorithmTreeQueue) Dequeue

func (t *MultiAlgorithmTreeQueue) Dequeue(dequeueArgs *DequeueArgs) (QueuePath, any)

Dequeue removes and returns an item from the front of the next appropriate Node in the MultiAlgorithmTreeQueue, as well as the path to the Node which that item was dequeued from. If DequeueArgs are passed, each QueuingAlgorithm's setup function is called with DequeueArgs to update its state. If DequeueArgs is nil, the dequeue operation will proceed without setting up QueuingAlgorithm state.

Either the root/self node or a child node is chosen according to the Node's QueuingAlgorithm. If the root node is chosen, an item will be dequeued from the front of its localQueue. If a child node is chosen, it is recursively dequeued from until a node selects its localQueue.

Nodes that satisfy IsEmpty after a dequeue operation are deleted as the recursion returns up the stack. This maintains structural guarantees relied upon to make IsEmpty() non-recursive.

func (*MultiAlgorithmTreeQueue) EnqueueBackByPath

func (t *MultiAlgorithmTreeQueue) EnqueueBackByPath(path QueuePath, v any) error

EnqueueBackByPath enqueues an item in the back of the local queue of the node located at a given path through the tree; nodes for the path are created as needed.

path is relative to the root node; providing a QueuePath beginning with "root" will create a child node of the root node which is also named "root."

func (*MultiAlgorithmTreeQueue) EnqueueFrontByPath

func (t *MultiAlgorithmTreeQueue) EnqueueFrontByPath(path QueuePath, v any) error

EnqueueFrontByPath enqueues an item in the front of the local queue of the Node located at a given path through the MultiAlgorithmTreeQueue; nodes for the path are created as needed.

Enqueueing to the front is intended only for items which were first enqueued to the back and then dequeued after reaching the front.

Re-enqueueing to the front is only intended for use in cases where a queue consumer fails to complete operations on the dequeued item, but failure is not yet final, and the operations should be retried by a subsequent queue consumer. A concrete example is when a queue consumer fails or disconnects for unrelated reasons while we are in the process of dequeuing a request for it.

path must be relative to the root node; providing a QueuePath beginning with "root" will create a child node of root which is also named "root."

func (*MultiAlgorithmTreeQueue) GetNode

func (t *MultiAlgorithmTreeQueue) GetNode(path QueuePath) *Node

func (*MultiAlgorithmTreeQueue) IsEmpty

func (t *MultiAlgorithmTreeQueue) IsEmpty() bool

func (*MultiAlgorithmTreeQueue) ItemCount

func (t *MultiAlgorithmTreeQueue) ItemCount() int

type Node

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

Node maintains node-specific information used to enqueue and dequeue to itself, such as a local queue, node height, references to its children, and position in queue. Note that the tenantQuerierAssignments QueuingAlgorithm largely disregards Node's queueOrder and queuePosition, managing analogous state instead, because shuffle-sharding + fairness requirements necessitate input from the querier.

func RootNode

func RootNode(tree *MultiAlgorithmTreeQueue) *Node

RootNode is a test utility

func (*Node) IsEmpty

func (n *Node) IsEmpty() bool

func (*Node) ItemCount

func (n *Node) ItemCount() int

ItemCount counts the queue items in the Node and in all its children, recursively.

func (*Node) Name

func (n *Node) Name() string

type QuerierID

type QuerierID string

type QuerierWorkerQueuePriorityAlgo

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

QuerierWorkerQueuePriorityAlgo implements QueuingAlgorithm by mapping worker IDs to a queue node to prioritize. Querier-workers' prioritized queue nodes are calculated by the integer WorkerID % len(nodeOrder). This distribution of workers across query component subtrees ensures that when one query component is experiencing high latency about 25% of querier-workers continue prioritizing queries for unaffected components.

This significantly outperforms the previous round-robin approach which simply rotated through the node order. Although a vanilla round-robin algorithm will select a given query-component node 1 / 4 of the time, in situations of high latency on a query component, the utilization of the querier-worker connections as measured by inflight query processing time will grow asymptotically to be dominated by the slow query component.

There are 4 possible query components: "ingester", "store-gateway", "ingester-and-store-gateway", and "unknown". When all 4 queue nodes exist, approximately 1 / 4 of the querier-workers are prioritized to each queue node. This algorithm requires a minimum of 4 querier-workers per querier to prevent queue starvation. The minimum is enforced in the queriers by overriding -querier.max-concurrent if necessary.

MultiAlgorithmTreeQueue always deletes empty leaf nodes and nodes with no children after a dequeue operation, and only recreates the queue nodes when a new query request is enqueued which requires that path through the tree. QuerierWorkerQueuePriorityAlgo responds by removing or re-adding the query component nodes to the nodeOrder. This has two implications for the distribution of workers across queue nodes:

1. The modulo operation may modulo the worker ID by 1, 2, 3, or 4 depending on the number of node types currently present in the nodeOrder, which can change which node a worker ID is prioritized for.
2. The nodeOrder changes as queues are deleted and re-created, so the worker ID-to-node mapping changes as the random enqueue order places query component nodes in different positions in the order.

These changes in nodeOrder guarantee that when the number of querier-workers is not evenly divisible by the number of query component nodes, through the randomized changes in nodeOrder over time, the workers are more evenly distributed across query component nodes than if length and order of the nodes were fixed.

A given worker ID is prioritized to *start* at a given queue node, but is not assigned strictly to that node. During any period without change to the nodeOrder, the same worker ID consistently starts at the same queue node, but moves on to other nodes if it cannot dequeue a request from the subtree of its first prioritized queue node. Continuing to search through other query-component nodes and their subtrees minimizes idle querier-worker capacity.

A querier-worker can process queries for nodes it has not prioritized when this QueuingAlgorithm is applied at the highest layer of the tree and the tenant-querier-shuffle-shard QueuingAlgorithm applied at the second layer of the tree. If shuffle-sharding is enabled, a querier-worker that prioritizes ingester-only queries may not find ingester-only queries for any tenant it is assigned to, and move on to the next query component subtree. E.g.:

1. This algorithm has nodeOrder: ["ingester", "store-gateway", "ingester-and-store-gateway", "unknown"].

2. A querier-worker with WorkerID 0 requests to dequeue; it prioritizes the "ingester" queue node.

3. The dequeue operation attempts to dequeue first from the child nodes of the "ingester" node, where each child node is a tenant-specific queue of ingester-only queries. The tenantQuerierAssignments QueuingAlgorithm checks if any of its tenant queue nodes is sharded to this querier, and finds none.

4. The dequeue operation walks back up to the QuerierWorkerQueuePriorityAlgo level, not having dequeued anything. The QuerierWorkerQueuePriorityAlgo moves on and selects the next query-component node in the nodeOrder, and recurs again to search that next subtree for tenant queue nodes sharded to this querier, from step 3, etc., until a dequeue-able tenant queue node is found, or every query component node subtree has been exhausted.

func NewQuerierWorkerQueuePriorityAlgo

func NewQuerierWorkerQueuePriorityAlgo() *QuerierWorkerQueuePriorityAlgo

type QueueIndex

type QueueIndex int //nolint:revive // disallows types beginning with package name

type QueuePath

type QueuePath []string //nolint:revive // disallows types beginning with package name

type QueuingAlgorithm

type QueuingAlgorithm interface {
	// contains filtered or unexported methods
}

QueuingAlgorithm represents the set of operations specific to different approaches to queuing/dequeuing. It is applied at the layer-level -- every Node at the same depth in a MultiAlgorithmTreeQueue shares the same QueuingAlgorithm, including any state in structs that implement QueuingAlgorithm.

type RoundRobinState

type RoundRobinState struct {
}

RoundRobinState is the simplest type of QueuingAlgorithm; nodes which use this QueuingAlgorithm and are at the same depth in a MultiAlgorithmTreeQueue do not share any state. When children are added to these nodes, they are placed at the "end" of the order from the perspective of the node's current queuePosition (e.g., if queuePosition is 3, a new child will be placed at index 2). Children are dequeued from using a simple round-robin ordering; queuePosition is incremented on every dequeue.

func NewRoundRobinState

func NewRoundRobinState() *RoundRobinState

type TenantID

type TenantID string

type TenantQuerierQueuingAlgorithm

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

TenantQuerierQueuingAlgorithm implements QueuingAlgorithm.

func NewTenantQuerierQueuingAlgorithm

func NewTenantQuerierQueuingAlgorithm() *TenantQuerierQueuingAlgorithm

func (*TenantQuerierQueuingAlgorithm) AddTenant

func (qa *TenantQuerierQueuingAlgorithm) AddTenant(tenantID string) int

AddTenant inserts a tenantID into tenantIDOrder, replacing the first empty string it finds, or appending to the end if no empty strings are found. It returns the index at which the tenantID was inserted.

func (*TenantQuerierQueuingAlgorithm) QueriersForTenant

func (qa *TenantQuerierQueuingAlgorithm) QueriersForTenant(tenantID string) map[QuerierID]struct{}

func (*TenantQuerierQueuingAlgorithm) SetQueriersForTenant

func (qa *TenantQuerierQueuingAlgorithm) SetQueriersForTenant(tenantID string, querierSet map[QuerierID]struct{})

func (*TenantQuerierQueuingAlgorithm) TenantIDOrder

func (qa *TenantQuerierQueuingAlgorithm) TenantIDOrder() []string

func (*TenantQuerierQueuingAlgorithm) TenantOrderIndex

func (qa *TenantQuerierQueuingAlgorithm) TenantOrderIndex() int

func (*TenantQuerierQueuingAlgorithm) TotalQueueSizeForTenant

func (qa *TenantQuerierQueuingAlgorithm) TotalQueueSizeForTenant(tenantID string) int

TotalQueueSizeForTenant counts up items for all the nodes in tenantNodes[tenantID] and returns the total. This is the total number of requests queued for that tenant.

type Tree

type Tree interface {
	EnqueueFrontByPath(QueuePath, any) error
	EnqueueBackByPath(QueuePath, any) error
	Dequeue(dequeueArgs *DequeueArgs) (QueuePath, any)
	GetNode(path QueuePath) *Node
	ItemCount() int
	IsEmpty() bool
}