diff --git a/pkg/volume/csi/nodeinfomanager/nodeinfomanager.go b/pkg/volume/csi/nodeinfomanager/nodeinfomanager.go index 1bff173c5fe..ba1bde76a2d 100644 --- a/pkg/volume/csi/nodeinfomanager/nodeinfomanager.go +++ b/pkg/volume/csi/nodeinfomanager/nodeinfomanager.go @@ -63,6 +63,7 @@ var ( // the Node and CSINode objects. type nodeInfoManager struct { nodeName types.NodeName + nodeID types.UID volumeHost volume.VolumeHost migratedPlugins map[string](func() bool) // lock protects changes to node. @@ -402,6 +403,10 @@ func (nim *nodeInfoManager) tryUpdateCSINode( return err } + if err = nim.ensureNodeOwnsCSINode(nodeInfo); err != nil { + return err + } + return nim.installDriverToCSINode(nodeInfo, driverName, driverNodeID, maxAttachLimit, topology) } @@ -430,6 +435,12 @@ func (nim *nodeInfoManager) tryInitializeCSINodeWithAnnotation(csiKubeClient cli nim.lock.Lock() defer nim.lock.Unlock() + node, err := csiKubeClient.CoreV1().Nodes().Get(context.TODO(), string(nim.nodeName), metav1.GetOptions{}) + if err != nil { + return err + } + nim.nodeID = node.UID + nodeInfo, err := csiKubeClient.StorageV1().CSINodes().Get(context.TODO(), string(nim.nodeName), metav1.GetOptions{}) if nodeInfo == nil || errors.IsNotFound(err) { // CreateCSINode will set the annotation @@ -439,6 +450,10 @@ func (nim *nodeInfoManager) tryInitializeCSINodeWithAnnotation(csiKubeClient cli return err } + if err = nim.ensureNodeOwnsCSINode(nodeInfo); err != nil { + return err + } + annotationModified := setMigrationAnnotation(nim.migratedPlugins, nodeInfo) if annotationModified { @@ -449,6 +464,44 @@ func (nim *nodeInfoManager) tryInitializeCSINodeWithAnnotation(csiKubeClient cli } +// ensureNodeOwnsCSINode will ensure that the current CSINode object is owned by the node represented by this nodeInfoManager. +// If not, it will delete the existing CSINode object and return an error. +func (nim *nodeInfoManager) ensureNodeOwnsCSINode(nodeInfo *storagev1.CSINode) error { + if ok, csiNodeOwnerID := nim.nodeOwnsCSINode(nodeInfo); !ok { + klog.V(2).Infof("existing CSINode %q is owned by different node (oldNodeID=%q, newNodeID=%q), cleaning up...", nodeInfo.Name, csiNodeOwnerID, nim.nodeID) + err := nim.DeleteCSINode() + if err != nil { + return fmt.Errorf("error deleting existing CSINode %q: %w", nodeInfo.Name, err) + } + // Returning now so that the next attempt can create a new CSINode object + return fmt.Errorf("CSINode %q was owned by different node (oldNodeID=%q, newNodeID=%q), deleted it", nodeInfo.Name, csiNodeOwnerID, nim.nodeID) + } + return nil +} + +// nodeOwnsCSINode checks if the CSINode object is owned by the node represented by this nodeInfoManager. +// It also returns the current owner UID. +func (nim *nodeInfoManager) nodeOwnsCSINode(nodeInfo *storagev1.CSINode) (bool, types.UID) { + var csiNodeOwnerID types.UID + var found bool + for _, ownerRef := range nodeInfo.OwnerReferences { + if ownerRef.Kind != nodeKind.Kind { + continue + } + + csiNodeOwnerID = ownerRef.UID + if ownerRef.Name != string(nim.nodeName) { + continue + } + + if csiNodeOwnerID == nim.nodeID { + found = true + break + } + } + return found, csiNodeOwnerID +} + func (nim *nodeInfoManager) CreateCSINode() (*storagev1.CSINode, error) { csiKubeClient := nim.volumeHost.GetKubeClient() @@ -456,11 +509,6 @@ func (nim *nodeInfoManager) CreateCSINode() (*storagev1.CSINode, error) { return nil, fmt.Errorf("error getting CSI client") } - node, err := csiKubeClient.CoreV1().Nodes().Get(context.TODO(), string(nim.nodeName), metav1.GetOptions{}) - if err != nil { - return nil, err - } - nodeInfo := &storagev1.CSINode{ ObjectMeta: metav1.ObjectMeta{ Name: string(nim.nodeName), @@ -468,8 +516,8 @@ func (nim *nodeInfoManager) CreateCSINode() (*storagev1.CSINode, error) { { APIVersion: nodeKind.Version, Kind: nodeKind.Kind, - Name: node.Name, - UID: node.UID, + Name: string(nim.nodeName), + UID: nim.nodeID, }, }, }, @@ -483,6 +531,16 @@ func (nim *nodeInfoManager) CreateCSINode() (*storagev1.CSINode, error) { return csiKubeClient.StorageV1().CSINodes().Create(context.TODO(), nodeInfo, metav1.CreateOptions{}) } +func (nim *nodeInfoManager) DeleteCSINode() error { + + csiKubeClient := nim.volumeHost.GetKubeClient() + if csiKubeClient == nil { + return fmt.Errorf("error getting CSI client") + } + + return csiKubeClient.StorageV1().CSINodes().Delete(context.TODO(), string(nim.nodeName), metav1.DeleteOptions{}) +} + func setMigrationAnnotation(migratedPlugins map[string](func() bool), nodeInfo *storagev1.CSINode) (modified bool) { if migratedPlugins == nil { return false diff --git a/pkg/volume/csi/nodeinfomanager/nodeinfomanager_test.go b/pkg/volume/csi/nodeinfomanager/nodeinfomanager_test.go index 7b414223c75..8e909d7e29c 100644 --- a/pkg/volume/csi/nodeinfomanager/nodeinfomanager_test.go +++ b/pkg/volume/csi/nodeinfomanager/nodeinfomanager_test.go @@ -56,6 +56,7 @@ type testcase struct { expectedCSINode *storage.CSINode expectFail bool hasModified bool + migratedPlugins map[string](func() bool) } type nodeIDMap map[string]string @@ -307,6 +308,83 @@ func TestInstallCSIDriver(t *testing.T) { }, }, }, + { + name: "pre-existing node info, but owned by previous node", + existingNode: func() *v1.Node { + node := generateNode(nil /*nodeIDs*/, nil /*labels*/, nil /*capacity*/) + node.UID = types.UID("node1") + return node + }(), + existingCSINode: func() *storage.CSINode { + csiNode := generateCSINode(nil /*nodeIDs*/, nil /*volumeLimits*/, nil /*topologyKeys*/) + csiNode.OwnerReferences[0].UID = types.UID("node2") + return csiNode + }(), + migratedPlugins: map[string](func() bool){ + "com.example.csi.driver1": func() bool { return true }, + }, + inputNodeID: "com.example.csi/csi-node1", + expectedNode: &v1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: "node1", + UID: types.UID("node1"), + Annotations: map[string]string{annotationKeyNodeID: marshall(nodeIDMap{"": "com.example.csi/csi-node1"})}, + }, + }, + expectedCSINode: func() *storage.CSINode { + csiNode := &storage.CSINode{ + ObjectMeta: getCSINodeObjectMeta(), + Spec: storage.CSINodeSpec{ + Drivers: []storage.CSINodeDriver{ + { + NodeID: "com.example.csi/csi-node1", + }, + }, + }, + } + csiNode.Annotations = map[string]string{v1.MigratedPluginsAnnotationKey: "com.example.csi.driver1"} + return csiNode + }(), + }, + { + name: "pre-existing node info with driver, but owned by previous node", + existingNode: func() *v1.Node { + node := generateNode(nil /*nodeIDs*/, nil /*labels*/, nil /*capacity*/) + node.UID = types.UID("node1") + return node + }(), + existingCSINode: func() *storage.CSINode { + csiNode := generateCSINode( + nodeIDMap{ + "com.example.csi.old-driver": "com.example.csi/csi-node2", + }, + nil /*volumeLimits*/, nil, /*topologyKeys*/ + ) + csiNode.OwnerReferences[0].UID = types.UID("node2") + return csiNode + }(), + driverName: "com.example.csi.driver1", + inputNodeID: "com.example.csi/csi-node1", + expectedNode: &v1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: "node1", + UID: types.UID("node1"), + Annotations: map[string]string{annotationKeyNodeID: marshall(nodeIDMap{"com.example.csi.driver1": "com.example.csi/csi-node1"})}, + }, + }, + expectedCSINode: &storage.CSINode{ + ObjectMeta: getCSINodeObjectMeta(), + Spec: storage.CSINodeSpec{ + Drivers: []storage.CSINodeDriver{ + { + // Only the new driver should be present because the old CSINode represented a previous node. + Name: "com.example.csi.driver1", + NodeID: "com.example.csi/csi-node1", + }, + }, + }, + }, + }, { name: "nil topology, empty node", driverName: "com.example.csi.driver1", @@ -1029,7 +1107,7 @@ func test(t *testing.T, addNodeInfo bool, testcases []testcase) { nil, nil, ) - nim := NewNodeInfoManager(types.NodeName(nodeName), host, nil) + nim := NewNodeInfoManager(types.NodeName(nodeName), host, tc.migratedPlugins) //// Act nim.CreateCSINode()