disk/ad_slot: return better GRPC error message

expected new message shown in Kubernetes events:
failed to reserve node i-xxxx for attach: still waiting for other disk(s) to finish attach/detach

Note that I don't specify GRPC status code intentionally, to let GRPC determine the code automatically based on error types.

Author: huww98

pkg/disk/attachdetach_slot.go

@@ -2,6 +2,7 @@ package disk
 
 import (
 	"context"
+	"errors"
 	"sync"
 )
 
@@ -73,7 +74,7 @@ func (s serialAD_DetachSlot) Acquire(ctx context.Context) error {
 	case s.slot <- struct{}{}:
 		return nil
 	case <-ctx.Done():
-		return ctx.Err()
+		return maybeWaitingAD(ctx.Err())
 	}
 }
 
@@ -85,7 +86,7 @@ func (s serialAD_AttachSlot) Acquire(ctx context.Context) error {
 	case s.slot <- struct{}{}:
 		return nil
 	case <-ctx.Done():
-		return ctx.Err()
+		return maybeWaitingAD(ctx.Err())
 	}
 }
 
@@ -120,7 +121,7 @@ func (s maxConcurrentSlot) Acquire(ctx context.Context) error {
 	case s.slots <- struct{}{}:
 		return nil
 	case <-ctx.Done():
-		return ctx.Err()
+		return maybeWaitingAD(ctx.Err())
 	}
 }
 
@@ -181,3 +182,20 @@ func NewSlots(detachConcurrency, attachConcurrency int) AttachDetachSlots {
 	}
 	return NewPerNodeSlots(makeSlot)
 }
+
+type waitingAD struct{}
+
+func (waitingAD) Error() string {
+	return "still waiting for other disk(s) to finish attach/detach"
+}
+
+func (waitingAD) Is(target error) bool {
+	return target == context.DeadlineExceeded
+}
+
+func maybeWaitingAD(err error) error {
+	if errors.Is(err, context.DeadlineExceeded) {
+		return waitingAD{}
+	}
+	return err
+}

pkg/disk/attachdetach_slot_test.go

@@ -171,3 +171,15 @@ func TestSerialDetach_NoRace(t *testing.T) {
 		t.Fatal("state not updated")
 	}
 }
+
+func TestWaitingADError(t *testing.T) {
+	s := NewSlots(1, 0).GetSlotFor("node1").Detach()
+	ctx := context.Background()
+	assert.NoError(t, s.Acquire(ctx))
+
+	ctx, cancel := context.WithTimeout(ctx, 10*time.Millisecond)
+	defer cancel()
+	err := s.Acquire(ctx)
+	assert.ErrorIs(t, err, waitingAD{})
+	assert.ErrorIs(t, err, context.DeadlineExceeded)
+}

pkg/disk/cloud.go

@@ -181,7 +181,7 @@ func (ad *DiskAttachDetach) attachDisk(ctx context.Context, diskID, nodeID strin
 
 	slot := ad.slots.GetSlotFor(nodeID).Attach()
 	if err := slot.Acquire(ctx); err != nil {
-		return "", status.Errorf(codes.Aborted, "AttachDisk: get ad-slot for disk %s failed: %v", diskID, err)
+		return "", fmt.Errorf("failed to reserve node %s for attach: %w", nodeID, err)
 	}
 	defer slot.Release()
 
@@ -442,7 +442,7 @@ func (ad *DiskAttachDetach) detachDisk(ctx context.Context, ecsClient *ecs.Clien
 	// NodeStageVolume/NodeUnstageVolume should be called by sequence
 	slot := ad.slots.GetSlotFor(nodeID).Detach()
 	if err := slot.Acquire(ctx); err != nil {
-		return status.Errorf(codes.Aborted, "DetachDisk: get ad-slot for disk %s failed: %v", diskID, err)
+		return fmt.Errorf("failed to reserve node %s for detach: %w", nodeID, err)
 	}
 	defer slot.Release()