6. 金丝雀与分批发布
三种发布模式
NeoSet 通过 partition 和 pause 的组合，支持三种发布模式：
| 模式 | 配置 | 行为 |
|---|---|---|
| 金丝雀 | partition: replicas-1 | 只发最后 1 个 Pod，观察 |
| 分批 | 逐步调小 partition | 先发 1，再发 3，再发全部 |
| 暂停 | pause: true | 任意时刻停住 |
金丝雀发布的完整流程
# 1. 改 image 触发发版，设 partition=4（只发 mysql-4）
kubectl patch neoset mysql --type='json' -p='[
{"op":"replace","path":"/spec/template/spec/containers/0/image","value":"mysql:8.1"},
{"op":"replace","path":"/spec/updateStrategy/partition","value":4}
]'
# 2. 看 mysql-4 的状态
kubectl get pods mysql-4 -w
# mysql-4 image 从 mysql:8.0 变成 mysql:8.1，容器重启，ready
# 3. 观察一段时间，没问题，继续发
kubectl patch neoset mysql --type='json' -p='[
{"op":"replace","path":"/spec/updateStrategy/partition","value":2}
]'
# mysql-3, mysql-2 原地升级
# 4. 全量
kubectl patch neoset mysql --type='json' -p='[
{"op":"replace","path":"/spec/updateStrategy/partition","value":0}
]'
暂停/恢复
发版到一半发现问题：
# 暂停
kubectl patch neoset mysql --type='json' -p='[
{"op":"replace","path":"/spec/updateStrategy/pause","value":true}
]'
# Controller 停在当前进度，不继续更新下一个 Pod
# 看日志确认没问题后恢复
kubectl patch neoset mysql --type='json' -p='[
{"op":"replace","path":"/spec/updateStrategy/pause","value":false}
]'
状态机实现
// internal/controller/state_machine.go
package controller
import (
"context"
corev1 "k8s.io/api/core/v1"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/log"
neov1alpha1 "neo.io/api/v1alpha1"
)
// reconcileUpdate 执行发版状态机
func (r *NeoSetReconciler) reconcileUpdate(
ctx context.Context,
neoset *neov1alpha1.NeoSet,
pods []corev1.Pod,
) (bool, error) {
logger := log.FromContext(ctx)
// 1. 暂停了——不动
if neoset.Spec.UpdateStrategy.Pause {
if neoset.Status.Phase != neov1alpha1.NeoSetPhasePaused {
neoset.Status.Phase = neov1alpha1.NeoSetPhasePaused
logger.Info("发版暂停", "updatedReplicas", neoset.Status.UpdatedReplicas)
}
return false, nil
}
// 2. 没有目标版本——不发版
if neoset.Status.TargetImage == "" {
return false, nil
}
// 3. 当前正在更新某个 Pod——等它就绪
if neoset.Status.CurrentUpdatePod != "" {
done, err := r.checkCurrentUpdate(ctx, neoset, pods)
return done, err
}
// 4. 找下一个该更新的 Pod
nextPod := r.findNextPodToUpdate(pods, neoset)
if nextPod == nil {
// 没有需要更新的 Pod 了——发版完成
if neoset.Status.UpdatedReplicas == *neoset.Spec.Replicas {
r.completeUpdate(neoset)
}
return true, nil
}
// 5. 检查 maxUnavailable 约束
if !r.canUpdateMore(pods, neoset) {
logger.V(1).Info("等待不可用 Pod 恢复", "reason", "maxUnavailable 约束")
return false, nil
}
// 6. 执行原地升级
updated, err := r.InPlaceUpdatePod(ctx, nextPod, neoset)
if err != nil {
return false, err
}
if updated {
neoset.Status.CurrentUpdatePod = nextPod.Name
// 标记阶段
if neoset.Status.UpdatedReplicas == 0 {
neoset.Status.Phase = neov1alpha1.NeoSetPhaseCanary
} else {
neoset.Status.Phase = neov1alpha1.NeoSetPhaseRollingOut
}
}
return true, nil
}
// checkCurrentUpdate 检查当前正在更新的 Pod 状态
func (r *NeoSetReconciler) checkCurrentUpdate(
ctx context.Context,
neoset *neov1alpha1.NeoSet,
pods []corev1.Pod,
) (bool, error) {
logger := log.FromContext(ctx)
currentPod := findPodByName(pods, neoset.Status.CurrentUpdatePod)
if currentPod == nil {
// Pod 被删了，清空状态
neoset.Status.CurrentUpdatePod = ""
return false, nil
}
// 就绪了
if IsPodReadyAfterUpdate(currentPod, neoset) {
neoset.Status.CurrentUpdatePod = ""
neoset.Status.UpdatedReplicas++
logger.Info("Pod 原地升级完成",
"pod", currentPod.Name,
"updatedReplicas", neoset.Status.UpdatedReplicas)
return true, nil
}
// 超时了
if IsUpdateTimedOut(currentPod, neoset) {
neoset.Status.Phase = neov1alpha1.NeoSetPhaseFailed
logger.Error(nil, "Pod 原地升级超时",
"pod", currentPod.Name)
return true, nil
}
// 还在等——不动，下一轮再看
return false, nil
}
// completeUpdate 发版完成
func (r *NeoSetReconciler) completeUpdate(neoset *neov1alpha1.NeoSet) {
neoset.Status.Phase = neov1alpha1.NeoSetPhaseCompleted
neoset.Status.CurrentImage = neoset.Status.TargetImage
neoset.Status.CurrentUpdatePod = ""
// partition 归零（可选——也可保留给用户做下一轮金丝雀）
}
// findNextPodToUpdate 找下一个该更新的 Pod
// 按序号从大到小，跳过序号 < partition 的
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
}
sorted := sortPodsByOrdinalDesc(pods)
for i := range sorted {
pod := &sorted[i]
ordinal := getOrdinal(pod)
if ordinal < partition {
continue
}
if !PodNeedsUpdate(pod, neoset) {
continue
}
// 跳过正在更新的
if pod.Name == neoset.Status.CurrentUpdatePod {
continue
}
return pod
}
return nil
}
// canUpdateMore 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)
}
unavailable := int32(0)
for i := range pods {
if !isPodReady(&pods[i]) {
unavailable++
}
}
return unavailable < maxUnavail
}
完整的 Reconcile 入口
func (r *NeoSetReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
logger := log.FromContext(ctx)
var neoset neov1alpha1.NeoSet
if err := r.Get(ctx, req.NamespacedName, &neoset); err != nil {
return ctrl.Result{}, client.IgnoreNotFound(err)
}
// 1. 检查 template 变化
templateHash := computeTemplateHash(neoset.Spec.Template)
if templateHash != neoset.Status.LastTemplateHash {
// template 变了，创建新版本
rev, err := r.CreateRevision(ctx, &neoset, templateHash)
if err != nil {
return ctrl.Result{}, err
}
neoset.Status.CurrentRevision = rev.Revision
neoset.Status.TargetImage = neoset.Spec.Template.Spec.Containers[0].Image
neoset.Status.Phase = neov1alpha1.NeoSetPhaseCanary
neoset.Status.UpdatedReplicas = 0
neoset.Status.LastTemplateHash = templateHash
neoset.Status.CurrentUpdatePod = ""
}
// 2. 列出所有 Pod
pods := r.listNeoSetPods(ctx, &neoset)
// 3. 扩缩容逻辑（下一章）
if err := r.reconcileReplicas(ctx, &neoset, pods); err != nil {
return ctrl.Result{}, err
}
// 4. 发版状态机
if neoset.Status.TargetImage != "" &&
neoset.Status.TargetImage != neoset.Status.CurrentImage {
_, err := r.reconcileUpdate(ctx, &neoset, pods)
if err != nil {
return ctrl.Result{}, err
}
}
// 5. 更新 status
if err := r.Status().Update(ctx, &neoset); err != nil {
return ctrl.Result{}, err
}
// 6. 如果还在发版中，5 秒后再来
if neoset.Status.Phase == neov1alpha1.NeoSetPhaseCanary ||
neoset.Status.Phase == neov1alpha1.NeoSetPhaseRollingOut {
return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
}
_ = logger
return ctrl.Result{}, nil
}
状态流转图
template 变了
│
▼
┌─────────┐
│ Canary │ (updatedReplicas=1, 在发金丝雀)
└────┬────┘
│
┌───────────┼───────────┐
│ │ │
pause=true 继续发版 超时/失败
│ │ │
▼ ▼ ▼
┌─────────┐ ┌────────────┐ ┌─────────┐
│ Paused │ │ RollingOut │ │ Failed │
└────┬────┘ └─────┬──────┘ └─────────┘
│ │
pause=false 全部更新完
│ │
└─────┬──────┘
▼
┌───────────┐
│ Completed │ (currentImage=targetImage)
└───────────┘
金丝雀 + 暂停的实战
# 1. 发版，金丝雀只发 1 个
kubectl patch neoset mysql --type='json' -p='[
{"op":"replace","path":"/spec/template/spec/containers/0/image","value":"mysql:8.1"},
{"op":"replace","path":"/spec/updateStrategy/partition","value":4}
]'
# 等待 mysql-4 原地升级完成
kubectl get neo mysql -w
# PHASE: Canary → Completed (1 个更新完)
# 2. 暂停，观察 mysql-4 的业务指标
kubectl patch neoset mysql --type='json' -p='[
{"op":"replace","path":"/spec/updateStrategy/pause","value":true}
]'
# PHASE: Paused
# 3. 观察 30 分钟，QPS 正常，继续
kubectl patch neoset mysql --type='json' -p='[
{"op":"replace","path":"/spec/updateStrategy/pause","value":false},
{"op":"replace","path":"/spec/updateStrategy/partition","value":2}
]'
# PHASE: RollingOut (发 mysql-3, mysql-2)
# 4. 发现错误率上升！紧急暂停
kubectl patch neoset mysql --type='json' -p='[
{"op":"replace","path":"/spec/updateStrategy/pause","value":true}
]'
# PHASE: Paused
# 5. 回滚到旧版本
kubectl rollout undo neoset mysql --to-revision=1
# 所有 Pod 原地改回 mysql:8.0
OnDelete：手动控制每一步
除了 InPlace（自动原地升级）和 Recreate（自动重建），NeoSet 还支持 OnDelete——Controller 不主动更新，只有用户手动删 Pod 时重建的才用新 template。
spec:
updateStrategy:
type: OnDelete # 手动删才更新
什么时候用 OnDelete
- 主从切换不能自动——要先手动切主再更新
- 灰度验证极谨慎——人工确认每一步
- 维护窗口——只在指定时间手动触发
代码实现
// reconcileUpdate 里加 OnDelete 分支
func (r *NeoSetReconciler) reconcileUpdate(
ctx context.Context,
neoset *neov1alpha1.NeoSet,
pods []corev1.Pod,
) (bool, error) {
switch neoset.Spec.UpdateStrategy.Type {
case neov1alpha1.OnDeleteUpdateStrategyType:
// OnDelete：不主动更新，只统计已更新的
updatedCount := int32(0)
for _, pod := range pods {
if IsPodUpdated(&pod, neoset) {
updatedCount++
}
}
neoset.Status.UpdatedReplicas = updatedCount
if updatedCount == *neoset.Spec.Replicas {
r.completeUpdate(neoset)
}
return true, nil
case neov1alpha1.InPlaceUpdateStrategyType:
return r.reconcileInPlaceUpdate(ctx, neoset, pods)
case neov1alpha1.RecreateUpdateStrategyType:
return r.reconcileRecreateUpdate(ctx, neoset, pods)
}
return false, nil
}
OnDelete 模式下 Controller 的 Reconcile 很简单——不主动 PATCH，只读 Pod 看有几个已经是新版本，更新 status。用户手动 kubectl delete pod mysql-2，Pod 被重建时用 spec.template 的新 image，下一轮 Reconcile 统计到 updatedReplicas 增加。
更新 Hooks：让业务介入发版
发版不只是换镜像——有时要先做业务操作。比如 MySQL 主从切换前要先 SET GLOBAL read_only=ON，更新后再 SET GLOBAL read_only=OFF。
NeoSet 的 lifecycle 字段支持 PreUpdate 和 PostUpdate 回调：
spec:
updateStrategy:
lifecycle:
preUpdate:
exec:
command: ["/bin/sh", "-c", "mysql -e 'SET GLOBAL read_only=ON'"]
postUpdate:
exec:
command: ["/bin/sh", "-c", "mysql -e 'SET GLOBAL read_only=OFF'"]
代码实现
// internal/controller/hooks.go
// RunPreUpdateHook 更新前在 Pod 里执行 hook
func (r *NeoSetReconciler) RunPreUpdateHook(
ctx context.Context,
pod *corev1.Pod,
neoset *neov1alpha1.NeoSet,
) error {
lifecycle := neoset.Spec.UpdateStrategy.Lifecycle
if lifecycle == nil || lifecycle.PreUpdate == nil {
return nil
}
hook := lifecycle.PreUpdate
logger := log.FromContext(ctx)
if hook.Exec != nil {
// 在 Pod 里执行命令
req := r.ExecClient.Post().
Namespace(pod.Namespace).
Resource("pods").
Name(pod.Name).
SubResource("exec").
VersionedParams(&corev1.PodExecOptions{
Container: pod.Spec.Containers[0].Name,
Command: hook.Exec.Command,
Stdout: true,
Stderr: true,
}, scheme.ParameterCodec)
exec, err := remotecommand.NewSPDYExecutor(r.RestConfig, "POST", req.URL())
if err != nil {
return err
}
err = exec.StreamWithContext(ctx, remotecommand.StreamOptions{
Stdout: os.Stdout,
Stderr: os.Stderr,
})
if err != nil {
return fmt.Errorf("preUpdate hook 失败: %w", err)
}
logger.Info("preUpdate hook 执行成功", "pod", pod.Name)
}
return nil
}
// RunPostUpdateHook 更新后在 Pod 里执行 hook
func (r *NeoSetReconciler) RunPostUpdateHook(
ctx context.Context,
pod *corev1.Pod,
neoset *neov1alpha1.NeoSet,
) error {
lifecycle := neoset.Spec.UpdateStrategy.Lifecycle
if lifecycle == nil || lifecycle.PostUpdate == nil {
return nil
}
// 等 Pod ready 后再执行 postUpdate
if !IsPodReadyAfterUpdate(pod, neoset) {
return nil // 还没 ready，等下一轮
}
// 逻辑跟 PreUpdate 一样，执行 PostUpdate 的命令
// ...
return nil
}
带 Hooks 的发版时序
Reconcile:
1. RunPreUpdateHook(mysql-4)
→ mysql -e 'SET GLOBAL read_only=ON'
→ mysql-4 变只读，流量转到从库
2. InPlaceUpdatePod(mysql-4)
→ PATCH image，容器重启
3. 等 mysql-4 ready
4. RunPostUpdateHook(mysql-4)
→ mysql -e 'SET GLOBAL read_only=OFF'
→ mysql-4 恢复读写，流量回来
5. 更新下一个 Pod
Hooks 让业务逻辑介入发版流程——比如主从切换、缓存预热、连接池刷新。不用写额外脚本，配置在 YAML 里就行。
前置阅读
- 实现原地升级 — 本章前置，原地升级是分批的基础
- ControllerRevision版本管理 — 回滚逻辑依赖版本管理
- StatefulSet控制循环 — Controller 系列第 4 章，partition 概念
思考题
问题 1：金丝雀阶段（partition=replicas-1）完成后，phase 变成 Completed 还是 Canary？为什么？
问题 2：暂停后用户改了 partition（比如从 4 改成 2），然后恢复（pause=false）。Controller 会从哪个 Pod 开始更新？
问题 3：maxUnavailable=1，但当前有 2 个 Pod 不 ready（1 个在更新中，1 个应用 bug）。Controller 会怎样？
解答
问题 1：Completed。partition=4 时，只有 mysql-4 需要更新。mysql-4 原地升级 ready 后，UpdatedReplicas=1。但 UpdatedReplicas == replicas 的判断是 1 == 5，不成立，所以 phase 不会变 Completed。实际上 phase 会停在 Canary——因为还有 4 个 Pod 没更新。Controller 只更新了 partition 范围内的 Pod，其余的不动。phase 变 Completed 的条件是"所有 Pod 都更新到 targetImage"，金丝雀只更新了 1 个，所以不是 Completed。
问题 2：从最大序号开始，跳过已更新的。mysql-4 已经是新版本（金丝雀发的），Controller 检查 partition=2，发现 mysql-3 和 mysql-2 还没更新，从 mysql-3 开始原地升级。mysql-4 不会再动（已经是目标版本）。
问题 3：Controller 停住不更新。canUpdateMore 算 unavailable=2，maxUnavailable=1，2 >= 1，返回 false。Controller 等那个应用 bug 的 Pod 恢复 ready 后才继续。这是保守策略——防止不可用 Pod 越来越多。如果应用 bug 的 Pod 一直不 ready，发版会卡住。解决办法：修 bug 让 Pod 恢复，或者手动 kubectl delete pod 逼它重建。
下一章
下一章实现用旧镜像扩容——发版途中扩容时，新 Pod 用稳定版本镜像。