Merge pull request kubernetes#80570 from klueska/upstream-add-topology-manager-to-devicemanager

Add support for Topology Manager to Device Manager

Author: k8s-ci-robot

pkg/kubelet/apis/deviceplugin/v1beta1/api.pb.go

@@ -74,7 +74,7 @@ message ListAndWatchResponse {
 }
 
 message TopologyInfo {
-    	NUMANode node = 1;
+	repeated NUMANode nodes = 1;
 }
 
 message NUMANode {

pkg/kubelet/apis/deviceplugin/v1beta1/api.proto

@@ -304,7 +304,8 @@ func NewContainerManager(mountUtil mount.Interface, cadvisorInterface cadvisor.I
 
 	klog.Infof("Creating device plugin manager: %t", devicePluginEnabled)
 	if devicePluginEnabled {
-		cm.deviceManager, err = devicemanager.NewManagerImpl()
+		cm.deviceManager, err = devicemanager.NewManagerImpl(numaNodeInfo, cm.topologyManager)
+		cm.topologyManager.AddHintProvider(cm.deviceManager)
 	} else {
 		cm.deviceManager, err = devicemanager.NewManagerStub()
 	}

pkg/kubelet/cm/container_manager_linux.go

@@ -50,12 +50,12 @@ func (a *cpuAccumulator) take(cpus cpuset.CPUSet) {
 
 // Returns true if the supplied socket is fully available in `topoDetails`.
 func (a *cpuAccumulator) isSocketFree(socketID int) bool {
-	return a.details.CPUsInSocket(socketID).Size() == a.topo.CPUsPerSocket()
+	return a.details.CPUsInSockets(socketID).Size() == a.topo.CPUsPerSocket()
 }
 
 // Returns true if the supplied core is fully available in `topoDetails`.
 func (a *cpuAccumulator) isCoreFree(coreID int) bool {
-	return a.details.CPUsInCore(coreID).Size() == a.topo.CPUsPerCore()
+	return a.details.CPUsInCores(coreID).Size() == a.topo.CPUsPerCore()
 }
 
 // Returns free socket IDs as a slice sorted by:
@@ -72,14 +72,14 @@ func (a *cpuAccumulator) freeCores() []int {
 	socketIDs := a.details.Sockets().ToSliceNoSort()
 	sort.Slice(socketIDs,
 		func(i, j int) bool {
-			iCores := a.details.CoresInSocket(socketIDs[i]).Filter(a.isCoreFree)
-			jCores := a.details.CoresInSocket(socketIDs[j]).Filter(a.isCoreFree)
+			iCores := a.details.CoresInSockets(socketIDs[i]).Filter(a.isCoreFree)
+			jCores := a.details.CoresInSockets(socketIDs[j]).Filter(a.isCoreFree)
 			return iCores.Size() < jCores.Size() || socketIDs[i] < socketIDs[j]
 		})
 
 	coreIDs := []int{}
 	for _, s := range socketIDs {
-		coreIDs = append(coreIDs, a.details.CoresInSocket(s).Filter(a.isCoreFree).ToSlice()...)
+		coreIDs = append(coreIDs, a.details.CoresInSockets(s).Filter(a.isCoreFree).ToSlice()...)
 	}
 	return coreIDs
 }
@@ -100,25 +100,25 @@ func (a *cpuAccumulator) freeCPUs() []int {
 			iCore := cores[i]
 			jCore := cores[j]
 
-			iCPUs := a.topo.CPUDetails.CPUsInCore(iCore).ToSlice()
-			jCPUs := a.topo.CPUDetails.CPUsInCore(jCore).ToSlice()
+			iCPUs := a.topo.CPUDetails.CPUsInCores(iCore).ToSlice()
+			jCPUs := a.topo.CPUDetails.CPUsInCores(jCore).ToSlice()
 
 			iSocket := a.topo.CPUDetails[iCPUs[0]].SocketID
 			jSocket := a.topo.CPUDetails[jCPUs[0]].SocketID
 
 			// Compute the number of CPUs in the result reside on the same socket
 			// as each core.
-			iSocketColoScore := a.topo.CPUDetails.CPUsInSocket(iSocket).Intersection(a.result).Size()
-			jSocketColoScore := a.topo.CPUDetails.CPUsInSocket(jSocket).Intersection(a.result).Size()
+			iSocketColoScore := a.topo.CPUDetails.CPUsInSockets(iSocket).Intersection(a.result).Size()
+			jSocketColoScore := a.topo.CPUDetails.CPUsInSockets(jSocket).Intersection(a.result).Size()
 
 			// Compute the number of available CPUs available on the same socket
 			// as each core.
-			iSocketFreeScore := a.details.CPUsInSocket(iSocket).Size()
-			jSocketFreeScore := a.details.CPUsInSocket(jSocket).Size()
+			iSocketFreeScore := a.details.CPUsInSockets(iSocket).Size()
+			jSocketFreeScore := a.details.CPUsInSockets(jSocket).Size()
 
 			// Compute the number of available CPUs on each core.
-			iCoreFreeScore := a.details.CPUsInCore(iCore).Size()
-			jCoreFreeScore := a.details.CPUsInCore(jCore).Size()
+			iCoreFreeScore := a.details.CPUsInCores(iCore).Size()
+			jCoreFreeScore := a.details.CPUsInCores(jCore).Size()
 
 			return iSocketColoScore > jSocketColoScore ||
 				iSocketFreeScore < jSocketFreeScore ||
@@ -129,7 +129,7 @@ func (a *cpuAccumulator) freeCPUs() []int {
 
 	// For each core, append sorted CPU IDs to result.
 	for _, core := range cores {
-		result = append(result, a.details.CPUsInCore(core).ToSlice()...)
+		result = append(result, a.details.CPUsInCores(core).ToSlice()...)
 	}
 	return result
 }
@@ -161,7 +161,7 @@ func takeByTopology(topo *topology.CPUTopology, availableCPUs cpuset.CPUSet, num
 	if acc.needs(acc.topo.CPUsPerSocket()) {
 		for _, s := range acc.freeSockets() {
 			klog.V(4).Infof("[cpumanager] takeByTopology: claiming socket [%d]", s)
-			acc.take(acc.details.CPUsInSocket(s))
+			acc.take(acc.details.CPUsInSockets(s))
 			if acc.isSatisfied() {
 				return acc.result, nil
 			}
@@ -176,7 +176,7 @@ func takeByTopology(topo *topology.CPUTopology, availableCPUs cpuset.CPUSet, num
 	if acc.needs(acc.topo.CPUsPerCore()) {
 		for _, c := range acc.freeCores() {
 			klog.V(4).Infof("[cpumanager] takeByTopology: claiming core [%d]", c)
-			acc.take(acc.details.CPUsInCore(c))
+			acc.take(acc.details.CPUsInCores(c))
 			if acc.isSatisfied() {
 				return acc.result, nil
 			}

pkg/kubelet/cm/cpumanager/cpu_assignment.go

@@ -258,7 +258,7 @@ func (p *staticPolicy) allocateCPUs(s state.State, numCPUs int, numaAffinity soc
 	if numaAffinity != nil {
 		alignedCPUs := cpuset.NewCPUSet()
 		for _, numaNodeID := range numaAffinity.GetSockets() {
-			alignedCPUs = alignedCPUs.Union(p.assignableCPUs(s).Intersection(p.topology.CPUDetails.CPUsInNUMANode(numaNodeID)))
+			alignedCPUs = alignedCPUs.Union(p.assignableCPUs(s).Intersection(p.topology.CPUDetails.CPUsInNUMANodes(numaNodeID)))
 		}
 
 		numAlignedToAlloc := alignedCPUs.Size()