port setNodeStatusMachineInfo to Setter abstraction, add test

pkg/kubelet/kubelet_node_status.go

@@ -19,7 +19,6 @@ package kubelet
 import (
 	"context"
 	"fmt"
-	"math"
 	"net"
 	goruntime "runtime"
 	"time"
@@ -36,7 +35,6 @@ import (
 	"k8s.io/kubernetes/pkg/cloudprovider"
 	"k8s.io/kubernetes/pkg/features"
 	kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
-	"k8s.io/kubernetes/pkg/kubelet/cadvisor"
 	"k8s.io/kubernetes/pkg/kubelet/events"
 	"k8s.io/kubernetes/pkg/kubelet/nodestatus"
 	"k8s.io/kubernetes/pkg/kubelet/util"
@@ -442,131 +440,9 @@ func (kl *Kubelet) recordNodeStatusEvent(eventType, event string) {
 	kl.recorder.Eventf(kl.nodeRef, eventType, event, "Node %s status is now: %s", kl.nodeName, event)
 }
 
-func (kl *Kubelet) setNodeStatusMachineInfo(node *v1.Node) {
-	// Note: avoid blindly overwriting the capacity in case opaque
-	//       resources are being advertised.
-	if node.Status.Capacity == nil {
-		node.Status.Capacity = v1.ResourceList{}
-	}
-
-	var devicePluginAllocatable v1.ResourceList
-	var devicePluginCapacity v1.ResourceList
-	var removedDevicePlugins []string
-
-	// TODO: Post NotReady if we cannot get MachineInfo from cAdvisor. This needs to start
-	// cAdvisor locally, e.g. for test-cmd.sh, and in integration test.
-	info, err := kl.GetCachedMachineInfo()
-	if err != nil {
-		// TODO(roberthbailey): This is required for test-cmd.sh to pass.
-		// See if the test should be updated instead.
-		node.Status.Capacity[v1.ResourceCPU] = *resource.NewMilliQuantity(0, resource.DecimalSI)
-		node.Status.Capacity[v1.ResourceMemory] = resource.MustParse("0Gi")
-		node.Status.Capacity[v1.ResourcePods] = *resource.NewQuantity(int64(kl.maxPods), resource.DecimalSI)
-		glog.Errorf("Error getting machine info: %v", err)
-	} else {
-		node.Status.NodeInfo.MachineID = info.MachineID
-		node.Status.NodeInfo.SystemUUID = info.SystemUUID
-
-		for rName, rCap := range cadvisor.CapacityFromMachineInfo(info) {
-			node.Status.Capacity[rName] = rCap
-		}
-
-		if kl.podsPerCore > 0 {
-			node.Status.Capacity[v1.ResourcePods] = *resource.NewQuantity(
-				int64(math.Min(float64(info.NumCores*kl.podsPerCore), float64(kl.maxPods))), resource.DecimalSI)
-		} else {
-			node.Status.Capacity[v1.ResourcePods] = *resource.NewQuantity(
-				int64(kl.maxPods), resource.DecimalSI)
-		}
-
-		if node.Status.NodeInfo.BootID != "" &&
-			node.Status.NodeInfo.BootID != info.BootID {
-			// TODO: This requires a transaction, either both node status is updated
-			// and event is recorded or neither should happen, see issue #6055.
-			kl.recorder.Eventf(kl.nodeRef, v1.EventTypeWarning, events.NodeRebooted,
-				"Node %s has been rebooted, boot id: %s", kl.nodeName, info.BootID)
-		}
-		node.Status.NodeInfo.BootID = info.BootID
-
-		if utilfeature.DefaultFeatureGate.Enabled(features.LocalStorageCapacityIsolation) {
-			// TODO: all the node resources should use GetCapacity instead of deriving the
-			// capacity for every node status request
-			initialCapacity := kl.containerManager.GetCapacity()
-			if initialCapacity != nil {
-				node.Status.Capacity[v1.ResourceEphemeralStorage] = initialCapacity[v1.ResourceEphemeralStorage]
-			}
-		}
-
-		devicePluginCapacity, devicePluginAllocatable, removedDevicePlugins = kl.containerManager.GetDevicePluginResourceCapacity()
-		if devicePluginCapacity != nil {
-			for k, v := range devicePluginCapacity {
-				if old, ok := node.Status.Capacity[k]; !ok || old.Value() != v.Value() {
-					glog.V(2).Infof("Update capacity for %s to %d", k, v.Value())
-				}
-				node.Status.Capacity[k] = v
-			}
-		}
-
-		for _, removedResource := range removedDevicePlugins {
-			glog.V(2).Infof("Set capacity for %s to 0 on device removal", removedResource)
-			// Set the capacity of the removed resource to 0 instead of
-			// removing the resource from the node status. This is to indicate
-			// that the resource is managed by device plugin and had been
-			// registered before.
-			//
-			// This is required to differentiate the device plugin managed
-			// resources and the cluster-level resources, which are absent in
-			// node status.
-			node.Status.Capacity[v1.ResourceName(removedResource)] = *resource.NewQuantity(int64(0), resource.DecimalSI)
-		}
-	}
-
-	// Set Allocatable.
-	if node.Status.Allocatable == nil {
-		node.Status.Allocatable = make(v1.ResourceList)
-	}
-	// Remove extended resources from allocatable that are no longer
-	// present in capacity.
-	for k := range node.Status.Allocatable {
-		_, found := node.Status.Capacity[k]
-		if !found && v1helper.IsExtendedResourceName(k) {
-			delete(node.Status.Allocatable, k)
-		}
-	}
-	allocatableReservation := kl.containerManager.GetNodeAllocatableReservation()
-	for k, v := range node.Status.Capacity {
-		value := *(v.Copy())
-		if res, exists := allocatableReservation[k]; exists {
-			value.Sub(res)
-		}
-		if value.Sign() < 0 {
-			// Negative Allocatable resources don't make sense.
-			value.Set(0)
-		}
-		node.Status.Allocatable[k] = value
-	}
-
-	if devicePluginAllocatable != nil {
-		for k, v := range devicePluginAllocatable {
-			if old, ok := node.Status.Allocatable[k]; !ok || old.Value() != v.Value() {
-				glog.V(2).Infof("Update allocatable for %s to %d", k, v.Value())
-			}
-			node.Status.Allocatable[k] = v
-		}
-	}
-	// for every huge page reservation, we need to remove it from allocatable memory
-	for k, v := range node.Status.Capacity {
-		if v1helper.IsHugePageResourceName(k) {
-			allocatableMemory := node.Status.Allocatable[v1.ResourceMemory]
-			value := *(v.Copy())
-			allocatableMemory.Sub(value)
-			if allocatableMemory.Sign() < 0 {
-				// Negative Allocatable resources don't make sense.
-				allocatableMemory.Set(0)
-			}
-			node.Status.Allocatable[v1.ResourceMemory] = allocatableMemory
-		}
-	}
+// recordEvent records an event for this node, the Kubelet's nodeRef is passed to the recorder
+func (kl *Kubelet) recordEvent(eventType, event, message string) {
+	kl.recorder.Eventf(kl.nodeRef, eventType, event, message)
 }
 
 // Set versioninfo for the node.
@@ -693,7 +569,8 @@ func (kl *Kubelet) defaultNodeStatusFuncs() []func(*v1.Node) error {
 	var setters []func(n *v1.Node) error
 	setters = append(setters,
 		nodestatus.NodeAddress(kl.nodeIP, kl.nodeIPValidator, kl.hostname, kl.externalCloudProvider, kl.cloud, nodeAddressesFunc),
-		withoutError(kl.setNodeStatusMachineInfo),
+		nodestatus.MachineInfo(string(kl.nodeName), kl.maxPods, kl.podsPerCore, kl.GetCachedMachineInfo, kl.containerManager.GetCapacity,
+			kl.containerManager.GetDevicePluginResourceCapacity, kl.containerManager.GetNodeAllocatableReservation, kl.recordEvent),
 		withoutError(kl.setNodeStatusVersionInfo),
 		withoutError(kl.setNodeStatusDaemonEndpoints),
 		withoutError(kl.setNodeStatusImages),

pkg/kubelet/kubelet_node_status_test.go

@@ -196,6 +196,10 @@ func TestUpdateNewNodeStatus(t *testing.T) {
 					v1.ResourceEphemeralStorage: *resource.NewQuantity(5000, resource.BinarySI),
 				},
 			}
+			// Since this test retroactively overrides the stub container manager,
+			// we have to regenerate default status setters.
+			kubelet.setNodeStatusFuncs = kubelet.defaultNodeStatusFuncs()
+
 			kubeClient := testKubelet.fakeKubeClient
 			existingNode := v1.Node{ObjectMeta: metav1.ObjectMeta{Name: testKubeletHostname}}
 			kubeClient.ReactionChain = fake.NewSimpleClientset(&v1.NodeList{Items: []v1.Node{existingNode}}).ReactionChain
@@ -329,6 +333,9 @@ func TestUpdateExistingNodeStatus(t *testing.T) {
 			v1.ResourceEphemeralStorage: *resource.NewQuantity(5000, resource.BinarySI),
 		},
 	}
+	// Since this test retroactively overrides the stub container manager,
+	// we have to regenerate default status setters.
+	kubelet.setNodeStatusFuncs = kubelet.defaultNodeStatusFuncs()
 
 	kubeClient := testKubelet.fakeKubeClient
 	existingNode := v1.Node{
@@ -595,6 +602,9 @@ func TestUpdateNodeStatusWithRuntimeStateError(t *testing.T) {
 			v1.ResourceEphemeralStorage: *resource.NewQuantity(20E9, resource.BinarySI),
 		},
 	}
+	// Since this test retroactively overrides the stub container manager,
+	// we have to regenerate default status setters.
+	kubelet.setNodeStatusFuncs = kubelet.defaultNodeStatusFuncs()
 
 	clock := testKubelet.fakeClock
 	kubeClient := testKubelet.fakeKubeClient
@@ -1036,6 +1046,10 @@ func TestUpdateNewNodeStatusTooLargeReservation(t *testing.T) {
 			v1.ResourceEphemeralStorage: *resource.NewQuantity(3000, resource.BinarySI),
 		},
 	}
+	// Since this test retroactively overrides the stub container manager,
+	// we have to regenerate default status setters.
+	kubelet.setNodeStatusFuncs = kubelet.defaultNodeStatusFuncs()
+
 	kubeClient := testKubelet.fakeKubeClient
 	existingNode := v1.Node{ObjectMeta: metav1.ObjectMeta{Name: testKubeletHostname}}
 	kubeClient.ReactionChain = fake.NewSimpleClientset(&v1.NodeList{Items: []v1.Node{existingNode}}).ReactionChain

pkg/kubelet/nodestatus/BUILD

@@ -6,16 +6,20 @@ go_library(
     importpath = "k8s.io/kubernetes/pkg/kubelet/nodestatus",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/apis/core/v1/helper:go_default_library",
         "//pkg/cloudprovider:go_default_library",
         "//pkg/features:go_default_library",
         "//pkg/kubelet/apis:go_default_library",
+        "//pkg/kubelet/cadvisor:go_default_library",
         "//pkg/kubelet/cm:go_default_library",
         "//pkg/kubelet/events:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/net:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/github.com/google/cadvisor/info/v1:go_default_library",
     ],
 )
 
@@ -46,6 +50,7 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/diff:go_default_library",
+        "//vendor/github.com/google/cadvisor/info/v1:go_default_library",
         "//vendor/github.com/stretchr/testify/assert:go_default_library",
         "//vendor/github.com/stretchr/testify/require:go_default_library",
     ],

pkg/kubelet/nodestatus/setters.go

@@ -18,17 +18,23 @@ package nodestatus
 
 import (
 	"fmt"
+	"math"
 	"net"
 	"strings"
 	"time"
 
+	cadvisorapiv1 "github.com/google/cadvisor/info/v1"
+
 	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilnet "k8s.io/apimachinery/pkg/util/net"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	v1helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
 	"k8s.io/kubernetes/pkg/cloudprovider"
 	"k8s.io/kubernetes/pkg/features"
 	kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
+	"k8s.io/kubernetes/pkg/kubelet/cadvisor"
 	"k8s.io/kubernetes/pkg/kubelet/cm"
 	"k8s.io/kubernetes/pkg/kubelet/events"
 
@@ -147,6 +153,145 @@ func NodeAddress(nodeIP net.IP, // typically Kubelet.nodeIP
 	}
 }
 
+// MachineInfo returns a Setter that updates machine-related information on the node.
+func MachineInfo(nodeName string,
+	maxPods int,
+	podsPerCore int,
+	machineInfoFunc func() (*cadvisorapiv1.MachineInfo, error), // typically Kubelet.GetCachedMachineInfo
+	capacityFunc func() v1.ResourceList, // typically Kubelet.containerManager.GetCapacity
+	devicePluginResourceCapacityFunc func() (v1.ResourceList, v1.ResourceList, []string), // typically Kubelet.containerManager.GetDevicePluginResourceCapacity
+	nodeAllocatableReservationFunc func() v1.ResourceList, // typically Kubelet.containerManager.GetNodeAllocatableReservation
+	recordEventFunc func(eventType, event, message string), // typically Kubelet.recordEvent
+) Setter {
+	return func(node *v1.Node) error {
+		// Note: avoid blindly overwriting the capacity in case opaque
+		//       resources are being advertised.
+		if node.Status.Capacity == nil {
+			node.Status.Capacity = v1.ResourceList{}
+		}
+
+		var devicePluginAllocatable v1.ResourceList
+		var devicePluginCapacity v1.ResourceList
+		var removedDevicePlugins []string
+
+		// TODO: Post NotReady if we cannot get MachineInfo from cAdvisor. This needs to start
+		// cAdvisor locally, e.g. for test-cmd.sh, and in integration test.
+		info, err := machineInfoFunc()
+		if err != nil {
+			// TODO(roberthbailey): This is required for test-cmd.sh to pass.
+			// See if the test should be updated instead.
+			node.Status.Capacity[v1.ResourceCPU] = *resource.NewMilliQuantity(0, resource.DecimalSI)
+			node.Status.Capacity[v1.ResourceMemory] = resource.MustParse("0Gi")
+			node.Status.Capacity[v1.ResourcePods] = *resource.NewQuantity(int64(maxPods), resource.DecimalSI)
+			glog.Errorf("Error getting machine info: %v", err)
+		} else {
+			node.Status.NodeInfo.MachineID = info.MachineID
+			node.Status.NodeInfo.SystemUUID = info.SystemUUID
+
+			for rName, rCap := range cadvisor.CapacityFromMachineInfo(info) {
+				node.Status.Capacity[rName] = rCap
+			}
+
+			if podsPerCore > 0 {
+				node.Status.Capacity[v1.ResourcePods] = *resource.NewQuantity(
+					int64(math.Min(float64(info.NumCores*podsPerCore), float64(maxPods))), resource.DecimalSI)
+			} else {
+				node.Status.Capacity[v1.ResourcePods] = *resource.NewQuantity(
+					int64(maxPods), resource.DecimalSI)
+			}
+
+			if node.Status.NodeInfo.BootID != "" &&
+				node.Status.NodeInfo.BootID != info.BootID {
+				// TODO: This requires a transaction, either both node status is updated
+				// and event is recorded or neither should happen, see issue #6055.
+				recordEventFunc(v1.EventTypeWarning, events.NodeRebooted,
+					fmt.Sprintf("Node %s has been rebooted, boot id: %s", nodeName, info.BootID))
+			}
+			node.Status.NodeInfo.BootID = info.BootID
+
+			if utilfeature.DefaultFeatureGate.Enabled(features.LocalStorageCapacityIsolation) {
+				// TODO: all the node resources should use ContainerManager.GetCapacity instead of deriving the
+				// capacity for every node status request
+				initialCapacity := capacityFunc()
+				if initialCapacity != nil {
+					node.Status.Capacity[v1.ResourceEphemeralStorage] = initialCapacity[v1.ResourceEphemeralStorage]
+				}
+			}
+
+			devicePluginCapacity, devicePluginAllocatable, removedDevicePlugins = devicePluginResourceCapacityFunc()
+			if devicePluginCapacity != nil {
+				for k, v := range devicePluginCapacity {
+					if old, ok := node.Status.Capacity[k]; !ok || old.Value() != v.Value() {
+						glog.V(2).Infof("Update capacity for %s to %d", k, v.Value())
+					}
+					node.Status.Capacity[k] = v
+				}
+			}
+
+			for _, removedResource := range removedDevicePlugins {
+				glog.V(2).Infof("Set capacity for %s to 0 on device removal", removedResource)
+				// Set the capacity of the removed resource to 0 instead of
+				// removing the resource from the node status. This is to indicate
+				// that the resource is managed by device plugin and had been
+				// registered before.
+				//
+				// This is required to differentiate the device plugin managed
+				// resources and the cluster-level resources, which are absent in
+				// node status.
+				node.Status.Capacity[v1.ResourceName(removedResource)] = *resource.NewQuantity(int64(0), resource.DecimalSI)
+			}
+		}
+
+		// Set Allocatable.
+		if node.Status.Allocatable == nil {
+			node.Status.Allocatable = make(v1.ResourceList)
+		}
+		// Remove extended resources from allocatable that are no longer
+		// present in capacity.
+		for k := range node.Status.Allocatable {
+			_, found := node.Status.Capacity[k]
+			if !found && v1helper.IsExtendedResourceName(k) {
+				delete(node.Status.Allocatable, k)
+			}
+		}
+		allocatableReservation := nodeAllocatableReservationFunc()
+		for k, v := range node.Status.Capacity {
+			value := *(v.Copy())
+			if res, exists := allocatableReservation[k]; exists {
+				value.Sub(res)
+			}
+			if value.Sign() < 0 {
+				// Negative Allocatable resources don't make sense.
+				value.Set(0)
+			}
+			node.Status.Allocatable[k] = value
+		}
+
+		if devicePluginAllocatable != nil {
+			for k, v := range devicePluginAllocatable {
+				if old, ok := node.Status.Allocatable[k]; !ok || old.Value() != v.Value() {
+					glog.V(2).Infof("Update allocatable for %s to %d", k, v.Value())
+				}
+				node.Status.Allocatable[k] = v
+			}
+		}
+		// for every huge page reservation, we need to remove it from allocatable memory
+		for k, v := range node.Status.Capacity {
+			if v1helper.IsHugePageResourceName(k) {
+				allocatableMemory := node.Status.Allocatable[v1.ResourceMemory]
+				value := *(v.Copy())
+				allocatableMemory.Sub(value)
+				if allocatableMemory.Sign() < 0 {
+					// Negative Allocatable resources don't make sense.
+					allocatableMemory.Set(0)
+				}
+				node.Status.Allocatable[v1.ResourceMemory] = allocatableMemory
+			}
+		}
+		return nil
+	}
+}
+
 // ReadyCondition returns a Setter that updates the v1.NodeReady condition on the node.
 func ReadyCondition(
 	nowFunc func() time.Time, // typically Kubelet.clock.Now