Repair a node's export policy rules when it registers

CHANGELOG.md

@@ -10,6 +10,7 @@
 
 - **Kubernetes:** Fixed race conditions and state handling for concurrent publish/unpublish, clone, cache, and backend update operations.
 - **Kubernetes:** Fixed export policy race conditions and concurrent publish/unpublish handling for subordinate volumes and read-only clones in ONTAP-NAS and ONTAP-NAS-Economy drivers.
+- **Kubernetes:** Fixed ONTAP-NAS and ONTAP-NAS-Economy drivers with `autoExportPolicy` never restoring a published node's export policy rules after the node changed IP address or the rules were removed; the rules are now repaired for every volume published to a node when that node registers (Issue [#1181](https://github.com/NetApp/trident/issues/1181)).
 - **Kubernetes:** Fixed ONTAP-NAS-Economy FlexVol discovery after MetroCluster failover when snapshot policies differ (Issue [#1082](https://github.com/NetApp/trident/issues/1082)).
 - **Kubernetes:** Fixed ONTAP-NAS-Economy volume delete retries that interrupted long-running qtree deletes (Issue [#1121](https://github.com/NetApp/trident/issues/1121)).
 - **Kubernetes:** Fixed ONTAP-SAN and ONTAP-SAN-Economy import and resize behavior, including `fsType` validation, volume metadata handling, and autogrow mode behavior.

core/common.go

@@ -307,6 +307,46 @@ func generateVolumePublication(volName string, publishInfo *models.VolumePublish
 	return vp
 }
 
+// volumeNodeAccessRepair asks a backend's driver to add any missing export rules on one volume's
+// policy for the nodes currently published to it. volConfig is a private copy, so the driver call
+// can run without holding any orchestrator lock.
+type volumeNodeAccessRepair struct {
+	backend   storage.Backend
+	volConfig *storage.VolumeConfig
+	nodes     []*models.Node
+}
+
+// applyNodeAccessRepairs runs each repair through the backend's driver, stopping early if ctx is
+// cancelled. The driver is called directly rather than through the backend wrapper, whose
+// "already up to date" short-circuit can be flipped by a concurrent publish while the repairs are
+// in flight. Failures are logged and skipped: the next registration of any node published to the
+// volume retries the repair.
+func applyNodeAccessRepairs(ctx context.Context, nodeName string, repairs []volumeNodeAccessRepair) {
+	for _, repair := range repairs {
+		if ctx.Err() != nil {
+			return
+		}
+		err := repair.backend.Driver().ReconcileVolumeNodeAccess(ctx, repair.volConfig, repair.nodes)
+		if err != nil {
+			Logc(ctx).WithError(err).WithFields(LogFields{
+				"node":    nodeName,
+				"volume":  repair.volConfig.Name,
+				"backend": repair.backend.Name(),
+			}).Warn("Could not repair export rules for a volume published to the node.")
+		}
+	}
+}
+
+// backendCanRepairNodeAccess reports whether b hosts per-volume export policies worth repairing:
+// it enforces publications and is still serving volumes.
+func backendCanRepairNodeAccess(b storage.Backend) bool {
+	if !b.CanEnablePublishEnforcement() {
+		return false
+	}
+	state := b.State()
+	return state.IsOnline() || state.IsDeleting()
+}
+
 // isDockerPluginMode returns true if the ENV variable config.DockerPluginModeEnvVariable is set
 func isDockerPluginMode() bool {
 	return os.Getenv(config.DockerPluginModeEnvVariable) != ""

core/concurrent_core.go

@@ -7381,9 +7381,147 @@ func (o *ConcurrentTridentOrchestrator) AddNode(
 	if invalidateErr := o.invalidateAllBackendNodeAccess(ctx); invalidateErr != nil {
 		Logc(ctx).WithError(invalidateErr).Error("Could not invalidate backend node access.")
 	}
+
+	// A node's per-volume export rules are otherwise written only by its own ControllerPublishVolume,
+	// which Kubernetes never repeats while a VolumeAttachment exists. Registration is the event a
+	// rebooted or re-addressed node always produces, so repair its volumes' rules here, off the
+	// request path and detached from the request's cancellation.
+	nodeName := node.Name
+	o.bgTasks.launch(context.WithoutCancel(ctx), func(ctx context.Context) {
+		o.repairNodeAccessForNode(ctx, nodeName)
+	})
 	return
 }
 
+// repairNodeAccessForNode adds any missing export rules, for every node concerned, on every volume
+// that nodeName is published to. It reads what it needs from the cache and calls the drivers with
+// no cache lock held.
+func (o *ConcurrentTridentOrchestrator) repairNodeAccessForNode(ctx context.Context, nodeName string) {
+	repairs, err := o.nodeAccessRepairsForNode(ctx, nodeName)
+	if err != nil {
+		Logc(ctx).WithError(err).WithField("node", nodeName).Warn(
+			"Could not collect the volumes published to the node for export rule repair.")
+		return
+	}
+	if len(repairs) == 0 {
+		return
+	}
+	Logc(ctx).WithFields(LogFields{
+		"node":    nodeName,
+		"volumes": len(repairs),
+	}).Debug("Repairing export rules for volumes published to the node.")
+	applyNodeAccessRepairs(ctx, nodeName, repairs)
+}
+
+// nodeAccessRepairsForNode collects one repair per volume that nodeName is published to. A
+// subordinate publication resolves to the share-source volume whose policy grants the access, and
+// each repair carries every node published to that volume, so a shared volume is repaired for all
+// of its nodes at once. Only backends that enforce publications take part.
+func (o *ConcurrentTridentOrchestrator) nodeAccessRepairsForNode(
+	ctx context.Context, nodeName string,
+) ([]volumeNodeAccessRepair, error) {
+	_, results, unlocker, err := db.Lock(ctx, db.Query(
+		db.ListVolumePublications(), db.ListNodes(), db.ListSubordinateVolumes()))
+	if err != nil {
+		unlocker()
+		return nil, err
+	}
+	publications, allNodes, subordinates := results[0].VolumePublications, results[0].Nodes,
+		results[0].SubordinateVolumes
+	unlocker()
+
+	shareSourceByName := make(map[string]string, len(subordinates))
+	for _, subordinate := range subordinates {
+		shareSourceByName[subordinate.Config.Name] = subordinate.Config.ShareSourceVolume
+	}
+	nodesByName := make(map[string]*models.Node, len(allNodes))
+	for _, n := range allNodes {
+		nodesByName[n.Name] = n
+	}
+
+	// Every node published to each hosting volume, and the hosting volumes this node is published to.
+	nodeNamesByVolume := make(map[string]map[string]struct{})
+	targetVolumes := make(map[string]struct{})
+	for _, pub := range publications {
+		hosting := pub.VolumeName
+		if source, ok := shareSourceByName[hosting]; ok {
+			hosting = source
+		}
+		if nodeNamesByVolume[hosting] == nil {
+			nodeNamesByVolume[hosting] = make(map[string]struct{})
+		}
+		nodeNamesByVolume[hosting][pub.NodeName] = struct{}{}
+		if pub.NodeName == nodeName {
+			targetVolumes[hosting] = struct{}{}
+		}
+	}
+	if len(targetVolumes) == 0 {
+		return nil, nil
+	}
+
+	volumeQueries := make([][]db.Subquery, 0, len(targetVolumes))
+	for name := range targetVolumes {
+		volumeQueries = append(volumeQueries, db.Query(db.ReadVolume(name)))
+	}
+	_, results, unlocker, err = db.Lock(ctx, volumeQueries...)
+	if err != nil {
+		unlocker()
+		return nil, err
+	}
+	volumes := make([]*storage.Volume, 0, len(results))
+	backendUUIDs := make(map[string]struct{})
+	for _, result := range results {
+		if volume := result.Volume.Read; volume != nil {
+			volumes = append(volumes, volume)
+			backendUUIDs[volume.BackendUUID] = struct{}{}
+		}
+	}
+	unlocker()
+	if len(volumes) == 0 {
+		return nil, nil
+	}
+
+	backendQueries := make([][]db.Subquery, 0, len(backendUUIDs))
+	for backendUUID := range backendUUIDs {
+		backendQueries = append(backendQueries, db.Query(db.ReadBackend(backendUUID)))
+	}
+	_, results, unlocker, err = db.Lock(ctx, backendQueries...)
+	if err != nil {
+		unlocker()
+		return nil, err
+	}
+	backendsByUUID := make(map[string]storage.Backend, len(results))
+	for _, result := range results {
+		if backend := result.Backend.Read; backend != nil {
+			backendsByUUID[backend.BackendUUID()] = backend
+		}
+	}
+	unlocker()
+
+	sort.Slice(volumes, func(i, j int) bool { return volumes[i].Config.Name < volumes[j].Config.Name })
+	repairs := make([]volumeNodeAccessRepair, 0, len(volumes))
+	for _, volume := range volumes {
+		backend, ok := backendsByUUID[volume.BackendUUID]
+		if !ok || !backendCanRepairNodeAccess(backend) {
+			continue
+		}
+		nodeNames := make([]string, 0, len(nodeNamesByVolume[volume.Config.Name]))
+		for name := range nodeNamesByVolume[volume.Config.Name] {
+			nodeNames = append(nodeNames, name)
+		}
+		sort.Strings(nodeNames)
+		nodes := make([]*models.Node, 0, len(nodeNames))
+		for _, name := range nodeNames {
+			if node := nodesByName[name]; node != nil {
+				nodes = append(nodes, node)
+			}
+		}
+		// ReadVolume hands back a deep copy, so the config is already private to this repair.
+		repairs = append(repairs, volumeNodeAccessRepair{backend: backend, volConfig: volume.Config, nodes: nodes})
+	}
+	return repairs, nil
+}
+
 // UpdateNode updates the publication state of a node. It does not create a new node if it does not exist.
 func (o *ConcurrentTridentOrchestrator) UpdateNode(
 	ctx context.Context, nodeName string, flags *models.NodePublicationStateFlags,

core/concurrent_core_test.go

@@ -13687,6 +13687,9 @@ func TestAddNodeConcurrentCore(t *testing.T) {
 			o := getConcurrentOrchestrator()
 			o.storeClient = mockStoreClient
 			o.lastNodeRegistrationTime = time.Time{}
+			// AddNode launches a background repair that reads the cache; stop it before the next
+			// case reinitializes the cache underneath it.
+			defer o.bgTasks.stop()
 
 			err := o.AddNode(testCtx, tt.newNode, func(_, _, _ string) {})
 
@@ -13704,6 +13707,153 @@ func TestAddNodeConcurrentCore(t *testing.T) {
 	}
 }
 
+func TestNodeAccessRepairsForNodeConcurrentCore(t *testing.T) {
+	const enforcingUUID, plainUUID = "uuid-enforcing", "uuid-plain"
+
+	tests := []struct {
+		name          string
+		node          string
+		publications  []*models.VolumePublication
+		expectVolumes []string
+		expectNodes   []string
+	}{
+		{
+			name: "RepairCarriesEveryNodePublishedToTheVolume",
+			node: "nodeA",
+			publications: []*models.VolumePublication{
+				getFakeVolumePublication("vol1", "nodeA"),
+				getFakeVolumePublication("vol1", "nodeB"),
+			},
+			expectVolumes: []string{"vol1"},
+			expectNodes:   []string{"nodeA", "nodeB"},
+		},
+		{
+			name: "SubordinatePublicationResolvesToTheSourceVolume",
+			node: "nodeA",
+			publications: []*models.VolumePublication{
+				getFakeVolumePublication("sub1", "nodeA"),
+				getFakeVolumePublication("vol1", "nodeB"),
+			},
+			expectVolumes: []string{"vol1"},
+			expectNodes:   []string{"nodeA", "nodeB"},
+		},
+		{
+			name: "SourceAndSubordinateProduceOneRepair",
+			node: "nodeA",
+			publications: []*models.VolumePublication{
+				getFakeVolumePublication("vol1", "nodeA"),
+				getFakeVolumePublication("sub1", "nodeA"),
+			},
+			expectVolumes: []string{"vol1"},
+			expectNodes:   []string{"nodeA"},
+		},
+		{
+			name: "UnregisteredNodeIsLeftOut",
+			node: "nodeA",
+			publications: []*models.VolumePublication{
+				getFakeVolumePublication("vol1", "nodeA"),
+				getFakeVolumePublication("vol1", "ghost"),
+			},
+			expectVolumes: []string{"vol1"},
+			expectNodes:   []string{"nodeA"},
+		},
+		{
+			name:          "BackendWithoutPublishEnforcementIsSkipped",
+			node:          "nodeA",
+			publications:  []*models.VolumePublication{getFakeVolumePublication("vol2", "nodeA")},
+			expectVolumes: []string{},
+		},
+		{
+			name:          "OtherNodesPublicationsAreNotRepaired",
+			node:          "nodeA",
+			publications:  []*models.VolumePublication{getFakeVolumePublication("vol1", "nodeB")},
+			expectVolumes: []string{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			db.Initialize()
+			mockCtrl := gomock.NewController(t)
+			enforcing := getMockBackend(mockCtrl, "enforcing", enforcingUUID)
+			enforcing.EXPECT().CanEnablePublishEnforcement().Return(true).AnyTimes()
+			plain := getMockBackend(mockCtrl, "plain", plainUUID)
+			plain.EXPECT().CanEnablePublishEnforcement().Return(false).AnyTimes()
+			addBackendsToCache(t, enforcing, plain)
+
+			source := getFakeVolume("vol1", enforcingUUID)
+			source.Config.ExportPolicy = "vol1"
+			addVolumesToCache(t, source, getFakeVolume("vol2", plainUUID))
+			addSubordinateVolumesToCache(t, &storage.Volume{
+				Config: &storage.VolumeConfig{Name: "sub1", ShareSourceVolume: "vol1"},
+				State:  storage.VolumeStateSubordinate,
+			})
+			addNodesToCache(t, getFakeNode("nodeA"), getFakeNode("nodeB"))
+			addVolumePublicationsToCache(t, tt.publications...)
+
+			repairs, err := getConcurrentOrchestrator().nodeAccessRepairsForNode(testCtx, tt.node)
+			require.NoError(t, err)
+
+			volumes := make([]string, 0, len(repairs))
+			for _, r := range repairs {
+				volumes = append(volumes, r.volConfig.Name)
+			}
+			assert.ElementsMatch(t, tt.expectVolumes, volumes)
+			if len(tt.expectVolumes) == 1 {
+				assert.Same(t, enforcing, repairs[0].backend)
+				assert.ElementsMatch(t, tt.expectNodes, nodeNamesOf(repairs[0].nodes))
+				assert.Equal(t, source.Config, repairs[0].volConfig)
+				assert.NotSame(t, source.Config, repairs[0].volConfig, "the driver must get a private copy")
+			}
+		})
+	}
+}
+
+// Registering a node must repair the export rules of the volumes it is published to, for every
+// node published to them, without holding any cache lock while the driver talks to ONTAP.
+func TestAddNodeConcurrentCore_RepairsExportRulesOfPublishedVolumes(t *testing.T) {
+	const backendUUID = "uuid1"
+	db.Initialize()
+	mockCtrl := gomock.NewController(t)
+
+	mockStoreClient := mockpersistentstore.NewMockStoreClient(mockCtrl)
+	mockStoreClient.EXPECT().AddOrUpdateNode(gomock.Any(), gomock.Any()).Return(nil)
+
+	driver := mockstorage.NewMockDriver(mockCtrl)
+	backend := getMockBackend(mockCtrl, "backend1", backendUUID)
+	backend.EXPECT().CanEnablePublishEnforcement().Return(true).AnyTimes()
+	backend.EXPECT().Driver().Return(driver).AnyTimes()
+	backend.EXPECT().InvalidateNodeAccess().AnyTimes()
+	addBackendsToCache(t, backend)
+
+	volume := getFakeVolume("vol1", backendUUID)
+	volume.Config.ExportPolicy = "vol1"
+	addVolumesToCache(t, volume)
+	addNodesToCache(t, getFakeNode("nodeB"))
+	addVolumePublicationsToCache(t,
+		getFakeVolumePublication("vol1", "nodeA"), getFakeVolumePublication("vol1", "nodeB"))
+
+	repaired := make(chan []*models.Node, 1)
+	driver.EXPECT().ReconcileVolumeNodeAccess(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+		func(_ context.Context, volConfig *storage.VolumeConfig, nodes []*models.Node) error {
+			assert.Equal(t, "vol1", volConfig.Name)
+			repaired <- nodes
+			return nil
+		})
+
+	o := getConcurrentOrchestrator()
+	o.storeClient = mockStoreClient
+	require.NoError(t, o.AddNode(testCtx, &models.Node{Name: "nodeA", IPs: []string{"10.0.0.5"}}, func(_, _, _ string) {}))
+	defer o.bgTasks.stop()
+
+	select {
+	case nodes := <-repaired:
+		assert.ElementsMatch(t, []string{"nodeA", "nodeB"}, nodeNamesOf(nodes))
+	case <-time.After(5 * time.Second):
+		t.Fatal("registering the node did not repair its published volume")
+	}
+}
+
 func TestUpdateNodeConcurrentCore(t *testing.T) {
 	tests := []struct {
 		name          string