NVME driver concurrency

Author: vivakeram

core/concurrent_core.go

@@ -3236,6 +3236,7 @@ func (o *ConcurrentTridentOrchestrator) publishVolume(ctx context.Context, volum
 	}
 
 	publishInfo.TridentUUID = o.uuid
+	publishInfo.BackendUUID = backend.BackendUUID()
 
 	// Enable publish enforcement if the backend supports it and the volume isn't already enforced
 	if publishEnforceable && !volume.Config.AccessInfo.PublishEnforcement {

storage_drivers/ontap/api/abstraction_error.go

@@ -18,6 +18,7 @@ const (
 	EXPORT_POLICY_NOT_FOUND                    = "1703954"
 	EXPORT_POLICY_RULE_EXISTS                  = "1704070"
 	CONSISTENCY_GROUP_SNAP_EXISTS_ERROR        = "53411921"
+	NVME_SUBSYSTEM_ALREADY_EXISTS              = "72090025"
 )
 
 // ///////////////////////////////////////////////////////////////////////////

storage_drivers/ontap/api/abstraction_rest.go

@@ -3460,6 +3460,28 @@ func (d OntapAPIREST) NVMeSubsystemCreate(ctx context.Context, subsystemName, co
 		Logc(ctx).Infof("Subsystem doesn't exist, creating new subsystem %v now.", subsystemName)
 		subsystem, err = d.api.NVMeSubsystemCreate(ctx, subsystemName, comment)
 		if err != nil {
+
+			errorResponse, extractErr := ExtractErrorResponse(ctx, err)
+			if extractErr != nil {
+				return nil, err
+			}
+			if errorResponse != nil && errorResponse.Error != nil {
+				errorCode := errorResponse.Error.Code
+				if errorCode != nil && *errorCode == NVME_SUBSYSTEM_ALREADY_EXISTS {
+					// If there is a conflict error, it means the subsystem got created in between the get and create calls
+					Logc(ctx).Infof("Subsystem %v already exists.", subsystemName)
+					subsystem, err = d.api.NVMeSubsystemGetByName(ctx, subsystemName, fields)
+					if err != nil {
+						Logc(ctx).Infof("Problem getting subsystem; %v", err)
+						return nil, err
+					}
+					if subsystem != nil {
+						return &NVMeSubsystem{UUID: *subsystem.UUID, Name: *subsystem.Name, NQN: *subsystem.TargetNqn}, nil
+					}
+					return nil, fmt.Errorf("unable to create subsystem %v", subsystemName)
+				}
+			}
+
 			return nil, err
 		}
 

storage_drivers/ontap/api/abstraction_rest_test.go

@@ -795,6 +795,41 @@ func TestNVMeSubsystemCreate(t *testing.T) {
 	mock.EXPECT().ClientConfig().Return(clientConfig).AnyTimes()
 	newsubsys, err = oapi.NVMeSubsystemCreate(ctx, subsystemName, subsystemComment)
 	assert.Error(t, err)
+
+	// case 6: Subsystem not present, create a new one but returned already exists error with code NVME_SUBSYSTEM_ALREADY_EXISTS
+	mock.EXPECT().NVMeSubsystemGetByName(ctx, subsystemName, gomock.Any()).Return(nil, nil).Times(1)
+	mock.EXPECT().NVMeSubsystemGetByName(ctx, subsystemName, gomock.Any()).Return(subsys, nil).Times(1)
+	mockErr := &nvme.NvmeSubsystemCreateDefault{
+		Payload: &models.ErrorResponse{
+			Error: &models.ReturnedError{
+				Code:    convert.ToPtr(api.NVME_SUBSYSTEM_ALREADY_EXISTS),
+				Message: convert.ToPtr("NVMe subsystem already exists"),
+			},
+		},
+	}
+	mock.EXPECT().NVMeSubsystemCreate(ctx, subsystemName, subsystemComment).Return(nil, mockErr)
+	mock.EXPECT().ClientConfig().Return(clientConfig).AnyTimes()
+	newsubsys, err = oapi.NVMeSubsystemCreate(ctx, subsystemName, subsystemComment)
+	assert.NoError(t, err)
+	assert.Equal(t, newsubsys.UUID, subsysUUID, "subsystem UUID does not match")
+	assert.Equal(t, newsubsys.Name, subsystemName, "subsystem name does not match")
+
+	// case 7: Subsystem not present, create a new one but returned already exists error with code NVME_SUBSYSTEM_ALREADY_EXISTS
+	mock.EXPECT().NVMeSubsystemGetByName(ctx, subsystemName, gomock.Any()).Return(nil, nil).Times(1)
+	mock.EXPECT().NVMeSubsystemGetByName(ctx, subsystemName, gomock.Any()).Return(nil, nil).Times(1)
+	mockErr = &nvme.NvmeSubsystemCreateDefault{
+		Payload: &models.ErrorResponse{
+			Error: &models.ReturnedError{
+				Code:    convert.ToPtr(api.NVME_SUBSYSTEM_ALREADY_EXISTS),
+				Message: convert.ToPtr("NVMe subsystem already exists"),
+			},
+		},
+	}
+	mock.EXPECT().NVMeSubsystemCreate(ctx, subsystemName, subsystemComment).Return(nil, mockErr)
+	mock.EXPECT().ClientConfig().Return(clientConfig).AnyTimes()
+	newsubsys, err = oapi.NVMeSubsystemCreate(ctx, subsystemName, subsystemComment)
+	assert.Error(t, err)
+	assert.Nil(t, newsubsys, "subsystem expected to be nil when error is returned")
 }
 
 func TestNVMeEnsureNamespaceMapped(t *testing.T) {

storage_drivers/ontap/ontap_common.go

@@ -152,10 +152,16 @@ var (
 	volumeNameRegex          = regexp.MustCompile(`\{+.*\.volume.Name[^{a-z]*\}+`)
 	volumeNameStartWithRegex = regexp.MustCompile(`^[A-Za-z_].*`)
 	smbShareDeleteACL        = map[string]string{DefaultSMBAccessControlUser: DefaultSMBAccessControlUserType}
-	lunMutex                 = locks.NewGCNamedMutex()
-	igroupMutex              = locks.NewGCNamedMutex()
 	exportPolicyMutex        = locks.NewGCNamedMutex()
 
+	// NOTE: The lock order should be lunMutex first, then igroupMutex
+	lunMutex    = locks.NewGCNamedMutex()
+	igroupMutex = locks.NewGCNamedMutex()
+
+	// NOTE: The lock order should be namespaceMutex first, then subsystemMutex
+	namespaceMutex = locks.NewGCNamedMutex()
+	subsystemMutex = locks.NewGCNamedMutex()
+
 	duringVolCloneAfterSnapCreation1 = fiji.Register("duringVolCloneAfterSnapCreation1", "ontap_common")
 	duringVolCloneAfterSnapCreation2 = fiji.Register("duringVolCloneAfterSnapCreation2", "ontap_common")
 )
@@ -5431,3 +5437,15 @@ func getUniqueNodeSpecificSubsystemName(
 
 	return completeSSName, finalSSName, nil
 }
+
+// lockNamespaceAndSubsystem acquires the namespace lock first and then the subsystem lock.
+// It returns a function that should be deferred to release the locks in reverse order.
+func lockNamespaceAndSubsystem(nsUUID, ssUUID string) func() {
+	namespaceMutex.Lock(nsUUID)
+	subsystemMutex.Lock(ssUUID)
+
+	return func() {
+		subsystemMutex.Unlock(ssUUID)
+		namespaceMutex.Unlock(nsUUID)
+	}
+}

storage_drivers/ontap/ontap_san_nvme.go

@@ -690,6 +690,10 @@ func (d *NVMeStorageDriver) Import(ctx context.Context, volConfig *storage.Volum
 		return fmt.Errorf("nvme namespace not found in volume %s", originalName)
 	}
 
+	// Lock the namespace until import is done.
+	namespaceMutex.Lock(nsInfo.UUID)
+	defer namespaceMutex.Unlock(nsInfo.UUID)
+
 	// The Namespace should be online
 	if nsInfo.State != "online" {
 		return fmt.Errorf("Namespace %s is not online", nsInfo.Name)
@@ -888,6 +892,7 @@ func (d *NVMeStorageDriver) Publish(
 
 	nsPath := volConfig.InternalID
 	nsUUID := volConfig.AccessInfo.NVMeNamespaceUUID
+
 	// For docker context, some of the attributes like fsType, luks needs to be
 	// fetched from namespace where they were stored while creating the namespace.
 	if d.Config.DriverContext == tridentconfig.ContextDocker {
@@ -929,17 +934,20 @@ func (d *NVMeStorageDriver) Publish(
 		}
 	}
 
-	// Checks if subsystem exists and creates a new one if not
-	subsystem, err := d.API.NVMeSubsystemCreate(ctx, ssName, ssName)
+	// If 2 concurrent requests try to create the same subsystem, one will succeed and ONTAP is guaranteed to return
+	// "already exists" error code for the other. So no need for locking around subsystem creation.
+	subsystem, err := d.createOrGetSubsystem(ctx, ssName)
 	if err != nil {
-		Logc(ctx).Errorf("subsystem create failed, %v", err)
 		return err
 	}
 
 	if subsystem == nil {
 		return fmt.Errorf("No subsystem returned after subsystem create")
 	}
 
+	unlock := lockNamespaceAndSubsystem(nsUUID, subsystem.UUID)
+	defer unlock()
+
 	// Fill important info in publishInfo
 	publishInfo.NVMeSubsystemNQN = subsystem.NQN
 	publishInfo.NVMeSubsystemUUID = subsystem.UUID
@@ -989,6 +997,9 @@ func (d *NVMeStorageDriver) Unpublish(
 	subsystemUUID := volConfig.AccessInfo.NVMeSubsystemUUID
 	namespaceUUID := volConfig.AccessInfo.NVMeNamespaceUUID
 
+	unlock := lockNamespaceAndSubsystem(namespaceUUID, subsystemUUID)
+	defer unlock()
+
 	removePublishInfo, err := d.API.NVMeEnsureNamespaceUnmapped(ctx, publishInfo.HostNQN, subsystemUUID, namespaceUUID)
 	if removePublishInfo {
 		volConfig.AccessInfo.NVMeTargetIPs = []string{}
@@ -1165,6 +1176,16 @@ func (d *NVMeStorageDriver) getStoragePoolAttributes(ctx context.Context) map[st
 	}
 }
 
+func (d *NVMeStorageDriver) createOrGetSubsystem(ctx context.Context, ssName string) (*api.NVMeSubsystem, error) {
+	// This checks if subsystem exists and creates it if not.
+	ss, err := d.API.NVMeSubsystemCreate(ctx, ssName, ssName)
+	if err != nil {
+		Logc(ctx).Errorf("subsystem create failed, %v", err)
+		return nil, err
+	}
+	return ss, nil
+}
+
 // GetVolumeOpts populates the volume properties required in volume create.
 func (d *NVMeStorageDriver) GetVolumeOpts(
 	ctx context.Context, volConfig *storage.VolumeConfig, requests map[string]sa.Request,