mirror of
https://github.com/hanzoai/traces.git
synced 2026-10-30 18:20:41 +00:00
b5930aa883
* Begin changes for backend-scheduler module * Comment breadcrumb * Initial compactor extensions and backendscheduler module * Add initial jsonnet * Some cleanup after discussion * Add basic integration test for jobs fetching * Flesh out some unit tests * Clean up and add test coverage * Improve coverage * Clean up dead ring code * Set default value for compactor job polling * Add log message for backend scheduling * Add note about fixing the compactor flow * Small operational fixes * Plumb client defaults * Fix job handling and module dependency chicken/egg/lizard * Update config manifest * Move CompactionBlockSelector * Revert tempodb compactor changes * Revert "Revert tempodb compactor changes" This reverts commit 239c4f2a9df2c289164321f427ac56fa4b7d92cc. * Add backendworker module and work around comapctor dependencies * Move job work and queue into separate package * Add jsonnet for backend-worker * Fix backendscheduler tests * Update docs manifest * Revert compaction method rename * Drop debug * Revert compactor module changes * Drop debug * Revert compactor test change * Fix mock interface after extension * Start e2e integration test * Lint and clean up * Move tmpDir creation to the top * Commit for sharing * Clean up debug * Working integration test * Implement backoff * Add manifest * Clean up comments * Clean up comments * Clean up comments * Clean up * Clean up comments * Clean up test * Clean up comment * Clean up * Ensure the queue is drained before schedule * Fix metric label * Ensure the poller context is canceled * Set default values for worker backoff locally * Fix metric label * Drop worker interval in favor of backoff * Fix prune handling * Improve work selection * Hack in a backend cache * Leverage existing blocklist measurement * Skip the work cache when listing tenants * Test CompactWithConfig * Move proto for new work to new file * Tenants with zero blocks have zero priority * Many things * Add note * Drop empty file * Plumb work config for pruning * Drop needless proto change * Update backendscheduler tests * Update integration tests * Fix config and generate doc manifest * Add config validation * Add more config validation * Implement backoff wrapper for update calls * Move tenant index-writitng to the backend-worker and shard on the ring * Include RingConfig * Generate manifest * Fix manifest instance_id * Drop workerID context injection from the GRPC request * Push job collection into work package * Push logic for skipping blocks which are already working into work package * Log the workerID * Return gRPC error on invalid job status * Metric worker retry calls * Improve test coverage for work * Improve test coverage for job * Fix test logic * Reduce surface area, improve test coverage, clean up error handling * Include new errors * Fix lint * Add tests for backendworker module * Drop comments * Fix backoff for Next() call on the worker to ensure backoff is honored with error return * Bring back outstanding block metric on scheduler * Make note about losing the last work time when the tenant is pushed back to the queue * Measure all tenants, not just ones which get scheduled * Continue scheduling if a single tenant was not enough work to load * thoughts Signed-off-by: Joe Elliott <[email protected]> * Add a timeout to cleanup jobs which have been running too long * Drop "last work" from tenant selector implementation for simplicity * Revert blocklist changes * Ensure locking on work.OnBlock * Clean up * Generate manifest * Clean up comment * Clean up * Clean up and handle a couple TODO * Metric jobs created * Fix job count per tenant * Fix signature change in grpc DialOption * Generate manifest * Clean up * Spellcheck * Add missing Prune call * Auto start jobs which are returned, metric job retries * Drop Type from NextRequest * Move marshal/unmarshal to Work and drop a mutex * Simplify mutex handling * Test for job reload * Clean up after rebase * Generate manifest --------- Signed-off-by: Joe Elliott <[email protected]> Co-authored-by: Joe Elliott <[email protected]>
88 lines
1.6 KiB
Go
88 lines
1.6 KiB
Go
package tenantselector
|
|
|
|
import (
|
|
"container/heap"
|
|
"sort"
|
|
)
|
|
|
|
type Item struct {
|
|
value string // tenantID
|
|
priority int
|
|
index int // Index in the heap.
|
|
}
|
|
|
|
func NewItem(value string, priority int) *Item {
|
|
return &Item{
|
|
value: value,
|
|
priority: priority,
|
|
}
|
|
}
|
|
|
|
func (i *Item) Value() string {
|
|
return i.value
|
|
}
|
|
|
|
func (i *Item) Priority() int {
|
|
return i.priority
|
|
}
|
|
|
|
var _ heap.Interface = (*PriorityQueue)(nil)
|
|
|
|
type PriorityQueue []*Item
|
|
|
|
func NewPriorityQueue() *PriorityQueue {
|
|
pq := make(PriorityQueue, 0)
|
|
heap.Init(&pq)
|
|
return &pq
|
|
}
|
|
|
|
func (pq PriorityQueue) Items() []Item {
|
|
items := make([]Item, len(pq))
|
|
for i, item := range pq {
|
|
items[i] = *item
|
|
}
|
|
|
|
sort.Slice(items, func(i, j int) bool {
|
|
return items[i].priority > items[j].priority
|
|
})
|
|
|
|
return items
|
|
}
|
|
|
|
func (pq PriorityQueue) Len() int { return len(pq) }
|
|
|
|
func (pq PriorityQueue) Less(i, j int) bool {
|
|
// We want Pop to give us the highest, not lowest, priority so we use greater than here.
|
|
return pq[i].priority > pq[j].priority
|
|
}
|
|
|
|
func (pq PriorityQueue) Swap(i, j int) {
|
|
pq[i], pq[j] = pq[j], pq[i]
|
|
pq[i].index = i
|
|
pq[j].index = j
|
|
}
|
|
|
|
func (pq *PriorityQueue) Push(x any) {
|
|
n := len(*pq)
|
|
item := x.(*Item)
|
|
item.index = n
|
|
*pq = append(*pq, item)
|
|
}
|
|
|
|
func (pq *PriorityQueue) Pop() any {
|
|
old := *pq
|
|
n := len(old)
|
|
item := old[n-1]
|
|
old[n-1] = nil // avoid memory leak
|
|
item.index = -1 // for safety
|
|
*pq = old[0 : n-1]
|
|
return item
|
|
}
|
|
|
|
// update modifies the priority and value of an Item in the queue.
|
|
func (pq *PriorityQueue) Update(item *Item, value string, priority int) {
|
|
item.value = value
|
|
item.priority = priority
|
|
heap.Fix(pq, item.index)
|
|
}
|