5. 实现原地升级

原地升级的本质

原生 STS 发版 = 删 Pod + 建 Pod。NeoSet 发版 = PATCH Pod 的 container image。

原生 STS:
  DELETE /pods/mysql-4
  CREATE /pods/mysql-4 (新 image)
  ↑ 新 IP、新 Volume 挂载、重新调度

NeoSet:
  PATCH /pods/mysql-4
  { "spec.containers[0].image": "mysql:8.1" }
  ↑ IP 不变、Volume 不动、不重新调度、容器重启

K8S 从 1.27 开始原生支持 Pod 的 container image 原地更新——PATCH spec.containers[*].image 后，kubelet 重启对应容器，不重建 Pod。

PATCH 的完整代码

// internal/controller/inplace.go
package controller

import (
    "context"
    "fmt"
    "strings"

    corev1 "k8s.io/api/core/v1"
    "k8s.io/apimachinery/pkg/api/meta"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/types"
    "sigs.k8s.io/controller-runtime/pkg/client"
    "sigs.k8s.io/controller-runtime/pkg/log"

    neov1alpha1 "neo.io/api/v1alpha1"
)

// InPlaceUpdatePod 原地更新 Pod 的 container image
// 返回 true 表示发生了更新，false 表示无需更新
func (r *NeoSetReconciler) InPlaceUpdatePod(
    ctx context.Context,
    pod *corev1.Pod,
    neoset *neov1alpha1.NeoSet,
) (bool, error) {
    logger := log.FromContext(ctx)

    targetTemplate := neoset.Spec.Template
    updated := false

    // 1. 构造 PATCH body——只改 image，不动其他字段
    //    用 strategic merge patch，避免覆盖整个 spec
    patch := map[string]interface{}{
        "spec": map[string]interface{}{
            "containers":    []interface{}{},
            "initContainers": []interface{}{},
        },
    }

    containers := patch["spec"].(map[string]interface{})["containers"].([]interface{})
    for i, c := range pod.Spec.Containers {
        target := findContainerByName(targetTemplate.Spec.Containers, c.Name)
        if target != nil && target.Image != c.Image {
            containers = append(containers, map[string]interface{}{
                "name":  c.Name,
                "image": target.Image,
            })
            updated = true
            logger.Info("原地更新 container",
                "pod", pod.Name,
                "container", c.Name,
                "oldImage", c.Image,
                "newImage", target.Image)
        }
    }
    patch["spec"].(map[string]interface{})["containers"] = containers

    // init containers 同理
    initContainers := patch["spec"].(map[string]interface{})["initContainers"].([]interface{})
    for i, c := range pod.Spec.InitContainers {
        target := findContainerByName(targetTemplate.Spec.InitContainers, c.Name)
        if target != nil && target.Image != c.Image {
            initContainers = append(initContainers, map[string]interface{}{
                "name":  c.Name,
                "image": target.Image,
            })
            updated = true
        }
    }
    patch["spec"].(map[string]interface{})["initContainers"] = initContainers

    if !updated {
        return false, nil
    }

    // 2. 序列化 PATCH body
    patchBytes, err := json.Marshal(patch)
    if err != nil {
        return false, fmt.Errorf("序列化 patch 失败: %w", err)
    }

    // 3. 执行 PATCH——用 strategic merge patch 类型
    //    这样只更新指定字段，不会覆盖整个 spec
    err = r.Patch(ctx, pod, client.RawPatch(types.StrategicMergePatchType, patchBytes))
    if err != nil {
        return false, fmt.Errorf("PATCH pod %s 失败: %w", pod.Name, err)
    }

    // 4. 打 annotation 记录更新时间和目标 image
    //    用于就绪判断时算超时
    annotationPatch := map[string]interface{}{
        "metadata": map[string]interface{}{
            "annotations": map[string]string{
                "neoset.neo.io/inplace-update-at":     metav1.Now().Format(time.RFC3339),
                "neoset.neo.io/inplace-target-image":  neoset.Status.TargetImage,
            },
        },
    }
    annotationBytes, _ := json.Marshal(annotationPatch)
    _ = r.Patch(ctx, pod, client.RawPatch(types.MergePatchType, annotationBytes))

    logger.Info("原地升级 PATCH 完成",
        "pod", pod.Name,
        "targetImage", neoset.Status.TargetImage)

    return true, nil
}

func findContainerByName(containers []corev1.Container, name string) *corev1.Container {
    for i := range containers {
        if containers[i].Name == name {
            return &containers[i]
        }
    }
    return nil
}

几个关键设计

Strategic Merge Patch——不用 MergePatchType（会覆盖整个 containers 数组），用 StrategicMergePatchType（按 name 匹配只改指定字段）。这样只改 image，不动 command/env/resources 等其他字段。

只改 image——原地升级的精髓是"最小变更"。只 PATCH image 字段，kubelet 收到后只重启对应容器。如果 PATCH 了 resources，会触发 Pod 的 resource 变更，某些字段不可变会导致 PATCH 失败。

annotation 记录更新时间——用于就绪超时判断。PATCH 后开始计时，如果 readinessTimeoutSeconds 内 Pod 没就绪，标记失败。

判断 Pod 是否需要更新

// PodNeedsUpdate 判断 Pod 的 image 是否跟目标版本一致
func PodNeedsUpdate(pod *corev1.Pod, neoset *neov1alpha1.NeoSet) bool {
    targetImage := neoset.Status.TargetImage
    if targetImage == "" {
        return false // 没有目标版本，不需要更新
    }

    for _, c := range pod.Spec.Containers {
        // 跳过 sidecar 容器（不是 NeoSet 管的）
        if !isManagedContainer(c.Name, neoset) {
            continue
        }
        if c.Image != targetImage {
            return true
        }
    }
    return false
}

// IsPodUpdated 判断 Pod 是否已经是目标版本
func IsPodUpdated(pod *corev1.Pod, neoset *neov1alpha1.NeoSet) bool {
    targetImage := neoset.Status.TargetImage
    if targetImage == "" {
        return true
    }

    for _, c := range pod.Spec.Containers {
        if !isManagedContainer(c.Name, neoset) {
            continue
        }
        if c.Image != targetImage {
            return false
        }
    }
    return true
}

func isManagedContainer(name string, neoset *neov1alpha1.NeoSet) bool {
    for _, c := range neoset.Spec.Template.Spec.Containers {
        if c.Name == name {
            return true
        }
    }
    return false
}

判断 Pod 是否就绪

原地更新后要等容器重启完成、就绪探针通过，才算这个 Pod 更新成功：

// IsPodReadyAfterUpdate 判断原地更新后 Pod 是否就绪
func IsPodReadyAfterUpdate(pod *corev1.Pod, neoset *neov1alpha1.NeoSet) bool {
    strategy := neoset.Spec.UpdateStrategy.InPlaceUpdateStrategy
    readinessType := ReadinessTypeProbe
    if strategy != nil && strategy.Readiness != "" {
        readinessType = strategy.Readiness
    }

    switch readinessType {
    case ReadinessTypeProbe:
        // Probe 模式：检查 containerStatuses 的 ready 字段
        // ready=true 表示 readinessProbe 通过
        return checkContainerReady(pod, neoset)

    case ReadinessTypeReady:
        // Ready 模式：检查 Pod 的 Ready 条件
        return isPodConditionTrue(pod, corev1.PodReady)

    default:
        return checkContainerReady(pod, neoset)
    }
}

func checkContainerReady(pod *corev1.Pod, neoset *neov1alpha1.NeoSet) bool {
    for _, cs := range pod.Status.ContainerStatuses {
        if !isManagedContainer(cs.Name, neoset) {
            continue
        }
        if !cs.Ready {
            return false
        }
    }
    return true
}

func isPodConditionTrue(pod *corev1.Pod, condType corev1.PodConditionType) bool {
    cond := meta.FindStatusCondition(pod.Status.Conditions, condType)
    return cond != nil && cond.Status == corev1.ConditionTrue
}

就绪超时判断

// IsUpdateTimedOut 判断原地更新是否超时
func IsUpdateTimedOut(pod *corev1.Pod, neoset *neov1alpha1.NeoSet) bool {
    updateAtStr := pod.Annotations["neoset.neo.io/inplace-update-at"]
    if updateAtStr == "" {
        return false // 没有更新记录
    }

    updateAt, err := time.Parse(time.RFC3339, updateAtStr)
    if err != nil {
        return false
    }

    timeout := int64(300) // 默认 5 分钟
    if neoset.Spec.UpdateStrategy.InPlaceUpdateStrategy != nil &&
       neoset.Spec.UpdateStrategy.InPlaceUpdateStrategy.ReadinessTimeoutSeconds != nil {
        timeout = *neoset.Spec.UpdateStrategy.InPlaceUpdateStrategy.ReadinessTimeoutSeconds
    }

    return time.Since(updateAt) > time.Duration(timeout)*time.Second
}

完整的 Reconcile 片段

func (r *NeoSetReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    var neoset neov1alpha1.NeoSet
    if err := r.Get(ctx, req.NamespacedName, &neoset); err != nil {
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }

    // ... 前面的版本管理逻辑 ...

    // 如果有目标版本且没暂停，执行原地升级
    if neoset.Status.TargetImage != "" && neoset.Status.Phase != neov1alpha1.NeoSetPhasePaused {
        pods := r.listNeoSetPods(ctx, &neoset)

        // 1. 检查当前正在更新的 Pod 是否就绪
        if neoset.Status.CurrentUpdatePod != "" {
            currentPod := findPodByName(pods, neoset.Status.CurrentUpdatePod)
            if currentPod != nil {
                if IsPodReadyAfterUpdate(currentPod, &neoset) {
                    // 就绪了，清空 CurrentUpdatePod，更新计数
                    neoset.Status.CurrentUpdatePod = ""
                    neoset.Status.UpdatedReplicas++
                } else if IsUpdateTimedOut(currentPod, &neoset) {
                    // 超时了，标记失败
                    neoset.Status.Phase = neov1alpha1.NeoSetPhaseFailed
                    return ctrl.Result{}, r.Status().Update(ctx, &neoset)
                } else {
                    // 还在等就绪，30 秒后再来
                    return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
                }
            }
        }

        // 2. 找下一个该更新的 Pod
        nextPod := r.findNextPodToUpdate(pods, &neoset)
        if nextPod != nil && neoset.Status.CurrentUpdatePod == "" {
            // 检查 maxUnavailable 约束
            if r.canUpdateMore(pods, &neoset) {
                updated, err := r.InPlaceUpdatePod(ctx, nextPod, &neoset)
                if err != nil {
                    return ctrl.Result{}, err
                }
                if updated {
                    neoset.Status.CurrentUpdatePod = nextPod.Name
                }
            }
        }

        // 3. 检查是否全部更新完
        if neoset.Status.UpdatedReplicas == *neoset.Spec.Replicas {
            neoset.Status.Phase = neov1alpha1.NeoSetPhaseCompleted
            neoset.Status.CurrentImage = neoset.Status.TargetImage
        }
    }

    return ctrl.Result{}, r.Status().Update(ctx, &neoset)
}

// findNextPodToUpdate 找下一个该更新的 Pod
// 按 partition 约束：只更新序号 >= partition 的 Pod
// 从最大序号开始往小更新（跟 STS 一样）
func (r *NeoSetReconciler) findNextPodToUpdate(pods []corev1.Pod, neoset *neov1alpha1.NeoSet) *corev1.Pod {
    partition := int32(0)
    if neoset.Spec.UpdateStrategy.Partition != nil {
        partition = *neoset.Spec.UpdateStrategy.Partition
    }

    // 按序号从大到小排序
    sortedPods := sortPodsByOrdinalDesc(pods)

    for i := range sortedPods {
        pod := &sortedPods[i]
        ordinal := getOrdinal(pod)
        if ordinal < partition {
            continue // 小于 partition 的不动
        }
        if !PodNeedsUpdate(pod, neoset) {
            continue // 已经是目标版本
        }
        return pod
    }
    return nil
}

// canUpdateMore 检查是否还能更新更多 Pod（maxUnavailable 约束）
func (r *NeoSetReconciler) canUpdateMore(pods []corev1.Pod, neoset *neov1alpha1.NeoSet) bool {
    maxUnavail := int32(1)
    if neoset.Spec.UpdateStrategy.MaxUnavailable != nil {
        val, _ := intstr.GetScaledValueFromIntOrPercent(
            neoset.Spec.UpdateStrategy.MaxUnavailable, int(*neoset.Spec.Replicas), true)
        maxUnavail = int32(val)
    }

    // 当前不可用的 Pod 数（正在更新中 + 不 ready 的）
    unavailable := int32(0)
    for _, pod := range pods {
        if !isPodReady(&pod) {
            unavailable++
        }
    }

    return unavailable < maxUnavail
}

原地升级的完整时序

Reconcile 第 1 轮:
  targetImage = mysql:8.1
  CurrentUpdatePod = "" (空)
  → findNextPodToUpdate → mysql-4 (最大序号)
  → canUpdateMore? unavailable=0 < maxUnavailable=1 ✓
  → InPlaceUpdatePod(mysql-4) → PATCH image
  → CurrentUpdatePod = "mysql-4"
  → UpdatedReplicas 还是 0

Reconcile 第 2 轮 (5 秒后):
  CurrentUpdatePod = "mysql-4"
  → IsPodReadyAfterUpdate(mysql-4)?
    → containerStatuses[0].Ready = false (还在重启)
  → IsUpdateTimedOut? 否
  → requeue 5 秒后再来

Reconcile 第 3 轮 (5 秒后):
  CurrentUpdatePod = "mysql-4"
  → IsPodReadyAfterUpdate(mysql-4)?
    → containerStatuses[0].Ready = true (重启完成，探针通过)
  → CurrentUpdatePod = ""
  → UpdatedReplicas = 1

Reconcile 第 4 轮:
  CurrentUpdatePod = "" (空)
  → findNextPodToUpdate → mysql-3
  → InPlaceUpdatePod(mysql-3)
  → CurrentUpdatePod = "mysql-3"
  ...

直到 UpdatedReplicas == replicas:
  Phase = Completed
  CurrentImage = TargetImage

优雅摘除：更新前排空连接

原地升级 PATCH image 后容器立刻重启——但此时 Pod 可能有活跃连接。直接重启会让连接断掉，业务感知。

参考 OpenKruise 的做法：PATCH 前先把 Pod 从 Endpoints 摘除，等连接排空再 PATCH。

不摘除:
  PATCH image → 容器重启 → 活跃连接全断 → ready → 流量回来
  ↑ 有损

优雅摘除:
  设 ReadinessGate=False → Endpoints 摘除 → 等 gracePeriod → PATCH image → ready → 设 ReadinessGate=True
  ↑ 无损

// internal/controller/graceful.go

// GracefulInPlaceUpdate 优雅原地升级
func (r *NeoSetReconciler) GracefulInPlaceUpdate(
    ctx context.Context,
    pod *corev1.Pod,
    neoset *neov1alpha1.NeoSet,
) error {
    logger := log.FromContext(ctx)

    graceful := neoset.Spec.UpdateStrategy.GracefulUpdate
    if graceful == nil || !graceful.RemoveFromEndpoints {
        // 没开优雅摘除，直接 PATCH
        _, err := r.InPlaceUpdatePod(ctx, pod, neoset)
        return err
    }

    // 1. 第一步：设 ReadinessGate=False，让 Endpoints 摘除
    if !hasReadinessGateFalse(pod, "InPlaceUpdateReady") {
        r.setReadinessGate(ctx, pod, "InPlaceUpdateReady", false)
        logger.Info("优雅摘除：设 ReadinessGate=False", "pod", pod.Name)
        return nil // 这一轮先摘除，下一轮再 PATCH
    }

    // 2. 第二步：等 gracePeriod 让连接排空
    gateSetTime := getReadinessGateSetTime(pod, "InPlaceUpdateReady")
    gracePeriod := time.Duration(graceful.GracePeriodSeconds) * time.Second
    if time.Since(gateSetTime) < gracePeriod {
        logger.V(1).Info("等待连接排空",
            "pod", pod.Name,
            "elapsed", time.Since(gateSetTime),
            "need", gracePeriod)
        return nil // 还没排空，等着
    }

    // 3. 第三步：连接排空，执行 PATCH
    _, err := r.InPlaceUpdatePod(ctx, pod, neoset)
    if err != nil {
        return err
    }

    logger.Info("优雅摘除完成，已 PATCH", "pod", pod.Name)
    return nil
}

优雅摘除的完整时序

Reconcile 第 1 轮:
  → 设 ReadinessGate(InPlaceUpdateReady)=False
  → Endpoints 摘除 mysql-4
  → return（等下一轮）

Reconcile 第 2 轮 (gracePeriod 内):
  → 检查时间，还没到 gracePeriod
  → return（继续等）

Reconcile 第 3 轮 (gracePeriod 到了):
  → 连接已排空
  → PATCH image
  → 容器重启

Reconcile 第 4 轮:
  → 容器 ready
  → 设 ReadinessGate(InPlaceUpdateReady)=True
  → Endpoints 加回 mysql-4
  → 流量恢复

资源热更新：不只改 image

K8S 1.27+ 支持原地改 container 的 resources（CPU/Memory），不重启容器。NeoSet 可以利用这个——不只原地改 image，还原地改资源配额。

// InPlaceUpdatePod 扩展：不只 PATCH image，还 PATCH resources
func (r *NeoSetReconciler) InPlaceUpdatePod(
    ctx context.Context,
    pod *corev1.Pod,
    neoset *neov1alpha1.NeoSet,
) (bool, error) {
    targetTemplate := neoset.Spec.Template
    updated := false

    containers := []interface{}{}
    for _, c := range pod.Spec.Containers {
        target := findContainerByName(targetTemplate.Spec.Containers, c.Name)
        if target == nil {
            continue
        }

        patch := map[string]interface{}{"name": c.Name}
        changed := false

        // 改 image
        if target.Image != c.Image {
            patch["image"] = target.Image
            changed = true
        }

        // 改 resources（CPU/Memory 热更新）
        if !equality.Semantic.DeepEqual(target.Resources.Requests, c.Resources.Requests) ||
           !equality.Semantic.DeepEqual(target.Resources.Limits, c.Resources.Limits) {
            patch["resources"] = map[string]interface{}{
                "requests": target.Resources.Requests,
                "limits":   target.Resources.Limits,
            }
            changed = true
        }

        if changed {
            containers = append(containers, patch)
            updated = true
        }
    }

    if !updated {
        return false, nil
    }

    // Strategic Merge Patch
    patchBody := map[string]interface{}{
        "spec": map[string]interface{}{"containers": containers},
    }
    patchBytes, _ := json.Marshal(patchBody)
    err := r.Patch(ctx, pod, client.RawPatch(types.StrategicMergePatchType, patchBytes))
    return updated, err
}

只改资源不重启容器

跟改 image 不同——改 resources 时 kubelet 不重启容器，只更新 cgroup 限制。容器进程无感知，CPU/Memory 限制实时生效。

改 image:  kubelet 停容器 → 拉新镜像 → 起新容器（有中断）
改 resources: kubelet 改 cgroup（无中断，实时生效）

这个特性很有用——比如 MySQL 负载高了，原地加 CPU 不用重启数据库。

前置阅读

设计CRD — 本章前置，status 里的 targetImage/currentImage
ControllerRevision版本管理 — 本章前置，targetImage 来自版本管理
原地升级 — Starter 第 15 章，原地升级概念
StatefulSet控制循环 — Controller 系列第 4 章

思考题

问题 1：为什么用 Strategic Merge Patch 而不是 Merge Patch？Merge Patch 会怎样？

问题 2：原地升级后容器重启，但 Pod 的 restartPolicy 默认是 Always。kubelet 重启容器跟原地升级"重启容器"有什么区别？

问题 3：canUpdateMore 检查 unavailable < maxUnavailable。但如果一个 Pod 因为应用 bug（不是更新导致）不 ready，会被算进 unavailable 吗？这会导致什么？

解答

问题 1：Merge Patch 会用 PATCH body 里的 containers 数组替换整个 spec.containers——Pod 里原本 3 个容器（mysql + 2 个 sidecar），PATCH body 里只写了 mysql，Merge Patch 会把 sidecar 删掉。Strategic Merge Patch 按 name 字段匹配——只更新写了的容器，没写的保持不变。原地升级只改 image，Strategic Merge Patch 是正确选择。

问题 2：restartPolicy: Always 的容器 CrashLoopBackOff 时，kubelet 重启容器是"容器进程崩了重启"。原地升级的重启是"image 变了，kubelet 主动停掉旧容器、用新 image 创建新容器"。区别在于：前者用同一个 image 重启，后者换了 image。从 Pod 视角看都是容器重启——IP 不变、Volume 不动、网络命名空间不变。但 containerStatuses 的 imageID 会变（新 image 的 digest），Controller 靠这个判断"新容器是否已经起来"。

问题 3：会。canUpdateMore 算 unavailable 时只要 !isPodReady 就算，不区分"正在更新中"和"应用 bug 导致不 ready"。如果一个 Pod 因为 OOM 反复重启，它会被算进 unavailable，挤占 maxUnavailable 配额——导致 Controller 不敢更新更多 Pod。这是保守策略——宁可慢一点也不冒险。如果要优化，可以加 annotation 标记"这个 Pod 是 NeoSet 正在更新的"，只把正在更新的算进 unavailable。

下一章我们实现金丝雀与分批发布——Partition + MaxUnavailable 状态机、暂停/恢复逻辑。

原地升级的本质​

PATCH 的完整代码​

几个关键设计​

判断 Pod 是否需要更新​

判断 Pod 是否就绪​

就绪超时判断​

完整的 Reconcile 片段​

原地升级的完整时序​

优雅摘除：更新前排空连接​

优雅摘除的完整时序​

资源热更新：不只改 image​

只改资源不重启容器​

前置阅读​

思考题​

解答​

下一章​