Merge pull request kubernetes#125296 from jsturtevant/windows-numa-support

Support CPU and Topology manager on Windows

Author: k8s-ci-robot

pkg/features/kube_features.go

@@ -715,6 +715,12 @@ const (
 	// Allows kube-proxy to run in Overlay mode for Windows
 	WinOverlay featuregate.Feature = "WinOverlay"
 
+	// owner: @jsturtevant
+	// kep: https://kep.k8s.io/4888
+	//
+	// Add CPU and Memory Affinity support to Windows nodes with CPUManager, MemoryManager and Topology manager
+	WindowsCPUAndMemoryAffinity featuregate.Feature = "WindowsCPUAndMemoryAffinity"
+
 	// owner: @marosset
 	// kep: https://kep.k8s.io/3503
 	//

pkg/features/versioned_kube_features.go

@@ -774,6 +774,10 @@ var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate
 		{Version: version.MustParse("1.20"), Default: true, PreRelease: featuregate.Beta},
 	},
 
+	WindowsCPUAndMemoryAffinity: {
+		{Version: version.MustParse("1.32"), Default: false, PreRelease: featuregate.Alpha},
+	},
+
 	WindowsHostNetwork: {
 		{Version: version.MustParse("1.26"), Default: true, PreRelease: featuregate.Alpha},
 	},

pkg/kubelet/cm/container_manager.go

@@ -198,6 +198,14 @@ type Status struct {
 	SoftRequirements error
 }
 
+func int64Slice(in []int) []int64 {
+	out := make([]int64, len(in))
+	for i := range in {
+		out[i] = int64(in[i])
+	}
+	return out
+}
+
 // parsePercentage parses the percentage string to numeric value.
 func parsePercentage(v string) (int64, error) {
 	if !strings.HasSuffix(v, "%") {

pkg/kubelet/cm/container_manager_linux.go

@@ -920,14 +920,6 @@ func (cm *containerManagerImpl) GetAllocatableDevices() []*podresourcesapi.Conta
 	return containerDevicesFromResourceDeviceInstances(cm.deviceManager.GetAllocatableDevices())
 }
 
-func int64Slice(in []int) []int64 {
-	out := make([]int64, len(in))
-	for i := range in {
-		out[i] = int64(in[i])
-	}
-	return out
-}
-
 func (cm *containerManagerImpl) GetCPUs(podUID, containerName string) []int64 {
 	if cm.cpuManager != nil {
 		return int64Slice(cm.cpuManager.GetExclusiveCPUs(podUID, containerName).UnsortedList())

pkg/kubelet/cm/container_manager_windows.go

@@ -25,6 +25,10 @@ package cm
 import (
 	"context"
 	"fmt"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	kubefeatures "k8s.io/kubernetes/pkg/features"
+	"k8s.io/kubernetes/pkg/kubelet/cm/memorymanager"
+	"sync"
 
 	"k8s.io/klog/v2"
 	"k8s.io/mount-utils"
@@ -38,10 +42,8 @@ import (
 	pluginwatcherapi "k8s.io/kubelet/pkg/apis/pluginregistration/v1"
 	podresourcesapi "k8s.io/kubelet/pkg/apis/podresources/v1"
 	"k8s.io/kubernetes/pkg/kubelet/cadvisor"
-	"k8s.io/kubernetes/pkg/kubelet/cm/admission"
 	"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"
@@ -63,12 +65,9 @@ type containerManagerImpl struct {
 	deviceManager devicemanager.Manager
 	// Interface for Topology resource co-ordination
 	topologyManager topologymanager.Manager
-}
-
-type noopWindowsResourceAllocator struct{}
-
-func (ra *noopWindowsResourceAllocator) Admit(attrs *lifecycle.PodAdmitAttributes) lifecycle.PodAdmitResult {
-	return admission.GetPodAdmitResult(nil)
+	cpuManager      cpumanager.Manager
+	nodeInfo        *v1.Node
+	sync.RWMutex
 }
 
 func (cm *containerManagerImpl) Start(ctx context.Context, node *v1.Node,
@@ -80,6 +79,8 @@ func (cm *containerManagerImpl) Start(ctx context.Context, node *v1.Node,
 	localStorageCapacityIsolation bool) error {
 	klog.V(2).InfoS("Starting Windows container manager")
 
+	cm.nodeInfo = node
+
 	if localStorageCapacityIsolation {
 		rootfs, err := cm.cadvisorInterface.RootFsInfo()
 		if err != nil {
@@ -92,6 +93,14 @@ func (cm *containerManagerImpl) Start(ctx context.Context, node *v1.Node,
 
 	containerMap, containerRunningSet := buildContainerMapAndRunningSetFromRuntime(ctx, runtimeService)
 
+	// Initialize CPU manager
+	if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.WindowsCPUAndMemoryAffinity) {
+		err := cm.cpuManager.Start(cpumanager.ActivePodsFunc(activePods), sourcesReady, podStatusProvider, runtimeService, containerMap)
+		if err != nil {
+			return fmt.Errorf("start cpu manager error: %v", err)
+		}
+	}
+
 	// Starts device manager.
 	if err := cm.deviceManager.Start(devicemanager.ActivePodsFunc(activePods), sourcesReady, containerMap, containerRunningSet); err != nil {
 		return err
@@ -117,7 +126,37 @@ func NewContainerManager(mountUtil mount.Interface, cadvisorInterface cadvisor.I
 		cadvisorInterface: cadvisorInterface,
 	}
 
-	cm.topologyManager = topologymanager.NewFakeManager()
+	if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.WindowsCPUAndMemoryAffinity) {
+		klog.InfoS("Creating topology manager")
+		cm.topologyManager, err = topologymanager.NewManager(machineInfo.Topology,
+			nodeConfig.TopologyManagerPolicy,
+			nodeConfig.TopologyManagerScope,
+			nodeConfig.TopologyManagerPolicyOptions)
+		if err != nil {
+			klog.ErrorS(err, "Failed to initialize topology manager")
+			return nil, err
+		}
+
+		klog.InfoS("Creating cpu manager")
+		cm.cpuManager, err = cpumanager.NewManager(
+			nodeConfig.CPUManagerPolicy,
+			nodeConfig.CPUManagerPolicyOptions,
+			nodeConfig.CPUManagerReconcilePeriod,
+			machineInfo,
+			nodeConfig.NodeAllocatableConfig.ReservedSystemCPUs,
+			cm.GetNodeAllocatableReservation(),
+			nodeConfig.KubeletRootDir,
+			cm.topologyManager,
+		)
+		if err != nil {
+			klog.ErrorS(err, "Failed to initialize cpu manager")
+			return nil, err
+		}
+		cm.topologyManager.AddHintProvider(cm.cpuManager)
+	} else {
+		cm.topologyManager = topologymanager.NewFakeManager()
+		cm.cpuManager = cpumanager.NewFakeManager()
+	}
 
 	klog.InfoS("Creating device plugin manager")
 	cm.deviceManager, err = devicemanager.NewManagerImpl(nil, cm.topologyManager)
@@ -134,7 +173,9 @@ func (cm *containerManagerImpl) SystemCgroupsLimit() v1.ResourceList {
 }
 
 func (cm *containerManagerImpl) GetNodeConfig() NodeConfig {
-	return NodeConfig{}
+	cm.RLock()
+	defer cm.RUnlock()
+	return cm.nodeConfig
 }
 
 func (cm *containerManagerImpl) GetMountedSubsystems() *CgroupSubsystems {
@@ -226,7 +267,7 @@ func (cm *containerManagerImpl) UpdatePluginResources(node *schedulerframework.N
 }
 
 func (cm *containerManagerImpl) InternalContainerLifecycle() InternalContainerLifecycle {
-	return &internalContainerLifecycleImpl{cpumanager.NewFakeManager(), memorymanager.NewFakeManager(), topologymanager.NewFakeManager()}
+	return &internalContainerLifecycleImpl{cm.cpuManager, memorymanager.NewFakeManager(), cm.topologyManager}
 }
 
 func (cm *containerManagerImpl) GetPodCgroupRoot() string {
@@ -246,18 +287,30 @@ func (cm *containerManagerImpl) ShouldResetExtendedResourceCapacity() bool {
 }
 
 func (cm *containerManagerImpl) GetAllocateResourcesPodAdmitHandler() lifecycle.PodAdmitHandler {
-	return &noopWindowsResourceAllocator{}
+	return cm.topologyManager
 }
 
 func (cm *containerManagerImpl) UpdateAllocatedDevices() {
 	return
 }
 
-func (cm *containerManagerImpl) GetCPUs(_, _ string) []int64 {
+func (cm *containerManagerImpl) GetCPUs(podUID, containerName string) []int64 {
+	if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.WindowsCPUAndMemoryAffinity) {
+		if cm.cpuManager != nil {
+			return int64Slice(cm.cpuManager.GetExclusiveCPUs(podUID, containerName).UnsortedList())
+		}
+		return []int64{}
+	}
 	return nil
 }
 
 func (cm *containerManagerImpl) GetAllocatableCPUs() []int64 {
+	if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.WindowsCPUAndMemoryAffinity) {
+		if cm.cpuManager != nil {
+			return int64Slice(cm.cpuManager.GetAllocatableCPUs().UnsortedList())
+		}
+		return []int64{}
+	}
 	return nil
 }
 

pkg/kubelet/cm/cpumanager/cpu_manager.go

@@ -512,21 +512,6 @@ func findContainerStatusByName(status *v1.PodStatus, name string) (*v1.Container
 	return nil, fmt.Errorf("unable to find status for container with name %v in pod status (it may not be running)", name)
 }
 
-func (m *manager) updateContainerCPUSet(ctx context.Context, containerID string, cpus cpuset.CPUSet) error {
-	// TODO: Consider adding a `ResourceConfigForContainer` helper in
-	// helpers_linux.go similar to what exists for pods.
-	// It would be better to pass the full container resources here instead of
-	// this patch-like partial resources.
-	return m.containerRuntime.UpdateContainerResources(
-		ctx,
-		containerID,
-		&runtimeapi.ContainerResources{
-			Linux: &runtimeapi.LinuxContainerResources{
-				CpusetCpus: cpus.String(),
-			},
-		})
-}
-
 func (m *manager) GetExclusiveCPUs(podUID, containerName string) cpuset.CPUSet {
 	if result, ok := m.state.GetCPUSet(podUID, containerName); ok {
 		return result

pkg/kubelet/cm/cpumanager/cpu_manager_others.go

@@ -0,0 +1,43 @@
+//go:build !windows
+// +build !windows
+
+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cpumanager
+
+import (
+	"context"
+
+	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
+	"k8s.io/utils/cpuset"
+)
+
+func (m *manager) updateContainerCPUSet(ctx context.Context, containerID string, cpus cpuset.CPUSet) error {
+	// TODO: Consider adding a `ResourceConfigForContainer` helper in
+	// helpers_linux.go similar to what exists for pods.
+	// It would be better to pass the full container resources here instead of
+	// this patch-like partial resources.
+
+	return m.containerRuntime.UpdateContainerResources(
+		ctx,
+		containerID,
+		&runtimeapi.ContainerResources{
+			Linux: &runtimeapi.LinuxContainerResources{
+				CpusetCpus: cpus.String(),
+			},
+		})
+}

pkg/kubelet/cm/cpumanager/cpu_manager_test.go

@@ -21,11 +21,16 @@ import (
 	"fmt"
 	"os"
 	"reflect"
+	"runtime"
 	"strconv"
 	"strings"
 	"testing"
 	"time"
 
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"k8s.io/kubernetes/pkg/features"
+
 	cadvisorapi "github.com/google/cadvisor/info/v1"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -263,6 +268,10 @@ func makeMultiContainerPodWithOptions(initCPUs, appCPUs []*containerOptions) *v1
 }
 
 func TestCPUManagerAdd(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WindowsCPUAndMemoryAffinity, true)
+	}
+
 	testPolicy, _ := NewStaticPolicy(
 		&topology.CPUTopology{
 			NumCPUs:    4,
@@ -347,6 +356,10 @@ func TestCPUManagerAdd(t *testing.T) {
 }
 
 func TestCPUManagerAddWithInitContainers(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WindowsCPUAndMemoryAffinity, true)
+	}
+
 	testCases := []struct {
 		description      string
 		topo             *topology.CPUTopology
@@ -598,6 +611,10 @@ func TestCPUManagerAddWithInitContainers(t *testing.T) {
 }
 
 func TestCPUManagerGenerate(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WindowsCPUAndMemoryAffinity, true)
+	}
+
 	testCases := []struct {
 		description                string
 		cpuPolicyName              string
@@ -703,6 +720,10 @@ func TestCPUManagerGenerate(t *testing.T) {
 }
 
 func TestCPUManagerRemove(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WindowsCPUAndMemoryAffinity, true)
+	}
+
 	containerID := "fakeID"
 	containerMap := containermap.NewContainerMap()
 
@@ -746,6 +767,10 @@ func TestCPUManagerRemove(t *testing.T) {
 }
 
 func TestReconcileState(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WindowsCPUAndMemoryAffinity, true)
+	}
+
 	testPolicy, _ := NewStaticPolicy(
 		&topology.CPUTopology{
 			NumCPUs:    8,
@@ -1269,6 +1294,10 @@ func TestReconcileState(t *testing.T) {
 // above test cases are without kubelet --reserved-cpus cmd option
 // the following tests are with --reserved-cpus configured
 func TestCPUManagerAddWithResvList(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WindowsCPUAndMemoryAffinity, true)
+	}
+
 	testPolicy, _ := NewStaticPolicy(
 		&topology.CPUTopology{
 			NumCPUs:    4,
@@ -1343,6 +1372,10 @@ func TestCPUManagerAddWithResvList(t *testing.T) {
 }
 
 func TestCPUManagerHandlePolicyOptions(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WindowsCPUAndMemoryAffinity, true)
+	}
+
 	testCases := []struct {
 		description      string
 		cpuPolicyName    string
@@ -1409,6 +1442,10 @@ func TestCPUManagerHandlePolicyOptions(t *testing.T) {
 }
 
 func TestCPUManagerGetAllocatableCPUs(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WindowsCPUAndMemoryAffinity, true)
+	}
+
 	nonePolicy, _ := NewNonePolicy(nil)
 	staticPolicy, _ := NewStaticPolicy(
 		&topology.CPUTopology{