UPSTREAM: <carry>: add management support to kubelet

UPSTREAM: <carry>: management workloads enhancement 741 UPSTREAM: <carry>: lower verbosity of managed workloads logging Support for managed workloads was introduced by PR#627. However, the the CPU manager reconcile loop now seems to flood kubelet log with "reconcileState: skipping pod; pod is managed" warnings. Lower the verbosity of these log messages. UPSTREAM: <carry>: set correctly static pods CPUs when workload partitioning is disabled UPSTREAM: <carry>: Remove reserved CPUs from default set Remove reserved CPUs from default set when workload partitioning is enabled. Co-Authored-By: Brent Rowsell <[email protected]> Signed-off-by: Artyom Lukianov <[email protected]> Signed-off-by: Don Penney <[email protected]> OpenShift-Rebase-Source: b762ced OpenShift-Rebase-Source: 63cf793 OpenShift-Rebase-Source: 32af64c UPSTREAM: <carry>: add management support to kubelet UPSTREAM: <carry>: OCPBUGS-29520: fix cpu manager default cpuset check in workload partitioned env (this can be squashed to 04070bb UPSTREAM: : add management support to kubelet) Workload partitioning makes the separation between reserved and workload cpus more strict. It is therefore expected the reserved cpus are NOT part of the default cpuset and the existing check was overzealous. First execution of kubelet after reboot never gets here as the cpuset is computed on line 209. However a kubelet restart without reboot skips this code, recovers from state file and runs the check on line 220. This was uncovered by decoupling the cpu manager state file cleanup from kubelet restart, doing it only once at reboot as part of OCPBUGS-24366 UPSTREAM: <carry>: add management workload check for guaranteed qos when static pods have workload partitioning enabled we should not alter their resources if they are Guaranteed QoS, this change adds a check for Guaranteed QoS Signed-off-by: ehila <[email protected]> test: add unit tests for error states Signed-off-by: ehila <[email protected]>
openshift · Nov 29, 2024 · 96dea4a · 96dea4a
1 parent 0f4e262
commit 96dea4a
Show file tree

Hide file tree

Showing 10 changed files with 1,017 additions and 16 deletions.
diff --git a/pkg/kubelet/cm/cpumanager/cpu_manager.go b/pkg/kubelet/cm/cpumanager/cpu_manager.go
@@ -35,6 +35,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/cm/topologymanager"
 	"k8s.io/kubernetes/pkg/kubelet/config"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
+	"k8s.io/kubernetes/pkg/kubelet/managed"
 	"k8s.io/kubernetes/pkg/kubelet/status"
 	"k8s.io/utils/cpuset"
 )
@@ -409,13 +410,18 @@ func (m *manager) reconcileState() (success []reconciledContainer, failure []rec
 	failure = []reconciledContainer{}
 
 	m.removeStaleState()
+	workloadEnabled := managed.IsEnabled()
 	for _, pod := range m.activePods() {
 		pstatus, ok := m.podStatusProvider.GetPodStatus(pod.UID)
 		if !ok {
 			klog.V(5).InfoS("ReconcileState: skipping pod; status not found", "pod", klog.KObj(pod))
 			failure = append(failure, reconciledContainer{pod.Name, "", ""})
 			continue
 		}
+		if enabled, _, _ := managed.IsPodManaged(pod); workloadEnabled && enabled {
+			klog.V(4).InfoS("[cpumanager] reconcileState: skipping pod; pod is managed (pod: %s)", pod.Name)
+			continue
+		}
 
 		allContainers := pod.Spec.InitContainers
 		allContainers = append(allContainers, pod.Spec.Containers...)

diff --git a/pkg/kubelet/cm/cpumanager/policy_static.go b/pkg/kubelet/cm/cpumanager/policy_static.go
@@ -29,6 +29,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/cm/cpumanager/topology"
 	"k8s.io/kubernetes/pkg/kubelet/cm/topologymanager"
 	"k8s.io/kubernetes/pkg/kubelet/cm/topologymanager/bitmask"
+	"k8s.io/kubernetes/pkg/kubelet/managed"
 	"k8s.io/kubernetes/pkg/kubelet/metrics"
 	"k8s.io/utils/cpuset"
 )
@@ -214,6 +215,10 @@ func (p *staticPolicy) validateState(s state.State) error {
 		// state is empty initialize
 		s.SetDefaultCPUSet(allCPUs)
 		klog.InfoS("Static policy initialized", "defaultCPUSet", allCPUs)
+		if managed.IsEnabled() {
+			defaultCpus := s.GetDefaultCPUSet().Difference(p.reservedCPUs)
+			s.SetDefaultCPUSet(defaultCpus)
+		}
 		return nil
 	}
 
@@ -227,7 +232,9 @@ func (p *staticPolicy) validateState(s state.State) error {
 				p.reservedCPUs.Intersection(tmpDefaultCPUset).String(), tmpDefaultCPUset.String())
 		}
 	} else {
-		if !p.reservedCPUs.Intersection(tmpDefaultCPUset).Equals(p.reservedCPUs) {
+		// 2. This only applies when managed mode is disabled. Active workload partitioning feature
+		//    removes the reserved cpus from the default cpu mask on purpose.
+		if !managed.IsEnabled() && !p.reservedCPUs.Intersection(tmpDefaultCPUset).Equals(p.reservedCPUs) {
 			return fmt.Errorf("not all reserved cpus: \"%s\" are present in defaultCpuSet: \"%s\"",
 				p.reservedCPUs.String(), tmpDefaultCPUset.String())
 		}
@@ -259,10 +266,17 @@ func (p *staticPolicy) validateState(s state.State) error {
 		}
 	}
 	totalKnownCPUs = totalKnownCPUs.Union(tmpCPUSets...)
-	if !totalKnownCPUs.Equals(allCPUs) {
-		return fmt.Errorf("current set of available CPUs \"%s\" doesn't match with CPUs in state \"%s\"",
-			allCPUs.String(), totalKnownCPUs.String())
+	availableCPUs := p.topology.CPUDetails.CPUs()
 
+	// CPU (workload) partitioning removes reserved cpus
+	// from the default mask intentionally
+	if managed.IsEnabled() {
+		availableCPUs = availableCPUs.Difference(p.reservedCPUs)
+	}
+
+	if !totalKnownCPUs.Equals(availableCPUs) {
+		return fmt.Errorf("current set of available CPUs \"%s\" doesn't match with CPUs in state \"%s\"",
+			availableCPUs.String(), totalKnownCPUs.String())
 	}
 
 	return nil

diff --git a/pkg/kubelet/cm/cpumanager/policy_static_test.go b/pkg/kubelet/cm/cpumanager/policy_static_test.go
@@ -21,6 +21,8 @@ import (
 	"reflect"
 	"testing"
 
+	"k8s.io/kubernetes/pkg/kubelet/managed"
+
 	v1 "k8s.io/api/core/v1"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
@@ -961,18 +963,19 @@ func TestTopologyAwareAllocateCPUs(t *testing.T) {
 // above test cases are without kubelet --reserved-cpus cmd option
 // the following tests are with --reserved-cpus configured
 type staticPolicyTestWithResvList struct {
-	description      string
-	topo             *topology.CPUTopology
-	numReservedCPUs  int
-	reserved         cpuset.CPUSet
-	cpuPolicyOptions map[string]string
-	stAssignments    state.ContainerCPUAssignments
-	stDefaultCPUSet  cpuset.CPUSet
-	pod              *v1.Pod
-	expErr           error
-	expNewErr        error
-	expCPUAlloc      bool
-	expCSet          cpuset.CPUSet
+	description         string
+	topo                *topology.CPUTopology
+	numReservedCPUs     int
+	reserved            cpuset.CPUSet
+	cpuPolicyOptions    map[string]string
+	stAssignments       state.ContainerCPUAssignments
+	stDefaultCPUSet     cpuset.CPUSet
+	pod                 *v1.Pod
+	expErr              error
+	expNewErr           error
+	expCPUAlloc         bool
+	expCSet             cpuset.CPUSet
+	managementPartition bool
 }
 
 func TestStaticPolicyStartWithResvList(t *testing.T) {
@@ -1024,9 +1027,31 @@ func TestStaticPolicyStartWithResvList(t *testing.T) {
 			stDefaultCPUSet: cpuset.New(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11),
 			expNewErr:       fmt.Errorf("[cpumanager] unable to reserve the required amount of CPUs (size of 0-1 did not equal 1)"),
 		},
+		{
+			description:         "reserved cores 0 & 6 are not present in available cpuset when management partitioning is enabled",
+			topo:                topoDualSocketHT,
+			numReservedCPUs:     2,
+			stAssignments:       state.ContainerCPUAssignments{},
+			managementPartition: true,
+			expCSet:             cpuset.New(1, 2, 3, 4, 5, 7, 8, 9, 10, 11),
+		},
+		{
+			description:         "reserved cores 0 & 6 are not present in available cpuset when management partitioning is enabled during recovery",
+			topo:                topoDualSocketHT,
+			numReservedCPUs:     2,
+			stAssignments:       state.ContainerCPUAssignments{},
+			stDefaultCPUSet:     cpuset.New(1, 2, 3, 4, 5, 7, 8, 9, 10, 11),
+			managementPartition: true,
+			expCSet:             cpuset.New(1, 2, 3, 4, 5, 7, 8, 9, 10, 11),
+		},
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.description, func(t *testing.T) {
+			wasManaged := managed.IsEnabled()
+			managed.TestOnlySetEnabled(testCase.managementPartition)
+			defer func() {
+				managed.TestOnlySetEnabled(wasManaged)
+			}()
 			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, pkgfeatures.CPUManagerPolicyAlphaOptions, true)
 			p, err := NewStaticPolicy(testCase.topo, testCase.numReservedCPUs, testCase.reserved, topologymanager.NewFakeManager(), testCase.cpuPolicyOptions)
 			if !reflect.DeepEqual(err, testCase.expNewErr) {

diff --git a/pkg/kubelet/cm/qos_container_manager_linux.go b/pkg/kubelet/cm/qos_container_manager_linux.go
@@ -35,6 +35,7 @@ import (
 	"k8s.io/component-helpers/resource"
 	v1qos "k8s.io/kubernetes/pkg/apis/core/v1/helper/qos"
 	kubefeatures "k8s.io/kubernetes/pkg/features"
+	"k8s.io/kubernetes/pkg/kubelet/managed"
 )
 
 const (
@@ -174,6 +175,9 @@ func (m *qosContainerManagerImpl) setCPUCgroupConfig(configs map[v1.PodQOSClass]
 	reuseReqs := make(v1.ResourceList, 4)
 	for i := range pods {
 		pod := pods[i]
+		if enabled, _, _ := managed.IsPodManaged(pod); enabled {
+			continue
+		}
 		qosClass := v1qos.GetPodQOS(pod)
 		if qosClass != v1.PodQOSBurstable {
 			// we only care about the burstable qos tier

diff --git a/pkg/kubelet/config/file.go b/pkg/kubelet/config/file.go
@@ -30,6 +30,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 	api "k8s.io/kubernetes/pkg/apis/core"
+	"k8s.io/kubernetes/pkg/kubelet/managed"
 	kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
 	utilio "k8s.io/utils/io"
 )
@@ -230,6 +231,16 @@ func (s *sourceFile) extractFromFile(filename string) (pod *v1.Pod, err error) {
 		if podErr != nil {
 			return pod, podErr
 		}
+		if managed.IsEnabled() {
+			if newPod, _, err := managed.ModifyStaticPodForPinnedManagement(pod); err != nil {
+				klog.V(2).Error(err, "Static Pod is managed but errored", "name", pod.ObjectMeta.Name, "namespace", pod.ObjectMeta.Namespace)
+			} else if newPod != nil {
+				klog.V(2).InfoS("Static Pod is managed. Using modified pod", "name", newPod.ObjectMeta.Name, "namespace", newPod.ObjectMeta.Namespace, "annotations", newPod.Annotations)
+				pod = newPod
+			} else {
+				klog.V(2).InfoS("Static Pod is not managed", "name", pod.ObjectMeta.Name, "namespace", pod.ObjectMeta.Namespace)
+			}
+		}
 		return pod, nil
 	}
 

diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go
@@ -95,6 +95,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/kuberuntime"
 	"k8s.io/kubernetes/pkg/kubelet/lifecycle"
 	"k8s.io/kubernetes/pkg/kubelet/logs"
+	"k8s.io/kubernetes/pkg/kubelet/managed"
 	"k8s.io/kubernetes/pkg/kubelet/metrics"
 	"k8s.io/kubernetes/pkg/kubelet/metrics/collectors"
 	"k8s.io/kubernetes/pkg/kubelet/network/dns"
@@ -668,6 +669,10 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 
 	klet.runtimeService = kubeDeps.RemoteRuntimeService
 
+	if managed.IsEnabled() {
+		klog.InfoS("Pinned Workload Management Enabled")
+	}
+
 	if kubeDeps.KubeClient != nil {
 		klet.runtimeClassManager = runtimeclass.NewManager(kubeDeps.KubeClient)
 	}

diff --git a/pkg/kubelet/kubelet_node_status.go b/pkg/kubelet/kubelet_node_status.go
@@ -42,7 +42,9 @@ import (
 	kubeletapis "k8s.io/kubelet/pkg/apis"
 	v1helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
 	"k8s.io/kubernetes/pkg/features"
+	"k8s.io/kubernetes/pkg/kubelet/cadvisor"
 	"k8s.io/kubernetes/pkg/kubelet/events"
+	"k8s.io/kubernetes/pkg/kubelet/managed"
 	"k8s.io/kubernetes/pkg/kubelet/nodestatus"
 	taintutil "k8s.io/kubernetes/pkg/util/taints"
 	volutil "k8s.io/kubernetes/pkg/volume/util"
@@ -131,6 +133,9 @@ func (kl *Kubelet) tryRegisterWithAPIServer(node *v1.Node) bool {
 	requiresUpdate = kl.updateDefaultLabels(node, existingNode) || requiresUpdate
 	requiresUpdate = kl.reconcileExtendedResource(node, existingNode) || requiresUpdate
 	requiresUpdate = kl.reconcileHugePageResource(node, existingNode) || requiresUpdate
+	if managed.IsEnabled() {
+		requiresUpdate = kl.addManagementNodeCapacity(node, existingNode) || requiresUpdate
+	}
 	if requiresUpdate {
 		if _, _, err := nodeutil.PatchNodeStatus(kl.kubeClient.CoreV1(), types.NodeName(kl.nodeName), originalNode, existingNode); err != nil {
 			klog.ErrorS(err, "Unable to reconcile node with API server,error updating node", "node", klog.KObj(node))
@@ -141,6 +146,25 @@ func (kl *Kubelet) tryRegisterWithAPIServer(node *v1.Node) bool {
 	return true
 }
 
+// addManagementNodeCapacity adds the managednode capacity to the node
+func (kl *Kubelet) addManagementNodeCapacity(initialNode, existingNode *v1.Node) bool {
+	updateDefaultResources(initialNode, existingNode)
+	machineInfo, err := kl.cadvisor.MachineInfo()
+	if err != nil {
+		klog.Errorf("Unable to calculate managed node capacity for %q: %v", kl.nodeName, err)
+		return false
+	}
+	cpuRequest := cadvisor.CapacityFromMachineInfo(machineInfo)[v1.ResourceCPU]
+	cpuRequestInMilli := cpuRequest.MilliValue()
+	newCPURequest := resource.NewMilliQuantity(cpuRequestInMilli*1000, cpuRequest.Format)
+	managedResourceName := managed.GenerateResourceName("management")
+	if existingCapacity, ok := existingNode.Status.Capacity[managedResourceName]; ok && existingCapacity.Equal(*newCPURequest) {
+		return false
+	}
+	existingNode.Status.Capacity[managedResourceName] = *newCPURequest
+	return true
+}
+
 // reconcileHugePageResource will update huge page capacity for each page size and remove huge page sizes no longer supported
 func (kl *Kubelet) reconcileHugePageResource(initialNode, existingNode *v1.Node) bool {
 	requiresUpdate := updateDefaultResources(initialNode, existingNode)
@@ -432,6 +456,9 @@ func (kl *Kubelet) initialNode(ctx context.Context) (*v1.Node, error) {
 			}
 		}
 	}
+	if managed.IsEnabled() {
+		kl.addManagementNodeCapacity(node, node)
+	}
 
 	kl.setNodeStatus(ctx, node)
 

diff --git a/pkg/kubelet/managed/cpu_shares.go b/pkg/kubelet/managed/cpu_shares.go
@@ -0,0 +1,30 @@
+package managed
+
+const (
+	// These limits are defined in the kernel:
+	// https://github.com/torvalds/linux/blob/0bddd227f3dc55975e2b8dfa7fc6f959b062a2c7/kernel/sched/sched.h#L427-L428
+	MinShares = 2
+	MaxShares = 262144
+
+	SharesPerCPU  = 1024
+	MilliCPUToCPU = 1000
+)
+
+// MilliCPUToShares converts the milliCPU to CFS shares.
+func MilliCPUToShares(milliCPU int64) uint64 {
+	if milliCPU == 0 {
+		// Docker converts zero milliCPU to unset, which maps to kernel default
+		// for unset: 1024. Return 2 here to really match kernel default for
+		// zero milliCPU.
+		return MinShares
+	}
+	// Conceptually (milliCPU / milliCPUToCPU) * sharesPerCPU, but factored to improve rounding.
+	shares := (milliCPU * SharesPerCPU) / MilliCPUToCPU
+	if shares < MinShares {
+		return MinShares
+	}
+	if shares > MaxShares {
+		return MaxShares
+	}
+	return uint64(shares)
+}