Merge pull request #1218 from huww98/disk-adslot-message

disk/ad_slot: return better GRPC error message

Author: k8s-ci-robot

pkg/common/interceptors.go

@@ -39,11 +39,37 @@ func instrumentGRPC(driverType string) grpc.UnaryServerInterceptor {
 }
 
 func recordExecTime(time time.Duration, method, driverType string, err error) {
+	// copied from google.golang.org/grpc/server.go
+	appStatus, ok := status.FromError(err)
+	if !ok {
+		// Convert non-status application error to a status error with code
+		// Unknown, but handle context errors specifically.
+		appStatus = status.FromContextError(err)
+	}
+
 	labels := prometheus.Labels{
 		metric.CsiGrpcExecTimeLabelMethod: method,
 		metric.CsiGrpcExecTimeLabelType:   driverType,
-		metric.CsiGrpcExecTimeLabelCode:   status.Code(err).String(),
+		metric.CsiGrpcExecTimeLabelCode:   appStatus.Code().String(),
 	}
 	metric.CsiGrpcExecTimeCollector.ExecCountMetric.With(labels).Inc()
 	metric.CsiGrpcExecTimeCollector.ExecTimeTotalMetric.With(labels).Add(time.Seconds())
 }
+
+// Timeout the request a little bit earlier, to get the error message out.
+// reduce the timeout by 1s or 10%, whichever is smaller.
+func earlyTimeout(ctx context.Context, req interface{}, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+	deadline, ok := ctx.Deadline()
+	if !ok {
+		return handler(ctx, req)
+	}
+	timeout := time.Until(deadline)
+	if time.Second < timeout/10 {
+		deadline = deadline.Add(-time.Second)
+	} else {
+		deadline = deadline.Add(-timeout / 10)
+	}
+	ctx, cancel := context.WithDeadline(ctx, deadline)
+	defer cancel()
+	return handler(ctx, req)
+}

pkg/common/interceptors_test.go

@@ -0,0 +1,42 @@
+package common
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestEarlyTimeout_None(t *testing.T) {
+	handler := func(ctx context.Context, _ interface{}) (interface{}, error) {
+		_, ok := ctx.Deadline()
+		assert.False(t, ok)
+		return struct{}{}, nil
+	}
+	earlyTimeout(context.Background(), nil, nil, handler)
+}
+
+func TestEarlyTimeout(t *testing.T) {
+	cases := []struct {
+		before, after time.Duration
+	}{
+		{60 * time.Second, 59 * time.Second},
+		{1 * time.Second, 900 * time.Millisecond},
+	}
+	for _, tc := range cases {
+		t.Run(tc.before.String(), func(t *testing.T) {
+			now := time.Now()
+			handler := func(ctx context.Context, _ interface{}) (interface{}, error) {
+				deadline, ok := ctx.Deadline()
+				assert.True(t, ok)
+				assert.WithinDuration(t, now.Add(tc.after), deadline, 10*time.Millisecond)
+				return struct{}{}, nil
+			}
+			ctx, cancel := context.WithDeadline(context.Background(), now.Add(tc.before))
+			t.Cleanup(cancel)
+			earlyTimeout(ctx, nil, nil, handler)
+		})
+	}
+
+}

pkg/common/start.go

@@ -54,7 +54,7 @@ func RunCSIServer(driverType, endpoint string, servers Servers) {
 	}
 
 	opts := []grpc.ServerOption{
-		grpc.ChainUnaryInterceptor(logGRPC, instrumentGRPC(driverType)),
+		grpc.ChainUnaryInterceptor(logGRPC, instrumentGRPC(driverType), earlyTimeout),
 	}
 	server := grpc.NewServer(opts...)
 

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

@@ -179,7 +179,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()
 
@@ -443,7 +443,7 @@ func (ad *DiskAttachDetach) detachDisk(ctx context.Context, ecsClient cloud.ECSI
 	// 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()