diff --git a/pkg/storage/unified/resource/server.go b/pkg/storage/unified/resource/server.go index 6e74bec40cd..35dc9de717b 100644 --- a/pkg/storage/unified/resource/server.go +++ b/pkg/storage/unified/resource/server.go @@ -146,7 +146,7 @@ type BlobSupport interface { } type QOSEnqueuer interface { - Enqueue(ctx context.Context, tenantID string, runnable func(ctx context.Context)) error + Enqueue(ctx context.Context, tenantID string, runnable func()) error } type BlobConfig struct { @@ -602,7 +602,7 @@ func (s *server) Create(ctx context.Context, req *resourcepb.CreateRequest) (*re res *resourcepb.CreateResponse err error ) - runErr := s.runInQueue(ctx, req.Key.Namespace, func(ctx context.Context) { + runErr := s.runInQueue(ctx, req.Key.Namespace, func() { res, err = s.create(ctx, user, req) }) if runErr != nil { @@ -656,7 +656,7 @@ func (s *server) Update(ctx context.Context, req *resourcepb.UpdateRequest) (*re res *resourcepb.UpdateResponse err error ) - runErr := s.runInQueue(ctx, req.Key.Namespace, func(ctx context.Context) { + runErr := s.runInQueue(ctx, req.Key.Namespace, func() { res, err = s.update(ctx, user, req) }) if runErr != nil { @@ -724,7 +724,7 @@ func (s *server) Delete(ctx context.Context, req *resourcepb.DeleteRequest) (*re err error ) - runErr := s.runInQueue(ctx, req.Key.Namespace, func(ctx context.Context) { + runErr := s.runInQueue(ctx, req.Key.Namespace, func() { res, err = s.delete(ctx, user, req) }) if runErr != nil { @@ -776,10 +776,6 @@ func (s *server) delete(ctx context.Context, user claims.AuthInfo, req *resource PreviousRV: latest.ResourceVersion, GUID: uuid.New().String(), } - requester, ok := claims.AuthInfoFrom(ctx) - if !ok { - return nil, apierrors.NewBadRequest("unable to get user") - } marker := &unstructured.Unstructured{} err = json.Unmarshal(latest.Value, marker) if err != nil { @@ -794,7 +790,7 @@ func (s *server) delete(ctx context.Context, user claims.AuthInfo, req *resource obj.SetUpdatedTimestamp(&now.Time) obj.SetManagedFields(nil) obj.SetFinalizers(nil) - obj.SetUpdatedBy(requester.GetUID()) + obj.SetUpdatedBy(user.GetUID()) obj.SetGeneration(utils.DeletedGeneration) obj.SetAnnotation(utils.AnnoKeyKubectlLastAppliedConfig, "") // clears it event.Value, err = marker.MarshalJSON() @@ -832,7 +828,7 @@ func (s *server) Read(ctx context.Context, req *resourcepb.ReadRequest) (*resour res *resourcepb.ReadResponse err error ) - runErr := s.runInQueue(ctx, req.Key.Namespace, func(ctx context.Context) { + runErr := s.runInQueue(ctx, req.Key.Namespace, func() { res, err = s.read(ctx, user, req) }) if runErr != nil { @@ -1339,7 +1335,7 @@ func (s *server) GetBlob(ctx context.Context, req *resourcepb.GetBlobRequest) (* return rsp, nil } -func (s *server) runInQueue(ctx context.Context, tenantID string, runnable func(ctx context.Context)) error { +func (s *server) runInQueue(ctx context.Context, tenantID string, runnable func()) error { boff := backoff.New(ctx, backoff.Config{ MinBackoff: DefaultMinBackoff, MaxBackoff: DefaultMaxBackoff, @@ -1351,9 +1347,9 @@ func (s *server) runInQueue(ctx context.Context, tenantID string, runnable func( err error ) wg.Add(1) - wrapped := func(ctx context.Context) { - runnable(ctx) - wg.Done() + wrapped := func() { + defer wg.Done() + runnable() } for boff.Ongoing() { err = s.queue.Enqueue(ctx, tenantID, wrapped) diff --git a/pkg/storage/unified/sql/server.go b/pkg/storage/unified/sql/server.go index 93f55900fd6..97163958972 100644 --- a/pkg/storage/unified/sql/server.go +++ b/pkg/storage/unified/sql/server.go @@ -21,8 +21,8 @@ import ( type QOSEnqueueDequeuer interface { services.Service - Enqueue(ctx context.Context, tenantID string, runnable func(ctx context.Context)) error - Dequeue(ctx context.Context) (func(ctx context.Context), error) + Enqueue(ctx context.Context, tenantID string, runnable func()) error + Dequeue(ctx context.Context) (func(), error) } // ServerOptions contains the options for creating a new ResourceServer diff --git a/pkg/util/scheduler/queue.go b/pkg/util/scheduler/queue.go index b065d92804f..f2c1458987f 100644 --- a/pkg/util/scheduler/queue.go +++ b/pkg/util/scheduler/queue.go @@ -25,7 +25,7 @@ var ErrMissingTenantID = errors.New("item requires TenantID") type tenantQueue struct { id string - items []func(ctx context.Context) + items []func() isActive bool } @@ -42,13 +42,13 @@ func (tq *tenantQueue) isEmpty() bool { func (tq *tenantQueue) isFull(maxSize int) bool { return maxSize > 0 && len(tq.items) >= maxSize } -func (tq *tenantQueue) addItem(runnable func(ctx context.Context)) { +func (tq *tenantQueue) addItem(runnable func()) { tq.items = append(tq.items, runnable) } type enqueueRequest struct { tenantID string - runnable func(ctx context.Context) + runnable func() respChan chan error } @@ -57,7 +57,7 @@ type dequeueRequest struct { } type dequeueResponse struct { - runnable func(ctx context.Context) + runnable func() err error } @@ -71,8 +71,8 @@ type activeTenantsLenRequest struct { type NoopQueue struct{} -func (*NoopQueue) Enqueue(ctx context.Context, _ string, runnable func(ctx context.Context)) error { - runnable(ctx) +func (*NoopQueue) Enqueue(ctx context.Context, _ string, runnable func()) error { + runnable() return nil } @@ -143,11 +143,11 @@ func NewQueue(opts *QueueOptions) *Queue { q.queueLength = promauto.With(opts.Registerer).NewGaugeVec(prometheus.GaugeOpts{ Name: "queue_length", Help: "Number of items in the queue", - }, []string{"namespace"}) + }, []string{"tenant"}) q.discardedRequests = promauto.With(opts.Registerer).NewCounterVec(prometheus.CounterOpts{ Name: "discarded_requests_total", Help: "Total number of discarded requests", - }, []string{"namespace", "reason"}) + }, []string{"tenant", "reason"}) q.enqueueDuration = promauto.With(opts.Registerer).NewHistogram(prometheus.HistogramOpts{ Name: "enqueue_duration_seconds", Help: "Duration of enqueue operation in seconds", @@ -200,7 +200,7 @@ func (q *Queue) handleEnqueueRequest(req enqueueRequest) { if !exists { tq = &tenantQueue{ id: req.tenantID, - items: make([]func(ctx context.Context), 0, 8), + items: make([]func(), 0, 8), } q.tenantQueues[req.tenantID] = tq } @@ -263,7 +263,7 @@ func (q *Queue) dispatcherLoop(ctx context.Context) error { // Enqueue adds a work item to the appropriate tenant's qos. // It blocks only if the dispatcher is busy or the tenant queue is full. -func (q *Queue) Enqueue(ctx context.Context, tenantID string, runnable func(ctx context.Context)) error { +func (q *Queue) Enqueue(ctx context.Context, tenantID string, runnable func()) error { if runnable == nil { return ErrNilRunnable } @@ -303,7 +303,7 @@ func (q *Queue) Enqueue(ctx context.Context, tenantID string, runnable func(ctx // Dequeue removes and returns a work item from the qos using linked-list round-robin. // It blocks until an item is available for any tenant, the queue is closed, // or the context is cancelled. -func (q *Queue) Dequeue(ctx context.Context) (func(ctx context.Context), error) { +func (q *Queue) Dequeue(ctx context.Context) (func(), error) { if q.State() != services.Running { return nil, ErrQueueClosed } diff --git a/pkg/util/scheduler/queue_test.go b/pkg/util/scheduler/queue_test.go index 1f56603bf3c..baf70a9b835 100644 --- a/pkg/util/scheduler/queue_test.go +++ b/pkg/util/scheduler/queue_test.go @@ -54,7 +54,7 @@ func TestQueue(t *testing.T) { // Enqueue items for i := 0; i < numItems; i++ { - err := q.Enqueue(ctx, tenantID, func(ctx context.Context) { + err := q.Enqueue(ctx, tenantID, func() { processed.Add(1) }) require.NoError(t, err, "Enqueue should succeed") @@ -72,7 +72,7 @@ func TestQueue(t *testing.T) { runnable, err := q.Dequeue(dequeueCtx) require.NoError(t, err, "Dequeue should succeed") require.NotNil(t, runnable, "Dequeued runnable should not be nil") - runnable(ctx) + runnable() }() } @@ -86,7 +86,7 @@ func TestQueue(t *testing.T) { require.NoError(t, services.StartAndAwaitRunning(ctx, qSimple)) for i := 0; i < numItems; i++ { - err := qSimple.Enqueue(ctx, tenantID, func(ctx context.Context) {}) + err := qSimple.Enqueue(ctx, tenantID, func() {}) require.NoError(t, err) } require.Equal(t, numItems, qSimple.Len(), "Queue length after enqueue (simple)") @@ -131,8 +131,8 @@ func TestQueue(t *testing.T) { var results []string var resultsMu sync.Mutex - makeRunnable := func(id string) func(ctx context.Context) { - return func(ctx context.Context) { + makeRunnable := func(id string) func() { + return func() { resultsMu.Lock() results = append(results, id) resultsMu.Unlock() @@ -163,7 +163,7 @@ func TestQueue(t *testing.T) { cancel() require.NoError(t, err, "Dequeue %d should succeed", i) require.NotNil(t, runnable, "Dequeued runnable %d should not be nil", i) - runnable(ctx) // Execute to record the tenant ID + runnable() // Execute to record the tenant ID } // Check execution order - should alternate between tenants @@ -187,16 +187,16 @@ func TestQueue(t *testing.T) { tenantID := "tenant-limited" // Enqueue up to the limit - err := q.Enqueue(ctx, tenantID, func(ctx context.Context) {}) + err := q.Enqueue(ctx, tenantID, func() {}) require.NoError(t, err) - err = q.Enqueue(ctx, tenantID, func(ctx context.Context) {}) + err = q.Enqueue(ctx, tenantID, func() {}) require.NoError(t, err) require.Equal(t, 2, q.Len()) require.Equal(t, 1, q.ActiveTenantsLen()) // Enqueue one more, expect error - err = q.Enqueue(ctx, tenantID, func(ctx context.Context) {}) + err = q.Enqueue(ctx, tenantID, func() {}) require.ErrorIs(t, err, ErrTenantQueueFull, "Expected ErrTenantQueueFull") // Len should still be 2 @@ -210,7 +210,7 @@ func TestQueue(t *testing.T) { require.Equal(t, 1, q.Len()) // Now enqueue should succeed again - err = q.Enqueue(ctx, tenantID, func(ctx context.Context) {}) + err = q.Enqueue(ctx, tenantID, func() {}) require.NoError(t, err, "Enqueue should succeed after dequeueing one item") require.Equal(t, 2, q.Len(), "Length should be back to 2") }) @@ -264,7 +264,7 @@ func TestQueue(t *testing.T) { require.NoError(t, services.StopAndAwaitTerminated(context.Background(), q)) // Now try to enqueue - should return ErrQueueClosed - err := q.Enqueue(context.Background(), "tenant-id", func(ctx context.Context) {}) + err := q.Enqueue(context.Background(), "tenant-id", func() {}) require.ErrorIs(t, err, ErrQueueClosed, "Enqueue after Stop should return ErrQueueClosed") }) @@ -272,7 +272,7 @@ func TestQueue(t *testing.T) { t.Parallel() ctx := context.Background() q := NewQueue(QueueOptionsWithDefaults(nil)) - err := q.Enqueue(ctx, "tenant-id", func(ctx context.Context) {}) + err := q.Enqueue(ctx, "tenant-id", func() {}) require.ErrorIs(t, err, ErrQueueClosed, "Enqueue before Start should return ErrQueueClosed") }) @@ -337,7 +337,7 @@ func TestQueue(t *testing.T) { } // Execute the runnable which will update our tracking - runnable(ctx) + runnable() // Check if we've processed all expected items mu.Lock() @@ -365,7 +365,7 @@ func TestQueue(t *testing.T) { for j := 0; j < itemsPerProducer; j++ { itemID := fmt.Sprintf("p%d-item%d", producerID, j) - err := q.Enqueue(ctx, tenantID, func(ctx context.Context) { + err := q.Enqueue(ctx, tenantID, func() { mu.Lock() processedItems[itemID] = 1 mu.Unlock() @@ -420,7 +420,7 @@ func TestQueue(t *testing.T) { // Enqueue a slow item for tenant A wg.Add(1) - err := q.Enqueue(ctx, tenantA, func(ctx context.Context) { + err := q.Enqueue(ctx, tenantA, func() { defer wg.Done() time.Sleep(300 * time.Millisecond) // Simulate slow processing completionOrder <- "A-slow" @@ -430,14 +430,14 @@ func TestQueue(t *testing.T) { // Enqueue regular items for other tenants for i := 0; i < 2; i++ { wg.Add(1) - err := q.Enqueue(ctx, tenantB, func(ctx context.Context) { + err := q.Enqueue(ctx, tenantB, func() { defer wg.Done() completionOrder <- fmt.Sprintf("B-%d", i) }) require.NoError(t, err) wg.Add(1) - err = q.Enqueue(ctx, tenantC, func(ctx context.Context) { + err = q.Enqueue(ctx, tenantC, func() { defer wg.Done() completionOrder <- fmt.Sprintf("C-%d", i) }) @@ -446,7 +446,7 @@ func TestQueue(t *testing.T) { // Enqueue another item for tenant A wg.Add(1) - err = q.Enqueue(ctx, tenantA, func(ctx context.Context) { + err = q.Enqueue(ctx, tenantA, func() { defer wg.Done() completionOrder <- "A-fast" }) @@ -462,7 +462,7 @@ func TestQueue(t *testing.T) { if err != nil { return } - runnable(ctx) + runnable() } }() } @@ -523,9 +523,9 @@ func TestQueue(t *testing.T) { require.NoError(t, q.AwaitRunning(context.Background()), "Queue should be running") // Enqueue items for different tenants - err := q.Enqueue(context.Background(), "tenant1", func(ctx context.Context) {}) + err := q.Enqueue(context.Background(), "tenant1", func() {}) require.NoError(t, err) - err = q.Enqueue(context.Background(), "tenant2", func(ctx context.Context) {}) + err = q.Enqueue(context.Background(), "tenant2", func() {}) require.NoError(t, err) // Check active tenants @@ -544,9 +544,9 @@ func TestQueue(t *testing.T) { require.NoError(t, q.AwaitRunning(context.Background()), "Queue should be running") // Enqueue items - err := q.Enqueue(context.Background(), "tenant1", func(ctx context.Context) {}) + err := q.Enqueue(context.Background(), "tenant1", func() {}) require.NoError(t, err) - err = q.Enqueue(context.Background(), "tenant1", func(ctx context.Context) {}) + err = q.Enqueue(context.Background(), "tenant1", func() {}) require.NoError(t, err) // Check queue length @@ -567,7 +567,7 @@ func TestQueue(t *testing.T) { processed := make(chan struct{}) // Enqueue an item that signals when processed - err := q.Enqueue(context.Background(), "tenant1", func(ctx context.Context) { + err := q.Enqueue(context.Background(), "tenant1", func() { close(processed) }) require.NoError(t, err) @@ -582,7 +582,7 @@ func TestQueue(t *testing.T) { runnable, err := q.Dequeue(ctx) require.NoError(t, err) require.NotNil(t, runnable) - runnable(ctx) + runnable() }() // Wait for the item to be processed @@ -599,7 +599,7 @@ func TestQueue(t *testing.T) { wg.Wait() // Check that the queue is closed - err = q.Enqueue(context.Background(), "tenant1", func(ctx context.Context) {}) + err = q.Enqueue(context.Background(), "tenant1", func() {}) require.ErrorIs(t, err, ErrQueueClosed) }) } diff --git a/pkg/util/scheduler/scheduler.go b/pkg/util/scheduler/scheduler.go index 6f781c2536e..f530f9522d0 100644 --- a/pkg/util/scheduler/scheduler.go +++ b/pkg/util/scheduler/scheduler.go @@ -26,7 +26,7 @@ const ( type WorkQueue interface { services.Service - Dequeue(ctx context.Context) (runnable func(ctx context.Context), err error) + Dequeue(ctx context.Context) (runnable func(), err error) } // Worker processes items from the QoS request queue @@ -62,7 +62,7 @@ func (w *Worker) dequeueWithRetries(ctx context.Context) error { for boff.Ongoing() { runnable, err := w.queue.Dequeue(ctx) if err == nil { - runnable(ctx) + runnable() break } diff --git a/pkg/util/scheduler/scheduler_bench_test.go b/pkg/util/scheduler/scheduler_bench_test.go index 64d79a2bd5d..5ac13880615 100644 --- a/pkg/util/scheduler/scheduler_bench_test.go +++ b/pkg/util/scheduler/scheduler_bench_test.go @@ -46,7 +46,7 @@ func benchScheduler(b *testing.B, numWorkers, numTenants, itemsPerTenant int) { for i := 0; i < numTenants; i++ { tenantID := tenantIDs[i] for j := 0; j < itemsPerTenant; j++ { - require.NoError(b, q.Enqueue(context.Background(), tenantID, func(_ context.Context) { + require.NoError(b, q.Enqueue(context.Background(), tenantID, func() { processed.Add(1) wg.Done() })) @@ -166,7 +166,7 @@ func BenchmarkSchedulerFairness(b *testing.B) { tenantID := tenantIDs[i] tenantIdx := i for j := 0; j < itemsPerTenant; j++ { - require.NoError(b, q.Enqueue(context.Background(), tenantID, func(_ context.Context) { + require.NoError(b, q.Enqueue(context.Background(), tenantID, func() { processedPerTenant[tenantIdx].Add(1) wg.Done() })) @@ -248,7 +248,7 @@ func BenchmarkSchedulerFairnessAlternating(b *testing.B) { for i := 0; i < numTenants; i++ { tenantID := tenantIDs[i] tenantIdx := i - require.NoError(b, q.Enqueue(context.Background(), tenantID, func(_ context.Context) { + require.NoError(b, q.Enqueue(context.Background(), tenantID, func() { processedPerTenant[tenantIdx].Add(1) wg.Done() })) diff --git a/pkg/util/scheduler/scheduler_test.go b/pkg/util/scheduler/scheduler_test.go index 9d185df39c7..55a1484f34a 100644 --- a/pkg/util/scheduler/scheduler_test.go +++ b/pkg/util/scheduler/scheduler_test.go @@ -151,7 +151,7 @@ func TestScheduler(t *testing.T) { itemID := i tenantIndex := itemID % 10 tenantID := fmt.Sprintf("tenant-%d", tenantIndex) - require.NoError(t, q.Enqueue(context.Background(), tenantID, func(_ context.Context) { + require.NoError(t, q.Enqueue(context.Background(), tenantID, func() { processed.Store(itemID, true) time.Sleep(10 * time.Millisecond) wg.Done() @@ -199,12 +199,12 @@ func TestScheduler(t *testing.T) { require.NoError(t, services.StartAndAwaitRunning(context.Background(), scheduler)) for i := 0; i < 5; i++ { - require.NoError(t, q.Enqueue(context.Background(), "tenant-1", func(_ context.Context) { + require.NoError(t, q.Enqueue(context.Background(), "tenant-1", func() { processed.Add(1) })) } - require.NoError(t, q.Enqueue(context.Background(), "tenant-1", func(_ context.Context) { + require.NoError(t, q.Enqueue(context.Background(), "tenant-1", func() { close(taskStarted) time.Sleep(1 * time.Second) processed.Add(1)