From 1abe8df6eeddc6b0e8d07d671f4609049143e272 Mon Sep 17 00:00:00 2001 From: Zhan Ziyang Date: Sat, 22 Aug 2026 15:34:09 +0800 Subject: [PATCH] feat: complete daemon operations and recovery tooling --- CHANGELOG.md | 33 +++ CLI_NAMING_AND_OPERATIONS_PLAN.md | 140 ----------- UPDATE_PLAN.md => PLAN.md | 144 +++++++++++ internal/backendexecutor/executor_test.go | 2 +- internal/backendstatus/container.go | 203 +++++++++++++++ internal/backendstatus/container_test.go | 291 ++++++++++++++++++++++ internal/backendstatus/diagnoser.go | 157 ++++++++++++ internal/backendstatus/native.go | 137 ++++++++++ internal/backendstatus/reconcile.go | 187 ++++++++++++++ internal/backendupdate/restart.go | 2 +- internal/backendupdate/updater.go | 15 +- internal/containerengine/engine.go | 4 +- internal/containerengine/moby.go | 25 +- internal/daemonclient/client.go | 42 ++++ internal/daemonserver/server.go | 56 ++++- internal/daemonserver/server_test.go | 77 +++++- internal/deploymentconfig/config.go | 8 +- internal/deploymentconfig/config_test.go | 22 ++ packaging/etc/yms-daemon/yms-daemon.toml | 47 ++++ 19 files changed, 1421 insertions(+), 171 deletions(-) create mode 100644 CHANGELOG.md delete mode 100644 CLI_NAMING_AND_OPERATIONS_PLAN.md rename UPDATE_PLAN.md => PLAN.md (94%) create mode 100644 internal/backendstatus/container.go create mode 100644 internal/backendstatus/container_test.go create mode 100644 internal/backendstatus/diagnoser.go create mode 100644 internal/backendstatus/native.go create mode 100644 internal/backendstatus/reconcile.go diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..415c9df --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,33 @@ +# Changelog + +yms-daemon 面向用户的版本历史，遵循 [Keep a Changelog](https://keepachangelog.com/zh-CN/1.1.0/) 风格与语义化版本（SemVer）。 + +当前版本：**0.1.0**（POC 阶段，尚未达到 RC/GA）。 + +版本号策略：POC 阶段沿用 `0.1.x`；功能达到 RC 时发布 `0.1.0-rc.N`（或按规模升 `0.2.0-rc.N`）；达到 GA 时发布 `1.0.0`，并把对应条目从 `[Unreleased]` 移入正式版本号。 + +## [Unreleased] + +### Added + +- 命令入口拆分：`ymsd`（守护进程，监听 Unix Socket）/ `ymsctl`（运维 CLI），chrony 风格分离。 +- backend native 更新：repack ZIP（`-f`）与直传 JAR（`--native-jar`）两种互斥输入。 +- backend container 更新：`--container-image`（pull 后冻结 repository digest，蓝绿切流，含全新机器首次部署）。 +- 同版本零停机重启：`ymsctl restart --service backend`（native + container）。 +- 历史查询：`ymsctl list [--limit/--service/--state/--json]`。 +- 诊断与对账：`ymsctl status` / `doctor` / `reconcile [--apply]`。 +- 宿主 Nginx 蓝绿切流：managed upstream 块 + `nginx -s reload`。 +- 服务端 SQLite 事务、步骤幂等与 backend container 部署记录。 +- 启动日志跟随：update / restart 期间实时输出容器启动日志，直到 Actuator 健康。 + +### Changed + +- 更新命令由旧 `yms-daemon update` 迁移为 `ymsctl update`。 +- `hostnginx.NewController` 简化为两参数，重载改走 `nginx -s reload`。 +- 容器停止旧容器改为显式超时（默认 30 秒后强制终止），可通过 `backend.stop_grace_seconds` 配置，覆盖原先 9000 秒 StopTimeout。 +- 示例配置 `yms-daemon.toml` 同时给出 native / container 两套 `[backend]` 块与切换说明。 + +### Known limitations + +- native 后端的 `reconcile --apply` 尚无自动修复项（JAR 链接漂移修复待实现）。 +- `rollback`、`frontend`/`nodeSsr`/`all` 更新入口、离线 repack 交付尚未实现。 diff --git a/CLI_NAMING_AND_OPERATIONS_PLAN.md b/CLI_NAMING_AND_OPERATIONS_PLAN.md deleted file mode 100644 index 8348e1c..0000000 --- a/CLI_NAMING_AND_OPERATIONS_PLAN.md +++ /dev/null @@ -1,140 +0,0 @@ -# YMS 命令入口与操作查询方案 - -> 状态：方案确认，暂不进入代码实现 - -## 1. 命令职责 - -命令入口采用 daemon/client 分离，命名参考 `chronyd / chronyc`： - -| 命令 | 职责 | 是否常驻 | -|---|---|---:| -| `ymsd` | 更新助手服务端，监听 Unix Socket，执行事务和恢复 | 是 | -| `ymsctl` | 运维和交付 CLI，提交更新、查询状态、诊断和回滚 | 否 | -| `yms-gui` | 客户 PC 图形化中转助手，名称后续确定 | 否 | - -`ymsd` 不承担交互式更新命令；更新请求由 `ymsctl` 通过 Unix Socket 提交。POC 阶段不对外提供 `serve` 子命令，systemd 的 `ExecStart` 直接启动 `/usr/sbin/ymsd`。 - -## 2. systemd 与安装 - -systemd unit 名称继续固定为： - -```text -yms-daemon.service -``` - -unit 的启动目标为： - -```text -/usr/sbin/ymsd -``` - -POC 阶段不保留 `yms-daemon` 命令兼容入口，直接采用新命名。更新命令统一使用： - -```bash -ymsctl update ... -ymsctl restart ... -``` - -## 3. ymsctl 命令层级 - -### 3.1 更新与回滚 - -```bash -ymsctl update --service backend -f -ymsctl update --service backend --native-jar -ymsctl update --service backend --container-image -ymsctl restart --service backend -ymsctl rollback --service backend --transaction -``` - -`--native-jar` 与 `--container-image` 互斥。`rollback` 必须指定明确的事务 ID 或已提交版本身份，不允许根据目录排序、文件修改时间或镜像 tag 猜测回滚目标。 - -### 3.2 历史操作查询 - -```bash -ymsctl list -ymsctl list --limit 50 -ymsctl list --state FAILED -ymsctl list --service backend -ymsctl list --json -``` - -默认显示最近 20 次操作，按创建时间倒序。每条记录至少展示： - -- 创建时间和结束时间； -- transaction ID； -- service； -- operation； -- 输入类型和版本或镜像 digest； -- 来源（CLI、PC client、Jenkins、server API）； -- 最终状态； -- 失败原因。 - -`--json` 输出稳定 JSON，供 Jenkins 和 PC client 使用。列表数据只读取 SQLite，不直接扫描 systemd、Docker 或 Nginx 生成历史记录。 - -### 3.3 当前状态与诊断 - -```bash -ymsctl status --service backend -ymsctl doctor --service backend -ymsctl reconcile --service backend -ymsctl reconcile --service backend --apply -``` - -- `status`：读取 SQLite 后校验当前 systemd、Docker、Nginx 和活动槽位状态； -- `doctor`：只读诊断，输出漂移项和建议动作； -- `reconcile`：只读生成修复计划； -- `reconcile --apply`：执行明确授权的修复动作，并创建可审计事务。 - -## 4. 状态漂移处理原则 - -SQLite 是事务事实来源，外部系统是待校验运行状态。发现不一致时默认拒绝更新，不自动覆盖现场。 - -只有以下类型允许自动修复： - -1. SQLite 记录的活动容器存在、身份匹配且健康，Nginx 仅指向错误槽位； -2. 非活动槽位容器存在、已停止且没有提交记录； -3. native 活动 JAR 链接与已提交事务明确记录的 release 不一致，且目标文件身份校验通过； -4. 已完成事务的临时文件可以按照事务凭据清理。 - -以下情况必须进入 `DRIFT_DETECTED`，等待人工确认： - -- SQLite 记录的活动容器已经不存在； -- 两个槽位同时运行且无法确定提交归属； -- SQLite 没有部署记录，但现场已经存在容器或运行中的 native 服务； -- Nginx 配置无法验证或存在非 daemon 管理的冲突； -- systemd unit、容器镜像 digest、JAR SHA-256 与事务记录不一致。 - -任何修复动作都必须先写入事务，再执行外部变更；修复失败时沿用现有回滚和恢复机制。 - -## 5. systemd 服务与 ymsd 的边界 - -`ymsd` 负责： - -- 持有进程锁； -- 打开 SQLite； -- 恢复未完成事务； -- 调用 native/container 执行器； -- 通过 Unix Socket 提供请求和进度响应； -- 写入 stdout 和本地日志文件。 - -`ymsctl` 负责： - -- 参数校验； -- 提交请求； -- 展示进度； -- 查询历史和当前状态； -- 发起显式诊断和修复。 - -`ymsctl` 不直接写 SQLite，不直接修改 Nginx，不直接调用 Docker 或 systemd。 - -## 6. 实施顺序 - -1. 增加 SQLite 最近操作查询接口； -2. 实现 `ymsctl list` 和 JSON 输出； -3. 拆分 `ymsd` 与 `ymsctl` 的编译入口； -4. 安装 `ymsd`、`ymsctl` 和兼容入口 `yms-daemon`； -5. 实现 `status` 只读一致性检查； -6. 实现 `doctor` 漂移报告； -7. 实现带事务审计的 `reconcile --apply`； -8. 最后确定 GUI 名称和 PC client 的调用协议。 diff --git a/UPDATE_PLAN.md b/PLAN.md similarity index 94% rename from UPDATE_PLAN.md rename to PLAN.md index 20e8458..9c25974 100644 --- a/UPDATE_PLAN.md +++ b/PLAN.md @@ -1790,3 +1790,147 @@ Kubernetes executor 不在每个业务 Pod 中运行 daemon。它负责： 37. 未提交文件的隔离目录、识别规则、保留期限和清理流程。 上述精确项冻结前，不修改现网更新脚本、systemd、Nginx 配置和发布包格式。 + + +--- + +# 附录：YMS 命令入口与操作查询方案 + +> 状态：方案确认。ymsd/ymsctl 拆分、list、status、doctor、reconcile 已实现；restart 的 container 后端已实现；rollback 待实现。 + +## 1. 命令职责 + +命令入口采用 daemon/client 分离，命名参考 `chronyd / chronyc`： + +| 命令 | 职责 | 是否常驻 | +|---|---|---:| +| `ymsd` | 更新助手服务端，监听 Unix Socket，执行事务和恢复 | 是 | +| `ymsctl` | 运维和交付 CLI，提交更新、查询状态、诊断和回滚 | 否 | +| `yms-gui` | 客户 PC 图形化中转助手，名称后续确定 | 否 | + +`ymsd` 不承担交互式更新命令；更新请求由 `ymsctl` 通过 Unix Socket 提交。POC 阶段不对外提供 `serve` 子命令，systemd 的 `ExecStart` 直接启动 `/usr/bin/ymsd`。 + +## 2. systemd 与安装 + +systemd unit 名称继续固定为： + +```text +yms-daemon.service +``` + +unit 的启动目标为： + +```text +/usr/bin/ymsd +``` + +POC 阶段不保留 `yms-daemon` 命令兼容入口，直接采用新命名。更新命令统一使用： + +```bash +ymsctl update ... +ymsctl restart ... +``` + +## 3. ymsctl 命令层级 + +### 3.1 更新与回滚 + +```bash +ymsctl update --service backend -f +ymsctl update --service backend --native-jar +ymsctl update --service backend --container-image +ymsctl restart --service backend +ymsctl rollback --service backend --transaction +``` + +`--native-jar` 与 `--container-image` 互斥。`rollback` 必须指定明确的事务 ID 或已提交版本身份，不允许根据目录排序、文件修改时间或镜像 tag 猜测回滚目标。 + +### 3.2 历史操作查询 + +```bash +ymsctl list +ymsctl list --limit 50 +ymsctl list --state FAILED +ymsctl list --service backend +ymsctl list --json +``` + +默认显示最近 20 次操作，按创建时间倒序。每条记录至少展示： + +- 创建时间和结束时间； +- transaction ID； +- service； +- operation； +- 输入类型和版本或镜像 digest； +- 来源（CLI、PC client、Jenkins、server API）； +- 最终状态； +- 失败原因。 + +`--json` 输出稳定 JSON，供 Jenkins 和 PC client 使用。列表数据只读取 SQLite，不直接扫描 systemd、Docker 或 Nginx 生成历史记录。 + +### 3.3 当前状态与诊断 + +```bash +ymsctl status --service backend +ymsctl doctor --service backend +ymsctl reconcile --service backend +ymsctl reconcile --service backend --apply +``` + +- `status`：读取 SQLite 后校验当前 systemd、Docker、Nginx 和活动槽位状态； +- `doctor`：只读诊断，输出漂移项和建议动作； +- `reconcile`：只读生成修复计划； +- `reconcile --apply`：执行明确授权的修复动作，并创建可审计事务。 + +## 4. 状态漂移处理原则 + +SQLite 是事务事实来源，外部系统是待校验运行状态。发现不一致时默认拒绝更新，不自动覆盖现场。 + +只有以下类型允许自动修复： + +1. SQLite 记录的活动容器存在、身份匹配且健康，Nginx 仅指向错误槽位； +2. 非活动槽位容器存在、已停止且没有提交记录； +3. native 活动 JAR 链接与已提交事务明确记录的 release 不一致，且目标文件身份校验通过； +4. 已完成事务的临时文件可以按照事务凭据清理。 + +以下情况必须进入 `DRIFT_DETECTED`，等待人工确认： + +- SQLite 记录的活动容器已经不存在； +- 两个槽位同时运行且无法确定提交归属； +- SQLite 没有部署记录，但现场已经存在容器或运行中的 native 服务； +- Nginx 配置无法验证或存在非 daemon 管理的冲突； +- systemd unit、容器镜像 digest、JAR SHA-256 与事务记录不一致。 + +任何修复动作都必须先写入事务，再执行外部变更；修复失败时沿用现有回滚和恢复机制。 + +## 5. systemd 服务与 ymsd 的边界 + +`ymsd` 负责： + +- 持有进程锁； +- 打开 SQLite； +- 恢复未完成事务； +- 调用 native/container 执行器； +- 通过 Unix Socket 提供请求和进度响应； +- 写入 stdout 和本地日志文件。 + +`ymsctl` 负责： + +- 参数校验； +- 提交请求； +- 展示进度； +- 查询历史和当前状态； +- 发起显式诊断和修复。 + +`ymsctl` 不直接写 SQLite，不直接修改 Nginx，不直接调用 Docker 或 systemd。 + +## 6. 实施顺序 + +1. 增加 SQLite 最近操作查询接口； +2. 实现 `ymsctl list` 和 JSON 输出； +3. 拆分 `ymsd` 与 `ymsctl` 的编译入口； +4. 安装 `ymsd`、`ymsctl` 和兼容入口 `yms-daemon`； +5. 实现 `status` 只读一致性检查； +6. 实现 `doctor` 漂移报告； +7. 实现带事务审计的 `reconcile --apply`； +8. 最后确定 GUI 名称和 PC client 的调用协议。 diff --git a/internal/backendexecutor/executor_test.go b/internal/backendexecutor/executor_test.go index e4b0f4e..3abde7e 100644 --- a/internal/backendexecutor/executor_test.go +++ b/internal/backendexecutor/executor_test.go @@ -451,7 +451,7 @@ func (e *fakeEngine) ContainerLogs(context.Context, string) (io.ReadCloser, erro } // StopContainer 实现容器引擎的 StopContainer，将指定容器标记为已退出状态，不存在时返回 ErrNotFound。 -func (e *fakeEngine) StopContainer(_ context.Context, name string) error { +func (e *fakeEngine) StopContainer(_ context.Context, name string, _ int) error { e.mu.Lock() defer e.mu.Unlock() record, exists := e.containers[name] diff --git a/internal/backendstatus/container.go b/internal/backendstatus/container.go new file mode 100644 index 0000000..678dee1 --- /dev/null +++ b/internal/backendstatus/container.go @@ -0,0 +1,203 @@ +package backendstatus + +import ( + "context" + "errors" + "fmt" + + "yms-daemon/internal/containerengine" + "yms-daemon/internal/daemonapi" + "yms-daemon/internal/deploymentconfig" + "yms-daemon/internal/transaction" +) + +// 诊断项稳定标识，供测试与 reconcile 识别，不参与展示。 +const ( + codeUndeployed = "container_backend_undeployed" + codeUndeployedWithHistory = "container_backend_undeployed_with_history" + codeContainerWithoutRecord = "container_present_without_deployment_record" + codeCommittedPortOK = "nginx_active_port" + codeNginxWrongPort = "nginx_wrong_port" + codeActiveContainerOK = "active_container_healthy" + codeActiveContainerMissing = "active_container_missing" + codeActiveContainerIdentity = "active_container_identity_mismatch" + codeActiveContainerNotRunning = "active_container_not_running" + codeInactiveContainerOK = "inactive_container_absent" + codeInactiveContainerResidual = "inactive_container_residual" + codeInactiveContainerRunning = "inactive_container_running" + codeDeploymentConfigMismatch = "deployment_record_config_mismatch" +) + +// diagnoseContainer 对 container 后端做只读诊断。 +// 事实来源是 backend_container_deployment 单例记录；现场状态由容器引擎与宿主 Nginx 提供。 +func (d *Diagnoser) diagnoseContainer(ctx context.Context) (daemonapi.Diagnosis, error) { + findings, err := d.containerGather(ctx) + if err != nil { + return daemonapi.Diagnosis{}, err + } + return toDiagnosis("backend", deploymentconfig.BackendTypeContainer, findings), nil +} + +// containerGather 读取 container 后端的部署记录、历史事务与宿主 Nginx 状态，生成全部诊断项。 +func (d *Diagnoser) containerGather(ctx context.Context) ([]finding, error) { + deployment, depErr := d.store.BackendContainerDeployment(ctx) + hasDeployment := depErr == nil + if depErr != nil && !errors.Is(depErr, transaction.ErrNotFound) { + return nil, fmt.Errorf("read backend container deployment: %w", depErr) + } + hasHistory, err := d.store.HasCommittedBackendContainerTransactionHistory(ctx) + if err != nil { + return nil, err + } + before, err := d.gateway.Read() + if err != nil { + return nil, fmt.Errorf("read host Nginx configuration: %w", err) + } + return d.containerFindings(ctx, deployment, hasDeployment, hasHistory, before.ActivePort) +} + +// containerFindings 生成 container 后端的全部诊断项。 +// deployment 部署记录，hasDeployment 表示是否存在该记录，hasHistory 表示是否存在历史 +// 已提交事务，activePort 是宿主 Nginx 当前指向的后端端口。 +func (d *Diagnoser) containerFindings(ctx context.Context, deployment transaction.BackendContainerDeployment, hasDeployment bool, hasHistory bool, activePort int) ([]finding, error) { + if !hasDeployment { + return d.containerUndeployedFindings(ctx, hasHistory) + } + + findings := make([]finding, 0, 6) + + committedSlot, err := d.config.Backend.SlotForPort(deployment.ActivePort) + if err != nil { + return nil, err + } + if deployment.ContainerName != committedSlot.ContainerName { + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelDrift, + code: codeDeploymentConfigMismatch, + message: fmt.Sprintf("部署记录容器 %s 与活动端口 %d 的槽位容器 %s 不一致", deployment.ContainerName, deployment.ActivePort, committedSlot.ContainerName), + }) + return findings, nil + } + + committedContainer, committedErr := d.engine.InspectContainer(ctx, deployment.ContainerName) + committedHealthy := false + switch { + case errors.Is(committedErr, containerengine.ErrNotFound): + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelDrift, + code: codeActiveContainerMissing, + message: "部署记录的活动容器 " + deployment.ContainerName + " 不存在", + }) + case committedErr != nil: + return nil, fmt.Errorf("inspect committed backend container %s: %w", deployment.ContainerName, committedErr) + case committedContainer.ID != deployment.ContainerID: + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelDrift, + code: codeActiveContainerIdentity, + message: "活动容器 " + deployment.ContainerName + " 身份与部署记录不一致", + }) + case !committedContainer.Running || committedContainer.Dead: + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelDrift, + code: codeActiveContainerNotRunning, + message: "活动容器 " + deployment.ContainerName + " 未运行", + }) + default: + committedHealthy = true + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelOK, + code: codeActiveContainerOK, + message: "活动容器 " + deployment.ContainerName + " 健康", + }) + } + + if activePort == deployment.ActivePort { + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelOK, + code: codeCommittedPortOK, + message: fmt.Sprintf("宿主 Nginx 指向部署记录的活动端口 %d", deployment.ActivePort), + }) + } else if committedHealthy { + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelFixable, + code: codeNginxWrongPort, + message: fmt.Sprintf("宿主 Nginx 指向 %d，但部署记录的活动端口是 %d", activePort, deployment.ActivePort), + action: fmt.Sprintf("将宿主 Nginx 切流到端口 %d", deployment.ActivePort), + fix: &fixAction{kind: fixSwitchPort, port: deployment.ActivePort}, + }) + } else { + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelDrift, + code: codeNginxWrongPort, + message: fmt.Sprintf("宿主 Nginx 指向 %d，与部署记录活动端口 %d 不一致，且活动容器不健康", activePort, deployment.ActivePort), + }) + } + + inactiveSlot, err := d.config.Backend.SlotForPort(otherPort(deployment.ActivePort)) + if err != nil { + return nil, err + } + inactiveContainer, inactiveErr := d.engine.InspectContainer(ctx, inactiveSlot.ContainerName) + switch { + case errors.Is(inactiveErr, containerengine.ErrNotFound): + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelOK, + code: codeInactiveContainerOK, + message: "非活动槽位容器 " + inactiveSlot.ContainerName + " 不存在", + }) + case inactiveErr != nil: + return nil, fmt.Errorf("inspect inactive backend container %s: %w", inactiveSlot.ContainerName, inactiveErr) + case inactiveContainer.Running && !inactiveContainer.Dead: + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelDrift, + code: codeInactiveContainerRunning, + message: "非活动槽位容器 " + inactiveSlot.ContainerName + " 意外处于运行状态", + }) + default: + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelFixable, + code: codeInactiveContainerResidual, + message: "非活动槽位容器 " + inactiveSlot.ContainerName + " 已停止，属于残留", + action: "移除残留的已停止容器 " + inactiveSlot.ContainerName, + fix: &fixAction{kind: fixRemoveContainer, container: inactiveSlot.ContainerName}, + }) + } + + return findings, nil +} + +// containerUndeployedFindings 处理不存在部署记录的情况，依据容器现场与历史记录判断是否为全新机器。 +func (d *Diagnoser) containerUndeployedFindings(ctx context.Context, hasHistory bool) ([]finding, error) { + var present []string + for _, port := range []int{deploymentconfig.BackendPort8080, deploymentconfig.BackendPort8081} { + slot, err := d.config.Backend.SlotForPort(port) + if err != nil { + return nil, err + } + if _, err := d.engine.InspectContainer(ctx, slot.ContainerName); err == nil { + present = append(present, slot.ContainerName) + } else if !errors.Is(err, containerengine.ErrNotFound) { + return nil, fmt.Errorf("inspect backend container %s: %w", slot.ContainerName, err) + } + } + + if len(present) > 0 { + return []finding{{ + level: daemonapi.DiagnosisLevelDrift, + code: codeContainerWithoutRecord, + message: fmt.Sprintf("SQLite 没有部署记录，但现场存在容器 %v", present), + }}, nil + } + if hasHistory { + return []finding{{ + level: daemonapi.DiagnosisLevelDrift, + code: codeUndeployedWithHistory, + message: "存在历史已提交事务，但缺少部署记录且两个槽位容器均不存在", + }}, nil + } + return []finding{{ + level: daemonapi.DiagnosisLevelOK, + code: codeUndeployed, + message: "全新机器，尚未部署 backend 容器", + }}, nil +} diff --git a/internal/backendstatus/container_test.go b/internal/backendstatus/container_test.go new file mode 100644 index 0000000..9115179 --- /dev/null +++ b/internal/backendstatus/container_test.go @@ -0,0 +1,291 @@ +package backendstatus + +import ( + "context" + "crypto/rand" + "io" + "log/slog" + "path/filepath" + "strings" + "testing" + + "yms-daemon/internal/containerengine" + "yms-daemon/internal/daemonapi" + "yms-daemon/internal/deploymentconfig" + "yms-daemon/internal/hostnginx" + "yms-daemon/internal/transaction" +) + +// TestContainerDiagnoseUndeployed 验证全新机器（无部署记录、无容器、无历史事务）被诊断为健康且未部署。 +func TestContainerDiagnoseUndeployed(t *testing.T) { + d, _ := newContainerDiagnoser(t, &fakeEngine{}, &fakeGateway{port: 8080}) + diagnosis, err := d.Diagnose(context.Background()) + if err != nil { + t.Fatalf("diagnose undeployed container backend: %v", err) + } + if !diagnosis.Healthy { + t.Fatalf("undeployed machine should be healthy: %+v", diagnosis) + } + if diagnosis.Items[0].Level != daemonapi.DiagnosisLevelOK || diagnosis.Items[0].Code != codeUndeployed { + t.Fatalf("unexpected undeployed finding: %+v", diagnosis.Items) + } +} + +// TestContainerDiagnoseHealthy 验证部署记录、活动容器与 Nginx 三者一致时被诊断为健康。 +func TestContainerDiagnoseHealthy(t *testing.T) { + engine := &fakeEngine{containers: map[string]containerengine.Container{ + "backend-8080": {ID: "container-8080", Name: "backend-8080", Running: true}, + }} + d, store := newContainerDiagnoser(t, engine, &fakeGateway{port: 8080}) + commitDeployment(t, store, transaction.BackendContainerDeployment{ + ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", ContainerID: "container-8080", + }) + + diagnosis, err := d.Diagnose(context.Background()) + if err != nil { + t.Fatalf("diagnose healthy container backend: %v", err) + } + if !diagnosis.Healthy { + t.Fatalf("healthy container backend reported drift: %+v", diagnosis) + } + for _, item := range diagnosis.Items { + if item.Level == daemonapi.DiagnosisLevelDrift || item.Level == daemonapi.DiagnosisLevelFixable { + t.Fatalf("unexpected non-ok finding: %+v", item) + } + } +} + +// TestContainerDiagnoseNginxWrongPort 验证 Nginx 指向错误端口而活动容器健康时判定为可自动修复。 +func TestContainerDiagnoseNginxWrongPort(t *testing.T) { + engine := &fakeEngine{containers: map[string]containerengine.Container{ + "backend-8080": {ID: "container-8080", Name: "backend-8080", Running: true}, + }} + d, store := newContainerDiagnoser(t, engine, &fakeGateway{port: 8081}) + commitDeployment(t, store, transaction.BackendContainerDeployment{ + ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", ContainerID: "container-8080", + }) + + diagnosis, err := d.Diagnose(context.Background()) + if err != nil { + t.Fatalf("diagnose nginx wrong port: %v", err) + } + if !diagnosis.Healthy { + t.Fatalf("nginx wrong port with healthy container should be fixable: %+v", diagnosis) + } + if !hasFinding(diagnosis, daemonapi.DiagnosisLevelFixable, codeNginxWrongPort) { + t.Fatalf("missing nginx wrong port finding: %+v", diagnosis.Items) + } +} + +// TestContainerDiagnoseActiveContainerMissing 验证部署记录的活动容器缺失时判定为漂移。 +func TestContainerDiagnoseActiveContainerMissing(t *testing.T) { + d, store := newContainerDiagnoser(t, &fakeEngine{}, &fakeGateway{port: 8080}) + commitDeployment(t, store, transaction.BackendContainerDeployment{ + ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", ContainerID: "container-8080", + }) + + diagnosis, err := d.Diagnose(context.Background()) + if err != nil { + t.Fatalf("diagnose missing active container: %v", err) + } + if diagnosis.Healthy { + t.Fatalf("missing active container should be drift: %+v", diagnosis) + } + if !hasFinding(diagnosis, daemonapi.DiagnosisLevelDrift, codeActiveContainerMissing) { + t.Fatalf("missing active container finding: %+v", diagnosis.Items) + } +} + +// TestContainerDiagnoseInactiveResidual 验证非活动槽位残留已停止容器时判定为可自动修复。 +func TestContainerDiagnoseInactiveResidual(t *testing.T) { + engine := &fakeEngine{containers: map[string]containerengine.Container{ + "backend-8080": {ID: "container-8080", Name: "backend-8080", Running: true}, + "backend-8081": {ID: "container-8081", Name: "backend-8081", Running: false}, + }} + d, store := newContainerDiagnoser(t, engine, &fakeGateway{port: 8080}) + commitDeployment(t, store, transaction.BackendContainerDeployment{ + ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", ContainerID: "container-8080", + }) + + diagnosis, err := d.Diagnose(context.Background()) + if err != nil { + t.Fatalf("diagnose inactive residual: %v", err) + } + if !hasFinding(diagnosis, daemonapi.DiagnosisLevelFixable, codeInactiveContainerResidual) { + t.Fatalf("missing inactive residual finding: %+v", diagnosis.Items) + } +} + +// TestContainerDiagnoseContainerWithoutRecord 验证无部署记录但现场存在容器时判定为漂移。 +func TestContainerDiagnoseContainerWithoutRecord(t *testing.T) { + engine := &fakeEngine{containers: map[string]containerengine.Container{ + "backend-8080": {ID: "container-8080", Name: "backend-8080", Running: true}, + }} + d, _ := newContainerDiagnoser(t, engine, &fakeGateway{port: 8080}) + + diagnosis, err := d.Diagnose(context.Background()) + if err != nil { + t.Fatalf("diagnose container without record: %v", err) + } + if diagnosis.Healthy { + t.Fatalf("container without record should be drift: %+v", diagnosis) + } + if !hasFinding(diagnosis, daemonapi.DiagnosisLevelDrift, codeContainerWithoutRecord) { + t.Fatalf("missing container-without-record finding: %+v", diagnosis.Items) + } +} + +// TestContainerReconcileApplyFixesNginx 验证 reconcile --apply 能把指向错误的 Nginx 切回记录端口。 +func TestContainerReconcileApplyFixesNginx(t *testing.T) { + engine := &fakeEngine{containers: map[string]containerengine.Container{ + "backend-8080": {ID: "container-8080", Name: "backend-8080", Running: true}, + }} + gateway := &fakeGateway{port: 8081} + d, store := newContainerDiagnoser(t, engine, gateway) + commitDeployment(t, store, transaction.BackendContainerDeployment{ + ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", ContainerID: "container-8080", + }) + + diagnosis, err := d.Reconcile(context.Background(), true) + if err != nil { + t.Fatalf("reconcile apply nginx: %v", err) + } + if gateway.port != 8080 { + t.Fatalf("reconcile did not switch Nginx to port 8080: port=%d", gateway.port) + } + if !diagnosis.Healthy { + t.Fatalf("reconcile result should be healthy: %+v", diagnosis) + } + if hasFinding(diagnosis, daemonapi.DiagnosisLevelFixable, codeNginxWrongPort) { + t.Fatalf("nginx wrong port should be fixed after reconcile: %+v", diagnosis.Items) + } +} + +// hasFinding 判断诊断结果中是否存在指定级别与稳定标识的诊断项。 +func hasFinding(diagnosis daemonapi.Diagnosis, level string, code string) bool { + for _, item := range diagnosis.Items { + if item.Level == level && item.Code == code { + return true + } + } + return false +} + +// newContainerDiagnoser 构造 container 后端的诊断器及其底层存储，返回诊断器与可写的存储。 +func newContainerDiagnoser(t *testing.T, engine *fakeEngine, gateway *fakeGateway) (*Diagnoser, *transaction.Store) { + t.Helper() + ctx := context.Background() + store, err := transaction.OpenStore(ctx, filepath.Join(t.TempDir(), "transactions.db")) + if err != nil { + t.Fatalf("open transaction store: %v", err) + } + t.Cleanup(func() { _ = store.Close() }) + coordinator, err := transaction.NewCoordinator(store, slog.New(slog.NewTextHandler(io.Discard, nil))) + if err != nil { + t.Fatalf("create transaction coordinator: %v", err) + } + config := deploymentconfig.Config{ + Daemon: deploymentconfig.Daemon{Environment: deploymentconfig.EnvironmentDev}, + Backend: deploymentconfig.Backend{ + Type: deploymentconfig.BackendTypeContainer, + SystemctlPath: "/bin/systemctl", + Slot: deploymentconfig.BackendSlots{ + Port8080: deploymentconfig.BackendSlot{ContainerName: "backend-8080", HealthEndpoint: "http://127.0.0.1:8080/yms/actuator/health"}, + Port8081: deploymentconfig.BackendSlot{ContainerName: "backend-8081", HealthEndpoint: "http://127.0.0.1:8081/yms/actuator/health"}, + }, + }, + } + diagnoser, err := New(config, store, coordinator, engine, nil, gateway) + if err != nil { + t.Fatalf("create diagnoser: %v", err) + } + return diagnoser, store +} + +// commitDeployment 通过合法的状态机推进写入一条 container 部署记录。 +func commitDeployment(t *testing.T, store *transaction.Store, deployment transaction.BackendContainerDeployment) { + t.Helper() + ctx := context.Background() + transactionID := "deploy-" + rand.Text() + if _, _, err := store.CreateTransaction(ctx, transaction.CreateRequest{ + ID: transactionID, + IdempotencyKey: "backend:container:test:" + rand.Text(), + Source: "test", + Service: "backend", + }); err != nil { + t.Fatalf("create deployment 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, "test"); err != nil { + t.Fatalf("advance deployment transaction: %v", err) + } + } + if _, err := store.CommitBackendContainerDeployment(ctx, transactionID, deployment, "test committed"); err != nil { + t.Fatalf("commit deployment: %v", err) + } +} + +// fakeGateway gateway 接口的内存实现，记录活动端口并在 Switch 时直接更新。 +type fakeGateway struct { + port int + content []byte +} + +// Read 返回当前快照与活动端口。 +func (g *fakeGateway) Read() (hostnginx.Snapshot, error) { + content := g.content + if content == nil { + content = []byte("http {\n upstream yms-server {\n # yms-update managed upstream begin\n server 127.0.0.1:8080 max_fails=1 fail_timeout=2s;\n server 127.0.0.1:8081 max_fails=1 fail_timeout=2s;\n # yms-update managed upstream end\n }\n}\n") + } + return hostnginx.Snapshot{Content: append([]byte(nil), content...), ActivePort: g.port}, nil +} + +// Switch 直接更新活动端口并返回切换前快照。 +func (g *fakeGateway) Switch(_ context.Context, port int) (hostnginx.Snapshot, error) { + previous, _ := g.Read() + g.port = port + return previous, nil +} + +// fakeEngine containerengine.Engine 的内存实现，维护容器表。 +type fakeEngine struct { + containers map[string]containerengine.Container + removed []string +} + +func (e *fakeEngine) Ping(context.Context) error { return nil } +func (e *fakeEngine) PullImage(context.Context, string) error { return nil } +func (e *fakeEngine) LoadImage(context.Context, io.Reader) error { return nil } +func (e *fakeEngine) InspectImage(context.Context, string) (containerengine.Image, error) { + return containerengine.Image{}, nil +} +func (e *fakeEngine) CreateContainer(context.Context, containerengine.ContainerSpec) (containerengine.Container, error) { + return containerengine.Container{}, nil +} +func (e *fakeEngine) StartContainer(context.Context, string) error { return nil } +func (e *fakeEngine) ContainerLogs(context.Context, string) (io.ReadCloser, error) { + return io.NopCloser(strings.NewReader("")), nil +} +func (e *fakeEngine) StopContainer(context.Context, string, int) error { return nil } +func (e *fakeEngine) InspectContainer(_ context.Context, name string) (containerengine.Container, error) { + record, found := e.containers[name] + if !found { + return containerengine.Container{}, containerengine.ErrNotFound + } + return record, nil +} +func (e *fakeEngine) RemoveContainer(_ context.Context, name string, _ bool) error { + delete(e.containers, name) + e.removed = append(e.removed, name) + return nil +} +func (e *fakeEngine) Close() error { return nil } + +var _ containerengine.Engine = (*fakeEngine)(nil) diff --git a/internal/backendstatus/diagnoser.go b/internal/backendstatus/diagnoser.go new file mode 100644 index 0000000..acf5ddd --- /dev/null +++ b/internal/backendstatus/diagnoser.go @@ -0,0 +1,157 @@ +// Package backendstatus 对 backend 组件的现场状态与 SQLite 事实来源做只读诊断，并按需执行自动修复。 +// +// 诊断遵循“SQLite 是事实来源、外部系统是待校验状态”的原则：status 与 doctor 只读， +// 不修改任何现场；reconcile 默认只生成修复计划，仅在显式要求 apply 时执行经过明确授权的 +// 自动修复动作，且每个修复动作都先写入事务再执行外部变更。 +package backendstatus + +import ( + "context" + "errors" + "fmt" + + "yms-daemon/internal/containerengine" + "yms-daemon/internal/daemonapi" + "yms-daemon/internal/deploymentconfig" + "yms-daemon/internal/hostnginx" + "yms-daemon/internal/systemd" + "yms-daemon/internal/transaction" +) + +// gateway 抽象宿主 Nginx 配置的读取与切流，便于在测试中替换。 +type gateway interface { + // Read 返回当前完整配置快照及其活动后端端口。 + Read() (hostnginx.Snapshot, error) + // Switch 把活动后端端口切到指定端口并重载 Nginx。 + Switch(context.Context, int) (hostnginx.Snapshot, error) +} + +// Diagnoser 持有执行诊断与自动修复所需的全部依赖。 +// 除 store、coordinator、gateway 外，其余依赖按组件运行类型选择性使用：container 使用 +// engine，native 使用 units，未使用的依赖可以为 nil。 +type Diagnoser struct { + // config 已校验的本地部署配置，决定组件运行类型与槽位信息。 + config deploymentconfig.Config + // store 服务端 SQLite 事务存储，是诊断的事实来源。 + store *transaction.Store + // coordinator 用于在 reconcile --apply 时以可恢复方式执行单个修复步骤。 + coordinator *transaction.Coordinator + // engine 容器引擎，container 后端诊断时使用。 + engine containerengine.Engine + // units systemd 管理器，native 后端诊断时使用。 + units systemd.Manager + // gateway 宿主 Nginx 配置控制器，用于读取活动端口与执行切流修复。 + gateway gateway +} + +// New 构造一个 Diagnoser 并校验依赖。 +// config 必须已经通过 deploymentconfig.Validate；store、coordinator、gateway 不能为空； +// 当 backend.type 为 container 时 engine 不能为空，为 native 时 units 不能为空。 +func New(config deploymentconfig.Config, store *transaction.Store, coordinator *transaction.Coordinator, engine containerengine.Engine, units systemd.Manager, gateway gateway) (*Diagnoser, error) { + if err := config.Validate(); err != nil { + return nil, err + } + if store == nil || coordinator == nil || gateway == nil { + return nil, errors.New("backend status dependencies are required") + } + switch config.Backend.Type { + case deploymentconfig.BackendTypeContainer: + if engine == nil { + return nil, errors.New("container backend diagnosis requires a container engine") + } + case deploymentconfig.BackendTypeNative: + if units == nil { + return nil, errors.New("native backend diagnosis requires a systemd manager") + } + default: + return nil, fmt.Errorf("unsupported backend.type %q", config.Backend.Type) + } + return &Diagnoser{config: config, store: store, coordinator: coordinator, engine: engine, units: units, gateway: gateway}, nil +} + +// Diagnose 执行只读诊断并返回结构化结果，不修改任何现场状态。 +// 返回的 daemonapi.Diagnosis 中，Healthy 为 true 表示不存在需要人工处理的漂移项。 +func (d *Diagnoser) Diagnose(ctx context.Context) (daemonapi.Diagnosis, error) { + switch d.config.Backend.Type { + case deploymentconfig.BackendTypeContainer: + return d.diagnoseContainer(ctx) + case deploymentconfig.BackendTypeNative: + return d.diagnoseNative(ctx) + default: + return daemonapi.Diagnosis{}, fmt.Errorf("unsupported backend.type %q", d.config.Backend.Type) + } +} + +// Reconcile 生成修复计划（apply 为 false）或执行自动修复（apply 为 true）。 +// apply 为 false 时行为与 Diagnose 一致，仅返回只读诊断；apply 为 true 时先诊断， +// 对每个可自动修复项执行修复动作，完成后重新诊断并返回修复后的结果。 +func (d *Diagnoser) Reconcile(ctx context.Context, apply bool) (daemonapi.Diagnosis, error) { + if !apply { + return d.Diagnose(ctx) + } + switch d.config.Backend.Type { + case deploymentconfig.BackendTypeContainer: + return d.reconcileContainer(ctx) + case deploymentconfig.BackendTypeNative: + return d.reconcileNative(ctx) + default: + return daemonapi.Diagnosis{}, fmt.Errorf("unsupported backend.type %q", d.config.Backend.Type) + } +} + +// otherPort 返回给定端口的另一个蓝绿槽位端口。 +func otherPort(port int) int { + if port == deploymentconfig.BackendPort8080 { + return deploymentconfig.BackendPort8081 + } + return deploymentconfig.BackendPort8080 +} + +// finding 内部诊断结论，比协议层的 DiagnosisItem 多携带一个可选的修复参数。 +type finding struct { + // level 诊断级别，取值 daemonapi.DiagnosisLevelOK / DiagnosisLevelFixable / DiagnosisLevelDrift。 + level string + // code 稳定标识，供测试与 reconcile 识别。 + code string + // message 人类可读的诊断描述。 + message string + // action 仅在 level 为 fixable 时给出建议动作描述。 + action string + // fix 描述可自动修复动作，仅 fixable 项非 nil。 + fix *fixAction +} + +// fixAction 描述一个可自动修复动作所需的精确参数。 +type fixAction struct { + // kind 修复动作种类，取值 fixSwitchPort 或 fixRemoveContainer。 + kind string + // port 切流修复的目标端口。 + port int + // container 容器移除修复的目标容器名。 + container string +} + +const ( + // fixSwitchPort 表示把宿主 Nginx 切流到指定端口。 + fixSwitchPort = "nginx-switch" + // fixRemoveContainer 表示移除残留的已停止容器。 + fixRemoveContainer = "container-remove" +) + +// toDiagnosis 把内部 finding 列表转换为协议层诊断结果，并据此计算 Healthy 标志。 +func toDiagnosis(service string, runtimeType string, findings []finding) daemonapi.Diagnosis { + items := make([]daemonapi.DiagnosisItem, 0, len(findings)) + healthy := true + for _, f := range findings { + if f.level == daemonapi.DiagnosisLevelDrift { + healthy = false + } + items = append(items, daemonapi.DiagnosisItem{ + Level: f.level, + Code: f.code, + Message: f.message, + Action: f.action, + }) + } + return daemonapi.Diagnosis{Service: service, Type: runtimeType, Healthy: healthy, Items: items} +} diff --git a/internal/backendstatus/native.go b/internal/backendstatus/native.go new file mode 100644 index 0000000..2cb3ba7 --- /dev/null +++ b/internal/backendstatus/native.go @@ -0,0 +1,137 @@ +package backendstatus + +import ( + "context" + "errors" + "fmt" + "os" + + "yms-daemon/internal/daemonapi" + "yms-daemon/internal/deploymentconfig" + "yms-daemon/internal/systemd" +) + +// 诊断项稳定标识，供测试与 reconcile 识别，不参与展示。 +const ( + codeNativeActivePortOK = "native_nginx_active_port" + codeNativeActiveUnitOK = "native_active_unit_running" + codeNativeActiveUnitDown = "native_active_unit_not_running" + codeNativeInactiveUnitOK = "native_inactive_unit_stopped" + codeNativeInactiveRunning = "native_inactive_unit_running" + codeNativeSlotJAROK = "native_slot_jar_link" + codeNativeSlotJARDrift = "native_slot_jar_link_drift" +) + +// diagnoseNative 对 native 后端做只读诊断。 +// native 没有部署单例记录，事实来源是宿主 Nginx 活动端口与 systemd 单元现场状态。 +func (d *Diagnoser) diagnoseNative(ctx context.Context) (daemonapi.Diagnosis, error) { + findings, err := d.nativeGather(ctx) + if err != nil { + return daemonapi.Diagnosis{}, err + } + return toDiagnosis("backend", deploymentconfig.BackendTypeNative, findings), nil +} + +// nativeGather 生成 native 后端的全部诊断项。 +func (d *Diagnoser) nativeGather(ctx context.Context) ([]finding, error) { + before, err := d.gateway.Read() + if err != nil { + return nil, fmt.Errorf("read host Nginx configuration: %w", err) + } + activePort := before.ActivePort + findings := make([]finding, 0, 6) + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelOK, + code: codeNativeActivePortOK, + message: fmt.Sprintf("宿主 Nginx 指向后端端口 %d", activePort), + }) + + activeSlot, err := d.config.Backend.SlotForPort(activePort) + if err != nil { + return nil, err + } + inactiveSlot, err := d.config.Backend.SlotForPort(otherPort(activePort)) + if err != nil { + return nil, err + } + + activeUnit, err := d.units.Inspect(ctx, activeSlot.Unit) + if err != nil { + return nil, fmt.Errorf("inspect native active unit %s: %w", activeSlot.Unit, err) + } + if unitRunning(activeUnit) { + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelOK, + code: codeNativeActiveUnitOK, + message: "活动单元 " + activeSlot.Unit + " 正在运行", + }) + } else { + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelDrift, + code: codeNativeActiveUnitDown, + message: "活动单元 " + activeSlot.Unit + " 未运行（ActiveState=" + activeUnit.ActiveState + "）", + }) + } + + inactiveUnit, err := d.units.Inspect(ctx, inactiveSlot.Unit) + if err != nil { + return nil, fmt.Errorf("inspect native inactive unit %s: %w", inactiveSlot.Unit, err) + } + if unitRunning(inactiveUnit) { + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelDrift, + code: codeNativeInactiveRunning, + message: "非活动单元 " + inactiveSlot.Unit + " 意外运行，两个槽位同时在线", + }) + } else { + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelOK, + code: codeNativeInactiveUnitOK, + message: "非活动单元 " + inactiveSlot.Unit + " 已停止", + }) + } + + if ok, detail := jarLinkStatus(activeSlot.JAR); ok { + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelOK, + code: codeNativeSlotJAROK, + message: "活动槽位 JAR 链接有效：" + detail, + }) + } else { + findings = append(findings, finding{ + level: daemonapi.DiagnosisLevelDrift, + code: codeNativeSlotJARDrift, + message: "活动槽位 JAR 链接异常：" + detail, + }) + } + + return findings, nil +} + +// unitRunning 判断 systemd 单元是否处于运行状态。 +func unitRunning(unit systemd.Unit) bool { + return unit.ActiveState != "inactive" && unit.ActiveState != "failed" +} + +// jarLinkStatus 检查槽位 JAR 路径是否为指向现存文件的符号链接。 +// 返回 ok 与人类可读的详情；JAR 缺失、不是符号链接或目标不存在时返回 ok=false。 +func jarLinkStatus(path string) (bool, string) { + info, err := os.Lstat(path) + if errors.Is(err, os.ErrNotExist) { + return false, "路径不存在" + } + if err != nil { + return false, "检查失败：" + err.Error() + } + if info.Mode()&os.ModeSymlink == 0 { + return false, "不是符号链接" + } + target, err := os.Readlink(path) + if err != nil { + return false, "读取链接目标失败：" + err.Error() + } + if _, err := os.Stat(path); err != nil { + return false, "链接目标不存在：" + target + } + return true, "指向 " + target +} diff --git a/internal/backendstatus/reconcile.go b/internal/backendstatus/reconcile.go new file mode 100644 index 0000000..d9d6072 --- /dev/null +++ b/internal/backendstatus/reconcile.go @@ -0,0 +1,187 @@ +package backendstatus + +import ( + "context" + "crypto/rand" + "encoding/json" + "errors" + "fmt" + "strconv" + + "yms-daemon/internal/containerengine" + "yms-daemon/internal/daemonapi" + "yms-daemon/internal/deploymentconfig" + "yms-daemon/internal/transaction" +) + +// reconcileSource 表示对账修复事务的来源标识。 +const reconcileSource = "local-cli" + +// reconcileContainer 执行 container 后端的对账：只读时返回修复计划，apply 时执行自动修复。 +func (d *Diagnoser) reconcileContainer(ctx context.Context) (daemonapi.Diagnosis, error) { + findings, err := d.containerGather(ctx) + if err != nil { + return daemonapi.Diagnosis{}, err + } + return d.applyFixes(ctx, deploymentconfig.BackendTypeContainer, findings) +} + +// reconcileNative 执行 native 后端的对账。当前 native 没有可自动修复项，仅返回只读诊断。 +func (d *Diagnoser) reconcileNative(ctx context.Context) (daemonapi.Diagnosis, error) { + findings, err := d.nativeGather(ctx) + if err != nil { + return daemonapi.Diagnosis{}, err + } + return d.applyFixes(ctx, deploymentconfig.BackendTypeNative, findings) +} + +// applyFixes 对诊断结果中的可自动修复项执行修复动作。 +// 若没有可修复项则直接返回只读诊断；否则创建对账事务，通过 coordinator 对每个修复动作 +// 先记录意图、再执行、再核对，全部成功后把事务推进到 COMMITTED，最后重新诊断返回修复后状态。 +func (d *Diagnoser) applyFixes(ctx context.Context, runtimeType string, findings []finding) (daemonapi.Diagnosis, error) { + var fixable []finding + for _, f := range findings { + if f.level == daemonapi.DiagnosisLevelFixable && f.fix != nil { + fixable = append(fixable, f) + } + } + if len(fixable) == 0 { + return toDiagnosis("backend", runtimeType, findings), nil + } + + transactionID := rand.Text() + codes := make([]string, 0, len(fixable)) + for _, f := range fixable { + codes = append(codes, f.code) + } + requestJSON, err := json.Marshal(struct { + Service string `json:"service"` + Type string `json:"type"` + Fixes []string `json:"fixes"` + }{"backend", runtimeType, codes}) + if err != nil { + return daemonapi.Diagnosis{}, fmt.Errorf("encode reconcile request: %w", err) + } + if _, _, err := d.store.CreateTransaction(ctx, transaction.CreateRequest{ + ID: transactionID, + IdempotencyKey: "backend:reconcile:" + rand.Text(), + Source: reconcileSource, + Service: "backend", + Request: requestJSON, + }); err != nil { + return daemonapi.Diagnosis{}, fmt.Errorf("create reconcile transaction: %w", err) + } + + for _, f := range fixable { + intent, operation, err := d.fixOperation(f) + if err != nil { + _, _ = d.store.Transition(ctx, transactionID, transaction.StateFailed, err.Error()) + return daemonapi.Diagnosis{}, err + } + if _, err := d.coordinator.ExecuteStep(ctx, transactionID, intent, operation); err != nil { + _, _ = d.store.Transition(ctx, transactionID, transaction.StateFailed, err.Error()) + return daemonapi.Diagnosis{}, fmt.Errorf("apply reconcile fix %s: %w", f.code, err) + } + } + + for _, next := range []transaction.State{ + transaction.StateValidating, + transaction.StatePrepared, + transaction.StateStarting, + transaction.StateSwitching, + transaction.StateVerifying, + transaction.StateDraining, + transaction.StateCommitted, + } { + if _, err := d.store.Transition(ctx, transactionID, next, "reconcile"); err != nil { + return daemonapi.Diagnosis{}, fmt.Errorf("commit reconcile transaction: %w", err) + } + } + + diagnosis, err := d.Diagnose(ctx) + if err != nil { + return daemonapi.Diagnosis{}, err + } + diagnosis.RepairApplied = true + diagnosis.RepairTransactionID = transactionID + return diagnosis, nil +} + +// fixOperation 把一条可修复诊断项转换为对账事务中可执行的步骤意图与操作。 +func (d *Diagnoser) fixOperation(f finding) (transaction.StepIntent, transaction.Operation, error) { + switch f.fix.kind { + case fixSwitchPort: + intent := reconcileIntent("reconcile.nginx.switch."+strconv.Itoa(f.fix.port), "switch host Nginx to committed backend port", struct { + Port int `json:"port"` + }{f.fix.port}) + return intent, &switchNginxOperation{gateway: d.gateway, port: f.fix.port}, nil + case fixRemoveContainer: + intent := reconcileIntent("reconcile.container.remove."+f.fix.container, "remove residual stopped backend container", struct { + ContainerName string `json:"containerName"` + }{f.fix.container}) + return intent, &removeContainerOperation{engine: d.engine, name: f.fix.container}, nil + default: + return transaction.StepIntent{}, nil, fmt.Errorf("unsupported reconcile fix kind %q", f.fix.kind) + } +} + +// reconcileIntent 构造对账修复步骤的持久化意图。 +func reconcileIntent(key, name string, value any) transaction.StepIntent { + payload, err := json.Marshal(value) + if err != nil { + panic(fmt.Sprintf("marshal reconcile intent: %v", err)) + } + return transaction.StepIntent{Key: key, Name: name, Intent: payload} +} + +// switchNginxOperation 把宿主 Nginx 切流到指定端口，实现 transaction.Operation。 +type switchNginxOperation struct { + gateway gateway + port int +} + +// Apply 执行宿主 Nginx 切流，切到当前已指向的端口时为空操作。 +func (o *switchNginxOperation) Apply(ctx context.Context) error { + _, err := o.gateway.Switch(ctx, o.port) + return err +} + +// Inspect 核对宿主 Nginx 当前活动端口是否等于目标端口。 +func (o *switchNginxOperation) Inspect(ctx context.Context) (transaction.Inspection, error) { + snapshot, err := o.gateway.Read() + if err != nil { + return transaction.Inspection{}, err + } + if snapshot.ActivePort == o.port { + return transaction.Inspection{Status: transaction.InspectionApplied}, nil + } + return transaction.Inspection{Status: transaction.InspectionNotApplied}, nil +} + +// removeContainerOperation 移除一个残留容器，实现 transaction.Operation。 +type removeContainerOperation struct { + engine containerengine.Engine + name string +} + +// Apply 强制移除目标容器，容器不存在时视为成功。 +func (o *removeContainerOperation) Apply(ctx context.Context) error { + err := o.engine.RemoveContainer(ctx, o.name, true) + if errors.Is(err, containerengine.ErrNotFound) { + return nil + } + return err +} + +// Inspect 核对目标容器是否已经不存在。 +func (o *removeContainerOperation) Inspect(ctx context.Context) (transaction.Inspection, error) { + if _, err := o.engine.InspectContainer(ctx, o.name); errors.Is(err, containerengine.ErrNotFound) { + return transaction.Inspection{Status: transaction.InspectionApplied}, nil + } else if err != nil { + return transaction.Inspection{}, err + } + return transaction.Inspection{Status: transaction.InspectionNotApplied}, nil +} + +var _ transaction.Operation = (*switchNginxOperation)(nil) +var _ transaction.Operation = (*removeContainerOperation)(nil) diff --git a/internal/backendupdate/restart.go b/internal/backendupdate/restart.go index d2cb3ff..c0e0cd5 100644 --- a/internal/backendupdate/restart.go +++ b/internal/backendupdate/restart.go @@ -22,7 +22,7 @@ import ( // 若已存在相关活动事务，则尝试复用并续跑。 func (u *Updater) Restart(ctx context.Context, report ProgressReporter) (transaction.Transaction, error) { if u.config.Backend.Type == deploymentconfig.BackendTypeContainer { - return transaction.Transaction{}, errors.New("restart is not implemented for container backend") + return u.RestartContainer(ctx, report) } reportProgress(report, Progress{Message: "Resolving the current native backend release"}) active, err := u.store.ActiveTransaction(ctx) diff --git a/internal/backendupdate/updater.go b/internal/backendupdate/updater.go index 0132d4f..fbd4ce6 100644 --- a/internal/backendupdate/updater.go +++ b/internal/backendupdate/updater.go @@ -76,6 +76,8 @@ type Updater struct { containerTmpSource string // containerTmpTarget 容器后端临时目录在容器内的目标路径。 containerTmpTarget string + // containerStopGraceSeconds 停止容器后端旧容器时，发送停止信号后到强制终止前的等待秒数。 + containerStopGraceSeconds int // logger 结构化日志记录器。 logger *slog.Logger // drain 切换后旧单元/旧容器的排空时长。 @@ -189,13 +191,18 @@ func NewContainer( if err != nil { return nil, err } + stopGraceSeconds := defaultContainerStopGraceSeconds + if config.Backend.StopGraceSeconds != nil { + stopGraceSeconds = *config.Backend.StopGraceSeconds + } return &Updater{ config: config, workRoot: workRoot, store: store, coordinator: coordinator, gateway: gateway, containerExecutor: executor, engine: engine, logger: logger, drain: drainDuration, - containerConfigSource: deploymentconfig.ContainerConfigSource, - containerConfigTarget: deploymentconfig.ContainerConfigTarget, - containerTmpSource: deploymentconfig.ContainerTmpSource, - containerTmpTarget: deploymentconfig.ContainerTmpTarget, + containerConfigSource: deploymentconfig.ContainerConfigSource, + containerConfigTarget: deploymentconfig.ContainerConfigTarget, + containerTmpSource: deploymentconfig.ContainerTmpSource, + containerTmpTarget: deploymentconfig.ContainerTmpTarget, + containerStopGraceSeconds: stopGraceSeconds, }, nil } diff --git a/internal/containerengine/engine.go b/internal/containerengine/engine.go index f858b1b..32a20c0 100644 --- a/internal/containerengine/engine.go +++ b/internal/containerengine/engine.go @@ -132,8 +132,8 @@ type Engine interface { StartContainer(context.Context, string) error // ContainerLogs 读取指定容器的日志流。 ContainerLogs(context.Context, string) (io.ReadCloser, error) - // StopContainer 停止指定容器。 - StopContainer(context.Context, string) error + // StopContainer 停止指定容器，timeoutSeconds 是发送停止信号后到强制终止前的等待秒数。 + StopContainer(context.Context, string, int) error // InspectContainer 检查容器并返回其运行时状态。 InspectContainer(context.Context, string) (Container, error) // RemoveContainer 移除指定容器，force 决定是否强制移除。 diff --git a/internal/containerengine/moby.go b/internal/containerengine/moby.go index 184bfad..b43ec47 100644 --- a/internal/containerengine/moby.go +++ b/internal/containerengine/moby.go @@ -1,7 +1,6 @@ package containerengine import ( - "bytes" "context" "encoding/json" "errors" @@ -164,24 +163,26 @@ func (e *MobyEngine) StartContainer(ctx context.Context, idOrName string) error return nil } -// ContainerLogs 读取指定容器的 stdout 与 stderr 日志， -// 将引擎的多路复用流解码后合并为一个只读流返回。 +// ContainerLogs 跟随读取指定容器的 stdout 与 stderr 日志，将引擎的多路复用流 +// 解码后合并为一个持续流返回，直到 ctx 取消或容器停止。 func (e *MobyEngine) ContainerLogs(ctx context.Context, idOrName string) (io.ReadCloser, error) { - stream, err := e.client.ContainerLogs(ctx, idOrName, client.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Tail: "all"}) + stream, err := e.client.ContainerLogs(ctx, idOrName, client.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Follow: true, Tail: "all"}) if err != nil { return nil, engineError("read container logs", err) } - defer stream.Close() - var output bytes.Buffer - if _, err := stdcopy.StdCopy(&output, &output, stream); err != nil { - return nil, fmt.Errorf("decode container logs: %w", err) - } - return io.NopCloser(bytes.NewReader(output.Bytes())), nil + reader, writer := io.Pipe() + go func() { + _, copyErr := stdcopy.StdCopy(writer, writer, stream) + _ = stream.Close() + _ = writer.CloseWithError(copyErr) + }() + return reader, nil } // StopContainer 停止指定容器，参数可为容器 ID 或名称。 -func (e *MobyEngine) StopContainer(ctx context.Context, idOrName string) error { - if _, err := e.client.ContainerStop(ctx, idOrName, client.ContainerStopOptions{}); err != nil { +// timeoutSeconds 是发送停止信号后到强制终止前的等待秒数，传给引擎以在超时后强制终止。 +func (e *MobyEngine) StopContainer(ctx context.Context, idOrName string, timeoutSeconds int) error { + if _, err := e.client.ContainerStop(ctx, idOrName, client.ContainerStopOptions{Timeout: &timeoutSeconds}); err != nil { return engineError("stop container", err) } return nil diff --git a/internal/daemonclient/client.go b/internal/daemonclient/client.go index 195240d..834ed03 100644 --- a/internal/daemonclient/client.go +++ b/internal/daemonclient/client.go @@ -55,6 +55,48 @@ func Restart(ctx context.Context, socketPath string, service string, progress fu return submit(ctx, socketPath, request, progress) } +// Status 向守护进程查询后端当前状态，返回结构化诊断结果。 +// socketPath 守护进程 Unix Socket 的绝对路径，service 是目标服务名。 +// 当路径非法、连接失败或守护进程返回错误时返回 error。 +func Status(ctx context.Context, socketPath string, service string) (daemonapi.Diagnosis, error) { + if !filepath.IsAbs(socketPath) { + return daemonapi.Diagnosis{}, errors.New("daemon socket path must be absolute") + } + return submitDiagnosis(ctx, socketPath, daemonapi.Request{Operation: daemonapi.OperationStatus, Service: service}) +} + +// Doctor 向守护进程提交一次只读诊断请求，返回全部漂移项与建议动作。 +// socketPath 守护进程 Unix Socket 的绝对路径，service 是目标服务名。 +// 当路径非法、连接失败或守护进程返回错误时返回 error。 +func Doctor(ctx context.Context, socketPath string, service string) (daemonapi.Diagnosis, error) { + if !filepath.IsAbs(socketPath) { + return daemonapi.Diagnosis{}, errors.New("daemon socket path must be absolute") + } + return submitDiagnosis(ctx, socketPath, daemonapi.Request{Operation: daemonapi.OperationDoctor, Service: service}) +} + +// Reconcile 向守护进程提交对账请求，apply 为 true 时执行自动修复，否则只生成修复计划。 +// socketPath 守护进程 Unix Socket 的绝对路径，service 是目标服务名。 +// 当路径非法、连接失败或守护进程返回错误时返回 error。 +func Reconcile(ctx context.Context, socketPath string, service string, apply bool) (daemonapi.Diagnosis, error) { + if !filepath.IsAbs(socketPath) { + return daemonapi.Diagnosis{}, errors.New("daemon socket path must be absolute") + } + return submitDiagnosis(ctx, socketPath, daemonapi.Request{Operation: daemonapi.OperationReconcile, Service: service, Apply: apply}) +} + +// submitDiagnosis 提交一次诊断类请求并解析结果中的诊断结构，供 status/doctor/reconcile 复用。 +func submitDiagnosis(ctx context.Context, socketPath string, request daemonapi.Request) (daemonapi.Diagnosis, error) { + response, err := submit(ctx, socketPath, request, nil) + if err != nil { + return daemonapi.Diagnosis{}, err + } + if response.Diagnosis == nil { + return daemonapi.Diagnosis{}, errors.New("daemon returned no diagnosis result") + } + return *response.Diagnosis, nil +} + // submit 通过 Unix Socket 把 request 发送给守护进程并读取响应流，供本包各公开函数复用。 // 它会以换行分隔的 JSON 编码请求，写完后关闭写端以向对端发送结束信号， // 随后循环解码响应：进度事件通过 progress 回调透传，最终结果则返回给调用方。 diff --git a/internal/daemonserver/server.go b/internal/daemonserver/server.go index 97aed37..cc281e9 100644 --- a/internal/daemonserver/server.go +++ b/internal/daemonserver/server.go @@ -38,6 +38,15 @@ type backendUpdater interface { Restart(context.Context, backendupdate.ProgressReporter) (transaction.Transaction, error) } +// backendDiagnoser 抽象了后端状态诊断与对账能力，供 status/doctor/reconcile 操作使用。 +// 它由 backendstatus 包在运行时注入，便于在测试中用轻量替身替换。 +type backendDiagnoser interface { + // Diagnose 执行只读诊断，返回结构化诊断结果。 + Diagnose(context.Context) (daemonapi.Diagnosis, error) + // Reconcile 生成修复计划或执行自动修复，apply 为 true 时执行修复动作。 + Reconcile(context.Context, bool) (daemonapi.Diagnosis, error) +} + // Server 持有本地 Unix Socket 并把收到的请求分派给事务编排器。 // 它负责连接的生命周期管理、请求解码与校验、执行进度回写以及最终结果的返回。 type Server struct { @@ -45,24 +54,29 @@ type Server struct { socketPath string // updater 后端更新与重启动作的实际执行者。 updater backendUpdater + // diagnoser 后端状态诊断与对账动作的实际执行者。 + diagnoser backendDiagnoser // logger 用于记录服务运行与请求处理过程中的日志。 logger *slog.Logger } // New 构造一个 Server 实例并校验入参。 -// socketPath 必须是规范化的绝对路径，updater 不能为空，logger 为空时回退为默认日志器。 -// 当路径非法或 updater 为空时返回 error。 -func New(socketPath string, updater backendUpdater, logger *slog.Logger) (*Server, error) { +// socketPath 必须是规范化的绝对路径，updater 与 diagnoser 不能为空，logger 为空时回退为默认日志器。 +// 当路径非法或依赖为空时返回 error。 +func New(socketPath string, updater backendUpdater, diagnoser backendDiagnoser, logger *slog.Logger) (*Server, error) { if !filepath.IsAbs(socketPath) || filepath.Clean(socketPath) != socketPath { return nil, errors.New("daemon Unix Socket path must be a clean absolute path") } if updater == nil { return nil, errors.New("backend updater is required") } + if diagnoser == nil { + return nil, errors.New("backend diagnoser is required") + } if logger == nil { logger = slog.Default() } - return &Server{socketPath: socketPath, updater: updater, logger: logger}, nil + return &Server{socketPath: socketPath, updater: updater, diagnoser: diagnoser, logger: logger}, nil } // Serve 监听 Unix Socket 直到 ctx 被取消，返回前会等待所有连接处理完成并清理 Socket 文件。 @@ -176,8 +190,24 @@ func (s *Server) handle(ctx context.Context, connection net.Conn) { return } record, updateErr = s.updater.Restart(ctx, report) + case daemonapi.OperationStatus, daemonapi.OperationDoctor: + if request.InputType != "" || request.File != "" || request.ImageReference != "" || request.Apply { + _ = s.writeResponse(connection, daemonapi.Response{Kind: daemonapi.ResponseResult, Error: request.Operation + " does not accept inputType, file, imageReference, or apply"}) + return + } + diagnosis, diagnosisErr := s.diagnoser.Diagnose(ctx) + s.writeDiagnosis(ctx, connection, request.Operation, diagnosis, diagnosisErr) + return + case daemonapi.OperationReconcile: + if request.InputType != "" || request.File != "" || request.ImageReference != "" { + _ = s.writeResponse(connection, daemonapi.Response{Kind: daemonapi.ResponseResult, Error: "reconcile does not accept inputType, file, or imageReference"}) + return + } + diagnosis, diagnosisErr := s.diagnoser.Reconcile(ctx, request.Apply) + s.writeDiagnosis(ctx, connection, request.Operation, diagnosis, diagnosisErr) + return default: - _ = s.writeResponse(connection, daemonapi.Response{Kind: daemonapi.ResponseResult, Error: "operation must be update or restart"}) + _ = s.writeResponse(connection, daemonapi.Response{Kind: daemonapi.ResponseResult, Error: "operation must be update, restart, status, doctor, or reconcile"}) return } response := daemonapi.Response{Kind: daemonapi.ResponseResult, TransactionID: record.ID, State: string(record.State)} @@ -197,6 +227,22 @@ func (s *Server) writeResponse(connection net.Conn, response daemonapi.Response) return json.NewEncoder(connection).Encode(response) } +// writeDiagnosis 把 status/doctor/reconcile 的诊断结果写回客户端。 +// 诊断失败时返回携带错误信息的响应并清空诊断结果，同时记录相应日志。 +func (s *Server) writeDiagnosis(ctx context.Context, connection net.Conn, operation string, diagnosis daemonapi.Diagnosis, err error) { + response := daemonapi.Response{Kind: daemonapi.ResponseResult, Diagnosis: &diagnosis} + if err != nil { + response.Diagnosis = nil + response.Error = err.Error() + s.logger.ErrorContext(ctx, "backend diagnosis failed", "operation", operation, "error", err) + } else { + s.logger.InfoContext(ctx, "backend diagnosis completed", "operation", operation, "healthy", diagnosis.Healthy, "items", len(diagnosis.Items)) + } + if writeErr := s.writeResponse(connection, response); writeErr != nil { + s.logger.ErrorContext(ctx, "write daemon diagnosis result", "operation", operation, "error", writeErr) + } +} + // decodeRequest 从 reader 中解码并校验一条客户端请求。 // 它限制请求体大小不超过 maximumRequestBytes，禁止未知字段，并要求请求只能包含一个 JSON 值， // 且所有字符串字段不得带有前后空白字符；任一条不满足时返回 error。 diff --git a/internal/daemonserver/server_test.go b/internal/daemonserver/server_test.go index 46436f2..007f3ba 100644 --- a/internal/daemonserver/server_test.go +++ b/internal/daemonserver/server_test.go @@ -22,7 +22,7 @@ import ( func TestServerAcceptsBackendUpdateThroughUnixSocket(t *testing.T) { socketPath := shortSocketPath(t) updater := &fakeUpdater{record: transaction.Transaction{ID: "transaction-01", State: transaction.StateCommitted}} - server, err := New(socketPath, updater, slog.New(slog.NewTextHandler(io.Discard, nil))) + server, err := New(socketPath, updater, &fakeDiagnoser{}, slog.New(slog.NewTextHandler(io.Discard, nil))) if err != nil { t.Fatalf("create daemon server: %v", err) } @@ -65,7 +65,7 @@ func TestServerAcceptsBackendUpdateThroughUnixSocket(t *testing.T) { func TestServerAcceptsDirectNativeJARThroughUnixSocket(t *testing.T) { socketPath := shortSocketPath(t) updater := &fakeUpdater{record: transaction.Transaction{ID: "transaction-direct-01", State: transaction.StateCommitted}} - server, err := New(socketPath, updater, slog.New(slog.NewTextHandler(io.Discard, nil))) + server, err := New(socketPath, updater, &fakeDiagnoser{}, slog.New(slog.NewTextHandler(io.Discard, nil))) if err != nil { t.Fatalf("create daemon server: %v", err) } @@ -93,7 +93,7 @@ func TestServerAcceptsDirectNativeJARThroughUnixSocket(t *testing.T) { func TestServerAcceptsContainerImageThroughUnixSocket(t *testing.T) { socketPath := shortSocketPath(t) updater := &fakeUpdater{record: transaction.Transaction{ID: "transaction-container-01", State: transaction.StateCommitted}} - server, err := New(socketPath, updater, slog.New(slog.NewTextHandler(io.Discard, nil))) + server, err := New(socketPath, updater, &fakeDiagnoser{}, slog.New(slog.NewTextHandler(io.Discard, nil))) if err != nil { t.Fatalf("create daemon server: %v", err) } @@ -121,7 +121,7 @@ func TestServerReturnsTransactionFailure(t *testing.T) { record: transaction.Transaction{ID: "transaction-02", State: transaction.StateRolledBack}, err: errors.New("health check failed"), } - server, err := New(socketPath, updater, slog.New(slog.NewTextHandler(io.Discard, nil))) + server, err := New(socketPath, updater, &fakeDiagnoser{}, slog.New(slog.NewTextHandler(io.Discard, nil))) if err != nil { t.Fatalf("create daemon server: %v", err) } @@ -143,7 +143,7 @@ func TestServerReturnsTransactionFailure(t *testing.T) { func TestServerAcceptsBackendRestartThroughUnixSocket(t *testing.T) { socketPath := shortSocketPath(t) updater := &fakeUpdater{record: transaction.Transaction{ID: "transaction-restart-01", State: transaction.StateCommitted}} - server, err := New(socketPath, updater, slog.New(slog.NewTextHandler(io.Discard, nil))) + server, err := New(socketPath, updater, &fakeDiagnoser{}, slog.New(slog.NewTextHandler(io.Discard, nil))) if err != nil { t.Fatalf("create daemon server: %v", err) } @@ -168,6 +168,55 @@ func TestServerAcceptsBackendRestartThroughUnixSocket(t *testing.T) { } } +// TestServerReturnsStatusThroughUnixSocket 验证服务端能通过 Unix Socket 处理 status 请求， +// 并把诊断结果返回给客户端。 +func TestServerReturnsStatusThroughUnixSocket(t *testing.T) { + socketPath := shortSocketPath(t) + updater := &fakeUpdater{} + diagnosis := daemonapi.Diagnosis{Service: "backend", Type: "container", Healthy: true, Items: []daemonapi.DiagnosisItem{{Level: daemonapi.DiagnosisLevelOK, Code: "ok", Message: "ok"}}} + diagnoser := &fakeDiagnoser{diagnosis: diagnosis} + server, err := New(socketPath, updater, diagnoser, slog.New(slog.NewTextHandler(io.Discard, nil))) + if err != nil { + t.Fatalf("create daemon server: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + serveResult := make(chan error, 1) + go func() { serveResult <- server.Serve(ctx) }() + waitForSocket(t, socketPath, serveResult) + + result, err := daemonclient.Status(context.Background(), socketPath, "backend") + if err != nil { + t.Fatalf("submit status: %v", err) + } + if result.Service != "backend" || !result.Healthy { + t.Fatalf("unexpected status result: %+v", result) + } +} + +// TestServerReconcileForwardsApply 验证 reconcile --apply 会把 apply 标志正确传递给诊断器。 +func TestServerReconcileForwardsApply(t *testing.T) { + socketPath := shortSocketPath(t) + updater := &fakeUpdater{} + diagnoser := &fakeDiagnoser{diagnosis: daemonapi.Diagnosis{Service: "backend", Type: "container", Healthy: true}} + server, err := New(socketPath, updater, diagnoser, slog.New(slog.NewTextHandler(io.Discard, nil))) + if err != nil { + t.Fatalf("create daemon server: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + serveResult := make(chan error, 1) + go func() { serveResult <- server.Serve(ctx) }() + waitForSocket(t, socketPath, serveResult) + + if _, err := daemonclient.Reconcile(context.Background(), socketPath, "backend", true); err != nil { + t.Fatalf("submit reconcile apply: %v", err) + } + if !diagnoser.applied { + t.Fatal("reconcile --apply was not forwarded to diagnoser") + } +} + // shortSocketPath 创建一个短目录并返回其下的 daemon.sock 路径，用于规避 Unix Socket 路径长度上限。 // 目录会在测试结束时自动清理。 func shortSocketPath(t *testing.T) string { @@ -259,3 +308,21 @@ func (u *fakeUpdater) Restart(_ context.Context, report backendupdate.ProgressRe } return u.record, u.err } + +// fakeDiagnoser backendDiagnoser 接口的测试替身，返回预设的空诊断结果。 +type fakeDiagnoser struct { + diagnosis daemonapi.Diagnosis + err error + applied bool +} + +// Diagnose 返回预设的诊断结果。 +func (d *fakeDiagnoser) Diagnose(context.Context) (daemonapi.Diagnosis, error) { + return d.diagnosis, d.err +} + +// Reconcile 返回预设的诊断结果，并记录 apply 标志。 +func (d *fakeDiagnoser) Reconcile(_ context.Context, apply bool) (daemonapi.Diagnosis, error) { + d.applied = apply + return d.diagnosis, d.err +} diff --git a/internal/deploymentconfig/config.go b/internal/deploymentconfig/config.go index 78912a0..1625d63 100644 --- a/internal/deploymentconfig/config.go +++ b/internal/deploymentconfig/config.go @@ -101,6 +101,9 @@ type Backend struct { ActiveJAR string `toml:"active_jar"` // SystemctlPath systemctl 可执行文件的绝对路径。 SystemctlPath string `toml:"systemctl_path"` + // StopGraceSeconds 容器后端停止旧容器时，发送停止信号后到强制终止前的等待秒数。 + // 可省略；省略时执行器使用默认值 30。原生后端不使用该字段。 + StopGraceSeconds *int `toml:"stop_grace_seconds"` // Slot 包含 8080、8081 两个受支持槽位的精确取值。 Slot BackendSlots `toml:"slot"` } @@ -187,7 +190,7 @@ func validateExactDocumentKeys(document []byte) error { if err != nil { return err } - if err := rejectUnknownKeys(backend, "backend", "type", "release_dir", "active_jar", "systemctl_path", "slot"); err != nil { + if err := rejectUnknownKeys(backend, "backend", "type", "release_dir", "active_jar", "systemctl_path", "stop_grace_seconds", "slot"); err != nil { return err } slots, err := exactTable(backend, "backend", "slot") @@ -264,6 +267,9 @@ func (c Config) Validate() error { if err := validateAbsolutePath("backend.systemctl_path", c.Backend.SystemctlPath); err != nil { return err } + if c.Backend.StopGraceSeconds != nil && *c.Backend.StopGraceSeconds <= 0 { + return errors.New("backend.stop_grace_seconds must be a positive integer") + } switch c.Backend.Type { case BackendTypeNative: if c.Backend.Slot.Port8080.ContainerName != "" || c.Backend.Slot.Port8081.ContainerName != "" { diff --git a/internal/deploymentconfig/config_test.go b/internal/deploymentconfig/config_test.go index ace018f..ee10d3f 100644 --- a/internal/deploymentconfig/config_test.go +++ b/internal/deploymentconfig/config_test.go @@ -91,6 +91,28 @@ func TestLoadValidContainerConfiguration(t *testing.T) { } } +// TestLoadAcceptsStopGraceSeconds 验证 backend.stop_grace_seconds 配置为合法正整数时 +// 能被接受，并解析到 Backend.StopGraceSeconds。 +func TestLoadAcceptsStopGraceSeconds(t *testing.T) { + content := strings.Replace(validContainerConfig, "systemctl_path = \"/bin/systemctl\"", "systemctl_path = \"/bin/systemctl\"\nstop_grace_seconds = 45", 1) + config, err := Load(writeConfig(t, content)) + if err != nil { + t.Fatalf("load configuration with stop_grace_seconds: %v", err) + } + if config.Backend.StopGraceSeconds == nil || *config.Backend.StopGraceSeconds != 45 { + t.Fatalf("unexpected stop_grace_seconds: %+v", config.Backend.StopGraceSeconds) + } +} + +// TestLoadRejectsNonPositiveStopGraceSeconds 验证 backend.stop_grace_seconds 配置为非正整数时被拒绝。 +func TestLoadRejectsNonPositiveStopGraceSeconds(t *testing.T) { + content := strings.Replace(validContainerConfig, "systemctl_path = \"/bin/systemctl\"", "systemctl_path = \"/bin/systemctl\"\nstop_grace_seconds = 0", 1) + _, err := Load(writeConfig(t, content)) + if err == nil || !strings.Contains(err.Error(), "backend.stop_grace_seconds") { + t.Fatalf("expected stop_grace_seconds rejection, got %v", err) + } +} + // TestLoadRejectsChangedContainerName 验证容器槽位的 container_name 被篡改时会被拒绝， // 且错误信息精确指向 backend.slot.8081.container_name 字段。 func TestLoadRejectsChangedContainerName(t *testing.T) { diff --git a/packaging/etc/yms-daemon/yms-daemon.toml b/packaging/etc/yms-daemon/yms-daemon.toml index 32b48bb..49f98c1 100644 --- a/packaging/etc/yms-daemon/yms-daemon.toml +++ b/packaging/etc/yms-daemon/yms-daemon.toml @@ -1,11 +1,42 @@ +# ============================================================================= +# yms-daemon 本机部署配置示例 +# ============================================================================= +# 后端支持两种运行类型，通过 [backend] 的 type 字段切换： +# type = "native" 原生部署：systemd 双实例 JVM + 版本化 JAR +# type = "container" 容器部署：Docker standalone 蓝绿容器 +# +# 快速切换步骤： +# 1. 注释掉当前使用的 [backend] 配置块，取消注释目标 [backend] 配置块； +# 2. 确认 [backend] 的 type 与目标一致（native 或 container）； +# 3. 重启 yms-daemon.service：systemctl restart yms-daemon.service。 +# +# 重要约束： +# - daemon 不根据现场状态推断运行类型，type 必须与现场实际部署一致； +# - native ↔ container 属于运行类型迁移，必须走独立迁移事务， +# 不能只修改本配置绕过；迁移执行器尚未实现。 +# ============================================================================= + [daemon] +# 机器级环境标记，只接受精确值 dev 或 prod： +# dev = 容器镜像从 Registry pull（开发环境） +# prod = 容器镜像从离线 repack ZIP load（客户现场） +# 该字段不改变 native 更新、健康检查、切流、事务或回滚行为。 environment = "prod" +# ============================================================================= +# 方式一：原生后端（默认启用） +# ============================================================================= [backend] type = "native" +# 原生后端 JAR 版本目录与当前生效兼容入口，与现有现场保持一致，勿改。 release_dir = "/home/yms/lib/releases" active_jar = "/home/yms/lib/glory-soft-yms.jar" +# systemctl 可执行文件的绝对路径，必须填写该服务器 command -v systemctl 的精确输出。 systemctl_path = "/bin/systemctl" +# 容器后端停止旧容器时，发送停止信号后到强制终止前的等待秒数。 +# 仅容器后端生效（原生后端忽略）；可省略，省略时默认 30 秒。 +# 配置后会覆盖容器创建时设置的 9000 秒 StopTimeout，避免旧容器优雅停机长期不退。 +# stop_grace_seconds = 30 [backend.slot.8080] unit = "yms-backend@8080.service" @@ -16,3 +47,19 @@ health_endpoint = "http://127.0.0.1:8080/yms/actuator/health" unit = "yms-backend@8081.service" jar = "/home/yms/lib/glory-soft-yms-8081.jar" health_endpoint = "http://127.0.0.1:8081/yms/actuator/health" + +# ============================================================================= +# 方式二：容器后端（切换到容器时，取消注释本块，并注释掉上面的方式一） +# ============================================================================= +# [backend] +# type = "container" +# systemctl_path = "/bin/systemctl" +# stop_grace_seconds = 30 +# +# [backend.slot.8080] +# container_name = "backend-8080" +# health_endpoint = "http://127.0.0.1:8080/yms/actuator/health" +# +# [backend.slot.8081] +# container_name = "backend-8081" +# health_endpoint = "http://127.0.0.1:8081/yms/actuator/health"