今日目标
基于 Cgroup 实现资源控制
如何实现
参考 k8s 官方文档, 有以下两种驱动
- cgroupfs
- systemd
cgroupfs 直接操作 /sys/fs/cgroup 下的虚拟文件接口, 而 systemd 则通过 D-Bus 总线与 systemd 通信，将操作 /sys/fs/cgroup 的任务委托给它
为了降低项目复杂度，目前只计划实现 cgroupfs 驱动且不考虑适配 Cgroup v1
将 Cgroup 控制的资源抽象为接口
type Resource interface {
// CgroupFile 返回 cgroup v2 控制器文件名（如 "cpu.weight"）。
CgroupFile() string
// CgroupValue 返回要写入文件的格式化字符串。
CgroupValue() string
// Validate 做边界检查，失败返回 ErrInvalidResource。
Validate() error
}
每一种资源最终都会变为字符流(CgroupValue())写入 cgroup 叶子节点对应的文件(CgroupFile())
- 内存
// MemoryLimit 单位是字节，对应 cgroup v2 的 memory.max。
// 不允许 0（v2 中 0 = "max" 即无限制），用户想解除限制就别传 -m。
type MemoryLimit int64
func (m MemoryLimit) CgroupFile() string { return "memory.max" }
func (m MemoryLimit) CgroupValue() string { return strconv.FormatInt(int64(m), 10) }
func (m MemoryLimit) Validate() error {
if m <= 0 {
return fmt.Errorf("%w: --memory must be > 0 bytes, got %d",
ErrInvalidResource, m)
}
return nil
}
- CPU占用率
// CPUMax 描述 cpu.max 的一行：每周期的 CPU 时间配额和周期长度。
// 对应 cgroup v2 cpu.max 文件格式 "$MAX $PERIOD"（单位均为微秒）。
type CPUMax struct {
Quota uint64 // 0 表示 "max"（不限制）
Period uint64 // 周期长度（微秒）
}
func (c CPUMax) CgroupFile() string { return "cpu.max" }
func (c CPUMax) CgroupValue() string {
if c.Quota == 0 {
return fmt.Sprintf("max %d", c.Period)
}
return fmt.Sprintf("%d %d", c.Quota, c.Period)
}
func (c CPUMax) Validate() error {
if c.Period == 0 {
return fmt.Errorf("%w: cpu-max period must be > 0", ErrInvalidResource)
}
return nil
}
驱动接口
// Driver 是 cgroup 后端抽象。cgroupfs 和 systemd 各自实现一份。
//
// 接口维度（不是底层路径）：
// - Create: 申请容器所属的 cgroup（scope 对 systemd 来说是 lazy：在 AddPID 时才创建）
// - Apply: 写入 Resource 指定的资源限额
// - AddPID: 把进程加进 cgroup；host PID 而不是容器内 PID
// - Remove: 销毁 cgroup；可恢复错误（残留进程）记 warning 不 fatal
type Driver interface {
Create(id, parent string) error
Apply(id, parent string, resources ...Resource) error
AddPID(id, parent string, pid int) error
Remove(id, parent string) error
Name() string
}
cgroupfs 实现
// cgroupRootFS 是 cgroup v2 统一层级的默认挂载点。
const cgroupRootFS = "/sys/fs/cgroup"
// cgroupParent 是 cgroupfs 驱动创建的父 cgroup 名（位于 cgroupRootFS 之下）。
// 类似 Docker 把所有容器 cgroup 放在 docker/ 之下。
const cgroupParent = "mydocker"
// cgroupfsDriver 直接读写 /sys/fs/cgroup 的实现。
// 所有方法都是 host-namespace 视角；调用者负责传宿主机 PID（不是容器内 PID）。
type cgroupfsDriver struct {
root string // "/sys/fs/cgroup"
log *slog.Logger
}
// NewCgroupfsDriver 在 v2 不可用时返回 ErrCgroupV2Unavailable，让上层 fail-fast。
func NewCgroupfsDriver() (Driver, error) {
if !CgroupV2Available() {
return nil, ErrCgroupV2Unavailable
}
return &cgroupfsDriver{root: cgroupRootFS, log: slog.Default()}, nil
}
// leafPath 拼出 id 对应的 cgroup 目录绝对路径。
// 我们不暴露 parent 参数——所有 cgroupfs 容器都挂在 cgroupParent 之下，
// 避免用户传任意路径把已存在的 cgroup 当成"父"用。
// cgroup v2 规定资源必须在叶子节点写入
func (d *cgroupfsDriver) leafPath(id string) string {
return filepath.Join(d.root, cgroupParent, id)
}
// Create 建出 cgroup 目录，并在父 cgroup 中开启 cpu/memory/pids 三个控制器。
//
// 关键：v2 要求父 cgroup 的 cgroup.subtree_control 显式 enable 对应 controller，
// 子 cgroup 才能拿到对应的文件。Create 每次都做幂等检查——只补缺失的。
func (d *cgroupfsDriver) Create(id, parent string) error {
if err := d.ensureParentCgroup(); err != nil {
return fmt.Errorf("ensure parent cgroup: %w", err)
}
leaf := d.leafPath(id)
if err := os.Mkdir(leaf, 0o755); err != nil && !errors.Is(err, os.ErrExist) {
return fmt.Errorf("mkdir %s: %w", leaf, err)
}
d.log.Info("cgroup created", "path", leaf)
return nil
}
// ensureParentCgroup 幂等创建父 cgroup 并补齐 subtree_control。
// 多次创建同名父 cgroup 是预期的——Docker daemon 每次起也只创建一次，
// 但我们这里做成幂等便于 mydocker 多次 run。
func (d *cgroupfsDriver) ensureParentCgroup() error {
parentDir := filepath.Join(d.root, cgroupParent)
if err := os.Mkdir(parentDir, 0o755); err != nil && !errors.Is(err, os.ErrExist) {
return fmt.Errorf("mkdir %s: %w", parentDir, err)
}
stPath := filepath.Join(parentDir, "cgroup.subtree_control")
current, err := os.ReadFile(stPath)
if err != nil {
// 第一次创建时文件可能还没刷出来；等一下再读。
return fmt.Errorf("read subtree_control: %w", err)
}
want := []string{"cpu", "memory", "pids"}
var toEnable []string
for _, w := range want {
if !strings.Contains(string(current), w) {
toEnable = append(toEnable, "+"+w)
}
}
if len(toEnable) == 0 {
return nil
}
// v2 的 subtree_control 一次接受空格分隔的 +list；不能重复 + 同名。
if err := os.WriteFile(stPath, []byte(strings.Join(toEnable, " ")), 0o644); err != nil {
return fmt.Errorf("enable controllers %v: %w", toEnable, err)
}
d.log.Info("cgroup subtree_control updated", "enabled", toEnable)
return nil
}
// Apply 把各项 Resource 落到 cgroup 控制器文件上。
// 每个 Resource 自行提供文件名和值，driver 只负责写入。
func (d *cgroupfsDriver) Apply(id, parent string, resources ...Resource) error {
if len(resources) == 0 {
return nil
}
leaf := d.leafPath(id)
for _, r := range resources {
if err := writeCgroupFile(leaf, r.CgroupFile(), r.CgroupValue()); err != nil {
return fmt.Errorf("set %s=%s: %w", r.CgroupFile(), r.CgroupValue(), err)
}
}
return nil
}
// AddPID 把宿主机视角的 pid 写入 cgroup.procs。
// 必须是 host PID——子进程在新 PID namespace 里的 1 在 cgroup 文件看来是同一个 task_struct。
func (d *cgroupfsDriver) AddPID(id, parent string, pid int) error {
leaf := d.leafPath(id)
if err := writeCgroupFile(leaf, "cgroup.procs", strconv.Itoa(pid)); err != nil {
return fmt.Errorf("add pid %d to cgroup.procs: %w", pid, err)
}
return nil
}
// Remove 删除 cgroup 目录。v2 要求目录为空才能 rmdir；
// 我们最多等 1 秒让 cgroup.events 显示 populated=0，再尝试。
// 真的还有残留进程（罕见，例如子进程 fork 了一个不归我们的进程且还活着），
// 打 warning 不报错——cgroup 会泄漏但下一次 mydocker run 还能用。
func (d *cgroupfsDriver) Remove(id, parent string) error {
leaf := d.leafPath(id)
// 等 1 秒：覆盖 Wait() 返回后 cgroup.procs 还没刷掉的窗口。
deadline := time.Now().Add(1 * time.Second)
for time.Now().Before(deadline) {
if isCgroupEmpty(leaf) {
break
}
time.Sleep(20 * time.Millisecond)
}
if err := os.Remove(leaf); err != nil {
if errors.Is(err, os.ErrNotExist) {
return nil
}
// EBUSY 说明还有进程：尝试 cgroup.kill（kernel 5.14+），再 rmdir。
if errors.Is(err, syscall.EBUSY) {
d.tryCgroupKill(leaf)
if err2 := os.Remove(leaf); err2 == nil {
return nil
}
}
d.log.Warn("cgroup remove failed; leaking", "path", leaf, "err", err)
return nil
}
return nil
}
// tryCgroupKill 写 1 到 cgroup.kill 让内核 SIGKILL 掉残留进程。
// 内核 < 5.14 不存在该文件，直接忽略。
func (d *cgroupfsDriver) tryCgroupKill(leaf string) {
f, err := os.OpenFile(filepath.Join(leaf, "cgroup.kill"), os.O_WRONLY, 0)
if err != nil {
return
}
defer f.Close()
_, _ = f.WriteString("1")
}
// isCgroupEmpty 读 cgroup.events 的 populated 字段判断目录是否还有进程。
// v2 标准接口，kernel 5.7+ 强制存在。
func isCgroupEmpty(leaf string) bool {
data, err := os.ReadFile(filepath.Join(leaf, "cgroup.events"))
if err != nil {
// 读不到就保守认为非空，避免 rmdir 撞 EBUSY。
return false
}
// 格式: "populated 0\n" 或 "populated 1\n"
for line := range strings.SplitSeq(string(data), "\n") {
if strings.HasPrefix(line, "populated ") {
return strings.HasSuffix(line, " 0")
}
}
return false
}
// writeCgroupFile 是 cgroup 文件写入的统一入口。
// 内部走 internal.WriteCgroupValue：单次写入 + read-back 校验。
// 读不到写入值时重试 maxRetries 次（见 internal/atomic.go）。
func writeCgroupFile(leaf, name, value string) error {
return internal.WriteCgroupValue(leaf, name, value)
}
// Name 满足 Driver 接口。
func (d *cgroupfsDriver) Name() string { return "cgroupfs" }
var _ = Driver(&cgroupfsDriver{})