public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: grpc resource delete error when qos enabled (#107560)

Browse Source
This commit is contained in:
Mustafa Sencer Özcan
2025-07-03 12:02:05 +02:00
committed by GitHub
parent 041c343a86
commit b7153d4d20
7 changed files with 57 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/storage/unified/resource/server.go
+10 -14
View File
@@ -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)
pkg/storage/unified/sql/server.go
+2 -2
View File
@@ -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
pkg/util/scheduler/queue.go
+11 -11
View File
@@ -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
}
pkg/util/scheduler/queue_test.go
+26 -26
View File
@@ -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)
})
}
pkg/util/scheduler/scheduler.go
+2 -2
View File
@@ -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
}
pkg/util/scheduler/scheduler_bench_test.go
+3 -3
View File
@@ -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()
}))
pkg/util/scheduler/scheduler_test.go
+3 -3
View File
@@ -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)