e2e: add nvmeof GroupLock stress tests for concurrent operations

Add e2e tests to validate nvmeof NodeServer GroupLock implementation
under concurrent NodeStage (Group A) and NodeUnstage (Group B)
operations. The tests ensure no deadlock occurs when multiple PVCs and
Pods are created and deleted simultaneously.

New helper file (nvmeof_helper.go) provides reusable functions for
concurrent PVC/Pod operations with proper error tracking.
Two test cases cover:
1) sequential concurrent batches (create all, then delete
all)

2) mixed operations with pre-created batch to guarantee
continuous Group A/B switching..

Signed-off-by: gadi-didi <gadi.didi@ibm.com>

Author: gadididi

e2e/nvmeof.go

@@ -18,7 +18,9 @@ package e2e
 
 import (
 	"context"
+	"fmt"
 
+	"github.com/google/uuid"
 	"github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	v1 "k8s.io/api/core/v1"
@@ -189,5 +191,125 @@ var _ = ginkgo.Describe("nvmeof", func() {
 			validateRBDImageCount(f, 0, nvmeofPool)
 			validateOmapCount(f, 0, rbdType, nvmeofPool, volumesType)
 		})
+
+		ginkgo.It("Test GroupLock: Concurrent Create/Delete Pods Only", func() {
+			// This test validates the GroupLock implementation in the NVMeoF NodeServer
+			// by creating and deleting multiple Pods (not PVCs) concurrently.
+			//
+			// Test flow:
+			// 1. Create 3 PVCs sequentially and validate they're Bound
+			// 2. Create 3 Pods concurrently using those PVCs (triggers NodeStage -> Group A lock)
+			// 3. Wait for all Pods to be Running
+			// 4. Delete all 3 Pods concurrently (triggers NodeUnstage -> Group B lock)
+			// 5. Delete all 3 PVCs sequentially
+			// 6. Verify no timeouts/deadlocks and all operations succeed
+			//
+			// This tests GroupLock in the NodeServer without involving ControllerServer operations.
+			totalCount := 3
+
+			ginkgo.By("Creating PVCs sequentially")
+			pvc, err := loadPVC(pvcPath)
+			Expect(err).ShouldNot(HaveOccurred())
+			pvc.Namespace = f.UniqueName
+			pvc.Spec.StorageClassName = &nvmeofStorageClass
+
+			pvcBaseName := uuid.NewString()
+			for i := range totalCount {
+				pvcName := fmt.Sprintf("%s-%d", pvcBaseName, i)
+				pvcCopy := pvc.DeepCopy()
+				pvcCopy.Name = pvcName
+
+				framework.Logf("Creating PVC %d/%d: %s", i+1, totalCount, pvcName)
+				err = createPVCAndvalidatePV(f.ClientSet, pvcCopy, deployTimeout)
+				Expect(err).ShouldNot(HaveOccurred())
+			}
+
+			ginkgo.By("Validating backend RBD images were created")
+			validateRBDImageCount(f, totalCount, nvmeofPool)
+			validateOmapCount(f, totalCount, rbdType, nvmeofPool, volumesType)
+
+			ginkgo.By("Creating Pods concurrently using those PVCs")
+			createResult := createConcurrentPods(totalCount, pvcBaseName, 0, appPath, f)
+
+			// Log any errors
+			if createResult.HasErrors() {
+				createResult.LogErrors()
+			}
+
+			// Verify all creations succeeded
+			Expect(createResult.failed).To(Equal(0),
+				"Expected all %d Pod create operations to succeed, but %d failed",
+				totalCount, createResult.failed)
+
+			ginkgo.By("Waiting for all Pods to be Running")
+			for i := range totalCount {
+				podName := fmt.Sprintf("%s-%d", createResult.uniqueName, i)
+				err = waitForPodInRunningState(podName, f.UniqueName, f.ClientSet, deployTimeout, noError)
+				Expect(err).ShouldNot(HaveOccurred())
+			}
+
+			ginkgo.By("Deleting Pods concurrently")
+			deleteResult := deleteConcurrentPods(createResult, f)
+
+			// Log any errors
+			if deleteResult.HasErrors() {
+				deleteResult.LogErrors()
+			}
+
+			// Verify all deletions succeeded
+			Expect(deleteResult.failed).To(Equal(0),
+				"Expected all %d Pod delete operations to succeed, but %d failed",
+				totalCount, deleteResult.failed)
+
+			ginkgo.By("Deleting PVCs sequentially")
+			for i := range totalCount {
+				pvcName := fmt.Sprintf("%s-%d", pvcBaseName, i)
+				pvcCopy := pvc.DeepCopy()
+				pvcCopy.Name = pvcName
+
+				framework.Logf("Deleting PVC %d/%d: %s", i+1, totalCount, pvcName)
+				err = deletePVCAndValidatePV(f.ClientSet, pvcCopy, deployTimeout)
+				Expect(err).ShouldNot(HaveOccurred())
+			}
+
+			ginkgo.By("Validating all backend RBD images were deleted")
+			validateRBDImageCount(f, 0, nvmeofPool)
+			validateOmapCount(f, 0, rbdType, nvmeofPool, volumesType)
+
+			framework.Logf("GroupLock test passed: %d concurrent Pod creates and %d concurrent Pod deletes completed successfully",
+				totalCount, totalCount)
+		})
+
+		ginkgo.It("Test GroupLock: Mixed Create/Delete Pods with Rapid Switching", func() {
+			// This test validates the GroupLock implementation under rapid switching
+			// between Group A (NodeStage) and Group B (NodeUnstage) operations.
+			//
+			// Test flow:
+			// 1. Create 15 PVCs sequentially
+			// 2. Create 5 Pods using PVCs 0-4 (Group A)
+			// 3. Concurrently: Create 5 Pods using PVCs 5-9 (Group A) + Delete previous 5 Pods (Group B)
+			// 4. Concurrently: Create 5 Pods using PVCs 10-14 (Group A) + Delete previous 5 Pods (Group B)
+			// 5. Delete final 5 Pods
+			// 6. Delete all 15 PVCs sequentially
+			//
+			// This tests rapid GroupLock switching between Group A and B in the NodeServer only,
+			// without involving ControllerServer operations.
+			totalCount := 15
+			batchSize := 5
+
+			ginkgo.By(fmt.Sprintf("Running Pods-only mixed test: %d total PVCs, batches of %d Pods",
+				totalCount, batchSize))
+
+			err := mixedCreateDeletePodsOnly(totalCount, batchSize, pvcPath, appPath, nvmeofStorageClass, f)
+			Expect(err).ShouldNot(HaveOccurred(),
+				"Mixed Pods-only operations should complete without errors")
+
+			ginkgo.By("Validating all backend RBD images were cleaned up")
+			validateRBDImageCount(f, 0, nvmeofPool)
+			validateOmapCount(f, 0, rbdType, nvmeofPool, volumesType)
+
+			framework.Logf("GroupLock Pods-only test passed: %d Pods created and deleted with rapid Group A/B switching",
+				totalCount)
+		})
 	})
 })