k8s 节点 CPU 升级，导致 kubelet 无法启动故障一例

事件背景

大家都知道 k8s 容量不够的时候，都是添加节点来解决问题。这几天有小伙伴在升级 k8s 容量的时候碰到一个问题，他将集群中某一个 node 节点的 CPU 做了升级，然后重启了这个 node 节点导致 kubelet 无法启动，然后大量 pod 被驱逐，报警电话响个不停。为了紧急恢复业务，果断参加故障恢复。

现象获取

在知道事件背景后，我登上了那个已经重启完毕的 node 节点，开始了一系列的网络测试，确认 node 这个宿主机到 Apiserver 和 Loadbalancer 的 ip 和 port 都是通的。随后赶紧看了下 kubelet 的日志，果不其然，一行日志让我看到问题点：

E112123:43:52.64455223453policy_static.go:158]"Staticpolicyinvalidstate,pleasedrainnodeandremovepolicystatefile"err="currentsetofavailableCPUs\"0-7\"doesn'tmatchwithCPUsinstate\"0-3\""
E112123:43:52.64456923453cpu_manager.go:230]"Policystarterror"err="currentsetofavailableCPUs\"0-7\"doesn'tmatchwithCPUsinstate\"0-3\""
E112123:43:52.64458723453kubelet.go:1431]"FailedtostartContainerManager"err="startcpumanagererror:currentsetofavailableCPUs\"0-7\"doesn'tmatchwithCPUsinstate\"0-3\""

说到这里，很多小伙伴会说：“就这？？”。

真的就这。是因为啥呢？

是因为 kubelet 启动参数里面有一个参数很重要：--cpu-manager-policy。表示 kubelet 在使用宿主机的 cpu 是什么逻辑策略。如果你设定为 static ，那么就会在参数 --root-dir 指定的目录下生成一个 cpu_manager_state 这样一个绑定文件。

cpu_manager_state 内容大致长得如下：

{"policyName":"static","defaultCpuSet":"0-7","checksum":14413152}

当你升级这个 k8s node 节点的 CPU 配置，并且使用了 static cpu 管理模式，那么 kubelet 会读取 cpu_manager_state 文件，然后跟现有的宿主运行的资源做对比，如果不一致，kubelet 就不会启动了。

原理分析

既然我们看到了具体现象和故障位置，不妨借着这个小问题我们一起开温馨下 k8s 的 cpu 管理规范。

官方文档如下：

https://kubernetes.io/zh-cn/docs/tasks/administer-cluster/cpu-management-policies/

当然我还想多少说点别的，关于 CPU Manager 整个架构，让小伙伴们有一个整体理解，能更加深入理解官方的 cpu 管理策略到底是做了些什么动作。

cpu-management-policies

CPU Manager 架构

CPU Manager 为满足条件的 Container 分配指定的 CPUs 时，会尽数按 CPU Topology 来分配，也就是参考 CPU Affinity，按如下的优先顺序进行 CPUs 选择：（Logic CPUs 就是 Hyperthreads）

• 如果 Container 要求的 Logic CPUs 数量不少于单块 CPU Socket 中 Logci CPUs 数量，那么会优先把整块 CPU Socket 中的 Logic CPUs 分配给 Container。
• 如果 Container 减余请求的 Logic CPU 数量不少于单块物理 CPU Core 提供的 Logic CPU 数量，那么会优先把整块物理 CPU Core 上的 Logic CPU 分配给 Container。

Container 托余请求的 Logic CPUs 则从按以下规则排列好的 Logic CPUs 列表中选择：

• 同一插槽上可用的 CPU 数量
• 同一核心上可用的 CPU 数量

参考代码：pkg/kubelet/cm/cpumanager/cpu_assignment.go

functakeByTopology(topo*topology.CPUTopology,availableCPUscpuset.CPUSet,numCPUsint)(cpuset.CPUSet,error){
acc:=newCPUAccumulator(topo,availableCPUs,numCPUs)
ifacc.isSatisfied(){
returnacc.result,nil
}
ifacc.isFailed(){
returncpuset.NewCPUSet(),fmt.Errorf("notenoughcpusavailabletosatisfyrequest")
}

//Algorithm:topology-awarebest-fit
//1.Acquirewholesockets,ifavailableandthecontainerrequiresat
//leastasocket's-worthofCPUs.
for_,s:=rangeacc.freeSockets(){
ifacc.needs(acc.topo.CPUsPerSocket()){
glog.V(4).Infof("[cpumanager]takeByTopology:claimingsocket[%d]",s)
acc.take(acc.details.CPUsInSocket(s))
ifacc.isSatisfied(){
returnacc.result,nil
}
}
}

//2.Acquirewholecores,ifavailableandthecontainerrequiresatleast
//acore's-worthofCPUs.
for_,c:=rangeacc.freeCores(){
ifacc.needs(acc.topo.CPUsPerCore()){
glog.V(4).Infof("[cpumanager]takeByTopology:claimingcore[%d]",c)
acc.take(acc.details.CPUsInCore(c))
ifacc.isSatisfied(){
returnacc.result,nil
}
}
}

//3.Acquiresinglethreads,preferringtofillpartially-allocatedcores
//onthesamesocketsasthewholecoreswehavealreadytakeninthis
//allocation.
for_,c:=rangeacc.freeCPUs(){
glog.V(4).Infof("[cpumanager]takeByTopology:claimingCPU[%d]",c)
ifacc.needs(1){
acc.take(cpuset.NewCPUSet(c))
}
ifacc.isSatisfied(){
returnacc.result,nil
}
}

returncpuset.NewCPUSet(),fmt.Errorf("failedtoallocatecpus")
}

发现 CPU Topology

参考代码：vendor/github.com/google/cadvisor/info/v1/machine.go

typeMachineInfostruct{
//Thenumberofcoresinthismachine.
NumCoresint`json:"num_cores"`

...

//MachineTopology
//Describescpu/memorylayoutandhierarchy.
Topology[]Node`json:"topology"`

...
}

typeNodestruct{
Idint`json:"node_id"`
//Per-nodememory
Memoryuint64`json:"memory"`
Cores[]Core`json:"cores"`
Caches[]Cache`json:"caches"`
}

cAdvisor 通过 GetTopology 完成 cpu 拓普信息生成，主要是读取宿主机上 /proc/cpuinfo 中信息来渲染 CPU Topology，通过读取 /sys/devices/system/cpu/cpu 来获得 cpu cache 信息。

参考代码：vendor/github.com/google/cadvisor/info/v1/machine.go

funcGetTopology(sysFssysfs.SysFs,cpuinfostring)([]info.Node,int,error){
nodes:=[]info.Node{}

...
returnnodes,numCores,nil
}

创建 pod 过程

对于前面提到的 static policy 情况下 Container 如何创建呢？kubelet 会为其选择约定的 cpu affinity 来为其选择最佳的 CPU Set。

Container 的创建时 CPU Manager 工作流程大致下：

1. Kuberuntime 调用容器运行时去创建容器。
2. Kuberuntime 将容器传递给 CPU Manager 处理。
3. CPU Manager 为 Container 按照静态策略进行处理。
4. CPU Manager 从当前 Shared Pool 中选择“最佳”Set 拓结构的 CPU，对于不满 Static Policy 的 Contianer，则返回 Shared Pool 中所有 CPU 组合的 Set。
5. CPU Manager 将针对容器的 CPUs 分配情况记录到 Checkpoint State 中，并从 Shared Pool 中删除刚刚分配的 CPUs。
6. CPU Manager 再从 state 中读取该 Container 的 CPU 分配信息，然后通过 UpdateContainerResources cRI 接口将其更新到 Cpuset Cgroups 中，包例如对于非 Static Policy Container。
7. Kuberuntime 调用容器运行时启动该容器。

参考代码：pkg/kubelet/cm/cpumanager/cpu_manager.go

func(m*manager)AddContainer(pod*v1.Pod,container*v1.Container,containerIDstring){
m.Lock()
deferm.Unlock()
ifcset,exists:=m.state.GetCPUSet(string(pod.UID),container.Name);exists{
m.lastUpdateState.SetCPUSet(string(pod.UID),container.Name,cset)
}
m.containerMap.Add(string(pod.UID),container.Name,containerID)
}

参考代码：pkg/kubelet/cm/cpumanager/policy_static.go

funcNewStaticPolicy(topology*topology.CPUTopology,numReservedCPUsint,reservedCPUscpuset.CPUSet,affinitytopologymanager.Store,cpuPolicyOptionsmap[string]string)(Policy,error){
opts,err:=NewStaticPolicyOptions(cpuPolicyOptions)
iferr!=nil{
returnnil,err
}

klog.InfoS("Staticpolicycreatedwithconfiguration","options",opts)

policy:=staticPolicy{
topology:topology,
affinity:affinity,
cpusToReuse:make(map[string]cpuset.CPUSet),
options:opts,
}

allCPUs:=topology.CPUDetails.CPUs()
varreservedcpuset.CPUSet
ifreservedCPUs.Size()>0{
reserved=reservedCPUs
}else{
//takeByTopologyallocatesCPUsassociatedwithlow-numberedcoresfrom
//allCPUs.
//
//Forexample:Givenasystemwith8CPUsavailableandHTenabled,
//ifnumReservedCPUs=2,thenreserved={0,4}
reserved,_=policy.takeByTopology(allCPUs,numReservedCPUs)
}

ifreserved.Size()!=numReservedCPUs{
err:=fmt.Errorf("[cpumanager]unabletoreservetherequiredamountofCPUs(sizeof%sdidnotequal%d)",reserved,numReservedCPUs)
returnnil,err
}

klog.InfoS("ReservedCPUsnotavailableforexclusiveassignment","reservedSize",reserved.Size(),"reserved",reserved)
policy.reserved=reserved

returnpolicy,nil
}

func(p*staticPolicy)Allocate(sstate.State,pod*v1.Pod,container*v1.Container)error{
ifnumCPUs:=p.guaranteedCPUs(pod,container);numCPUs!=0{
klog.InfoS("Staticpolicy:Allocate","pod",klog.KObj(pod),"containerName",container.Name)
//containerbelongsinanexclusivelyallocatedpool

ifp.options.FullPhysicalCPUsOnly((numCPUs%p.topology.CPUsPerCore())!=0){
//SinceCPUManagerhasbeenenabledrequestingstrictSMTalignment,itmeansaguaranteedpodcanonlybeadmitted
//iftheCPUrequestedisamultipleofthenumberofvirtualcpusperphysicalcores.
//IncaseCPUrequestisnotamultipleofthenumberofvirtualcpusperphysicalcoresthePodwillbeput
//inFailedstate,withSMTAlignmentErrorasreason.Sincetheallocationhappensintermsofphysicalcores
//andtheschedulerisresponsibleforensuringthattheworkloadgoestoanodethathasenoughCPUs,
//thepodwouldbeplacedonanodewherethereareenoughphysicalcoresavailabletobeallocated.
//Justlikethebehaviourincaseofstaticpolicy,takeByTopologywilltrytofirstallocateCPUsfromthesamesocket
//andonlyincasetherequestcannotbesattisfiedonasinglesocket,CPUallocationisdoneforaworkloadtooccupyall
//CPUsonaphysicalcore.Allocationofindividualthreadswouldneverhavetooccur.
returnSMTAlignmentError{
RequestedCPUs:numCPUs,
CpusPerCore:p.topology.CPUsPerCore(),
}
}
ifcpuset,ok:=s.GetCPUSet(string(pod.UID),container.Name);ok{
p.updateCPUsToReuse(pod,container,cpuset)
klog.InfoS("Staticpolicy:containeralreadypresentinstate,skipping","pod",klog.KObj(pod),"containerName",container.Name)
returnnil
}

//CallTopologyManagertogetthealignedsocketaffinityacrossallhintproviders.
hint:=p.affinity.GetAffinity(string(pod.UID),container.Name)
klog.InfoS("TopologyAffinity","pod",klog.KObj(pod),"containerName",container.Name,"affinity",hint)

//AllocateCPUsaccordingtotheNUMAaffinitycontainedinthehint.
cpuset,err:=p.allocateCPUs(s,numCPUs,hint.NUMANodeAffinity,p.cpusToReuse[string(pod.UID)])
iferr!=nil{
klog.ErrorS(err,"UnabletoallocateCPUs","pod",klog.KObj(pod),"containerName",container.Name,"numCPUs",numCPUs)
returnerr
}
s.SetCPUSet(string(pod.UID),container.Name,cpuset)
p.updateCPUsToReuse(pod,container,cpuset)

}
//containerbelongsinthesharedpool(nothingtodo;usedefaultcpuset)
returnnil
}

func(p*staticPolicy)allocateCPUs(sstate.State,numCPUsint,numaAffinitybitmask.BitMask,reusableCPUscpuset.CPUSet)(cpuset.CPUSet,error){
klog.InfoS("AllocateCPUs","numCPUs",numCPUs,"socket",numaAffinity)

allocatableCPUs:=p.GetAllocatableCPUs(s).Union(reusableCPUs)

//IftherearealignedCPUsinnumaAffinity,attempttotakethosefirst.
result:=cpuset.NewCPUSet()
ifnumaAffinity!=nil{
alignedCPUs:=cpuset.NewCPUSet()
for_,numaNodeID:=rangenumaAffinity.GetBits(){
alignedCPUs=alignedCPUs.Union(allocatableCPUs.Intersection(p.topology.CPUDetails.CPUsInNUMANodes(numaNodeID)))
}

numAlignedToAlloc:=alignedCPUs.Size()
ifnumCPUs<numAlignedToAlloc{
numAlignedToAlloc=numCPUs
}

alignedCPUs,err:=p.takeByTopology(alignedCPUs,numAlignedToAlloc)
iferr!=nil{
returncpuset.NewCPUSet(),err
}

result=result.Union(alignedCPUs)
}

//GetanyremainingCPUsfromwhat'sleftoverafterattemptingtograbalignedones.
remainingCPUs,err:=p.takeByTopology(allocatableCPUs.Difference(result),numCPUs-result.Size())
iferr!=nil{
returncpuset.NewCPUSet(),err
}
result=result.Union(remainingCPUs)

//RemoveallocatedCPUsfromthesharedCPUSet.
s.SetDefaultCPUSet(s.GetDefaultCPUSet().Difference(result))

klog.InfoS("AllocateCPUs","result",result)
returnresult,nil
}

删除 pod 过程

当这些通过 CPU Managers 分配 CPUs 的 Container 要删除时，CPU Manager 工作流大致如下：

1. Kuberuntime 会调用 CPU Manager 去按静态策略中定义分发处理。
2. CPU Manager 将容器分配的 Cpu Set 重新归还到 Shared Pool 中。
3. Kuberuntime 调用容器运行时移除该容器。
4. CPU Manager 会异步进行协调循环，为使用共享池中的 Cpus 容器更新 CPU 集合。

参考代码：pkg/kubelet/cm/cpumanager/cpu_manager.go

func(m*manager)RemoveContainer(containerIDstring)error{
m.Lock()
deferm.Unlock()

err:=m.policyRemoveContainerByID(containerID)
iferr!=nil{
klog.ErrorS(err,"RemoveContainererror")
returnerr
}

returnnil
}

参考代码：pkg/kubelet/cm/cpumanager/policy_static.go

func(p*staticPolicy)RemoveContainer(sstate.State,podUIDstring,containerNamestring)error{
klog.InfoS("Staticpolicy:RemoveContainer","podUID",podUID,"containerName",containerName)
iftoRelease,ok:=s.GetCPUSet(podUID,containerName);ok{
s.Delete(podUID,containerName)
//Mutatethesharedpool,addingreleasedcpus.
s.SetDefaultCPUSet(s.GetDefaultCPUSet().Union(toRelease))
}
returnnil
}

处理方法

知道了异常的原因和以及具体原因，解决办法也非常好弄就两步：

1. 删除原有 cpu_manager_state 文件
2. 重启 kubelet