e2e_node: DRA: reimplement call blocking

test/e2e/dra/test-driver/app/kubeletplugin.go

@@ -54,7 +54,8 @@ type ExamplePlugin struct {
 	prepared       map[ClaimID]any
 	gRPCCalls      []GRPCCall
 
-	block bool
+	blockPrepareResourcesMutex   sync.Mutex
+	blockUnprepareResourcesMutex sync.Mutex
 
 	prepareResourcesFailure   error
 	failPrepareResourcesMutex sync.Mutex
@@ -168,16 +169,20 @@ func (ex *ExamplePlugin) IsRegistered() bool {
 	return status.PluginRegistered
 }
 
-// Block sets a flag to block Node[Un]PrepareResources
+// BlockNodePrepareResources locks blockPrepareResourcesMutex and returns unlocking function for it
-// to emulate time consuming or stuck calls
+func (ex *ExamplePlugin) BlockNodePrepareResources() func() {
-func (ex *ExamplePlugin) Block() {
+	ex.blockPrepareResourcesMutex.Lock()
-	ex.block = true
+	return func() {
+		ex.blockPrepareResourcesMutex.Unlock()
+	}
 }
 
-func (ex *ExamplePlugin) withLock(mutex *sync.Mutex, f func()) {
+// BlockNodeUnprepareResources locks blockUnprepareResourcesMutex and returns unlocking function for it
-	mutex.Lock()
+func (ex *ExamplePlugin) BlockNodeUnprepareResources() func() {
-	f()
+	ex.blockUnprepareResourcesMutex.Lock()
-	mutex.Unlock()
+	return func() {
+		ex.blockUnprepareResourcesMutex.Unlock()
+	}
 }
 
 // SetNodePrepareResourcesFailureMode sets the failure mode for NodePrepareResources call
@@ -227,15 +232,10 @@ func (ex *ExamplePlugin) getUnprepareResourcesFailure() error {
 func (ex *ExamplePlugin) nodePrepareResource(ctx context.Context, claimName string, claimUID string, resourceHandle string, structuredResourceHandle []*resourceapi.StructuredResourceHandle) ([]string, error) {
 	logger := klog.FromContext(ctx)
 
-	// Block to emulate plugin stuckness or slowness.
-	// By default the call will not be blocked as ex.block = false.
-	if ex.block {
-		<-ctx.Done()
-		return nil, ctx.Err()
-	}
-
 	ex.mutex.Lock()
 	defer ex.mutex.Unlock()
+	ex.blockPrepareResourcesMutex.Lock()
+	defer ex.blockPrepareResourcesMutex.Unlock()
 
 	deviceName := "claim-" + claimUID
 	vendor := ex.driverName
@@ -385,14 +385,10 @@ func (ex *ExamplePlugin) NodePrepareResources(ctx context.Context, req *drapbv1a
 // NodePrepareResource. It's idempotent, therefore it is not an error when that
 // file is already gone.
 func (ex *ExamplePlugin) nodeUnprepareResource(ctx context.Context, claimName string, claimUID string, resourceHandle string, structuredResourceHandle []*resourceapi.StructuredResourceHandle) error {
-	logger := klog.FromContext(ctx)
+	ex.blockUnprepareResourcesMutex.Lock()
+	defer ex.blockUnprepareResourcesMutex.Unlock()
 
-	// Block to emulate plugin stuckness or slowness.
+	logger := klog.FromContext(ctx)
-	// By default the call will not be blocked as ex.block = false.
-	if ex.block {
-		<-ctx.Done()
-		return ctx.Err()
-	}
 
 	filePath := ex.getJSONFilePath(claimUID)
 	if err := ex.fileOps.Remove(filePath); err != nil {

test/e2e_node/dra_test.go

@@ -39,7 +39,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
-	dra "k8s.io/kubernetes/pkg/kubelet/cm/dra/plugin"
 	admissionapi "k8s.io/pod-security-admission/api"
 
 	"k8s.io/kubernetes/test/e2e/feature"
@@ -111,9 +110,9 @@ var _ = framework.SIGDescribe("node")("DRA", feature.DynamicResourceAllocation,
 		})
 
 		ginkgo.It("must keep pod in pending state if NodePrepareResources times out", func(ctx context.Context) {
-			ginkgo.By("set delay for the NodePrepareResources call")
+			unblock := kubeletPlugin.BlockNodePrepareResources()
-			kubeletPlugin.Block()
+			defer unblock()
-			pod := createTestObjects(ctx, f.ClientSet, getNodeName(ctx, f), f.Namespace.Name, "draclass", "external-claim", "drapod", true)
+			pod := createTestObjects(ctx, f.ClientSet, getNodeName(ctx, f), f.Namespace.Name, "draclass", "external-claim", "drapod", true, []string{driverName})
 
 			ginkgo.By("wait for pod to be in Pending state")
 			err := e2epod.WaitForPodCondition(ctx, f.ClientSet, f.Namespace.Name, pod.Name, "Pending", framework.PodStartShortTimeout, func(pod *v1.Pod) (bool, error) {
@@ -121,9 +120,6 @@ var _ = framework.SIGDescribe("node")("DRA", feature.DynamicResourceAllocation,
 			})
 			framework.ExpectNoError(err)
 
-			ginkgo.By("wait for NodePrepareResources call")
-			gomega.Eventually(kubeletPlugin.GetGRPCCalls).WithTimeout(dra.PluginClientTimeout * 2).Should(testdriver.NodePrepareResourcesSucceeded)
-
 			// TODO: Check condition or event when implemented
 			// see https://github.com/kubernetes/kubernetes/issues/118468 for details
 			ginkgo.By("check that pod is consistently in Pending state")