github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-17 11:28:58 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #131098 from DataDog/fricounet/upstream/race-csinode-creation

Browse Source 
Wait for CSINode removal to register CSI plugin
This commit is contained in:
Kubernetes Prow Robot
2025-10-23 11:35:35 -07:00
committed by GitHub
parent ab34215bab 28507d0e92
commit 7498131d5e
2 changed files with 144 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
pkg/volume/csi/nodeinfomanager/nodeinfomanager.go
View File

@@ -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

80
pkg/volume/csi/nodeinfomanager/nodeinfomanager_test.go
View File

@@ -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()