Merge pull request #42254 from justinsb/volumes_dont_leak_nodestatusupdateneeded

pkg/controller/volume/attachdetach/statusupdater/BUILD

@@ -22,6 +22,7 @@ go_library(
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/strategicpatch:go_default_library",
     ],
 )

pkg/controller/volume/attachdetach/statusupdater/node_status_updater.go

@@ -26,6 +26,7 @@ import (
 
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
@@ -83,65 +84,79 @@ func (nsu *nodeStatusUpdater) UpdateNodeStatuses() error {
 			continue
 		}
 
-		clonedNode, err := api.Scheme.DeepCopy(nodeObj)
-		if err != nil {
-			return fmt.Errorf("error cloning node %q: %v",
+		if err := nsu.updateNodeStatus(nodeName, nodeObj, attachedVolumes); err != nil {
+			// If update node status fails, reset flag statusUpdateNeeded back to true
+			// to indicate this node status needs to be updated again
+			nsu.actualStateOfWorld.SetNodeStatusUpdateNeeded(nodeName)
+
+			glog.V(2).Infof(
+				"Could not update node status for %q; re-marking for update. %v",
 				nodeName,
 				err)
-		}
 
-		node, ok := clonedNode.(*v1.Node)
-		if !ok || node == nil {
-			return fmt.Errorf(
-				"failed to cast %q object %#v to Node",
-				nodeName,
-				clonedNode)
+			// We currently always return immediately on error
+			return err
 		}
+	}
+	return nil
+}
 
-		// TODO: Change to pkg/util/node.UpdateNodeStatus.
-		oldData, err := json.Marshal(node)
-		if err != nil {
-			return fmt.Errorf(
-				"failed to Marshal oldData for node %q. %v",
-				nodeName,
-				err)
-		}
+func (nsu *nodeStatusUpdater) updateNodeStatus(nodeName types.NodeName, nodeObj *v1.Node, attachedVolumes []v1.AttachedVolume) error {
+	clonedNode, err := api.Scheme.DeepCopy(nodeObj)
+	if err != nil {
+		return fmt.Errorf("error cloning node %q: %v",
+			nodeName,
+			err)
+	}
 
-		node.Status.VolumesAttached = attachedVolumes
+	node, ok := clonedNode.(*v1.Node)
+	if !ok || node == nil {
+		return fmt.Errorf(
+			"failed to cast %q object %#v to Node",
+			nodeName,
+			clonedNode)
+	}
 
-		newData, err := json.Marshal(node)
-		if err != nil {
-			return fmt.Errorf(
-				"failed to Marshal newData for node %q. %v",
-				nodeName,
-				err)
-		}
+	// TODO: Change to pkg/util/node.UpdateNodeStatus.
+	oldData, err := json.Marshal(node)
+	if err != nil {
+		return fmt.Errorf(
+			"failed to Marshal oldData for node %q. %v",
+			nodeName,
+			err)
+	}
 
-		patchBytes, err :=
-			strategicpatch.CreateTwoWayMergePatch(oldData, newData, node)
-		if err != nil {
-			return fmt.Errorf(
-				"failed to CreateTwoWayMergePatch for node %q. %v",
-				nodeName,
-				err)
-		}
+	node.Status.VolumesAttached = attachedVolumes
 
-		_, err = nsu.kubeClient.Core().Nodes().PatchStatus(string(nodeName), patchBytes)
-		if err != nil {
-			// If update node status fails, reset flag statusUpdateNeeded back to true
-			// to indicate this node status needs to be updated again
-			nsu.actualStateOfWorld.SetNodeStatusUpdateNeeded(nodeName)
-			return fmt.Errorf(
-				"failed to kubeClient.Core().Nodes().Patch for node %q. %v",
-				nodeName,
-				err)
-		}
-		glog.V(2).Infof(
-			"Updating status for node %q succeeded. patchBytes: %q VolumesAttached: %v",
+	newData, err := json.Marshal(node)
+	if err != nil {
+		return fmt.Errorf(
+			"failed to Marshal newData for node %q. %v",
 			nodeName,
-			string(patchBytes),
-			node.Status.VolumesAttached)
+			err)
+	}
 
+	patchBytes, err :=
+		strategicpatch.CreateTwoWayMergePatch(oldData, newData, node)
+	if err != nil {
+		return fmt.Errorf(
+			"failed to CreateTwoWayMergePatch for node %q. %v",
+			nodeName,
+			err)
 	}
+
+	_, err = nsu.kubeClient.Core().Nodes().PatchStatus(string(nodeName), patchBytes)
+	if err != nil {
+		return fmt.Errorf(
+			"failed to kubeClient.Core().Nodes().Patch for node %q. %v",
+			nodeName,
+			err)
+	}
+	glog.V(4).Infof(
+		"Updating status for node %q succeeded. patchBytes: %q VolumesAttached: %v",
+		nodeName,
+		string(patchBytes),
+		node.Status.VolumesAttached)
+
 	return nil
 }