Reese/yms-daemon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: clarify reconciliation and preserve container image tags

Browse Source
This commit is contained in:
Zhan Ziyang
2026-08-22 15:33:27 +08:00
parent f536987a7e
commit a0a4a7556a
7 changed files with 677 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/backendupdate/container.go
+160 -19
View File
@@ -23,6 +23,15 @@ import (
// inputTypeContainerImage 标识以容器镜像方式执行后端更新的输入类型。
const inputTypeContainerImage = "container-image"
// inputTypeContainerRestart 标识容器后端同版本重启的输入类型，用于事务恢复时区分
// 普通更新与重启，从而在重启时复用本地镜像而不重新拉取。
const inputTypeContainerRestart = "container-restart"
// defaultContainerStopGraceSeconds 停止旧容器时，发送停止信号后到强制终止前的默认等待秒数，
// 当 backend.stop_grace_seconds 未配置时使用。该值覆盖容器创建时设置的 9000 秒 StopTimeout，
// 避免旧容器因优雅停机超时长期无法退出。
const defaultContainerStopGraceSeconds = 30
// persistedContainerRequest 容器后端更新请求的持久化形态，以 JSON 存入事务
// 记录。事务在提交、重试或补偿时依赖这些字段重建执行上下文。
type persistedContainerRequest struct {
@@ -69,19 +78,8 @@ func (u *Updater) UpdateContainerImage(ctx context.Context, imageReference strin
if strings.TrimSpace(imageReference) != imageReference || imageReference == "" {
return transaction.Transaction{}, errors.New("exact container image reference is required")
}
configInfo, err := os.Lstat(u.containerConfigSource)
if err != nil {
return transaction.Transaction{}, fmt.Errorf("inspect backend configuration %s: %w", u.containerConfigSource, err)
}
if !configInfo.Mode().IsRegular() || configInfo.Mode()&os.ModeSymlink != 0 {
return transaction.Transaction{}, fmt.Errorf("backend configuration is not a direct regular file: %s", u.containerConfigSource)
}
tmpInfo, err := os.Lstat(u.containerTmpSource)
if err != nil {
return transaction.Transaction{}, fmt.Errorf("inspect backend temporary directory %s: %w", u.containerTmpSource, err)
}
if !tmpInfo.IsDir() || tmpInfo.Mode()&os.ModeSymlink != 0 {
return transaction.Transaction{}, fmt.Errorf("backend temporary path is not a direct directory: %s", u.containerTmpSource)
if err := u.validateContainerInputs(); err != nil {
return transaction.Transaction{}, err
}
active, err := u.store.ActiveTransaction(ctx)
@@ -183,6 +181,142 @@ func (u *Updater) UpdateContainerImage(ctx context.Context, imageReference strin
return u.runContainerUpdate(ctx, record, request, created, report)
}
// RestartContainer 使用当前已提交部署记录的镜像执行一次同版本零停机轮转：它复用本地
// 镜像（不重新拉取），把同一镜像部署到非活动槽位，健康检查通过后切换流量并停止旧容器。
//
// 参数 ctx 用于控制整个重启过程的取消；report 用于回传实时进度，可为 nil。返回值为本次
// 重启对应的事务记录以及错误。若不存在已提交部署记录、现场与部署记录不一致，或存在其他
// 未完成事务，则拒绝新建重启；存在未完成的重启事务时复用续跑。
func (u *Updater) RestartContainer(ctx context.Context, report ProgressReporter) (transaction.Transaction, error) {
if u.containerExecutor == nil || u.engine == nil {
return transaction.Transaction{}, errors.New("container backend updater is not configured")
}
if err := u.validateContainerInputs(); err != nil {
return transaction.Transaction{}, err
}
deployment, err := u.store.BackendContainerDeployment(ctx)
if err != nil {
return transaction.Transaction{}, fmt.Errorf("container backend restart requires a committed deployment: %w", err)
}
active, err := u.store.ActiveTransaction(ctx)
if err == nil {
var request persistedContainerRequest
if decodeErr := decodeContainerRequest(active.Request, &request); decodeErr != nil {
return active, decodeErr
}
if active.Service != serviceBackend || request.InputType != inputTypeContainerRestart {
return active, &transaction.ActiveTransactionError{TransactionID: active.ID}
}
return u.runContainerUpdate(ctx, active, request, false, report)
}
if !errors.Is(err, transaction.ErrNotFound) {
return transaction.Transaction{}, err
}
before, err := u.gateway.Read()
if err != nil {
return transaction.Transaction{}, err
}
if before.ActivePort != deployment.ActivePort {
return transaction.Transaction{}, fmt.Errorf("container backend restart requires gateway active port %d, got %d", deployment.ActivePort, before.ActivePort)
}
activeContainer, err := u.engine.InspectContainer(ctx, deployment.ContainerName)
if err != nil {
return transaction.Transaction{}, fmt.Errorf("inspect active backend container %s: %w", deployment.ContainerName, err)
}
if activeContainer.ID != deployment.ContainerID {
return transaction.Transaction{}, fmt.Errorf("active backend container %s identity does not match committed deployment", deployment.ContainerName)
}
if !activeContainer.Running || activeContainer.Dead {
return transaction.Transaction{}, fmt.Errorf("active backend container %s is not running", deployment.ContainerName)
}
image, err := u.engine.InspectImage(ctx, activeContainer.ImageReference)
if err != nil {
return transaction.Transaction{}, fmt.Errorf("inspect active backend image %s: %w", activeContainer.ImageReference, err)
}
if image.Platform.OS == "" || image.Platform.Architecture == "" {
return transaction.Transaction{}, errors.New("active backend image does not report an exact platform")
}
targetPort := otherPort(deployment.ActivePort)
targetSlot, err := u.config.Backend.SlotForPort(targetPort)
if err != nil {
return transaction.Transaction{}, err
}
previousSlot, err := u.config.Backend.SlotForPort(deployment.ActivePort)
if err != nil {
return transaction.Transaction{}, err
}
afterContent, err := hostnginx.RenderBackendPort(before.Content, targetPort)
if err != nil {
return transaction.Transaction{}, err
}
transactionID := rand.Text()
transactionRoot := filepath.Join(u.workRoot, transactionID)
if err := os.MkdirAll(transactionRoot, 0o750); err != nil {
return transaction.Transaction{}, fmt.Errorf("create container backend transaction directory: %w", err)
}
beforePath := filepath.Join(transactionRoot, "gateway.before.conf")
afterPath := filepath.Join(transactionRoot, "gateway.after.conf")
if err := writeImmutableFile(beforePath, before.Content, 0o640); err != nil {
return transaction.Transaction{}, err
}
if err := writeImmutableFile(afterPath, afterContent, 0o640); err != nil {
return transaction.Transaction{}, err
}
request := persistedContainerRequest{
InputType: inputTypeContainerRestart, ImageReference: activeContainer.ImageReference,
ImmutableReference: activeContainer.ImageReference, ImageDigest: deployment.ImageDigest, Platform: image.Platform,
TargetPort: targetPort, TargetContainer: targetSlot.ContainerName,
PreviousPort: deployment.ActivePort, PreviousContainer: previousSlot.ContainerName,
TargetHealthEndpoint: targetSlot.HealthEndpoint,
StartLog: true,
GatewayBeforePath: beforePath, GatewayAfterPath: afterPath,
GatewayReceiptPath: filepath.Join(transactionRoot, "gateway.applied"),
}
content, err := json.Marshal(request)
if err != nil {
return transaction.Transaction{}, fmt.Errorf("encode container backend restart request: %w", err)
}
record, created, err := u.store.CreateTransaction(ctx, transaction.CreateRequest{
ID: transactionID, IdempotencyKey: serviceBackend + ":restart:" + rand.Text(),
Source: sourceLocalCLI, Service: serviceBackend, Request: content,
})
if err != nil {
return transaction.Transaction{}, err
}
if !created {
if err := decodeContainerRequest(record.Request, &request); err != nil {
return record, err
}
}
return u.runContainerUpdate(ctx, record, request, created, report)
}
// validateContainerInputs 校验容器后端的配置源文件与临时目录满足部署前置条件：
// 配置源必须是直接普通文件，临时目录必须是直接目录。
func (u *Updater) validateContainerInputs() error {
configInfo, err := os.Lstat(u.containerConfigSource)
if err != nil {
return fmt.Errorf("inspect backend configuration %s: %w", u.containerConfigSource, err)
}
if !configInfo.Mode().IsRegular() || configInfo.Mode()&os.ModeSymlink != 0 {
return fmt.Errorf("backend configuration is not a direct regular file: %s", u.containerConfigSource)
}
tmpInfo, err := os.Lstat(u.containerTmpSource)
if err != nil {
return fmt.Errorf("inspect backend temporary directory %s: %w", u.containerTmpSource, err)
}
if !tmpInfo.IsDir() || tmpInfo.Mode()&os.ModeSymlink != 0 {
return fmt.Errorf("backend temporary path is not a direct directory: %s", u.containerTmpSource)
}
return nil
}
// resolveContainerSlots 在选择目标槽位前，核对已提交的部署、网关以及两个精确
// 容器名是否一致。全新安装没有部署记录、没有已提交的容器事务历史，也没有槽位
// 容器，会从非路由槽位启动，这样只有健康检查通过后才会暴露流量。
@@ -318,9 +452,14 @@ func (u *Updater) runContainerUpdate(ctx context.Context, record transaction.Tra
if record.State.Terminal() {
return terminalResult(record)
}
imageAcquisition := backendexecutor.ImageAcquisitionPull
if request.InputType == inputTypeContainerRestart {
imageAcquisition = backendexecutor.ImageAcquisitionPresent
}
executorRequest := backendexecutor.Request{
ImageAcquisition: backendexecutor.ImageAcquisitionPull,
ImageReference: request.ImmutableReference, ExpectedImageDigest: request.ImageDigest, Platform: request.Platform,
ImageAcquisition: imageAcquisition,
ImageReference: request.ImmutableReference, DisplayImageReference: request.ImageReference,
ExpectedImageDigest: request.ImageDigest, Platform: request.Platform,
ContainerName: request.TargetContainer, Port: request.TargetPort,
PortEnvironmentKey: deploymentconfig.ContainerPortEnvironment,
ConfigSource: u.containerConfigSource, ConfigTarget: u.containerConfigTarget,
@@ -400,7 +539,7 @@ func (u *Updater) switchAndCommitContainer(ctx context.Context, transactionID st
if err := waitContext(ctx, u.drain); err != nil {
return err
}
operation := &containerStopOperation{engine: u.engine, name: request.PreviousContainer}
operation := &containerStopOperation{engine: u.engine, name: request.PreviousContainer, stopGraceSeconds: u.containerStopGraceSeconds}
if _, err := u.coordinator.ExecuteStep(ctx, transactionID, stopPreviousContainerIntent(request), operation); err != nil {
return err
}
@@ -412,7 +551,7 @@ func (u *Updater) switchAndCommitContainer(ctx context.Context, transactionID st
if !target.Running || target.Dead || target.ID == "" {
return fmt.Errorf("target backend container %s is not running with an exact identity", request.TargetContainer)
}
if target.ImageReference != request.ImmutableReference {
if target.ImageReference != request.ImmutableReference && target.ImageReference != request.ImageReference {
return fmt.Errorf("target backend container %s image reference does not match the transaction", request.TargetContainer)
}
_, err = u.store.CommitBackendContainerDeployment(ctx, transactionID, transaction.BackendContainerDeployment{
@@ -450,7 +589,7 @@ func (u *Updater) rollbackContainer(ctx context.Context, transactionID string, r
}
gateway := &gatewayOperation{controller: u.gateway, before: after, after: before, receiptPath: request.GatewayReceiptPath + ".restore"}
_, gatewayErr := u.coordinator.ExecuteStep(ctx, transactionID, containerGatewayRestoreIntent(request), gateway)
stop := &containerStopOperation{engine: u.engine, name: request.TargetContainer}
stop := &containerStopOperation{engine: u.engine, name: request.TargetContainer, stopGraceSeconds: u.containerStopGraceSeconds}
_, stopErr := u.coordinator.ExecuteStep(ctx, transactionID, stopTargetContainerIntent(request), stop)
if err := errors.Join(gatewayErr, stopErr); err != nil {
return errors.Join(cause, err)
@@ -465,11 +604,13 @@ type containerStopOperation struct {
engine containerengine.Engine
// name 待停止的容器名。
name string
// stopGraceSeconds 发送停止信号后到强制终止前的等待秒数。
stopGraceSeconds int
}
// Apply 停止指定容器，若容器不存在则视为已满足（幂等成功）。
func (o *containerStopOperation) Apply(ctx context.Context) error {
err := o.engine.StopContainer(ctx, o.name)
err := o.engine.StopContainer(ctx, o.name, o.stopGraceSeconds)
if errors.Is(err, containerengine.ErrNotFound) {
return nil
}
internal/backendupdate/container_test.go
+96 -4
View File
@@ -190,6 +190,94 @@ func TestContainerUpdaterRejectsMissingCommittedContainer(t *testing.T) {
}
}
// TestContainerUpdaterRestartsContainerWithLocalImage 验证 container 后端重启会复用当前已提交部署
// 记录的镜像：轮转到非活跃槽位、健康检查通过后切流并停止旧容器，且整个过程不重新拉取镜像。
func TestContainerUpdaterRestartsContainerWithLocalImage(t *testing.T) {
ctx := context.Background()
updater, store, engine, gateway := newContainerUpdaterFixture(t, map[string]containerengine.Container{
"backend-8080": {
ID: "committed-backend-8080", Name: "backend-8080", Running: true,
ImageReference: "harbor.ymswell.asia/ymswell/glory-ymswell@" + containerTestDigest,
},
}, 0)
commitContainerDeployment(t, store, transaction.BackendContainerDeployment{
ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: containerTestDigest, ContainerID: "committed-backend-8080",
})
record, err := updater.RestartContainer(ctx, nil)
if err != nil {
t.Fatalf("restart container backend: %v", err)
}
if record.State != transaction.StateCommitted || gateway.snapshot.ActivePort != 8081 {
t.Fatalf("unexpected committed container restart: record=%+v gateway=%+v", record, gateway.snapshot)
}
if engine.containers["backend-8080"].Running {
t.Fatal("previous backend container is still running after restart")
}
if engine.pullCalls != 0 {
t.Fatalf("restart must reuse the local image without pulling: pulls=%d", engine.pullCalls)
}
assertCommittedContainerDeployment(t, store, record.ID, 8081, "backend-8081", "container-id-backend-8081")
}
// TestContainerUpdaterRestartRejectsWithoutDeployment 验证不存在已提交部署记录时，container 后端重启被拒绝。
func TestContainerUpdaterRestartRejectsWithoutDeployment(t *testing.T) {
updater, _, _, _ := newContainerUpdaterFixture(t, map[string]containerengine.Container{}, 0)
_, err := updater.RestartContainer(context.Background(), nil)
if err == nil || !strings.Contains(err.Error(), "requires a committed deployment") {
t.Fatalf("unexpected restart without deployment: %v", err)
}
}
// TestContainerUpdaterRestartRejectsGatewayDrift 验证宿主 Nginx 活动端口与部署记录不一致时，container 后端重启被拒绝。
func TestContainerUpdaterRestartRejectsGatewayDrift(t *testing.T) {
updater, store, _, gateway := newContainerUpdaterFixture(t, map[string]containerengine.Container{
"backend-8080": {
ID: "committed-backend-8080", Name: "backend-8080", Running: true,
ImageReference: "harbor.ymswell.asia/ymswell/glory-ymswell@" + containerTestDigest,
},
}, 0)
commitContainerDeployment(t, store, transaction.BackendContainerDeployment{
ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: containerTestDigest, ContainerID: "committed-backend-8080",
})
gateway.snapshot.ActivePort = 8081
_, err := updater.RestartContainer(context.Background(), nil)
if err == nil || !strings.Contains(err.Error(), "requires gateway active port 8080, got 8081") {
t.Fatalf("unexpected restart with gateway drift: %v", err)
}
}
// commitContainerDeployment 通过合法的状态机推进写入一条 container 部署记录，供重启测试预置现场。
func commitContainerDeployment(t *testing.T, store *transaction.Store, deployment transaction.BackendContainerDeployment) {
t.Helper()
ctx := context.Background()
transactionID := "seed-" + deployment.ContainerName
_, _, err := store.CreateTransaction(ctx, transaction.CreateRequest{
ID: transactionID, IdempotencyKey: "backend:container:seed:" + deployment.ContainerName, Source: sourceLocalCLI,
Service: serviceBackend, Request: []byte(`{"inputType":"container-image"}`),
})
if err != nil {
t.Fatalf("create seed backend transaction: %v", err)
}
for _, state := range []transaction.State{
transaction.StateValidating,
transaction.StatePrepared,
transaction.StateStarting,
transaction.StateSwitching,
transaction.StateVerifying,
transaction.StateDraining,
} {
if _, err := store.Transition(ctx, transactionID, state, "seed committed deployment"); err != nil {
t.Fatalf("transition seed backend transaction to %s: %v", state, err)
}
}
if _, err := store.CommitBackendContainerDeployment(ctx, transactionID, deployment, "seed committed backend container deployment"); err != nil {
t.Fatalf("commit seed backend deployment: %v", err)
}
}
// newContainerUpdaterFixture 构造容器更新器的测试夹具，返回更新器、事务存储、假容器引擎和内存网关。
// containers 指定引擎初始存在的容器；drain 指定旧容器停止前的排水等待时长。
func newContainerUpdaterFixture(
@@ -263,7 +351,7 @@ func newContainerUpdaterFixture(
// 环境变量和挂载点均符合容器更新契约。
func assertCreatedContainerSpec(t *testing.T, request containerengine.ContainerSpec) {
t.Helper()
if request.Name != "backend-8081" || request.ImageReference != "harbor.ymswell.asia/ymswell/glory-ymswell@"+containerTestDigest {
if request.Name != "backend-8081" || request.ImageReference != containerTestImage {
t.Fatalf("unexpected target container identity: %+v", request)
}
if request.NetworkMode != "host" || request.RestartPolicy.Name != "no" || request.User != "0:0" {
@@ -325,13 +413,17 @@ type containerUpdateEngine struct {
stopped []string
lastCreateSpec containerengine.ContainerSpec
healthChecks int
pullCalls int
}
// Ping 返回 nil，模拟引擎连通性检查始终成功。
func (e *containerUpdateEngine) Ping(context.Context) error { return nil }
// PullImage 返回 nil，模拟镜像拉取始终成功。
func (e *containerUpdateEngine) PullImage(context.Context, string) error { return nil }
// PullImage 记录一次拉取调用并返回 nil，模拟镜像拉取始终成功。
func (e *containerUpdateEngine) PullImage(context.Context, string) error {
e.pullCalls++
return nil
}
// LoadImage 返回 nil，模拟从流加载镜像始终成功。
func (e *containerUpdateEngine) LoadImage(context.Context, io.Reader) error { return nil }
@@ -381,7 +473,7 @@ func (e *containerUpdateEngine) ContainerLogs(context.Context, string) (io.ReadC
}
// StopContainer 记录被停止的容器名称并将该容器标记为已退出；若容器不存在则返回 containerengine.ErrNotFound。
func (e *containerUpdateEngine) StopContainer(_ context.Context, name string) error {
func (e *containerUpdateEngine) StopContainer(_ context.Context, name string, _ int) error {
e.stopped = append(e.stopped, name)
record, found := e.containers[name]
if !found {