set AllocatedResourcesStatus in the Pod Status

Author: SergeyKanzhelev

api/openapi-spec/swagger.json

@@ -6817,6 +6817,7 @@
         "properties": {
           "name": {
             "default": "",
+            "description": "Name of the resource. Must be unique within the pod and match one of the resources from the pod spec.",
             "type": "string"
           },
           "resources": {

api/openapi-spec/v3/api__v1_openapi.json

@@ -1172,22 +1172,6 @@ func rroInUse(podSpec *api.PodSpec) bool {
 	return inUse
 }
 
-func allocatedResourcesStatusInUse(podSpec *api.PodStatus) bool {
-	if podSpec == nil {
-		return false
-	}
-	inUse := func(csl []api.ContainerStatus) bool {
-		for _, cs := range csl {
-			if len(cs.AllocatedResourcesStatus) > 0 {
-				return true
-			}
-		}
-		return false
-	}
-
-	return inUse(podSpec.ContainerStatuses) || inUse(podSpec.InitContainerStatuses) || inUse(podSpec.EphemeralContainerStatuses)
-}
-
 func dropDisabledClusterTrustBundleProjection(podSpec, oldPodSpec *api.PodSpec) {
 	if utilfeature.DefaultFeatureGate.Enabled(features.ClusterTrustBundleProjection) {
 		return

pkg/api/pod/util.go

@@ -8209,11 +8209,10 @@ func validateContainerStatusNoAllocatedResourcesStatus(containerStatuses []core.
 	allErrors := field.ErrorList{}
 
 	for i, containerStatus := range containerStatuses {
-		if containerStatus.AllocatedResourcesStatus == nil {
+		if len(containerStatus.AllocatedResourcesStatus) == 0 {
 			continue
-		} else {
-			allErrors = append(allErrors, field.Forbidden(fldPath.Index(i).Child("allocatedResourcesStatus"), "cannot be set for a container status"))
 		}
+		allErrors = append(allErrors, field.Forbidden(fldPath.Index(i).Child("allocatedResourcesStatus"), "must not be specified in container status"))
 	}
 
 	return allErrors
@@ -8263,12 +8262,18 @@ func validateContainerStatusAllocatedResourcesStatus(containerStatuses []core.Co
 			uniqueResources := sets.New[core.ResourceID]()
 			// check resource IDs are unique
 			for k, r := range allocatedResource.Resources {
-				if r.Health != core.ResourceHealthStatusHealthy && r.Health != core.ResourceHealthStatusUnhealthy && r.Health != core.ResourceHealthStatusUnknown {
-					allErrors = append(allErrors, field.Invalid(fldPath.Index(i).Child("allocatedResourcesStatus").Index(j).Child("resources").Index(k).Child("health"), r.Health, "must be one of Healthy, Unhealthy, Unknown"))
+
+				var supportedResourceHealthValues = sets.New(
+					core.ResourceHealthStatusHealthy,
+					core.ResourceHealthStatusUnhealthy,
+					core.ResourceHealthStatusUnknown)
+
+				if !supportedResourceHealthValues.Has(r.Health) {
+					allErrors = append(allErrors, field.NotSupported(fldPath.Index(i).Child("allocatedResourcesStatus").Index(j).Child("resources").Index(k).Child("health"), r.Health, sets.List(supportedResourceHealthValues)))
 				}
 
 				if uniqueResources.Has(r.ResourceID) {
-					allErrors = append(allErrors, field.Invalid(fldPath.Index(i).Child("allocatedResourcesStatus").Index(j).Child("resources").Index(k).Child("resourceID"), r.ResourceID, "must be unique"))
+					allErrors = append(allErrors, field.Duplicate(fldPath.Index(i).Child("allocatedResourcesStatus").Index(j).Child("resources").Index(k).Child("resourceID"), r.ResourceID))
 				} else {
 					uniqueResources.Insert(r.ResourceID)
 				}

pkg/apis/core/validation/validation.go

@@ -24457,11 +24457,11 @@ func TestValidateContainerStatusNoAllocatedResourcesStatus(t *testing.T) {
 
 	errs := validateContainerStatusNoAllocatedResourcesStatus(containerStatuses, fldPath)
 
-	assert.Equal(t, 2, len(errs))
+	assert.Len(t, errs, 2)
 	assert.Equal(t, "spec.containers[1].allocatedResourcesStatus", errs[0].Field)
-	assert.Equal(t, "cannot be set for a container status", errs[0].Detail)
+	assert.Equal(t, "must not be specified in container status", errs[0].Detail)
 	assert.Equal(t, "spec.containers[2].allocatedResourcesStatus", errs[1].Field)
-	assert.Equal(t, "cannot be set for a container status", errs[1].Detail)
+	assert.Equal(t, "must not be specified in container status", errs[1].Detail)
 }
 
 func TestValidateContainerStatusAllocatedResourcesStatus(t *testing.T) {
@@ -24580,7 +24580,7 @@ func TestValidateContainerStatusAllocatedResourcesStatus(t *testing.T) {
 				},
 			},
 			wantFieldErrors: field.ErrorList{
-				field.Invalid(fldPath.Index(0).Child("allocatedResourcesStatus").Index(0).Child("resources").Index(1).Child("resourceID"), core.ResourceID("resource-1"), "must be unique"),
+				field.Duplicate(fldPath.Index(0).Child("allocatedResourcesStatus").Index(0).Child("resources").Index(1).Child("resourceID"), core.ResourceID("resource-1")),
 			},
 		},
 
@@ -24619,6 +24619,38 @@ func TestValidateContainerStatusAllocatedResourcesStatus(t *testing.T) {
 				field.Invalid(fldPath.Index(0).Child("allocatedResourcesStatus").Index(1).Child("name"), core.ResourceName("test.device/test2"), "must match one of the container's resource requirements"),
 			},
 		},
+
+		"don't allow health status outside the known values": {
+			containers: []core.Container{
+				{
+					Name: "container-1",
+					Resources: core.ResourceRequirements{
+						Requests: core.ResourceList{
+							"test.device/test": resource.MustParse("1"),
+						},
+					},
+				},
+			},
+			containerStatuses: []core.ContainerStatus{
+				{
+					Name: "container-1",
+					AllocatedResourcesStatus: []core.ResourceStatus{
+						{
+							Name: "test.device/test",
+							Resources: []core.ResourceHealth{
+								{
+									ResourceID: "resource-1",
+									Health:     "invalid-health-value",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantFieldErrors: field.ErrorList{
+				field.NotSupported(fldPath.Index(0).Child("allocatedResourcesStatus").Index(0).Child("resources").Index(0).Child("health"), core.ResourceHealthStatus("invalid-health-value"), []string{"Healthy", "Unhealthy", "Unknown"}),
+			},
+		},
 	}
 	for name, tt := range testCases {
 		t.Run(name, func(t *testing.T) {

pkg/apis/core/validation/validation_test.go

@@ -664,6 +664,13 @@ const (
 	// No effect for other cases such as using serverTLSbootstap.
 	ReloadKubeletServerCertificateFile featuregate.Feature = "ReloadKubeletServerCertificateFile"
 
+	// owner: @SergeyKanzhelev
+	// kep: https://kep.k8s.io/4680
+	// alpha: v1.31
+	//
+	// Adds the AllocatedResourcesStatus to the container status.
+	ResourceHealthStatus featuregate.Feature = "ResourceHealthStatus"
+
 	// owner: @mikedanese
 	// alpha: v1.7
 	// beta: v1.12
@@ -1150,6 +1157,8 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 
 	ReloadKubeletServerCertificateFile: {Default: true, PreRelease: featuregate.Beta},
 
+	ResourceHealthStatus: {Default: false, PreRelease: featuregate.Alpha},
+
 	RotateKubeletServerCertificate: {Default: true, PreRelease: featuregate.Beta},
 
 	RuntimeClassInImageCriAPI: {Default: false, PreRelease: featuregate.Alpha},

pkg/features/kube_features.go

@@ -32,6 +32,7 @@ import (
 	kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
 	"k8s.io/kubernetes/pkg/kubelet/apis/podresources"
 	"k8s.io/kubernetes/pkg/kubelet/cm/devicemanager"
+	"k8s.io/kubernetes/pkg/kubelet/cm/resourceupdates"
 	"k8s.io/kubernetes/pkg/kubelet/config"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	evictionapi "k8s.io/kubernetes/pkg/kubelet/eviction/api"
@@ -132,6 +133,12 @@ type ContainerManager interface {
 	// might need to unprepare resources.
 	PodMightNeedToUnprepareResources(UID types.UID) bool
 
+	// UpdateAllocatedResourcesStatus updates the status of allocated resources for the pod.
+	UpdateAllocatedResourcesStatus(pod *v1.Pod, status *v1.PodStatus)
+
+	// Updates returns a channel that receives an Update when the device changed its status.
+	Updates() <-chan resourceupdates.Update
+
 	// Implements the PodResources Provider API
 	podresources.CPUsProvider
 	podresources.DevicesProvider

pkg/kubelet/cm/container_manager.go

@@ -53,6 +53,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/cm/dra"
 	"k8s.io/kubernetes/pkg/kubelet/cm/memorymanager"
 	memorymanagerstate "k8s.io/kubernetes/pkg/kubelet/cm/memorymanager/state"
+	"k8s.io/kubernetes/pkg/kubelet/cm/resourceupdates"
 	"k8s.io/kubernetes/pkg/kubelet/cm/topologymanager"
 	cmutil "k8s.io/kubernetes/pkg/kubelet/cm/util"
 	"k8s.io/kubernetes/pkg/kubelet/config"
@@ -1027,3 +1028,16 @@ func (cm *containerManagerImpl) UnprepareDynamicResources(pod *v1.Pod) error {
 func (cm *containerManagerImpl) PodMightNeedToUnprepareResources(UID types.UID) bool {
 	return cm.draManager.PodMightNeedToUnprepareResources(UID)
 }
+
+func (cm *containerManagerImpl) UpdateAllocatedResourcesStatus(pod *v1.Pod, status *v1.PodStatus) {
+
+	// For now we only support Device Plugin
+	cm.deviceManager.UpdateAllocatedResourcesStatus(pod, status)
+
+	// TODO(SergeyKanzhelev, https://kep.k8s.io/4680): add support for DRA resources which is planned for the next iteration of a KEP.
+}
+
+func (cm *containerManagerImpl) Updates() <-chan resourceupdates.Update {
+	// TODO(SergeyKanzhelev, https://kep.k8s.io/4680): add support for DRA resources, for now only use device plugin updates. DRA support is planned for the next iteration of a KEP.
+	return cm.deviceManager.Updates()
+}

pkg/kubelet/cm/container_manager_linux.go

@@ -28,6 +28,7 @@ import (
 	podresourcesapi "k8s.io/kubelet/pkg/apis/podresources/v1"
 	"k8s.io/kubernetes/pkg/kubelet/cm/cpumanager"
 	"k8s.io/kubernetes/pkg/kubelet/cm/memorymanager"
+	"k8s.io/kubernetes/pkg/kubelet/cm/resourceupdates"
 	"k8s.io/kubernetes/pkg/kubelet/cm/topologymanager"
 	"k8s.io/kubernetes/pkg/kubelet/config"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
@@ -181,6 +182,13 @@ func (cm *containerManagerStub) PodMightNeedToUnprepareResources(UID types.UID)
 	return false
 }
 
+func (cm *containerManagerStub) UpdateAllocatedResourcesStatus(pod *v1.Pod, status *v1.PodStatus) {
+}
+
+func (cm *containerManagerStub) Updates() <-chan resourceupdates.Update {
+	return nil
+}
+
 func NewStubContainerManager() ContainerManager {
 	return &containerManagerStub{shouldResetExtendedResourceCapacity: false}
 }

pkg/kubelet/cm/container_manager_stub.go

@@ -41,6 +41,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/cm/cpumanager"
 	"k8s.io/kubernetes/pkg/kubelet/cm/devicemanager"
 	"k8s.io/kubernetes/pkg/kubelet/cm/memorymanager"
+	"k8s.io/kubernetes/pkg/kubelet/cm/resourceupdates"
 	"k8s.io/kubernetes/pkg/kubelet/cm/topologymanager"
 	"k8s.io/kubernetes/pkg/kubelet/config"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
@@ -205,6 +206,19 @@ func (cm *containerManagerImpl) GetResources(pod *v1.Pod, container *v1.Containe
 	return opts, nil
 }
 
+func (cm *containerManagerImpl) UpdateAllocatedResourcesStatus(pod *v1.Pod, status *v1.PodStatus) {
+	// For now we only support Device Plugin
+
+	cm.deviceManager.UpdateAllocatedResourcesStatus(pod, status)
+
+	// TODO(SergeyKanzhelev, https://kep.k8s.io/4680): add support for DRA resources when DRA supports Windows
+}
+
+func (cm *containerManagerImpl) Updates() <-chan resourceupdates.Update {
+	// TODO(SergeyKanzhelev, https://kep.k8s.io/4680): add support for DRA resources, for now only use device plugin updates
+	return cm.deviceManager.Updates()
+}
+
 func (cm *containerManagerImpl) UpdatePluginResources(node *schedulerframework.NodeInfo, attrs *lifecycle.PodAdmitAttributes) error {
 	return cm.deviceManager.UpdatePluginResources(node, attrs)
 }