wip

pkg/volume/csi/nodeinfomanager/nodeinfomanager.go

@@ -74,7 +74,7 @@ type nodeUpdateFunc func(*v1.Node) (newNode *v1.Node, updated bool, err error)
 
 // Interface implements an interface for managing labels of a node
 type Interface interface {
-	CreateCSINode() (*storagev1.CSINode, error)
+	CreateCSINode(node *v1.Node) (*storagev1.CSINode, error)
 
 	// Updates or Creates the CSINode object with annotations for CSI Migration
 	InitializeCSINodeWithAnnotation() error
@@ -378,17 +378,36 @@ func (nim *nodeInfoManager) tryUpdateCSINode(
 	maxAttachLimit int64,
 	topology map[string]string) error {
 
+	node, err := csiKubeClient.CoreV1().Nodes().Get(context.TODO(), string(nim.nodeName), metav1.GetOptions{})
+	if err != nil {
+		return err
+	}
+
 	nodeInfo, err := csiKubeClient.StorageV1().CSINodes().Get(context.TODO(), string(nim.nodeName), metav1.GetOptions{})
 	if nodeInfo == nil || errors.IsNotFound(err) {
-		nodeInfo, err = nim.CreateCSINode()
+		nodeInfo, err = nim.CreateCSINode(node)
 	}
+
 	if err != nil {
 		return err
 	}
 
+	if !nodeOwnsCSINode(node, nodeInfo) {
+		return fmt.Errorf("CSINode %q is owned by different node", nodeInfo.Name)
+	}
+
 	return nim.installDriverToCSINode(nodeInfo, driverName, driverNodeID, maxAttachLimit, topology)
 }
 
+func nodeOwnsCSINode(node *v1.Node, nodeInfo *storagev1.CSINode) bool {
+	for _, ownerRef := range nodeInfo.OwnerReferences {
+		if ownerRef.Kind == nodeKind.Kind && ownerRef.Name == node.Name && ownerRef.UID == node.UID {
+			return true
+		}
+	}
+	return false
+}
+
 func (nim *nodeInfoManager) InitializeCSINodeWithAnnotation() error {
 	csiKubeClient := nim.volumeHost.GetKubeClient()
 	if csiKubeClient == nil {
@@ -411,15 +430,24 @@ func (nim *nodeInfoManager) InitializeCSINodeWithAnnotation() error {
 }
 
 func (nim *nodeInfoManager) tryInitializeCSINodeWithAnnotation(csiKubeClient clientset.Interface) error {
+	node, err := csiKubeClient.CoreV1().Nodes().Get(context.TODO(), string(nim.nodeName), metav1.GetOptions{})
+	if err != nil {
+		return err
+	}
+
 	nodeInfo, err := csiKubeClient.StorageV1().CSINodes().Get(context.TODO(), string(nim.nodeName), metav1.GetOptions{})
 	if nodeInfo == nil || errors.IsNotFound(err) {
 		// CreateCSINode will set the annotation
-		_, err = nim.CreateCSINode()
+		_, err = nim.CreateCSINode(node)
 		return err
 	} else if err != nil {
 		return err
 	}
 
+	if !nodeOwnsCSINode(node, nodeInfo) {
+		return fmt.Errorf("CSINode %q is owned by different node", nodeInfo.Name)
+	}
+
 	annotationModified := setMigrationAnnotation(nim.migratedPlugins, nodeInfo)
 
 	if annotationModified {
@@ -430,18 +458,13 @@ func (nim *nodeInfoManager) tryInitializeCSINodeWithAnnotation(csiKubeClient cli
 
 }
 
-func (nim *nodeInfoManager) CreateCSINode() (*storagev1.CSINode, error) {
+func (nim *nodeInfoManager) CreateCSINode(node *v1.Node) (*storagev1.CSINode, error) {
 
 	csiKubeClient := nim.volumeHost.GetKubeClient()
 	if csiKubeClient == nil {
 		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),

pkg/volume/csi/nodeinfomanager/nodeinfomanager_test.go

@@ -307,6 +307,21 @@ 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
+			}(),
+			inputNodeID: "com.example.csi/csi-node1",
+			expectFail:  true,
+		},
 		{
 			name:          "nil topology, empty node",
 			driverName:    "com.example.csi.driver1",
@@ -972,7 +987,7 @@ func TestInstallCSIDriverExistingAnnotation(t *testing.T) {
 		nim := NewNodeInfoManager(types.NodeName(nodeName), host, nil)
 
 		// Act
-		_, err = nim.CreateCSINode()
+		_, err = nim.CreateCSINode(tc.existingNode)
 		if err != nil {
 			t.Errorf("expected no error from creating CSINodeinfo but got: %v", err)
 			continue
@@ -1032,7 +1047,7 @@ func test(t *testing.T, addNodeInfo bool, testcases []testcase) {
 		nim := NewNodeInfoManager(types.NodeName(nodeName), host, nil)
 
 		//// Act
-		nim.CreateCSINode()
+		nim.CreateCSINode(tc.existingNode)
 		if addNodeInfo {
 			err = nim.InstallCSIDriver(tc.driverName, tc.inputNodeID, tc.inputVolumeLimit, tc.inputTopology)
 		} else {