啃 K8s 源码：一个 Pod 到底是怎么存进 etcd 的，从 RESTStorage.Create 到事务 Put 全流程

前言：那个让我看了三天源码的诡异 bug

生产环境有次报 Pod “Pending”，但 kubectl describe 一看 status 字段完全是空的——不是 Pending、不是 Running，是字面意义上的空 status。

调度器拒绝调度（因为 status.phase 不对），controller 也不更新——整个 Pod 就这么卡死在 etcd 里。

排查到最后发现是定制的 mutating webhook 把 pod.Status.Phase 给改成了 ""（空字符串），APIServer 写存储时保留了这个值，导致 controller 集体迷惑。

让我意外的是：APIServer 在写 etcd 前，明明有一道 PrepareForCreate 把 status 设为 PodPending 的逻辑——为什么没起作用？

啃了三天源码才搞明白：webhook 是在 admission 阶段改的，已经过了 PrepareForCreate。APIServer 的执行顺序是有讲究的——准入控制（webhook）跑在 strategy 之后。这次踩坑让我彻底搞清楚了从 RESTStorage.Create 到 etcd Put 之间到底发生了什么。

这篇就把这条链路逐行拆开讲，并把每个容易踩坑的点标出来。

本节重点

• kube-apiserver create Pod 时数据保存的完整链路
• 从 RESTStorage.Create 到 etcd Txn().Put() 之间的核心步骤
• 隐藏在 PrepareForCreate 里的 QoS 计算逻辑
• dryRun、加密 Transformer、事务一致性的实现细节

一、整体链路总览

先看一张全景图，知道我们要走哪些站：

APIServer 收到 POST /api/v1/namespaces/default/pods
│
▼
┌──────────────────────────────────┐
│ HTTP handler chain │
│ (认证 → 鉴权 → 准入控制 webhook) │
└─────────────┬────────────────────┘
▼
┌──────────────────────────────────┐
│ podStorage.Create() │
│ (PodStorage 的 REST.Create) │
└─────────────┬────────────────────┘
▼
┌──────────────────────────────────┐
│ genericregistry.Store.Create() │ ◀── 本节重点
│ ① BeginCreate (可选钩子) │
│ ② BeforeCreate │
│ └─ Strategy.PrepareForCreate │
│ └─ 设 Pending、算 QoS │
│ ③ Validate │
│ ④ Storage.Create (写 etcd) │
│ ⑤ AfterCreate / Decorator │
└─────────────┬────────────────────┘
▼
┌──────────────────────────────────┐
│ DryRunnableStorage.Create │
│ └─ dryRun 拦截层 │
└─────────────┬────────────────────┘
▼
┌──────────────────────────────────┐
│ etcd3.store.Create │
│ ① 序列化 (runtime.Encode) │
│ ② Transformer.TransformToStorage│ ← 加密在这
│ ③ Txn().If(notFound).Put() │ ← 事务 Put
└─────────────┬────────────────────┘
▼
etcd v3 存储

接下来一段一段拆。

二、Pod 的 RESTStorage 长什么样？

上节讲到每种资源对应一个 RESTStorage，定义了"如何跟存储打交道"。Pod 的位置：pkg/registry/core/pod/storage/storage.go

// REST implements a RESTStorage for pods
type REST struct {
*genericregistry.Store // 嵌入通用 Store
proxyTransport http.RoundTripper // exec/log 用的代理
}

主 store 的初始化（上节看过的代码）：

store := &genericregistry.Store{
NewFunc: func() runtime.Object { return &api.Pod{} },
NewListFunc: func() runtime.Object { return &api.PodList{} },
PredicateFunc: registrypod.MatchPod,
DefaultQualifiedResource: api.Resource("pods"),

CreateStrategy: registrypod.Strategy, // ★ 关键
UpdateStrategy: registrypod.Strategy,
DeleteStrategy: registrypod.Strategy,
ResetFieldsStrategy: registrypod.Strategy,
ReturnDeletedObject: true,

TableConvertor: printerstorage.TableConvertor{
TableGenerator: printers.NewTableGenerator().With(printersinternal.AddHandlers),
},
}

重点是 CreateStrategy: registrypod.Strategy——Pod 创建时所有"业务逻辑"都在这个 Strategy 里。

💡 Strategy 是什么？ 上节讲过，Strategy 模式让通用 Store 处理 CRUD，资源专属的校验/默认值/转换通过 Strategy 注入。Pod 的 Strategy 就在 pkg/registry/core/pod/strategy.go。

三、Store.Create 主流程逐行解读

位置：staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go

这是 K8s 所有资源 Create 的统一入口——Pod、Service、ConfigMap、CRD 都走这里。

3.1 第①步：BeginCreate（事务前钩子）

if e.BeginCreate != nil {
fn, err := e.BeginCreate(ctx, obj, options)
if err != nil {
return nil, err
}
finishCreate = fn
defer func() {
finishCreate(ctx, false)
}()
}

BeginCreate 是个可选的事务钩子，让 Strategy 能"开启一个事务"，并在 defer 里收尾（成功 commit 或失败 rollback）。

💡 大部分资源没有 BeginCreate。它主要给一些需要跨资源事务的场景留的口子，比如某些 Aggregated APIServer 实现里会用。Pod 自己没用这个。

3.2 第②步：BeforeCreate（PrepareForCreate + Validate）

if err := rest.BeforeCreate(e.CreateStrategy, ctx, obj); err != nil {
return nil, err
}

这个 BeforeCreate 不是简单一行——它内部干了4 件事，看下简化版源码：

// 简化版的 BeforeCreate 实现
func BeforeCreate(strategy RESTCreateStrategy, ctx context.Context, obj runtime.Object) error {
// ① 调 Strategy 的 PrepareForCreate（补默认值、改字段）
strategy.PrepareForCreate(ctx, obj)

// ② 生成 UID
if err := EnsureObjectMeta(obj); err != nil { … }

// ③ 调 Strategy.Validate（业务校验）
if errs := strategy.Validate(ctx, obj); len(errs) > 0 {
return errors.NewInvalid(…)
}

// ④ 处理 Canonicalize（标准化字段）
strategy.Canonicalize(obj)

return nil
}

🚨 开头那个 bug 就在这步发生：PrepareForCreate 把 status 设为 Pending 没错，但它在准入控制 webhook 之前跑。我们的 webhook 又在 PrepareForCreate 之后把 status 改回空——APIServer 信任 webhook，不会再纠正。

正确做法：webhook 应该只改 spec，不要碰 status。需要改 status 应该等 Pod 创建后通过 /status 子资源单独改。

3.3 Pod Strategy.PrepareForCreate 干了什么

位置：pkg/registry/core/pod/strategy.go

func (podStrategy) PrepareForCreate(ctx context.Context, obj runtime.Object) {
pod := obj.(*api.Pod)

// 1. 强制设状态为 Pending
pod.Status = api.PodStatus{
Phase: api.PodPending,
QOSClass: qos.GetPodQOS(pod), // ★ 关键：算 QoS
}

// 2. 丢掉禁用 feature gate 的字段
podutil.DropDisabledPodFields(pod, nil)

// 3. 处理 seccomp 字段的版本兼容
applySeccompVersionSkew(pod)
}

三件事：

💡 DropDisabledPodFields 是 K8s 的"防穿越"机制：如果你的集群关了某个 feature，但有人传了相关字段，APIServer 会静默丢弃而不是报错。好处是新客户端能向后兼容老集群；坏处是字段被丢了你都不知道——所以排查"字段莫名其妙消失"问题时，先检查 feature gate。

四、深入：QoS 是怎么算出来的？

Pod 的 QoS 分类（Guaranteed / Burstable / BestEffort）就是在 PrepareForCreate 里被打上去的。

4.1 QoS 三档简介

BestEffort ←────── 优先级递增 ──────→ Guaranteed

💡 QoS 的两个本质影响：

4.2 GetPodQOS 源码

位置：pkg/apis/core/helper/qos/qos.go

第一步：遍历所有容器，累加 requests：

for _, container := range allContainers {
// process requests
for name, quantity := range container.Resources.Requests {
if !isSupportedQoSComputeResource(name) {
continue
}
if quantity.Cmp(zeroQuantity) == 1 {
delta := quantity.DeepCopy()
if _, exists := requests[name]; !exists {
requests[name] = delta
} else {
delta.Add(requests[name])
requests[name] = delta
}
}
}

第二步：累加 limits：

// process limits
qosLimitsFound := sets.NewString()
for name, quantity := range container.Resources.Limits {
if !isSupportedQoSComputeResource(name) {
continue
}
if quantity.Cmp(zeroQuantity) == 1 {
qosLimitsFound.Insert(string(name))
delta := quantity.DeepCopy()
if _, exists := limits[name]; !exists {
limits[name] = delta
} else {
delta.Add(limits[name])
limits[name] = delta
}
}
}
}

第三步：判定 QoS 类：

// 规则 1: 都没设 → BestEffort
if len(requests) == 0 && len(limits) == 0 {
return core.PodQOSBestEffort
}

// 规则 2: limits == requests 且齐全 → Guaranteed
if isGuaranteed && len(requests) == len(limits) {
return core.PodQOSGuaranteed
}

// 规则 3: 其他 → Burstable
return core.PodQOSBurstable

4.3 实战例子

# 例 1: Guaranteed
resources:
requests:
cpu: 100m
memory: 100Mi
limits:
cpu: 100m # 和 requests 完全相等
memory: 100Mi

# 例 2: Burstable（最常见）
resources:
requests:
cpu: 100m
memory: 100Mi
limits:
cpu: 1000m # limits > requests
memory: 2500Mi

# 例 3: BestEffort
# 完全不写 resources 或者 requests/limits 全为 0

🚨 生产实战建议：

• 核心业务务必 Guaranteed：保证 OOM 时最后被杀。把 requests 和 limits 调成一样，付出的代价是预留资源稍多
• 大部分应用用 Burstable 即可，灵活+省资源
• 永远不要用 BestEffort 跑生产应用——节点稍微紧张就会被杀，且没有任何 SLA 保障

🤔 冷知识：很多人不知道，requests == limits 时还有个隐藏福利——kubelet 会启用 CPU 静态分配策略（static policy），把 CPU 核心独占绑定，避免上下文切换抖动。这对延迟敏感的服务（如交易系统）很重要。

五、第③④步：Storage.Create 写入存储

回到 Store.Create 主流程：

out := e.NewFunc()
if err := e.Storage.Create(ctx, key, obj, out, ttl, dryrun.IsDryRun(options.DryRun)); err != nil {
err = storeerr.InterpretCreateError(err, qualifiedResource, name)
err = rest.CheckGeneratedNameError(ctx, e.CreateStrategy, err, obj)
if !apierrors.IsAlreadyExists(err) {
klog.Warningf("failed to create %s: %v", qualifiedResource, err)
}
return nil, err
}

几个关键点：

5.1 DryRunnableStorage：dryRun 在这里拦截

位置：staging/src/k8s.io/apiserver/pkg/registry/generic/registry/dryrun.go

func (s *DryRunnableStorage) Create(ctx context.Context, key string,
obj, out runtime.Object, ttl uint64, dryRun bool) error {

if dryRun {
// 仅做对象解码到 out，不写 etcd
if err := s.copyInto(obj, out); err != nil {
return err
}
// 模拟一个 resourceVersion 以满足客户端
return s.Versioner.UpdateObject(out, 0)
}
return s.Storage.Create(ctx, key, obj, out, ttl)
}

💡 dryRun 的意义：让 webhook、validation、PrepareForCreate 全部跑一遍，但不真的落 etcd——非常适合 CI 流水线做合规检查。

5.2 etcd3.store.Create：真正的写入

位置：staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go

func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error {
// ① 计算最终 key
preparedKey, err := s.prepareKey(key)
if err != nil {
return err
}

// ② 序列化为 []byte（默认 protobuf）
data, err := runtime.Encode(s.codec, obj)
if err != nil {
return err
}

// ③ 加 TTL（如果有）
opts, err := s.ttlOpts(ctx, int64(ttl))
if err != nil {
return err
}

// ④ Transformer 加密
newData, err := s.transformer.TransformToStorage(ctx, data, authenticatedDataString(preparedKey))
if err != nil {
return storage.NewInternalError(err.Error())
}

// ⑤ 用事务 Put（!exists 才允许写入）
startTime := time.Now()
txnResp, err := s.client.KV.Txn(ctx).If(
notFound(preparedKey),
).Then(
clientv3.OpPut(preparedKey, string(newData), opts…),
).Commit()
metrics.RecordEtcdRequest("create", s.groupResourceString, err, startTime)
if err != nil {
return err
}
if !txnResp.Succeeded {
return storage.NewKeyExistsError(preparedKey, 0)
}
…
}

逐项拆：

① 序列化：runtime.Encode

K8s 内部用的不是 JSON，是 protobuf——比 JSON 快、紧凑（约小 30%）。

data, err := runtime.Encode(s.codec, obj)

s.codec 在 RESTStorage 初始化时注入（上节讲过的 StorageConfig.Codec）。

💡 CRD 用 JSON，原生资源用 protobuf——这就是为什么大量自定义资源会比原生资源占更多 etcd 空间。

② Transformer：透明加密

newData, err := s.transformer.TransformToStorage(ctx, data, authenticatedDataString(preparedKey))

这是 K8s 的静态数据加密入口。如果集群配了 EncryptionConfiguration（比如对 Secret 启用 AES-CBC 加密），这一步会把序列化后的 bytes 加密。

# /etc/kubernetes/encryption-config.yaml
apiVersion: apiserver.config.k8s.io/v1
kind: EncryptionConfiguration
resources:
– resources:
– secrets
providers:
– aescbc:
keys:
– name: key1
secret: <base64–encoded–32–byte–key>
– identity: {} # 兜底：不加密

🚨 生产踩坑：Secret 加密一定要配——否则 etcd snapshot 泄露 = Secret 全裸。但加密 key 必须备份——丢了等于所有加密数据永久丢失。

③ 为什么用 Txn 而不是 Put？

s.client.KV.Txn(ctx).If(
notFound(preparedKey), // 条件：key 必须不存在
).Then(
clientv3.OpPut(preparedKey, string(newData), opts…),
).Commit()

K8s 不直接用 etcd 的 Put，而是用 事务（Txn）+ notFound 条件：

• 如果 key 已存在 → 事务失败 → 返回 AlreadyExists 错误
• 如果 key 不存在 → 事务成功 → 写入数据

💡 为什么不用 Put？ etcd 的 Put 是覆盖语义——如果 key 已存在会被覆盖。而 K8s 的 Create 必须保证不能覆盖已有对象——所以用 Txn 实现原子的 CAS（Compare-And-Set）。

这也是为什么并发创建同名 Pod 时，第二个会拿到 AlreadyExists 错误——etcd Txn 层就拦住了。

5.3 第⑤步：写入后回填 + Decorator

putResp := txnResp.Responses[0].GetResponsePut()
err = decode(s.codec, s.versioner, data, out, putResp.Header.Revision)

decode 把刚写的数据反序列化回 out 对象，并把 etcd 返回的 Revision 填充到 out.ResourceVersion——客户端拿到的 Pod 对象就有 resourceVersion: "12345" 字段了，后续 watch、update 都靠它。

最后回到 Store.Create：

if e.Decorator != nil {
e.Decorator(out)
}

if e.AfterCreate != nil {
e.AfterCreate(out, options)
}

fn = finishCreate
return out, nil

• Decorator：可选的对象装饰器，少数资源会用（比如给 Pod 加上 ephemeral 信息）
• AfterCreate：可选钩子，目前主线 K8s 资源几乎都不用

整个流程结束，APIServer 把 out（带 resourceVersion 的完整 Pod 对象）通过 HTTP 200 返回给 kubectl。

六、完整数据流：从 kubectl 到 etcd

kubectl create -f pod.yaml
│
│ POST /api/v1/namespaces/default/pods
▼
┌──────────────────────┐
│ APIServer HTTP 入口 │
├──────────────────────┤
│ 1. 认证 (Authn) │ ← 谁在调用？
│ 2. 鉴权 (Authz) │ ← 有权限吗？RBAC
│ 3. Mutating Admission│ ← webhook 改 obj（⚠️ 别碰 status！）
│ 4. Schema Validation │ ← OpenAPI 校验
│ 5. Validating Admiss.│ ← webhook 校验
└──────────┬───────────┘
▼
┌──────────────────────┐
│ podStorage.Create() │ Pod 专属入口
└──────────┬───────────┘
▼
┌──────────────────────────────────────┐
│ genericregistry.Store.Create │
├──────────────────────────────────────┤
│ ① BeginCreate (可选) │
│ ② BeforeCreate │
│ ├─ PrepareForCreate │
│ │ ├─ Status = Pending │
│ │ ├─ QOSClass = GetPodQOS(pod) │
│ │ └─ DropDisabledPodFields │
│ ├─ EnsureObjectMeta (生成 UID) │
│ ├─ Validate (业务规则) │
│ └─ Canonicalize │
│ ③ Storage.Create(key, obj, out) │
│ ④ Decorator (可选) │
│ ⑤ AfterCreate (可选) │
└──────────┬───────────────────────────┘
▼
┌──────────────────────────────────────┐
│ DryRunnableStorage.Create │
├──────────────────────────────────────┤
│ if dryRun: 模拟返回，不写 etcd │
│ else: 透传到下层 │
└──────────┬───────────────────────────┘
▼
┌──────────────────────────────────────┐
│ etcd3.store.Create │
├──────────────────────────────────────┤
│ ① prepareKey → /registry/pods/… │
│ ② Encode(protobuf) │
│ ③ Transformer.TransformToStorage │ ← 加密（可选）
│ ④ Txn().If(!exist).Then(Put).Commit │ ← 原子写入
│ ⑤ decode(out)，填充 ResourceVersion │
└──────────┬───────────────────────────┘
▼
etcd v3 集群
│
│ Raft 共识 + 持久化
▼
磁盘 (boltdb)

写入后：

• watch 触发：所有 watch /pods 的客户端（scheduler、kubelet、controller-manager）收到 ADDED 事件
• scheduler 调度：把 Pod 绑定到合适的 Node
• kubelet 拉取：被调度的 Node 上的 kubelet 拉到 Pod，开始创建容器

七、踩坑实录：Pod 创建链路上 8 个常见坑

八、思考题

九、本节小结

这节啃完，你应该掌握了：

• 整条链路：Pod 从 HTTP 进来 → BeforeCreate → Strategy.PrepareForCreate → Validate → DryRunnable → etcd3.store → Txn Put
• PrepareForCreate：强制设 Pending、计算 QoS、丢弃禁用字段
• QoS 算法：requests vs limits 的对比规则，三档的实战影响
• 存储层细节：protobuf 序列化、Transformer 加密、Txn 事务实现 CAS
• dryRun 在哪一层拦截：DryRunnableStorage，所有上层逻辑都跑，只是不真的落库

下一节我们会继续往后看：APIServer 的限流策略（MaxInFlight、APF）——一个高 QPS 集群里 APIServer 怎么不被打爆。

参考资料

• Kubernetes 官方源码 (kubernetes/kubernetes)
• staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go
• pkg/registry/core/pod/strategy.go
• pkg/apis/core/helper/qos/qos.go
• staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go
• Kubernetes 文档 – QoS 类
• Kubernetes 文档 – 静态数据加密
• etcd Txn API 文档