我是 LEE翻伺，老李材泄，一個在 IT 行業(yè)摸爬滾打 16 年的技術(shù)老兵。
事件背景
大家都知道 k8s 容量不夠的時候穆趴，都是添加節(jié)點來解決問題脸爱。這幾天有小伙伴在升級 k8s 容量的時候碰到一個問題，他將集群中某一個 node 節(jié)點的 CPU 做了升級未妹，然后重啟了這個 node 節(jié)點導(dǎo)致 kubelet 無法啟動簿废，然后大量 pod 被驅(qū)逐，報警電話響個不停络它。為了緊急恢復(fù)業(yè)務(wù)族檬，果斷參加故障恢復(fù)。
現(xiàn)象獲取
在知道事件背景后化戳，我登上了那個已經(jīng)重啟完畢的 node 節(jié)點单料，開始了一系列的網(wǎng)絡(luò)測試埋凯，確認 node 這個宿主機到 Apiserver 和 Loadbalancer 的 ip 和 port 都是通的。隨后趕緊看了下 kubelet 的日志扫尖，果不其然白对，一行日志讓我看到問題點：
E1121 23:43:52.644552 23453 policy_static.go:158] "Static policy invalid state, please drain node and remove policy state file" err="current set of available CPUs \"0-7\" doesn't match with CPUs in state \"0-3\""
E1121 23:43:52.644569 23453 cpu_manager.go:230] "Policy start error" err="current set of available CPUs \"0-7\" doesn't match with CPUs in state \"0-3\""
E1121 23:43:52.644587 23453 kubelet.go:1431] "Failed to start ContainerManager" err="start cpu manager error: current set of available CPUs \"0-7\" doesn't match with CPUs in state \"0-3\""
說到這里，很多小伙伴會說：“就這换怖？甩恼？”。 真的就這沉颂。是因為啥呢条摸？ 是因為 kubelet 啟動參數(shù)里面有一個參數(shù)很重要：--cpu-manager-policy。表示 kubelet 在使用宿主機的 cpu 是什么邏輯策略铸屉。如果你設(shè)定為 static 钉蒲，那么就會在參數(shù) --root-dir 指定的目錄下生成一個 cpu_manager_state 這樣一個綁定文件。
cpu_manager_state 內(nèi)容大致長得如下：
{ "policyName": "static", "defaultCpuSet": "0-7", "checksum": 14413152 }
當你升級這個 k8s node 節(jié)點的 CPU 配置彻坛，并且使用了 static cpu 管理模式顷啼，那么 kubelet 會讀取 cpu_manager_state 文件，然后跟現(xiàn)有的宿主運行的資源做對比小压，如果不一致线梗，kubelet 就不會啟動了椰于。
原理分析
既然我們看到了具體現(xiàn)象和故障位置怠益，不妨借著這個小問題我們一起開溫馨下 k8s 的 cpu 管理規(guī)范。
官方文檔如下：
https://kubernetes.io/zh-cn/docs/tasks/administer-cluster/cpu-management-policies/
當然我還想多少說點別的瘾婿，關(guān)于 CPU Manager 整個架構(gòu)蜻牢，讓小伙伴們有一個整體理解，能更加深入理解官方的 cpu 管理策略到底是做了些什么動作偏陪。
cpu-management-policies
CPU Manager 架構(gòu)
CPU Manager 為滿足條件的 Container 分配指定的 CPUs 時抢呆，會盡數(shù)按 CPU Topology 來分配，也就是參考 CPU Affinity笛谦，按如下的優(yōu)先順序進行 CPUs 選擇：（Logic CPUs 就是 Hyperthreads）
- 如果 Container 要求的 Logic CPUs 數(shù)量不少于單塊 CPU Socket 中 Logci CPUs 數(shù)量抱虐，那么會優(yōu)先把整塊 CPU Socket 中的 Logic CPUs 分配給 Container。
- 如果 Container 減余請求的 Logic CPU 數(shù)量不少于單塊物理 CPU Core 提供的 Logic CPU 數(shù)量饥脑，那么會優(yōu)先把整塊物理 CPU Core 上的 Logic CPU 分配給 Container恳邀。
Container 托余請求的 Logic CPUs 則從按以下規(guī)則排列好的 Logic CPUs 列表中選擇：
- 同一插槽上可用的 CPU 數(shù)量
- 同一核心上可用的 CPU 數(shù)量
參考代碼： pkg/kubelet/cm/cpumanager/cpu_assignment.go
func takeByTopology(topo *topology.CPUTopology, availableCPUs cpuset.CPUSet, numCPUs int) (cpuset.CPUSet, error) {
acc := newCPUAccumulator(topo, availableCPUs, numCPUs)
if acc.isSatisfied() {
return acc.result, nil
}
if acc.isFailed() {
return cpuset.NewCPUSet(), fmt.Errorf("not enough cpus available to satisfy request")
}
// Algorithm: topology-aware best-fit
// 1. Acquire whole sockets, if available and the container requires at
// least a socket's-worth of CPUs.
for _, s := range acc.freeSockets() {
if acc.needs(acc.topo.CPUsPerSocket()) {
glog.V(4).Infof("[cpumanager] takeByTopology: claiming socket [%d]", s)
acc.take(acc.details.CPUsInSocket(s))
if acc.isSatisfied() {
return acc.result, nil
}
}
}
// 2. Acquire whole cores, if available and the container requires at least
// a core's-worth of CPUs.
for _, c := range acc.freeCores() {
if acc.needs(acc.topo.CPUsPerCore()) {
glog.V(4).Infof("[cpumanager] takeByTopology: claiming core [%d]", c)
acc.take(acc.details.CPUsInCore(c))
if acc.isSatisfied() {
return acc.result, nil
}
}
}
// 3. Acquire single threads, preferring to fill partially-allocated cores
// on the same sockets as the whole cores we have already taken in this
// allocation.
for _, c := range acc.freeCPUs() {
glog.V(4).Infof("[cpumanager] takeByTopology: claiming CPU [%d]", c)
if acc.needs(1) {
acc.take(cpuset.NewCPUSet(c))
}
if acc.isSatisfied() {
return acc.result, nil
}
}
return cpuset.NewCPUSet(), fmt.Errorf("failed to allocate cpus")
}
發(fā)現(xiàn) CPU Topology
參考代碼： vendor/github.com/google/cadvisor/info/v1/machine.go
type MachineInfo struct {
// The number of cores in this machine.
NumCores int `json:"num_cores"`
...
// Machine Topology
// Describes cpu/memory layout and hierarchy.
Topology []Node `json:"topology"`
...
}
type Node struct {
Id int `json:"node_id"`
// Per-node memory
Memory uint64 `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
func GetTopology(sysFs sysfs.SysFs, cpuinfo string) ([]info.Node, int, error) {
nodes := []info.Node{}
...
return nodes, numCores, nil
}
創(chuàng)建 pod 過程
對于前面提到的 static policy 情況下 Container 如何創(chuàng)建呢？kubelet 會為其選擇約定的 cpu affinity 來為其選擇最佳的 CPU Set笋颤。
Container 的創(chuàng)建時 CPU Manager 工作流程大致下：
- Kuberuntime 調(diào)用容器運行時去創(chuàng)建容器乳附。
- Kuberuntime 將容器傳遞給 CPU Manager 處理。
- CPU Manager 為 Container 按照靜態(tài)策略進行處理。
- CPU Manager 從當前 Shared Pool 中選擇“最佳”Set 拓結(jié)構(gòu)的 CPU赋除，對于不滿 Static Policy 的 Contianer阱缓，則返回 Shared Pool 中所有 CPU 組合的 Set。
- CPU Manager 將針對容器的 CPUs 分配情況記錄到 Checkpoint State 中举农，并從 Shared Pool 中刪除剛剛分配的 CPUs茬祷。
- CPU Manager 再從 state 中讀取該 Container 的 CPU 分配信息，然后通過 UpdateContainerResources cRI 接口將其更新到 Cpuset Cgroups 中并蝗，包例如對于非 Static Policy Container祭犯。
- Kuberuntime 調(diào)用容器運行時啟動該容器。
參考代碼： pkg/kubelet/cm/cpumanager/cpu_manager.go
func (m *manager) AddContainer(pod *v1.Pod, container *v1.Container, containerID string) {
m.Lock()
defer m.Unlock()
if cset, 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
func NewStaticPolicy(topology *topology.CPUTopology, numReservedCPUs int, reservedCPUs cpuset.CPUSet, affinity topologymanager.Store, cpuPolicyOptions map[string]string) (Policy, error) {
opts, err := NewStaticPolicyOptions(cpuPolicyOptions)
if err != nil {
return nil, err
}
klog.InfoS("Static policy created with configuration", "options", opts)
policy := &staticPolicy{
topology: topology,
affinity: affinity,
cpusToReuse: make(map[string]cpuset.CPUSet),
options: opts,
}
allCPUs := topology.CPUDetails.CPUs()
var reserved cpuset.CPUSet
if reservedCPUs.Size() > 0 {
reserved = reservedCPUs
} else {
// takeByTopology allocates CPUs associated with low-numbered cores from
// allCPUs.
//
// For example: Given a system with 8 CPUs available and HT enabled,
// if numReservedCPUs=2, then reserved={0,4}
reserved, _ = policy.takeByTopology(allCPUs, numReservedCPUs)
}
if reserved.Size() != numReservedCPUs {
err := fmt.Errorf("[cpumanager] unable to reserve the required amount of CPUs (size of %s did not equal %d)", reserved, numReservedCPUs)
return nil, err
}
klog.InfoS("Reserved CPUs not available for exclusive assignment", "reservedSize", reserved.Size(), "reserved", reserved)
policy.reserved = reserved
return policy, nil
}
func (p *staticPolicy) Allocate(s state.State, pod *v1.Pod, container *v1.Container) error {
if numCPUs := p.guaranteedCPUs(pod, container); numCPUs != 0 {
klog.InfoS("Static policy: Allocate", "pod", klog.KObj(pod), "containerName", container.Name)
// container belongs in an exclusively allocated pool
if p.options.FullPhysicalCPUsOnly && ((numCPUs % p.topology.CPUsPerCore()) != 0) {
// Since CPU Manager has been enabled requesting strict SMT alignment, it means a guaranteed pod can only be admitted
// if the CPU requested is a multiple of the number of virtual cpus per physical cores.
// In case CPU request is not a multiple of the number of virtual cpus per physical cores the Pod will be put
// in Failed state, with SMTAlignmentError as reason. Since the allocation happens in terms of physical cores
// and the scheduler is responsible for ensuring that the workload goes to a node that has enough CPUs,
// the pod would be placed on a node where there are enough physical cores available to be allocated.
// Just like the behaviour in case of static policy, takeByTopology will try to first allocate CPUs from the same socket
// and only in case the request cannot be sattisfied on a single socket, CPU allocation is done for a workload to occupy all
// CPUs on a physical core. Allocation of individual threads would never have to occur.
return SMTAlignmentError{
RequestedCPUs: numCPUs,
CpusPerCore: p.topology.CPUsPerCore(),
}
}
if cpuset, ok := s.GetCPUSet(string(pod.UID), container.Name); ok {
p.updateCPUsToReuse(pod, container, cpuset)
klog.InfoS("Static policy: container already present in state, skipping", "pod", klog.KObj(pod), "containerName", container.Name)
return nil
}
// Call Topology Manager to get the aligned socket affinity across all hint providers.
hint := p.affinity.GetAffinity(string(pod.UID), container.Name)
klog.InfoS("Topology Affinity", "pod", klog.KObj(pod), "containerName", container.Name, "affinity", hint)
// Allocate CPUs according to the NUMA affinity contained in the hint.
cpuset, err := p.allocateCPUs(s, numCPUs, hint.NUMANodeAffinity, p.cpusToReuse[string(pod.UID)])
if err != nil {
klog.ErrorS(err, "Unable to allocate CPUs", "pod", klog.KObj(pod), "containerName", container.Name, "numCPUs", numCPUs)
return err
}
s.SetCPUSet(string(pod.UID), container.Name, cpuset)
p.updateCPUsToReuse(pod, container, cpuset)
}
// container belongs in the shared pool (nothing to do; use default cpuset)
return nil
}
func (p *staticPolicy) allocateCPUs(s state.State, numCPUs int, numaAffinity bitmask.BitMask, reusableCPUs cpuset.CPUSet) (cpuset.CPUSet, error) {
klog.InfoS("AllocateCPUs", "numCPUs", numCPUs, "socket", numaAffinity)
allocatableCPUs := p.GetAllocatableCPUs(s).Union(reusableCPUs)
// If there are aligned CPUs in numaAffinity, attempt to take those first.
result := cpuset.NewCPUSet()
if numaAffinity != nil {
alignedCPUs := cpuset.NewCPUSet()
for _, numaNodeID := range numaAffinity.GetBits() {
alignedCPUs = alignedCPUs.Union(allocatableCPUs.Intersection(p.topology.CPUDetails.CPUsInNUMANodes(numaNodeID)))
}
numAlignedToAlloc := alignedCPUs.Size()
if numCPUs < numAlignedToAlloc {
numAlignedToAlloc = numCPUs
}
alignedCPUs, err := p.takeByTopology(alignedCPUs, numAlignedToAlloc)
if err != nil {
return cpuset.NewCPUSet(), err
}
result = result.Union(alignedCPUs)
}
// Get any remaining CPUs from what's leftover after attempting to grab aligned ones.
remainingCPUs, err := p.takeByTopology(allocatableCPUs.Difference(result), numCPUs-result.Size())
if err != nil {
return cpuset.NewCPUSet(), err
}
result = result.Union(remainingCPUs)
// Remove allocated CPUs from the shared CPUSet.
s.SetDefaultCPUSet(s.GetDefaultCPUSet().Difference(result))
klog.InfoS("AllocateCPUs", "result", result)
return result, nil
}
刪除 pod 過程
當這些通過 CPU Managers 分配 CPUs 的 Container 要刪除時滚停，CPU Manager 工作流大致如下：
- Kuberuntime 會調(diào)用 CPU Manager 去按靜態(tài)策略中定義分發(fā)處理沃粗。
- CPU Manager 將容器分配的 Cpu Set 重新歸還到 Shared Pool 中。
- Kuberuntime 調(diào)用容器運行時移除該容器键畴。
- CPU Manager 會異步進行協(xié)調(diào)循環(huán)最盅，為使用共享池中的 Cpus 容器更新 CPU 集合。
參考代碼： pkg/kubelet/cm/cpumanager/cpu_manager.go
func (m *manager) RemoveContainer(containerID string) error {
m.Lock()
defer m.Unlock()
err := m.policyRemoveContainerByID(containerID)
if err != nil {
klog.ErrorS(err, "RemoveContainer error")
return err
}
return nil
}
參考代碼： pkg/kubelet/cm/cpumanager/policy_static.go
func (p *staticPolicy) RemoveContainer(s state.State, podUID string, containerName string) error {
klog.InfoS("Static policy: RemoveContainer", "podUID", podUID, "containerName", containerName)
if toRelease, ok := s.GetCPUSet(podUID, containerName); ok {
s.Delete(podUID, containerName)
// Mutate the shared pool, adding released cpus.
s.SetDefaultCPUSet(s.GetDefaultCPUSet().Union(toRelease))
}
return nil
}
處理方法
知道了異常的原因和以及具體原因起惕，解決辦法也非常好弄就兩步：
- 刪除原有 cpu_manager_state 文件
- 重啟 kubelet
