Merge pull request #60159 from jingxu97/Feb/localephemeralBeta

pkg/features/kube_features.go

@@ -98,7 +98,7 @@ const (
 	PersistentLocalVolumes utilfeature.Feature = "PersistentLocalVolumes"
 
 	// owner: @jinxu
-	// alpha: v1.7
+	// beta: v1.10
 	//
 	// New local storage types to support local storage capacity isolation
 	LocalStorageCapacityIsolation utilfeature.Feature = "LocalStorageCapacityIsolation"
@@ -287,7 +287,7 @@ var defaultKubernetesFeatureGates = map[utilfeature.Feature]utilfeature.FeatureS
 	RotateKubeletServerCertificate:              {Default: false, PreRelease: utilfeature.Alpha},
 	RotateKubeletClientCertificate:              {Default: true, PreRelease: utilfeature.Beta},
 	PersistentLocalVolumes:                      {Default: true, PreRelease: utilfeature.Beta},
-	LocalStorageCapacityIsolation:               {Default: false, PreRelease: utilfeature.Alpha},
+	LocalStorageCapacityIsolation:               {Default: true, PreRelease: utilfeature.Beta},
 	HugePages:                                   {Default: true, PreRelease: utilfeature.Beta},
 	DebugContainers:                             {Default: false, PreRelease: utilfeature.Alpha},
 	PodShareProcessNamespace:                    {Default: false, PreRelease: utilfeature.Alpha},

pkg/kubelet/cm/BUILD

@@ -94,6 +94,7 @@ go_library(
         "//pkg/scheduler/schedulercache:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
@@ -140,7 +141,6 @@ go_library(
             "//vendor/github.com/opencontainers/runc/libcontainer/cgroups/fs:go_default_library",
             "//vendor/github.com/opencontainers/runc/libcontainer/cgroups/systemd:go_default_library",
             "//vendor/github.com/opencontainers/runc/libcontainer/configs:go_default_library",
-            "//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
             "//vendor/k8s.io/apimachinery/pkg/util/errors:go_default_library",
             "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
             "//vendor/k8s.io/client-go/tools/record:go_default_library",

pkg/kubelet/cm/container_manager_stub.go

@@ -20,6 +20,7 @@ import (
 	"github.com/golang/glog"
 	"k8s.io/api/core/v1"
 
+	"k8s.io/apimachinery/pkg/api/resource"
 	internalapi "k8s.io/kubernetes/pkg/kubelet/apis/cri"
 	"k8s.io/kubernetes/pkg/kubelet/cm/cpumanager"
 	"k8s.io/kubernetes/pkg/kubelet/config"
@@ -67,7 +68,12 @@ func (cm *containerManagerStub) GetNodeAllocatableReservation() v1.ResourceList
 }
 
 func (cm *containerManagerStub) GetCapacity() v1.ResourceList {
-	return nil
+	c := v1.ResourceList{
+		v1.ResourceEphemeralStorage: *resource.NewQuantity(
+			int64(0),
+			resource.BinarySI),
+	}
+	return c
 }
 
 func (cm *containerManagerStub) GetDevicePluginResourceCapacity() (v1.ResourceList, v1.ResourceList, []string) {

pkg/kubelet/images/BUILD

@@ -21,6 +21,7 @@ go_library(
         "//pkg/kubelet/apis/stats/v1alpha1:go_default_library",
         "//pkg/kubelet/container:go_default_library",
         "//pkg/kubelet/events:go_default_library",
+        "//pkg/kubelet/util/sliceutils:go_default_library",
         "//pkg/util/parsers:go_default_library",
         "//vendor/github.com/docker/distribution/reference:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",

pkg/kubelet/images/image_gc_manager.go

@@ -34,6 +34,7 @@ import (
 	statsapi "k8s.io/kubernetes/pkg/kubelet/apis/stats/v1alpha1"
 	"k8s.io/kubernetes/pkg/kubelet/container"
 	"k8s.io/kubernetes/pkg/kubelet/events"
+	"k8s.io/kubernetes/pkg/kubelet/util/sliceutils"
 )
 
 // StatsProvider is an interface for fetching stats used during image garbage
@@ -120,11 +121,15 @@ func (i *imageCache) set(images []container.Image) {
 	i.images = images
 }
 
-// get gets image list from image cache.
+// get gets a sorted (by image size) image list from image cache.
+// There is a potentical data race in this function. See PR #60448
+// Because there is deepcopy function available currently, move sort
+// function inside this function
 func (i *imageCache) get() []container.Image {
-	i.RLock()
-	defer i.RUnlock()
-	return append([]container.Image{}, i.images...)
+	i.Lock()
+	defer i.Unlock()
+	sort.Sort(sliceutils.ByImageSize(i.images))
+	return i.images
 }
 
 // Information about the images we track.

pkg/kubelet/kubelet.go

@@ -560,11 +560,11 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 		return nil, err
 	}
 	klet.networkPlugin = plug
-
-	machineInfo, err := klet.GetCachedMachineInfo()
+	machineInfo, err := klet.cadvisor.MachineInfo()
 	if err != nil {
 		return nil, err
 	}
+	klet.machineInfo = machineInfo
 
 	imageBackOff := flowcontrol.NewBackOff(backOffPeriod, MaxContainerBackOff)
 
@@ -1338,8 +1338,6 @@ func (kl *Kubelet) initializeRuntimeDependentModules() {
 		// TODO(random-liu): Add backoff logic in the babysitter
 		glog.Fatalf("Failed to start cAdvisor %v", err)
 	}
-	// eviction manager must start after cadvisor because it needs to know if the container runtime has a dedicated imagefs
-	kl.evictionManager.Start(kl.StatsProvider, kl.GetActivePods, kl.podResourcesAreReclaimed, evictionMonitoringPeriod)
 
 	// trigger on-demand stats collection once so that we have capacity information for ephemeral storage.
 	// ignore any errors, since if stats collection is not successful, the container manager will fail to start below.
@@ -1355,6 +1353,9 @@ func (kl *Kubelet) initializeRuntimeDependentModules() {
 		// Fail kubelet and rely on the babysitter to retry starting kubelet.
 		glog.Fatalf("Failed to start ContainerManager %v", err)
 	}
+	// eviction manager must start after cadvisor because it needs to know if the container runtime has a dedicated imagefs
+	kl.evictionManager.Start(kl.StatsProvider, kl.GetActivePods, kl.podResourcesAreReclaimed, evictionMonitoringPeriod)
+
 	// container log manager must start after container runtime is up to retrieve information from container runtime
 	// and inform container to reopen log file after log rotation.
 	kl.containerLogManager.Start()

pkg/kubelet/kubelet_getters.go

@@ -281,12 +281,5 @@ func (kl *Kubelet) GetVersionInfo() (*cadvisorapiv1.VersionInfo, error) {
 
 // GetCachedMachineInfo assumes that the machine info can't change without a reboot
 func (kl *Kubelet) GetCachedMachineInfo() (*cadvisorapiv1.MachineInfo, error) {
-	if kl.machineInfo == nil {
-		info, err := kl.cadvisor.MachineInfo()
-		if err != nil {
-			return nil, err
-		}
-		kl.machineInfo = info
-	}
 	return kl.machineInfo, nil
 }

pkg/kubelet/kubelet_node_status.go

@@ -22,8 +22,8 @@ import (
 	"math"
 	"net"
 	goruntime "runtime"
-	"sort"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/golang/glog"
@@ -42,7 +42,6 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/cadvisor"
 	"k8s.io/kubernetes/pkg/kubelet/events"
 	"k8s.io/kubernetes/pkg/kubelet/util"
-	"k8s.io/kubernetes/pkg/kubelet/util/sliceutils"
 	"k8s.io/kubernetes/pkg/scheduler/algorithm"
 	nodeutil "k8s.io/kubernetes/pkg/util/node"
 	"k8s.io/kubernetes/pkg/version"
@@ -702,7 +701,6 @@ func (kl *Kubelet) setNodeStatusImages(node *v1.Node) {
 		return
 	}
 	// sort the images from max to min, and only set top N images into the node status.
-	sort.Sort(sliceutils.ByImageSize(containerImages))
 	if maxImagesInNodeStatus < len(containerImages) {
 		containerImages = containerImages[0:maxImagesInNodeStatus]
 	}
@@ -773,7 +771,6 @@ func (kl *Kubelet) setNodeReadyCondition(node *v1.Node) {
 			LastHeartbeatTime: currentTime,
 		}
 	}
-
 	// Append AppArmor status if it's enabled.
 	// TODO(tallclair): This is a temporary message until node feature reporting is added.
 	if newNodeReadyCondition.Status == v1.ConditionTrue &&
@@ -1019,10 +1016,15 @@ func (kl *Kubelet) setNodeOODCondition(node *v1.Node) {
 
 // Maintains Node.Spec.Unschedulable value from previous run of tryUpdateNodeStatus()
 // TODO: why is this a package var?
-var oldNodeUnschedulable bool
+var (
+	oldNodeUnschedulable     bool
+	oldNodeUnschedulableLock sync.Mutex
+)
 
 // record if node schedulable change.
 func (kl *Kubelet) recordNodeSchedulableEvent(node *v1.Node) {
+	oldNodeUnschedulableLock.Lock()
+	defer oldNodeUnschedulableLock.Unlock()
 	if oldNodeUnschedulable != node.Spec.Unschedulable {
 		if node.Spec.Unschedulable {
 			kl.recordNodeStatusEvent(v1.EventTypeNormal, events.NodeNotSchedulable)

pkg/kubelet/kubelet_test.go

@@ -264,6 +264,15 @@ func newTestKubeletWithImageList(
 		ImageGCManager:   imageGCManager,
 	}
 	kubelet.containerLogManager = logs.NewStubContainerLogManager()
+	containerGCPolicy := kubecontainer.ContainerGCPolicy{
+		MinAge:             time.Duration(0),
+		MaxPerPodContainer: 1,
+		MaxContainers:      -1,
+	}
+	containerGC, err := kubecontainer.NewContainerGC(fakeRuntime, containerGCPolicy, kubelet.sourcesReady)
+	assert.NoError(t, err)
+	kubelet.containerGC = containerGC
+
 	fakeClock := clock.NewFakeClock(time.Now())
 	kubelet.backOff = flowcontrol.NewBackOff(time.Second, time.Minute)
 	kubelet.backOff.Clock = fakeClock

test/e2e/scheduling/resource_quota.go

@@ -46,83 +46,6 @@ var extendedResourceName string = "example.com/dongle"
 var _ = SIGDescribe("ResourceQuota", func() {
 	f := framework.NewDefaultFramework("resourcequota")
 
-	BeforeEach(func() {
-		// only run the tests when LocalStorageCapacityIsolation feature is enabled
-		framework.SkipUnlessLocalEphemeralStorageEnabled()
-	})
-
-	It("should create a ResourceQuota and capture the life of a pod.", func() {
-		By("Creating a ResourceQuota")
-		quotaName := "test-quota"
-		resourceQuota := newTestResourceQuotaForEphemeralStorage(quotaName)
-		resourceQuota, err := createResourceQuota(f.ClientSet, f.Namespace.Name, resourceQuota)
-		Expect(err).NotTo(HaveOccurred())
-
-		defer func() {
-			By("Removing resourceQuota")
-			err = deleteResourceQuota(f.ClientSet, f.Namespace.Name, resourceQuota.Name)
-			Expect(err).NotTo(HaveOccurred())
-		}()
-
-		By("Ensuring resource quota status is calculated")
-		usedResources := v1.ResourceList{}
-		usedResources[v1.ResourceQuotas] = resource.MustParse("1")
-		err = waitForResourceQuota(f.ClientSet, f.Namespace.Name, quotaName, usedResources)
-		Expect(err).NotTo(HaveOccurred())
-
-		By("Creating a Pod that fits quota")
-		podName := "test-pod"
-		requests := v1.ResourceList{}
-		requests[v1.ResourceEphemeralStorage] = resource.MustParse("300Mi")
-		pod := newTestPodForQuota(f, podName, requests, v1.ResourceList{})
-		pod, err = f.ClientSet.CoreV1().Pods(f.Namespace.Name).Create(pod)
-		Expect(err).NotTo(HaveOccurred())
-		podToUpdate := pod
-
-		defer func() {
-			By("Deleting the pod")
-			err = f.ClientSet.CoreV1().Pods(f.Namespace.Name).Delete(pod.Name, metav1.NewDeleteOptions(0))
-			Expect(err).NotTo(HaveOccurred())
-		}()
-
-		By("Ensuring ResourceQuota status captures the pod usage")
-		usedResources[v1.ResourceQuotas] = resource.MustParse("1")
-		usedResources[v1.ResourcePods] = resource.MustParse("1")
-		usedResources[v1.ResourceEphemeralStorage] = requests[v1.ResourceEphemeralStorage]
-		err = waitForResourceQuota(f.ClientSet, f.Namespace.Name, quotaName, usedResources)
-		Expect(err).NotTo(HaveOccurred())
-
-		By("Not allowing a pod to be created that exceeds remaining quota")
-		requests = v1.ResourceList{}
-		requests[v1.ResourceEphemeralStorage] = resource.MustParse("300Mi")
-		pod = newTestPodForQuota(f, "fail-pod", requests, v1.ResourceList{})
-		pod, err = f.ClientSet.CoreV1().Pods(f.Namespace.Name).Create(pod)
-		Expect(err).To(HaveOccurred())
-
-		By("Ensuring a pod cannot update its resource requirements")
-		// a pod cannot dynamically update its resource requirements.
-		requests = v1.ResourceList{}
-		requests[v1.ResourceEphemeralStorage] = resource.MustParse("100Mi")
-		podToUpdate.Spec.Containers[0].Resources.Requests = requests
-		_, err = f.ClientSet.CoreV1().Pods(f.Namespace.Name).Update(podToUpdate)
-		Expect(err).To(HaveOccurred())
-
-		By("Ensuring attempts to update pod resource requirements did not change quota usage")
-		err = waitForResourceQuota(f.ClientSet, f.Namespace.Name, quotaName, usedResources)
-		Expect(err).NotTo(HaveOccurred())
-
-		By("Ensuring resource quota status released the pod usage")
-		usedResources[v1.ResourceQuotas] = resource.MustParse("1")
-		usedResources[v1.ResourcePods] = resource.MustParse("0")
-		usedResources[v1.ResourceEphemeralStorage] = resource.MustParse("0")
-		err = waitForResourceQuota(f.ClientSet, f.Namespace.Name, quotaName, usedResources)
-		Expect(err).NotTo(HaveOccurred())
-	})
-})
-
-var _ = SIGDescribe("ResourceQuota", func() {
-	f := framework.NewDefaultFramework("resourcequota")
-
 	It("should create a ResourceQuota and ensure its status is promptly calculated.", func() {
 		By("Creating a ResourceQuota")
 		quotaName := "test-quota"
@@ -372,6 +295,7 @@ var _ = SIGDescribe("ResourceQuota", func() {
 		limits := v1.ResourceList{}
 		requests[v1.ResourceCPU] = resource.MustParse("500m")
 		requests[v1.ResourceMemory] = resource.MustParse("252Mi")
+		requests[v1.ResourceEphemeralStorage] = resource.MustParse("30Gi")
 		requests[v1.ResourceName(extendedResourceName)] = resource.MustParse("2")
 		limits[v1.ResourceName(extendedResourceName)] = resource.MustParse("2")
 		pod := newTestPodForQuota(f, podName, requests, limits)
@@ -384,6 +308,7 @@ var _ = SIGDescribe("ResourceQuota", func() {
 		usedResources[v1.ResourcePods] = resource.MustParse("1")
 		usedResources[v1.ResourceCPU] = requests[v1.ResourceCPU]
 		usedResources[v1.ResourceMemory] = requests[v1.ResourceMemory]
+		usedResources[v1.ResourceEphemeralStorage] = requests[v1.ResourceEphemeralStorage]
 		usedResources[v1.ResourceName(v1.DefaultResourceRequestsPrefix+extendedResourceName)] = requests[v1.ResourceName(extendedResourceName)]
 		err = waitForResourceQuota(f.ClientSet, f.Namespace.Name, quotaName, usedResources)
 		Expect(err).NotTo(HaveOccurred())
@@ -401,6 +326,7 @@ var _ = SIGDescribe("ResourceQuota", func() {
 		limits = v1.ResourceList{}
 		requests[v1.ResourceCPU] = resource.MustParse("500m")
 		requests[v1.ResourceMemory] = resource.MustParse("100Mi")
+		requests[v1.ResourceEphemeralStorage] = resource.MustParse("30Gi")
 		requests[v1.ResourceName(extendedResourceName)] = resource.MustParse("2")
 		limits[v1.ResourceName(extendedResourceName)] = resource.MustParse("2")
 		pod = newTestPodForQuota(f, "fail-pod-for-extended-resource", requests, limits)
@@ -412,6 +338,7 @@ var _ = SIGDescribe("ResourceQuota", func() {
 		requests = v1.ResourceList{}
 		requests[v1.ResourceCPU] = resource.MustParse("100m")
 		requests[v1.ResourceMemory] = resource.MustParse("100Mi")
+		requests[v1.ResourceEphemeralStorage] = resource.MustParse("10Gi")
 		podToUpdate.Spec.Containers[0].Resources.Requests = requests
 		_, err = f.ClientSet.CoreV1().Pods(f.Namespace.Name).Update(podToUpdate)
 		Expect(err).To(HaveOccurred())
@@ -429,6 +356,7 @@ var _ = SIGDescribe("ResourceQuota", func() {
 		usedResources[v1.ResourcePods] = resource.MustParse("0")
 		usedResources[v1.ResourceCPU] = resource.MustParse("0")
 		usedResources[v1.ResourceMemory] = resource.MustParse("0")
+		usedResources[v1.ResourceEphemeralStorage] = resource.MustParse("0")
 		usedResources[v1.ResourceName(v1.DefaultResourceRequestsPrefix+extendedResourceName)] = resource.MustParse("0")
 		err = waitForResourceQuota(f.ClientSet, f.Namespace.Name, quotaName, usedResources)
 		Expect(err).NotTo(HaveOccurred())
@@ -821,10 +749,11 @@ func newTestResourceQuotaWithScope(name string, scope v1.ResourceQuotaScope) *v1
 	}
 }
 
-// newTestResourceQuotaForEphemeralStorage returns a quota that enforces default constraints for testing alpha feature LocalStorageCapacityIsolation
+// newTestResourceQuotaForEphemeralStorage returns a quota that enforces default constraints for testing feature LocalStorageCapacityIsolation
 func newTestResourceQuotaForEphemeralStorage(name string) *v1.ResourceQuota {
 	hard := v1.ResourceList{}
 	hard[v1.ResourceEphemeralStorage] = resource.MustParse("500Mi")
+	hard[v1.ResourceQuotas] = resource.MustParse("1")
 	return &v1.ResourceQuota{
 		ObjectMeta: metav1.ObjectMeta{Name: name},
 		Spec:       v1.ResourceQuotaSpec{Hard: hard},
@@ -846,6 +775,7 @@ func newTestResourceQuota(name string) *v1.ResourceQuota {
 	hard[v1.ResourceSecrets] = resource.MustParse("10")
 	hard[v1.ResourcePersistentVolumeClaims] = resource.MustParse("10")
 	hard[v1.ResourceRequestsStorage] = resource.MustParse("10Gi")
+	hard[v1.ResourceEphemeralStorage] = resource.MustParse("50Gi")
 	hard[core.V1ResourceByStorageClass(classGold, v1.ResourcePersistentVolumeClaims)] = resource.MustParse("10")
 	hard[core.V1ResourceByStorageClass(classGold, v1.ResourceRequestsStorage)] = resource.MustParse("10Gi")
 	// test quota on discovered resource type