本文介绍如何在不依赖 kubectl 的前提下，通过读取 Linux cgroups（v1/v2）接口，在容器内部动态获取 Kubernetes 为当前 Pod 分配的 CPU 核心数，并提供健壮的 Python 实现与关键注意事项。
本文介绍如何在不依赖 `kubectl` 的前提下，通过读取 linux cgroups（v1/v2）接口，在容器内部动态获取 kubernetes 为当前 pod 分配的 cpu 核心数，并提供健壮的 python 实现与关键注意事项。
在 Kubernetes 中，Pod 的 CPU 资源限制（limits）和请求（requests）由 kubelet 通过底层 cgroups 机制进行实际管控。由于容器共享宿主机内核，且 cgroups 文件系统默认挂载到 /sys/fs/cgroup/，因此从容器内部直接读取 cgroups 参数是获取分配 CPU 资源最可靠、最轻量的方式——无需网络调用、不依赖集群权限，也完全兼容离线或受限环境。
Kubernetes 支持两种 cgroups 版本，需分别适配：
- cgroups v1：查看 /sys/fs/cgroup/cpu/cpu.cfs_quota_us（Q）和 /sys/fs/cgroup/cpu/cpu.cfs_period_us（P），计算公式为 ceil(Q / P)。例如：Q = 400000, P = 100000 → 4 cores；若 Q = -1，表示无 CPU 限制，应退回到宿主机总核数（如 multiprocessing.cpu_count()）。
- cgroups v2：统一使用 /sys/fs/cgroup/cpu.max，格式为 "Q P"（空格分隔）。当 Q == "max" 时，同样表示无硬性限制，需回退；否则解析整数后同样应用 ceil(Q / P)。
以下是一个生产就绪的 Python 3 实现，具备版本自动探测、路径容错、边界条件处理与清晰回退逻辑：
import typing
import multiprocessing
import math
from pathlib import Path
def get_cpu_cores_via_cgroups() -> typing.Optional[int]:
"""获取当前容器被分配的 CPU 核心数（基于 cgroups v1/v2）"""
cfs_period_path = "/sys/fs/cgroup/cpu/cpu.cfs_period_us"
cfs_quota_path = "/sys/fs/cgroup/cpu/cpu.cfs_quota_us"
cpu_max_path = "/sys/fs/cgroup/cpu.max"
return _get_cpu_cores(cfs_period_path, cfs_quota_path, cpu_max_path)
def _get_cpu_cores(
cfs_period_path: str,
cfs_quota_path: str,
cpu_max_path: str
) -> typing.Optional[int]:
# 优先尝试 cgroups v1
cfs_period = Path(cfs_period_path)
cfs_quota = Path(cfs_quota_path)
if cfs_period.exists() and cfs_quota.exists():
try:
p = int(cfs_period.read_text().strip())
q = int(cfs_quota.read_text().strip())
if q > 0 and p > 0:
return math.ceil(q / p)
elif q == -1: # unlimited in v1
return multiprocessing.cpu_count()
except (ValueError, OSError):
pass
# fallback to cgroups v2
cpu_max = Path(cpu_max_path)
if cpu_max.exists():
try:
line = cpu_max.read_text().strip()
return cpu_cores_from_cgroups_v2_cpu_max(line)
except (OSError, ValueError):
pass
# 最终兜底：返回宿主机总逻辑核数（保守但安全）
return multiprocessing.cpu_count()
def cpu_cores_from_cgroups_v2_cpu_max(line: str) -> typing.Optional[int]:
parts = line.split()
if len(parts) != 2:
return None
quota_str, period_str = parts[0], parts[1]
try:
if quota_str == "max":
return multiprocessing.cpu_count()
q, p = int(quota_str), int(period_str)
return math.ceil(q / p) if q > 0 and p > 0 else None
except ValueError:
return None✅ 关键注意事项：
- 该方法获取的是 CPU limits 对应的硬性配额（即 spec.containers[].resources.limits.cpu），而非 requests；如需 requests，需结合 Downward API 注入环境变量（如 FIELD_REF=spec.containers[0].resources.requests.cpu）。
- 确保容器运行时启用 cgroups（绝大多数主流运行时如 containerd、CRI-O 默认启用）；若使用 --cgroup-driver=systemd，路径可能为 /sys/fs/cgroup/system.slice/...，但 Kubernetes 默认仍挂载标准路径。
- multiprocessing.cpu_count() 返回的是宿主机总逻辑 CPU 数，仅作为无限制时的合理回退，不可用于资源感知调度。
- 在多容器 Pod 中，各容器独立受其自身 cgroups 限制，因此该函数应在目标容器内执行。
掌握此方法，可让应用（如 JVM、NumPy、Ray、Dask）自动适配分配资源，避免过度申请或性能浪费，是云原生应用弹性伸缩与资源感知的关键基础能力。
