From 7755da72e7b20c0817b5c5faa6d5776d292b4835 Mon Sep 17 00:00:00 2001 From: Zhan Ziyang Date: Sun, 16 Aug 2026 17:12:06 +0800 Subject: [PATCH] feat: backend executor implement --- .gitignore | 4 +- RUNBOOK_CONTAINER_BACKEND.md | 85 ++++ RUNBOOK_NATIVE_BACKEND.md | 66 ++- UPDATE_PLAN.md | 48 +- YMS_DAEMON_REPORT.drawio | 337 ++++++++++++ internal/backendexecutor/executor.go | 150 ++++-- internal/backendexecutor/executor_test.go | 101 ++-- internal/backendexecutor/operations.go | 58 ++- internal/backendupdate/container.go | 479 ++++++++++++++++++ internal/backendupdate/container_test.go | 366 +++++++++++++ internal/backendupdate/restart.go | 4 + internal/backendupdate/updater.go | 84 ++- internal/backendupdate/updater_test.go | 10 +- internal/containerengine/engine.go | 48 +- internal/containerengine/moby.go | 30 +- internal/daemonapi/protocol.go | 22 +- internal/daemonclient/client.go | 11 + internal/daemonserver/server.go | 21 +- internal/daemonserver/server_test.go | 33 ++ internal/deploymentconfig/config.go | 105 +++- internal/deploymentconfig/config_test.go | 80 ++- .../backend_container_deployment.go | 185 +++++++ internal/transaction/coordinator.go | 23 +- internal/transaction/coordinator_test.go | 50 ++ internal/transaction/store.go | 120 ++++- internal/transaction/store_test.go | 143 ++++++ main.go | 63 ++- main_test.go | 36 +- packaging/etc/yms-daemon/yms-daemon.toml | 3 + packaging/rpm/build-rpm.sh | 102 ++++ packaging/rpm/yms-daemon.spec | 79 +++ 31 files changed, 2762 insertions(+), 184 deletions(-) create mode 100644 RUNBOOK_CONTAINER_BACKEND.md create mode 100644 YMS_DAEMON_REPORT.drawio create mode 100644 internal/backendupdate/container.go create mode 100644 internal/backendupdate/container_test.go create mode 100644 internal/transaction/backend_container_deployment.go create mode 100755 packaging/rpm/build-rpm.sh create mode 100644 packaging/rpm/yms-daemon.spec diff --git a/.gitignore b/.gitignore index c37e21a..959f5f0 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,4 @@ .idea -yms-daemon \ No newline at end of file +yms-daemon +/dist/ +/.build/ diff --git a/RUNBOOK_CONTAINER_BACKEND.md b/RUNBOOK_CONTAINER_BACKEND.md new file mode 100644 index 0000000..701fa87 --- /dev/null +++ b/RUNBOOK_CONTAINER_BACKEND.md @@ -0,0 +1,85 @@ +# Container backend 运行手册 + +本文只描述已经实现的 Docker standalone backend 更新路径。开发环境通过 Registry pull 获取镜像；客户离线环境的 repack ZIP load 尚未接入 CLI。 + +## 1. daemon 配置 + +`/etc/yms-daemon/yms-daemon.toml` 使用以下精确内容。`systemctl_path` 仍用于 host Nginx reload，必须填写该服务器 `command -v systemctl` 的精确输出。 + +```toml +[daemon] +environment = "dev" + +[backend] +type = "container" +systemctl_path = "/bin/systemctl" + +[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" +``` + +## 2. 固定运行参数 + +daemon 创建 backend 容器时固定使用： + +- `--network host` +- root 用户 `0:0` +- restart policy `no` +- `SERVER_PORT=8080` 或 `SERVER_PORT=8081` +- `SPRING_CONFIG_LOCATION=file:/app/config/yms.yaml` +- `/home/yms/conf/yms.yaml:/app/config/yms.yaml:ro` +- `/home/yms/tmp:/home/yms/tmp` +- 容器 graceful stop 上限 `9000` 秒，与 Spring Boot 的 `150m` graceful shutdown 配置对齐 + +启动 daemon 前，以下路径必须已经存在： + +```text +/home/yms/conf/yms.yaml +/home/yms/tmp +``` + +`yms.yaml` 必须是普通文件；`/home/yms/tmp` 必须是非符号链接目录。`SPRING_CONFIG_LOCATION` 会替换 Spring Boot 的默认配置搜索位置，因此该文件必须同时包含 Spring Boot 配置和原 config 文件中的业务配置，不再依赖 JAR 内配置文件追加合并。 + +## 3. 已有 container 部署的更新 + +host Nginx 当前指向 8080 时，`backend-8080` 必须正在运行；当前指向 8081 时，`backend-8081` 必须正在运行。daemon 会拒绝容器名、端口和 Nginx 状态无法对齐的更新。 + +测试命令： + +```bash +yms-daemon update \ + --service backend \ + --container-image harbor.ymswell.asia/ymswell/glory-ymswell:20260813-184902-a37bf50d-v1.1.8.1 +``` + +daemon 执行以下事务： + +1. pull 完整 tag 引用。 +2. 读取与仓库名精确匹配的 RepoDigest，并将后续步骤固定到 `repository@sha256:...`。 +3. 删除非活动 slot 上次留下的已停止容器。 +4. 创建并启动非活动 slot。 +5. 最长等待 120 秒，检查 `/yms/actuator/health` 的顶层 `status` 为 `UP`。 +6. 原子更新 host Nginx 配置并 reload。 +7. 排空旧 slot，随后 graceful stop 旧容器。 +8. 提交事务；旧容器保留为 stopped 状态，下次该 slot 被使用时删除。 + +## 4. 全新机器首次部署 + +同一条 `update --container-image` 命令可以完成全新机器的首次部署，但必须同时满足以下事实： + +- SQLite 中不存在已提交的 container backend 部署记录。 +- SQLite 中不存在历史 `COMMITTED` container backend 事务。 +- `backend-8080` 和 `backend-8081` 均不存在。 + +daemon 以 Nginx 当前活动端口为基准，选择当前未接流的槽位，完成容器创建、启动和健康检查后再切换 Nginx。首次部署没有旧容器，因此不会等待 drain，也不会执行停止旧容器；事务提交时会在同一个 SQLite 写事务内记录活动端口、容器名、镜像 digest、容器 ID 和事务 ID。 + +如果 SQLite 已存在部署记录或历史 `COMMITTED` container backend 事务，而现场容器缺失，daemon 按状态漂移拒绝更新。daemon 不会用“两槽容器均不存在”这一项单独推断全新机器。`FAILED`、`ROLLED_BACK` 首装事务不代表机器曾成功部署；再次人工执行同一更新命令时，daemon 会恢复未完成的回滚。回滚到 `ROLLED_BACK` 后再次执行命令，daemon 会清理已停止的目标容器并创建新事务，允许使用同一镜像重新部署。 + +## 5. native 到 container 的边界 + +普通 `update --container-image` 不会把 native systemd backend 自动识别成 container。native 到 container 必须进入独立迁移事务；该迁移执行器尚未实现。不得通过修改 `backend.type` 绕过迁移事务。 diff --git a/RUNBOOK_NATIVE_BACKEND.md b/RUNBOOK_NATIVE_BACKEND.md index 0fb003b..f65f8b7 100644 --- a/RUNBOOK_NATIVE_BACKEND.md +++ b/RUNBOOK_NATIVE_BACKEND.md @@ -2,7 +2,61 @@ 本文只适用于已经安装 `yms-backend@.service`、使用宿主机 Nginx managed upstream、当前由旧 `yms.service` 或 `ymsback.service` 接流的开发环境。 -## 1. 构建 ARM64 二进制 +## 1. 构建 RPM + +ARM64： + +```bash +./packaging/rpm/build-rpm.sh --goarch arm64 +``` + +x86-64： + +```bash +./packaging/rpm/build-rpm.sh --goarch amd64 +``` + +RPM 输出到 `dist/`。需要指定包版本和发布序号时使用： + +```bash +./packaging/rpm/build-rpm.sh \ + --goarch arm64 \ + --version 0.1.0 \ + --release 20260816010000 +``` + +安装或升级： + +```bash +rpm -Uvh dist/<完整RPM文件名>.rpm +systemctl enable --now yms-daemon.service +``` + +RPM 不会自动启动或重启 `yms-daemon.service`，避免安装恰好发生在更新事务执行期间。首次安装执行 `systemctl enable --now`；升级后由交付人员在确认没有更新事务执行时运行： + +```bash +systemctl restart yms-daemon.service +``` + +`/etc/yms-daemon/yms-daemon.toml` 使用 `noreplace` 语义，升级不会覆盖现场配置。RPM 会安装 `/usr/bin/yms-daemon`、两个 systemd unit、tmpfiles 配置，并创建 `/home/yms/dump`。 + +RPM 示例配置使用客户环境值： + +```toml +[daemon] +environment = "prod" +``` + +开发机必须明确改为： + +```toml +[daemon] +environment = "dev" +``` + +从不包含 `[daemon]` 的旧配置升级时，必须先加入上述完整 table，再重启 `yms-daemon.service`。`environment` 只选择未来 container 更新使用 Registry pull 还是 repack ZIP load，不改变当前 native backend 行为。 + +## 2. 直接构建 ARM64 二进制 在 `yms-daemon` 源码目录执行： @@ -10,7 +64,7 @@ CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -trimpath -o yms-daemon . ``` -## 2. 安装文件 +## 3. 手工安装文件 把构建结果和 `packaging` 目录传到服务器后，以 root 执行： @@ -26,7 +80,7 @@ systemctl daemon-reload systemctl enable --now yms-daemon.service ``` -## 3. 更新前检查 +## 4. 更新前检查 ```bash systemctl status yms-daemon.service --no-pager -l @@ -36,7 +90,7 @@ curl -fsS http://127.0.0.1:8081/yms/actuator/health 确认健康响应顶层 `status` 为 `UP`。第一次更新前不要手工停止 `ymsback.service`；daemon 需要读取当前接流端口和旧 unit 状态完成受控迁移。 -## 4. 执行更新 +## 5. 执行更新 客户 repack ZIP 模式： @@ -110,7 +164,7 @@ transaction=<事务ID> state=COMMITTED journalctl -u yms-daemon.service -f ``` -## 5. 零停机重启 backend +## 6. 零停机重启 backend 修改 `/home/yms/bin/env/yms.env` 等 backend 启动配置后，执行： @@ -128,7 +182,7 @@ yms-daemon restart --service backend --quite restart 进行中如果 daemon 重启，再次执行同一命令会恢复唯一未完成的 backend restart 事务。存在其他未完成 update 事务时，restart 拒绝创建新事务并返回该活动事务 ID。 -## 6. 更新或重启后检查 +## 7. 更新或重启后检查 ```bash systemctl show --no-pager \ diff --git a/UPDATE_PLAN.md b/UPDATE_PLAN.md index ea25d1e..3fce0f8 100644 --- a/UPDATE_PLAN.md +++ b/UPDATE_PLAN.md @@ -254,6 +254,9 @@ RPM 安装阶段创建 `/home/yms/dump`，属主和用户组固定为 `root:root daemon 本机配置文件固定为 `/etc/yms-daemon/yms-daemon.toml`。第一阶段已经冻结并实现 backend native 配置： ```toml +[daemon] +environment = "prod" + [backend] type = "native" release_dir = "/home/yms/lib/releases" @@ -273,12 +276,34 @@ health_endpoint = "http://127.0.0.1:8081/yms/actuator/health" 字段规则固定为： +- `daemon.environment` 是机器级行为标记，只接受精确值 `dev`、`prod`；`dev` 使 container 从 Registry pull，`prod` 使 container 从 repack ZIP load。该字段不改变 native 更新、健康检查、切流、事务或回滚行为。 - table 和 key 必须与上例逐字一致；大小写不同、未知字段、未知槽位和重复字段全部拒绝。 -- `backend.type` 第一阶段只接受精确值 `native`；container 配置在其完整 table 和 key 冻结后扩展，daemon 不根据现场状态自动选择。 +- `backend.type` 接受精确值 `native`、`container`；daemon 不根据现场状态自动选择。 - `release_dir`、`active_jar`、两个槽位的 `unit`、`jar` 和 `health_endpoint` 必须与上例完全相同。 - `systemctl_path` 必须由现场明确填写干净的绝对路径；当前现场填写 `/bin/systemctl`，精确检查结果为 `/usr/bin/systemctl` 的服务器填写 `/usr/bin/systemctl`。 - loader 不补默认值，不改写路径，不改写大小写，不根据端口拼接 unit、JAR 路径或健康地址。 -- RPM 安装示例文件位于 `packaging/etc/yms-daemon/yms-daemon.toml`；RPM spec 接入时将其安装到固定路径。 +- RPM 安装示例文件位于 `packaging/etc/yms-daemon/yms-daemon.toml`；RPM spec 已将其以 `noreplace` 语义安装到固定路径，跨架构打包入口为 `packaging/rpm/build-rpm.sh`。 + +backend container 开发环境配置已经冻结为： + +```toml +[daemon] +environment = "dev" + +[backend] +type = "container" +systemctl_path = "/bin/systemctl" + +[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" +``` + +container slot 固定挂载 `/home/yms/conf/yms.yaml:/app/config/yms.yaml:ro` 和 `/home/yms/tmp:/home/yms/tmp`，固定注入 `SERVER_PORT` 与 `SPRING_CONFIG_LOCATION=file:/app/config/yms.yaml`，固定使用 host network、root 用户、restart policy `no` 和 9000 秒 graceful stop 上限。外挂 `yms.yaml` 是唯一 Spring Boot 配置入口，必须同时包含 Spring Boot 配置和原 config 文件中的业务配置，不再追加加载 JAR 内默认配置。开发更新入口固定为 `yms-daemon update --service backend --container-image <完整tag引用>`；daemon pull 后读取实际 RepoDigest，并将事务后续步骤固定到 digest 引用。 ## 3. 总体架构 @@ -692,6 +717,8 @@ OpenResty 配置目录整体以 bind mount 提供给 gateway。不能只 bind mo 配置检查失败时不得替换当前配置。reload 失败时旧 worker 继续使用旧配置，daemon 恢复更新前配置并再次检查。 +全新机器首次部署必须同时满足：SQLite 不存在已提交的 container backend 部署记录、不存在历史 `COMMITTED` container backend 事务，并且 `backend-8080`、`backend-8081` 均不存在。daemon 在 Nginx 当前未接流的槽位创建并启动首个容器，健康检查通过后切流并提交部署记录；因为没有旧容器，所以不执行 drain 和停止步骤。已提交记录、Nginx 和容器现场任一不一致时拒绝更新，不得降级为首次部署；`FAILED`、`ROLLED_BACK` 首装事务不表示机器曾成功部署（见第 17 节决策 51）。 + native backend 使用相同切流步骤，但准备阶段替换为： ```text @@ -879,6 +906,8 @@ frontend 源归档本身不进入当前客户 ZIP，不能在客户 manifest 中 两种方式在进入执行器前必须归一为同一份经过校验的组件安装描述，后续步骤完全相同。 +container 镜像获取方式只由本机 TOML 的 `daemon.environment` 选择：精确值 `dev` 执行 Registry pull，精确值 `prod` 执行 repack ZIP load。该标记不附带其他 CLI、签名、更新或重启限制。 + ### 7.4 环境策略差异 | 策略 | 开发环境 | 客户现场 | @@ -1746,12 +1775,15 @@ Kubernetes executor 不在每个业务 Pod 中运行 daemon。它负责： 41. 当前客户服务器的 systemctl 可执行文件绝对路径为 `/bin/systemctl`；其他服务器使用其本机明确配置的绝对路径，daemon 不自行尝试替代路径。 42. native backend unit 使用 `Type=simple` 直接执行 `/usr/bin/java`，不调用启动脚本；堆转储目录固定为 `/home/yms/dump`，其属主和用户组固定为 `root:root`、权限固定为 `0755` 并由 RPM 创建；JVM 错误日志固定为 `/home/yms/log/hs_err_pid%p.log`，stdout 和 stderr 进入 journal，`TimeoutStopSec` 固定为 `150min`。 43. daemon 本机配置文件固定为 `/etc/yms-daemon/yms-daemon.toml`，格式固定为 TOML；使用纯 Go 的 `github.com/pelletier/go-toml/v2` 严格解码，并在解码前逐字校验完整键树，拒绝大小写变化、未知字段、未知槽位和重复字段，保持 `CGO_ENABLED=0`。 -44. 第一阶段 backend native 配置精确使用 `[backend]`、`[backend.slot.8080]` 和 `[backend.slot.8081]`；key、必填规则和精确值以 2.9 节为准，loader 不提供配置默认值。 -45. daemon 服务端路径固定为：Unix Socket `/run/yms-daemon/yms-daemon.sock`、进程锁 `/run/yms-daemon/yms-daemon.lock`、SQLite `/var/lib/yms-daemon/yms-daemon.db`、事务工作目录 `/var/lib/yms-daemon/work`、本地日志 `/var/log/yms-daemon/yms-daemon.log`。 -46. 当前 native 过渡 gateway 固定管理 `/etc/nginx/nginx.conf` 中唯一一组 `# yms-update managed upstream begin/end` 标记，Nginx 可执行文件固定为 `/usr/sbin/nginx`，systemd unit 固定为 `nginx.service`；配置完整原子替换后执行 `/usr/sbin/nginx -t` 和 `systemctl reload`，任一步失败恢复替换前完整配置。 -47. 第一次旧 unit 运行态迁移固定映射为 8080=`yms.service`、8081=`ymsback.service`；Nginx 当前接流端口对应的旧 unit 和 `yms-backend@.service` 必须恰好一个处于运行态，否则拒绝更新。 -48. backend native 更新输入分为 `repack-zip` 和 `native-jar`；前者由 `-f` 提交，后者由 `--native-jar` 提交，二者互斥且不通过扩展名推断。直传 JAR 使用完整文件 SHA-256 作为幂等身份，在 `/home/yms/lib/releases` 下保存为 `direct//<原始JAR文件名>`；目录名精确取完整小写十六进制 SHA-256 的前 12 位，完整 SHA-256 继续用于幂等和内容校验，目录冲突时拒绝覆盖。直传 JAR 复用与 repack ZIP 相同的双槽、健康检查、切流、提交和补偿流程。 -49. backend native 零停机重启命令固定为 `yms-daemon restart --service backend`，可附加 `--quite`。restart 使用当前接流 release 创建独立事务并轮转到非活动槽，不调用 `systemctl restart`，也不重新上传 JAR；当前 JAR 已在 release 目录时直接复用，兼容入口仍是普通文件时按内容身份导入 release 目录。未完成 restart 由同一命令恢复，其他未完成事务阻止新 restart。 +44. `daemon.environment` 固定为必填机器级字段，只接受精确值 `dev`、`prod`；它只选择 container 镜像获取行为，`dev` 为 Registry pull，`prod` 为 repack ZIP load。 +45. 第一阶段 backend native 配置精确使用 `[backend]`、`[backend.slot.8080]` 和 `[backend.slot.8081]`；key、必填规则和精确值以 2.9 节为准，loader 不提供配置默认值。 +46. daemon 服务端路径固定为：Unix Socket `/run/yms-daemon/yms-daemon.sock`、进程锁 `/run/yms-daemon/yms-daemon.lock`、SQLite `/var/lib/yms-daemon/yms-daemon.db`、事务工作目录 `/var/lib/yms-daemon/work`、本地日志 `/var/log/yms-daemon/yms-daemon.log`。 +47. 当前 native 过渡 gateway 固定管理 `/etc/nginx/nginx.conf` 中唯一一组 `# yms-update managed upstream begin/end` 标记，Nginx 可执行文件固定为 `/usr/sbin/nginx`，systemd unit 固定为 `nginx.service`；配置完整原子替换后执行 `/usr/sbin/nginx -t` 和 `systemctl reload`，任一步失败恢复替换前完整配置。 +48. 第一次旧 unit 运行态迁移固定映射为 8080=`yms.service`、8081=`ymsback.service`；Nginx 当前接流端口对应的旧 unit 和 `yms-backend@.service` 必须恰好一个处于运行态，否则拒绝更新。 +49. backend native 更新输入分为 `repack-zip` 和 `native-jar`；前者由 `-f` 提交，后者由 `--native-jar` 提交，二者互斥且不通过扩展名推断。直传 JAR 使用完整文件 SHA-256 作为幂等身份，在 `/home/yms/lib/releases` 下保存为 `direct//<原始JAR文件名>`；目录名精确取完整小写十六进制 SHA-256 的前 12 位，完整 SHA-256 继续用于幂等和内容校验，目录冲突时拒绝覆盖。直传 JAR 复用与 repack ZIP 相同的双槽、健康检查、切流、提交和补偿流程。 +50. backend native 零停机重启命令固定为 `yms-daemon restart --service backend`，可附加 `--quite`。restart 使用当前接流 release 创建独立事务并轮转到非活动槽，不调用 `systemctl restart`，也不重新上传 JAR；当前 JAR 已在 release 目录时直接复用，兼容入口仍是普通文件时按内容身份导入 release 目录。未完成 restart 由同一命令恢复，其他未完成事务阻止新 restart。 +51. container backend 的 `update --service backend --container-image` 仅在 SQLite 不存在已提交部署记录、不存在历史 `COMMITTED` container backend 事务，且 `backend-8080`、`backend-8081` 都不存在时按首次部署处理：目标固定为 Nginx 当前未接流的槽位，容器启动并通过健康检查后执行一次正常切流，随后原子提交事务状态和活动容器部署记录；因为不存在 previous 容器，所以不执行 drain 和停止步骤。存在已提交记录或历史 `COMMITTED` container backend 事务时，容器、记录和 Nginx 任一不一致均按状态漂移拒绝更新，不静默按首次部署处理；`FAILED`、`ROLLED_BACK` 首装事务不构成成功部署证据。该语义只覆盖 container backend，native backend 首装保持现状。 +52. 外部步骤一次执行失败后，同一条命令在本次调用内不自动重试。交付人员修复 Nginx、systemd、容器运行时或文件系统后再次人工执行命令时，协调器必须先 inspect 已失败步骤：`APPLIED` 直接补记成功，`NOT_APPLIED` 才重新 Apply，`UNKNOWN` 继续拒绝推进。回滚完成为 `ROLLED_BACK` 后，再次提交相同载荷时原子归档旧事务幂等键并创建新事务；历史 `COMMITTED` 事务仍保持幂等返回。container 首装回滚遗留的已停止目标容器由新事务准备阶段检查并清理，不要求人工删除。 ## 18. 当前必须继续研讨的精确问题 diff --git a/YMS_DAEMON_REPORT.drawio b/YMS_DAEMON_REPORT.drawio new file mode 100644 index 0000000..b311c95 --- /dev/null +++ b/YMS_DAEMON_REPORT.drawio @@ -0,0 +1,337 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/internal/backendexecutor/executor.go b/internal/backendexecutor/executor.go index 6508d7a..a46a4f2 100644 --- a/internal/backendexecutor/executor.go +++ b/internal/backendexecutor/executor.go @@ -24,31 +24,44 @@ import ( ) const ( - healthPath = "/yms/actuator/health" - healthTimeout = 120 * time.Second - healthInterval = time.Second - hostNetworkMode = "host" - bindMountType = "bind" - stepLoadImage = "backend.image.load" - stepCreateContainer = "backend.container.create" - stepStartContainer = "backend.container.start" - stepCheckHealth = "backend.container.health" + healthPath = "/yms/actuator/health" + healthTimeout = 120 * time.Second + healthInterval = time.Second + containerStopTimeoutSeconds = 150 * 60 + hostNetworkMode = "host" + bindMountType = "bind" + stepLoadImage = "backend.image.load" + stepPullImage = "backend.image.pull" + stepRemoveContainer = "backend.container.remove-inactive" + stepCreateContainer = "backend.container.create" + stepStartContainer = "backend.container.start" + stepCheckHealth = "backend.container.health" +) + +const ( + ImageAcquisitionLoad = "load" + ImageAcquisitionPull = "pull" ) // Request contains exact values supplied by the update package and local deployment configuration. // ImageReference is opaque: the executor never extracts meaning from its tag. type Request struct { - ArchivePath string - ImageReference string - ExpectedImageDigest string - Platform containerengine.Platform - ContainerName string - Port int - PortEnvironmentKey string - ConfigSource string - ConfigTarget string - RestartPolicy containerengine.RestartPolicy - HealthEndpoint string + ImageAcquisition string + ArchivePath string + ImageReference string + ExpectedImageDigest string + Platform containerengine.Platform + ContainerName string + Port int + PortEnvironmentKey string + ConfigSource string + ConfigTarget string + TmpSource string + TmpTarget string + ConfigEnvironmentKey string + ConfigLocation string + RestartPolicy containerengine.RestartPolicy + HealthEndpoint string } // Executor drives the persisted transaction up to SWITCHING after the new container is healthy. @@ -115,10 +128,10 @@ func (e *Executor) run(ctx context.Context, transactionID string, request Reques case transaction.StateStarting: // 重放 PREPARED 步骤会核对持久化意图，阻止恢复时换入另一组请求参数。 if err := e.prepare(ctx, transactionID, request); err != nil { - return e.failUnlessRecoverable(ctx, transactionID, err) + return err } if err := e.startAndCheck(ctx, transactionID, request); err != nil { - return e.failUnlessRecoverable(ctx, transactionID, err) + return err } if _, err := e.store.Transition(ctx, transactionID, transaction.StateSwitching, "backend container is healthy"); err != nil { return err @@ -144,12 +157,17 @@ func (e *Executor) validate(ctx context.Context, request Request) error { if err := validateRequest(request); err != nil { return err } - if err := regularFile(request.ArchivePath, "image archive"); err != nil { - return err + if request.ImageAcquisition == ImageAcquisitionLoad { + if err := regularFile(request.ArchivePath, "image archive"); err != nil { + return err + } } if err := regularFile(request.ConfigSource, "backend configuration"); err != nil { return err } + if err := directDirectory(request.TmpSource, "backend temporary directory"); err != nil { + return err + } if err := e.engine.Ping(ctx); err != nil { return err } @@ -157,21 +175,27 @@ func (e *Executor) validate(ctx context.Context, request Request) error { } func (e *Executor) prepare(ctx context.Context, transactionID string, request Request) error { - loadOperation := &loadImageOperation{ - engine: e.engine, - archivePath: request.ArchivePath, - imageReference: request.ImageReference, - expectedDigest: request.ExpectedImageDigest, - platform: request.Platform, - } - if _, err := e.coordinator.ExecuteStep(ctx, transactionID, loadIntent(request), loadOperation); err != nil { - return err + switch request.ImageAcquisition { + case ImageAcquisitionLoad: + operation := &loadImageOperation{engine: e.engine, archivePath: request.ArchivePath, imageReference: request.ImageReference, expectedDigest: request.ExpectedImageDigest, platform: request.Platform} + if _, err := e.coordinator.ExecuteStep(ctx, transactionID, loadIntent(request), operation); err != nil { + return err + } + case ImageAcquisitionPull: + operation := &pullImageOperation{engine: e.engine, imageReference: request.ImageReference, expectedDigest: request.ExpectedImageDigest, platform: request.Platform} + if _, err := e.coordinator.ExecuteStep(ctx, transactionID, pullIntent(request), operation); err != nil { + return err + } } image, err := e.engine.InspectImage(ctx, request.ImageReference) if err != nil { return fmt.Errorf("inspect prepared backend image: %w", err) } + removeOperation := &removeContainerOperation{engine: e.engine, name: request.ContainerName} + if _, err := e.coordinator.ExecuteStep(ctx, transactionID, removeIntent(request), removeOperation); err != nil { + return err + } createOperation := &createContainerOperation{ engine: e.engine, expectedImage: image, @@ -198,8 +222,17 @@ func (e *Executor) startAndCheck(ctx context.Context, transactionID string, requ } func validateRequest(request Request) error { - if !filepath.IsAbs(request.ArchivePath) { - return errors.New("image archive path must be absolute") + switch request.ImageAcquisition { + case ImageAcquisitionLoad: + if !filepath.IsAbs(request.ArchivePath) { + return errors.New("image archive path must be absolute") + } + case ImageAcquisitionPull: + if request.ArchivePath != "" { + return errors.New("pull image acquisition does not accept an archive path") + } + default: + return fmt.Errorf("unsupported image acquisition: %q", request.ImageAcquisition) } if request.ImageReference == "" || strings.TrimSpace(request.ImageReference) != request.ImageReference { return errors.New("exact image reference is required") @@ -222,6 +255,15 @@ func validateRequest(request Request) error { if !filepath.IsAbs(request.ConfigSource) || !filepath.IsAbs(request.ConfigTarget) { return errors.New("backend configuration source and target must be absolute paths") } + if !filepath.IsAbs(request.TmpSource) || !filepath.IsAbs(request.TmpTarget) { + return errors.New("backend temporary source and target must be absolute paths") + } + if request.ConfigEnvironmentKey == "" || strings.Contains(request.ConfigEnvironmentKey, "=") || strings.TrimSpace(request.ConfigEnvironmentKey) != request.ConfigEnvironmentKey { + return errors.New("exact backend configuration environment key is required") + } + if request.ConfigLocation == "" || strings.TrimSpace(request.ConfigLocation) != request.ConfigLocation { + return errors.New("exact backend configuration location is required") + } if err := validateRestartPolicy(request.RestartPolicy); err != nil { return err } @@ -262,6 +304,17 @@ func regularFile(path, description string) error { return nil } +func directDirectory(path, description string) error { + info, err := os.Lstat(path) + if err != nil { + return fmt.Errorf("inspect %s %s: %w", description, path, err) + } + if !info.IsDir() || info.Mode()&os.ModeSymlink != 0 { + return fmt.Errorf("%s is not a direct directory: %s", description, path) + } + return nil +} + func containerSpec(request Request) containerengine.ContainerSpec { return containerengine.ContainerSpec{ Name: request.ContainerName, @@ -269,18 +322,27 @@ func containerSpec(request Request) containerengine.ContainerSpec { Platform: request.Platform, Environment: []string{ request.PortEnvironmentKey + "=" + strconv.Itoa(request.Port), + request.ConfigEnvironmentKey + "=" + request.ConfigLocation, }, NetworkMode: hostNetworkMode, RestartPolicy: request.RestartPolicy, - Mounts: []containerengine.Mount{{ - Type: bindMountType, - Source: request.ConfigSource, - Target: request.ConfigTarget, - ReadOnly: true, - }}, + Mounts: []containerengine.Mount{ + {Type: bindMountType, Source: request.ConfigSource, Target: request.ConfigTarget, ReadOnly: true}, + {Type: bindMountType, Source: request.TmpSource, Target: request.TmpTarget}, + }, + User: "0:0", + StopTimeoutSeconds: containerStopTimeoutSeconds, } } +func pullIntent(request Request) transaction.StepIntent { + return intent(stepPullImage, "pull and verify backend image", struct { + ImageReference string `json:"imageReference"` + ImageDigest string `json:"imageDigest"` + Platform containerengine.Platform `json:"platform"` + }{request.ImageReference, request.ExpectedImageDigest, request.Platform}) +} + func loadIntent(request Request) transaction.StepIntent { return intent(stepLoadImage, "load and verify backend image", struct { ArchivePath string `json:"archivePath"` @@ -297,6 +359,12 @@ func createIntent(request Request, imageID string) transaction.StepIntent { }{containerSpec(request), imageID}) } +func removeIntent(request Request) transaction.StepIntent { + return intent(stepRemoveContainer, "remove inactive backend container", struct { + ContainerName string `json:"containerName"` + }{request.ContainerName}) +} + func startIntent(request Request) transaction.StepIntent { return intent(stepStartContainer, "start inactive backend container", struct { ContainerName string `json:"containerName"` diff --git a/internal/backendexecutor/executor_test.go b/internal/backendexecutor/executor_test.go index 070a67a..6377a05 100644 --- a/internal/backendexecutor/executor_test.go +++ b/internal/backendexecutor/executor_test.go @@ -59,7 +59,7 @@ func TestExecutorPreservesOpaqueImageTagsAndReachesSwitching(t *testing.T) { engine.mu.Unlock() t.Fatalf("image reference changed: got %q want %q", engine.lastCreateSpec.ImageReference, imageReference) } - if !slices.Contains(engine.lastCreateSpec.Environment, "SERVER_PORT=8081") { + if !slices.Contains(engine.lastCreateSpec.Environment, "SERVER_PORT=8081") || !slices.Contains(engine.lastCreateSpec.Environment, "SPRING_CONFIG_LOCATION=file:/app/config/yms.yaml") { engine.mu.Unlock() t.Fatalf("missing explicit port environment: %+v", engine.lastCreateSpec.Environment) } @@ -97,6 +97,13 @@ func TestExecutorRecoversRecordedCreateIntentWithoutRepeatingCreate(t *testing.T if _, err := store.CompleteStep(ctx, record.ID, loadStep.Key, transaction.StepSucceeded, []byte(`{"loaded":true}`), ""); err != nil { t.Fatalf("complete image step: %v", err) } + removeStep := removeIntent(request) + if _, _, err := store.RecordStepIntent(ctx, record.ID, removeStep); err != nil { + t.Fatalf("record completed remove intent: %v", err) + } + if _, err := store.CompleteStep(ctx, record.ID, removeStep.Key, transaction.StepSucceeded, []byte(`{"absent":true}`), ""); err != nil { + t.Fatalf("complete remove step: %v", err) + } createStep := createIntent(request, engine.loadedImage.ID) if _, _, err := store.RecordStepIntent(ctx, record.ID, createStep); err != nil { t.Fatalf("record crash-window create intent: %v", err) @@ -108,7 +115,7 @@ func TestExecutorRecoversRecordedCreateIntentWithoutRepeatingCreate(t *testing.T engine.mu.Lock() calls := engine.callCounts() engine.mu.Unlock() - if calls.load != 0 || calls.create != 0 || calls.start != 1 { + if calls.load != 0 || calls.remove != 0 || calls.create != 0 || calls.start != 1 { t.Fatalf("unexpected recovery calls: %+v", calls) } current, err := store.Transaction(ctx, record.ID) @@ -135,7 +142,7 @@ func TestExecutorMarksValidationFailureTerminal(t *testing.T) { } } -func TestExecutorKeepsPreparedStateForConflictingContainerInspection(t *testing.T) { +func TestExecutorReplacesInactiveContainerWithConflictingImage(t *testing.T) { ctx := context.Background() store, coordinator := testTransactionKernel(t) request := testRequest(t, testRepository+":20260814-093609-d7ed70f0") @@ -147,14 +154,18 @@ func TestExecutorKeepsPreparedStateForConflictingContainerInspection(t *testing. executor := testExecutor(t, store, coordinator, engine, healthyResponse) record := createTransaction(t, store, "container-conflict") - err := executor.Run(ctx, record.ID, request) - var uncertain *transaction.UncertainStepError - if !errors.As(err, &uncertain) { - t.Fatalf("expected uncertain container step, got %v", err) + if err := executor.Run(ctx, record.ID, request); err != nil { + t.Fatalf("replace inactive container: %v", err) } current, readErr := store.Transaction(ctx, record.ID) - if readErr != nil || current.State != transaction.StatePrepared { - t.Fatalf("conflict did not preserve prepared state: record=%+v err=%v", current, readErr) + if readErr != nil || current.State != transaction.StateSwitching { + t.Fatalf("replacement did not reach switching: record=%+v err=%v", current, readErr) + } + engine.mu.Lock() + calls := engine.callCounts() + engine.mu.Unlock() + if calls.remove != 1 || calls.create != 1 { + t.Fatalf("inactive replacement calls mismatch: %+v", calls) } } @@ -222,17 +233,22 @@ func testRequest(t *testing.T, imageReference string) Request { t.Fatalf("write backend configuration: %v", err) } return Request{ - ArchivePath: archivePath, - ImageReference: imageReference, - ExpectedImageDigest: testDigest, - Platform: containerengine.Platform{OS: "linux", Architecture: "amd64"}, - ContainerName: "explicit-backend-8081", - Port: 8081, - PortEnvironmentKey: "SERVER_PORT", - ConfigSource: configPath, - ConfigTarget: "/app/config/application.yaml", - RestartPolicy: containerengine.RestartPolicy{Name: "unless-stopped"}, - HealthEndpoint: "http://127.0.0.1:8081/yms/actuator/health", + ImageAcquisition: ImageAcquisitionLoad, + ArchivePath: archivePath, + ImageReference: imageReference, + ExpectedImageDigest: testDigest, + Platform: containerengine.Platform{OS: "linux", Architecture: "amd64"}, + ContainerName: "explicit-backend-8081", + Port: 8081, + PortEnvironmentKey: "SERVER_PORT", + ConfigSource: configPath, + ConfigTarget: "/app/config/yms.yaml", + TmpSource: directory, + TmpTarget: "/home/yms/tmp", + ConfigEnvironmentKey: "SPRING_CONFIG_LOCATION", + ConfigLocation: "file:/app/config/yms.yaml", + RestartPolicy: containerengine.RestartPolicy{Name: "unless-stopped"}, + HealthEndpoint: "http://127.0.0.1:8081/yms/actuator/health", } } @@ -347,6 +363,14 @@ func (e *fakeEngine) LoadImage(_ context.Context, input io.Reader) error { return nil } +func (e *fakeEngine) PullImage(context.Context, string) error { + e.mu.Lock() + defer e.mu.Unlock() + e.loadCalls++ + e.imageAvailable = true + return nil +} + func (e *fakeEngine) InspectImage(context.Context, string) (containerengine.Image, error) { e.mu.Lock() defer e.mu.Unlock() @@ -384,6 +408,19 @@ func (e *fakeEngine) StartContainer(_ context.Context, name string) error { return nil } +func (e *fakeEngine) StopContainer(_ context.Context, name string) error { + e.mu.Lock() + defer e.mu.Unlock() + record, exists := e.containers[name] + if !exists { + return containerengine.ErrNotFound + } + record.Running = false + record.Status = "exited" + e.containers[name] = record + return nil +} + func (e *fakeEngine) InspectContainer(_ context.Context, name string) (containerengine.Container, error) { e.mu.Lock() defer e.mu.Unlock() @@ -409,17 +446,19 @@ func (e *fakeEngine) Close() error { return nil } func (e *fakeEngine) containerFromSpec(spec containerengine.ContainerSpec, running bool) containerengine.Container { return containerengine.Container{ - ID: "container-id-" + spec.Name, - Name: spec.Name, - ImageID: e.loadedImage.ID, - ImageReference: spec.ImageReference, - Platform: spec.Platform.OS + "/" + spec.Platform.Architecture, - Running: running, - Status: "created", - Environment: append([]string(nil), spec.Environment...), - NetworkMode: spec.NetworkMode, - RestartPolicy: spec.RestartPolicy, - Mounts: append([]containerengine.Mount(nil), spec.Mounts...), + ID: "container-id-" + spec.Name, + Name: spec.Name, + ImageID: e.loadedImage.ID, + ImageReference: spec.ImageReference, + Platform: spec.Platform.OS + "/" + spec.Platform.Architecture, + Running: running, + Status: "created", + Environment: append([]string(nil), spec.Environment...), + NetworkMode: spec.NetworkMode, + RestartPolicy: spec.RestartPolicy, + Mounts: append([]containerengine.Mount(nil), spec.Mounts...), + User: spec.User, + StopTimeoutSeconds: spec.StopTimeoutSeconds, } } diff --git a/internal/backendexecutor/operations.go b/internal/backendexecutor/operations.go index 6265029..06f9638 100644 --- a/internal/backendexecutor/operations.go +++ b/internal/backendexecutor/operations.go @@ -23,6 +23,36 @@ type loadImageOperation struct { platform containerengine.Platform } +type pullImageOperation struct { + engine containerengine.Engine + imageReference string + expectedDigest string + platform containerengine.Platform +} + +func (o *pullImageOperation) Apply(ctx context.Context) error { + return o.engine.PullImage(ctx, o.imageReference) +} + +func (o *pullImageOperation) Inspect(ctx context.Context) (transaction.Inspection, error) { + image, err := o.engine.InspectImage(ctx, o.imageReference) + if errors.Is(err, containerengine.ErrNotFound) { + return transaction.Inspection{Status: transaction.InspectionNotApplied}, nil + } + if err != nil { + return transaction.Inspection{}, err + } + matches, err := imageMatches(image, o.expectedDigest, o.platform) + if err != nil { + return transaction.Inspection{}, err + } + result := resultJSON(image) + if !matches { + return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: result}, nil + } + return transaction.Inspection{Status: transaction.InspectionApplied, Result: result}, nil +} + func (o *loadImageOperation) Apply(ctx context.Context) error { archive, err := os.Open(o.archivePath) if err != nil { @@ -62,6 +92,30 @@ type createContainerOperation struct { spec containerengine.ContainerSpec } +type removeContainerOperation struct { + engine containerengine.Engine + name string +} + +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 +} + +func (o *removeContainerOperation) Inspect(ctx context.Context) (transaction.Inspection, error) { + record, err := o.engine.InspectContainer(ctx, o.name) + if errors.Is(err, containerengine.ErrNotFound) { + return transaction.Inspection{Status: transaction.InspectionApplied}, nil + } + if err != nil { + return transaction.Inspection{}, err + } + return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: containerResult(record)}, nil +} + func (o *createContainerOperation) Apply(ctx context.Context) error { _, err := o.engine.CreateContainer(ctx, o.spec) return err @@ -168,7 +222,7 @@ func healthInspection(report healthcheck.ActuatorReport, ready bool, err error) } func containerMatches(record containerengine.Container, expectedImageID string, spec containerengine.ContainerSpec) bool { - if record.ImageID != expectedImageID || record.NetworkMode != spec.NetworkMode || record.RestartPolicy != spec.RestartPolicy { + if record.ImageID != expectedImageID || record.NetworkMode != spec.NetworkMode || record.RestartPolicy != spec.RestartPolicy || record.User != spec.User || record.StopTimeoutSeconds != spec.StopTimeoutSeconds { return false } for _, expected := range spec.Environment { @@ -195,6 +249,8 @@ func containerResult(record containerengine.Container) json.RawMessage { } var _ transaction.Operation = (*loadImageOperation)(nil) +var _ transaction.Operation = (*pullImageOperation)(nil) +var _ transaction.Operation = (*removeContainerOperation)(nil) var _ transaction.Operation = (*createContainerOperation)(nil) var _ transaction.Operation = (*startContainerOperation)(nil) var _ transaction.Operation = (*healthOperation)(nil) diff --git a/internal/backendupdate/container.go b/internal/backendupdate/container.go new file mode 100644 index 0000000..2093e14 --- /dev/null +++ b/internal/backendupdate/container.go @@ -0,0 +1,479 @@ +package backendupdate + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + + "github.com/distribution/reference" + + "yms-daemon/internal/backendexecutor" + "yms-daemon/internal/containerengine" + "yms-daemon/internal/deploymentconfig" + "yms-daemon/internal/hostnginx" + "yms-daemon/internal/transaction" +) + +const inputTypeContainerImage = "container-image" + +type persistedContainerRequest struct { + InputType string `json:"inputType"` + ImageReference string `json:"imageReference"` + ImmutableReference string `json:"immutableReference"` + ImageDigest string `json:"imageDigest"` + Platform containerengine.Platform `json:"platform"` + TargetPort int `json:"targetPort"` + TargetContainer string `json:"targetContainer"` + PreviousPort int `json:"previousPort"` + PreviousContainer string `json:"previousContainer"` + TargetHealthEndpoint string `json:"targetHealthEndpoint"` + GatewayBeforePath string `json:"gatewayBeforePath"` + GatewayAfterPath string `json:"gatewayAfterPath"` + GatewayReceiptPath string `json:"gatewayReceiptPath"` +} + +// UpdateContainerImage pulls one development image, freezes its repository +// digest, and updates the inactive Docker backend slot. +func (u *Updater) UpdateContainerImage(ctx context.Context, imageReference string, report ProgressReporter) (transaction.Transaction, error) { + if u.containerExecutor == nil || u.engine == nil { + return transaction.Transaction{}, errors.New("container backend updater is not configured") + } + 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) + } + + 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 != inputTypeContainerImage { + return active, &transaction.ActiveTransactionError{TransactionID: active.ID} + } + if request.ImageReference != imageReference { + 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 + } + + reportProgress(report, Progress{Message: "Pulling backend image " + imageReference}) + resolved, err := u.pullAndResolveImage(ctx, imageReference) + if err != nil { + return transaction.Transaction{}, err + } + reportProgress(report, Progress{Message: "Backend image resolved: " + resolved.ImmutableReference}) + + before, err := u.gateway.Read() + if err != nil { + return transaction.Transaction{}, err + } + activeSlot, err := u.config.Backend.SlotForPort(before.ActivePort) + if err != nil { + return transaction.Transaction{}, err + } + deployment, deploymentErr := u.store.BackendContainerDeployment(ctx) + hasDeployment := deploymentErr == nil + if deploymentErr != nil && !errors.Is(deploymentErr, transaction.ErrNotFound) { + return transaction.Transaction{}, deploymentErr + } + hasHistory, err := u.store.HasCommittedBackendContainerTransactionHistory(ctx) + if err != nil { + return transaction.Transaction{}, err + } + targetPort, targetSlot, previousContainer, err := u.resolveContainerSlots( + ctx, + before.ActivePort, + activeSlot, + deployment, + hasDeployment, + hasHistory, + ) + 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: inputTypeContainerImage, ImageReference: imageReference, + ImmutableReference: resolved.ImmutableReference, ImageDigest: resolved.Digest, Platform: resolved.Platform, + TargetPort: targetPort, TargetContainer: targetSlot.ContainerName, + PreviousPort: before.ActivePort, PreviousContainer: previousContainer, + TargetHealthEndpoint: targetSlot.HealthEndpoint, + 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 update request: %w", err) + } + record, created, err := u.store.CreateTransaction(ctx, transaction.CreateRequest{ + ID: transactionID, IdempotencyKey: serviceBackend + ":container:" + resolved.Digest, + 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) +} + +// resolveContainerSlots reconciles the committed deployment, gateway and both +// exact container names before selecting a target. A fresh installation has no +// deployment row, no committed container transaction history and no slot +// containers. It starts on the non-routed slot so traffic is exposed only after +// health passes. +func (u *Updater) resolveContainerSlots( + ctx context.Context, + activePort int, + activeSlot deploymentconfig.BackendSlot, + deployment transaction.BackendContainerDeployment, + hasDeployment bool, + hasHistory bool, +) (int, deploymentconfig.BackendSlot, string, error) { + inactivePort := otherPort(activePort) + inactiveSlot, err := u.config.Backend.SlotForPort(inactivePort) + if err != nil { + return 0, deploymentconfig.BackendSlot{}, "", err + } + active, activeErr := u.engine.InspectContainer(ctx, activeSlot.ContainerName) + if activeErr != nil && !errors.Is(activeErr, containerengine.ErrNotFound) { + return 0, deploymentconfig.BackendSlot{}, "", fmt.Errorf("inspect active backend container %s: %w", activeSlot.ContainerName, activeErr) + } + inactive, inactiveErr := u.engine.InspectContainer(ctx, inactiveSlot.ContainerName) + if inactiveErr != nil && !errors.Is(inactiveErr, containerengine.ErrNotFound) { + return 0, deploymentconfig.BackendSlot{}, "", fmt.Errorf("inspect inactive backend container %s: %w", inactiveSlot.ContainerName, inactiveErr) + } + activeFound := activeErr == nil + inactiveFound := inactiveErr == nil + + if hasDeployment { + if deployment.ActivePort != activePort { + return 0, deploymentconfig.BackendSlot{}, "", fmt.Errorf("committed backend container port %d does not match gateway active port %d", deployment.ActivePort, activePort) + } + if deployment.ContainerName != activeSlot.ContainerName { + return 0, deploymentconfig.BackendSlot{}, "", fmt.Errorf("committed backend container %s does not match gateway slot container %s", deployment.ContainerName, activeSlot.ContainerName) + } + if !activeFound { + return 0, deploymentconfig.BackendSlot{}, "", fmt.Errorf("committed active backend container %s is missing", activeSlot.ContainerName) + } + if active.ID != deployment.ContainerID { + return 0, deploymentconfig.BackendSlot{}, "", fmt.Errorf("active backend container %s identity does not match committed deployment", activeSlot.ContainerName) + } + } + + if activeFound { + if !active.Running || active.Dead { + return 0, deploymentconfig.BackendSlot{}, "", fmt.Errorf("active backend container %s is not running", activeSlot.ContainerName) + } + if inactiveFound && inactive.Running && !inactive.Dead { + return 0, deploymentconfig.BackendSlot{}, "", fmt.Errorf("inactive backend container %s is unexpectedly running", inactiveSlot.ContainerName) + } + return inactivePort, inactiveSlot, activeSlot.ContainerName, nil + } + if hasHistory { + return 0, deploymentconfig.BackendSlot{}, "", fmt.Errorf("active backend container %s is missing on a server with committed backend container transaction history", activeSlot.ContainerName) + } + if inactiveFound && inactive.Running && !inactive.Dead { + return 0, deploymentconfig.BackendSlot{}, "", fmt.Errorf("active backend container %s is missing but inactive container %s is running", activeSlot.ContainerName, inactiveSlot.ContainerName) + } + return inactivePort, inactiveSlot, "", nil +} + +type resolvedImage struct { + ImmutableReference string + Digest string + Platform containerengine.Platform +} + +func (u *Updater) pullAndResolveImage(ctx context.Context, imageReference string) (resolvedImage, error) { + named, err := reference.ParseNormalizedNamed(imageReference) + if err != nil { + return resolvedImage{}, fmt.Errorf("parse container image reference: %w", err) + } + if _, ok := named.(reference.Tagged); !ok { + return resolvedImage{}, errors.New("--container-image requires a tag-qualified image reference") + } + if err := u.engine.Ping(ctx); err != nil { + return resolvedImage{}, err + } + if err := u.engine.PullImage(ctx, imageReference); err != nil { + return resolvedImage{}, err + } + image, err := u.engine.InspectImage(ctx, imageReference) + if err != nil { + return resolvedImage{}, fmt.Errorf("inspect pulled backend image: %w", err) + } + if image.Platform.OS == "" || image.Platform.Architecture == "" { + return resolvedImage{}, errors.New("pulled backend image does not report an exact platform") + } + repository := reference.TrimNamed(named).Name() + matches := make(map[string]string) + for _, value := range image.RepoDigests { + digested, err := reference.ParseNormalizedNamed(value) + if err != nil { + return resolvedImage{}, fmt.Errorf("parse pulled repository digest %q: %w", value, err) + } + withDigest, ok := digested.(reference.Digested) + if !ok || reference.TrimNamed(digested).Name() != repository { + continue + } + matches[withDigest.Digest().String()] = value + } + if len(matches) == 0 { + return resolvedImage{}, fmt.Errorf("pulled backend image has no repository digest for %s", repository) + } + if len(matches) != 1 { + return resolvedImage{}, fmt.Errorf("pulled backend image has multiple repository digests for %s", repository) + } + for digest, immutableReference := range matches { + return resolvedImage{ImmutableReference: immutableReference, Digest: digest, Platform: image.Platform}, nil + } + return resolvedImage{}, errors.New("repository digest resolution produced no result") +} + +func (u *Updater) runContainerUpdate(ctx context.Context, record transaction.Transaction, request persistedContainerRequest, created bool, report ProgressReporter) (transaction.Transaction, error) { + message := "Resuming container backend transaction " + record.ID + if created { + message = "Created container backend transaction " + record.ID + } + reportProgress(report, Progress{TransactionID: record.ID, State: record.State, Message: message}) + if record.State.Terminal() { + return terminalResult(record) + } + executorRequest := backendexecutor.Request{ + ImageAcquisition: backendexecutor.ImageAcquisitionPull, + ImageReference: request.ImmutableReference, ExpectedImageDigest: request.ImageDigest, Platform: request.Platform, + ContainerName: request.TargetContainer, Port: request.TargetPort, + PortEnvironmentKey: deploymentconfig.ContainerPortEnvironment, + ConfigSource: u.containerConfigSource, ConfigTarget: u.containerConfigTarget, + TmpSource: u.containerTmpSource, TmpTarget: u.containerTmpTarget, + ConfigEnvironmentKey: deploymentconfig.ContainerConfigEnvironment, + ConfigLocation: deploymentconfig.ContainerConfigLocation, + RestartPolicy: containerengine.RestartPolicy{Name: "no"}, + HealthEndpoint: request.TargetHealthEndpoint, + } + switch record.State { + case transaction.StateCreated, transaction.StateValidating, transaction.StatePrepared, transaction.StateStarting: + if err := u.containerExecutor.Run(ctx, record.ID, executorRequest); err != nil { + current, readErr := u.store.Transaction(ctx, record.ID) + if readErr != nil || current.State == transaction.StateFailed { + return u.currentWithError(ctx, record.ID, errors.Join(err, readErr)) + } + before, beforeErr := readGatewaySnapshot(request.GatewayBeforePath, request.PreviousPort) + after, afterErr := readGatewaySnapshot(request.GatewayAfterPath, request.TargetPort) + if snapshotErr := errors.Join(beforeErr, afterErr); snapshotErr != nil { + return u.currentWithError(ctx, record.ID, errors.Join(err, snapshotErr)) + } + rollbackErr := u.rollbackContainer(ctx, record.ID, request, before, after, err) + return u.currentWithError(ctx, record.ID, rollbackErr) + } + case transaction.StateSwitching, transaction.StateVerifying, transaction.StateDraining, transaction.StateRollingBack: + default: + return u.currentWithError(ctx, record.ID, fmt.Errorf("container backend update cannot resume transaction %s in state %s", record.ID, record.State)) + } + if err := u.switchAndCommitContainer(ctx, record.ID, request, report); err != nil { + return u.currentWithError(ctx, record.ID, err) + } + return u.store.Transaction(ctx, record.ID) +} + +func (u *Updater) switchAndCommitContainer(ctx context.Context, transactionID string, request persistedContainerRequest, report ProgressReporter) error { + before, err := readGatewaySnapshot(request.GatewayBeforePath, request.PreviousPort) + if err != nil { + return err + } + after, err := readGatewaySnapshot(request.GatewayAfterPath, request.TargetPort) + if err != nil { + return err + } + for { + record, err := u.store.Transaction(ctx, transactionID) + if err != nil { + return err + } + switch record.State { + case transaction.StateSwitching: + reportProgress(report, Progress{TransactionID: transactionID, State: record.State, Message: fmt.Sprintf("Switching host Nginx backend traffic to port %d", request.TargetPort)}) + operation := &gatewayOperation{controller: u.gateway, before: before, after: after, receiptPath: request.GatewayReceiptPath} + if _, err := u.coordinator.ExecuteStep(ctx, transactionID, containerGatewaySwitchIntent(request), operation); err != nil { + return u.rollbackContainer(ctx, transactionID, request, before, after, err) + } + if _, err := u.store.Transition(ctx, transactionID, transaction.StateVerifying, "host Nginx now routes backend traffic to the healthy container slot"); err != nil { + return err + } + case transaction.StateVerifying: + message := "container backend switch verified" + if request.PreviousContainer != "" { + message += "; previous container draining" + } + if _, err := u.store.Transition(ctx, transactionID, transaction.StateDraining, message); err != nil { + return err + } + case transaction.StateDraining: + if request.PreviousContainer != "" { + reportProgress(report, Progress{TransactionID: transactionID, State: record.State, Message: fmt.Sprintf("Draining previous backend container for %s", u.drain)}) + if err := waitContext(ctx, u.drain); err != nil { + return err + } + operation := &containerStopOperation{engine: u.engine, name: request.PreviousContainer} + if _, err := u.coordinator.ExecuteStep(ctx, transactionID, stopPreviousContainerIntent(request), operation); err != nil { + return err + } + } + target, err := u.engine.InspectContainer(ctx, request.TargetContainer) + if err != nil { + return fmt.Errorf("inspect target backend container before commit: %w", err) + } + 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 { + return fmt.Errorf("target backend container %s image reference does not match the transaction", request.TargetContainer) + } + _, err = u.store.CommitBackendContainerDeployment(ctx, transactionID, transaction.BackendContainerDeployment{ + ActivePort: request.TargetPort, + ContainerName: request.TargetContainer, + ImageDigest: request.ImageDigest, + ContainerID: target.ID, + }, "container backend update committed") + if err == nil { + reportProgress(report, Progress{TransactionID: transactionID, State: transaction.StateCommitted, Message: "Container backend update committed"}) + } + return err + case transaction.StateCommitted: + return nil + case transaction.StateRollingBack: + return u.rollbackContainer(ctx, transactionID, request, before, after, errors.New("resuming container backend rollback")) + default: + return fmt.Errorf("container backend commit cannot continue transaction %s in state %s", transactionID, record.State) + } + } +} + +func (u *Updater) rollbackContainer(ctx context.Context, transactionID string, request persistedContainerRequest, before hostnginx.Snapshot, after hostnginx.Snapshot, cause error) error { + record, err := u.store.Transaction(ctx, transactionID) + if err != nil { + return errors.Join(cause, err) + } + if record.State != transaction.StateRollingBack { + if _, err := u.store.Transition(ctx, transactionID, transaction.StateRollingBack, "container backend compensation started"); err != nil { + return errors.Join(cause, err) + } + } + 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} + _, stopErr := u.coordinator.ExecuteStep(ctx, transactionID, stopTargetContainerIntent(request), stop) + if err := errors.Join(gatewayErr, stopErr); err != nil { + return errors.Join(cause, err) + } + _, transitionErr := u.store.Transition(ctx, transactionID, transaction.StateRolledBack, "container backend compensation completed") + return errors.Join(cause, transitionErr) +} + +type containerStopOperation struct { + engine containerengine.Engine + name string +} + +func (o *containerStopOperation) Apply(ctx context.Context) error { + err := o.engine.StopContainer(ctx, o.name) + if errors.Is(err, containerengine.ErrNotFound) { + return nil + } + return err +} + +func (o *containerStopOperation) Inspect(ctx context.Context) (transaction.Inspection, error) { + record, err := o.engine.InspectContainer(ctx, o.name) + if errors.Is(err, containerengine.ErrNotFound) { + return transaction.Inspection{Status: transaction.InspectionApplied}, nil + } + if err != nil { + return transaction.Inspection{}, err + } + result, _ := json.Marshal(record) + if !record.Running || record.Dead { + return transaction.Inspection{Status: transaction.InspectionApplied, Result: result}, nil + } + return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: result}, nil +} + +func containerGatewaySwitchIntent(request persistedContainerRequest) transaction.StepIntent { + return stepIntent("backend.container.gateway.switch", "switch host Nginx to container backend", struct { + BeforePort int `json:"beforePort"` + AfterPort int `json:"afterPort"` + }{request.PreviousPort, request.TargetPort}) +} + +func containerGatewayRestoreIntent(request persistedContainerRequest) transaction.StepIntent { + return stepIntent("backend.container.gateway.restore", "restore host Nginx after container backend failure", struct { + Port int `json:"port"` + }{request.PreviousPort}) +} + +func stopPreviousContainerIntent(request persistedContainerRequest) transaction.StepIntent { + return stepIntent("backend.previous-container.stop", "stop previous backend container after drain", struct { + ContainerName string `json:"containerName"` + }{request.PreviousContainer}) +} + +func stopTargetContainerIntent(request persistedContainerRequest) transaction.StepIntent { + return stepIntent("backend.target-container.stop", "stop compensated backend container", struct { + ContainerName string `json:"containerName"` + }{request.TargetContainer}) +} + +func decodeContainerRequest(content json.RawMessage, request *persistedContainerRequest) error { + decoder := json.NewDecoder(bytes.NewReader(content)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(request); err != nil { + return fmt.Errorf("decode persisted container backend request: %w", err) + } + return nil +} + +var _ transaction.Operation = (*containerStopOperation)(nil) diff --git a/internal/backendupdate/container_test.go b/internal/backendupdate/container_test.go new file mode 100644 index 0000000..d80f16e --- /dev/null +++ b/internal/backendupdate/container_test.go @@ -0,0 +1,366 @@ +package backendupdate + +import ( + "bytes" + "context" + "errors" + "io" + "log/slog" + "net/http" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "yms-daemon/internal/backendexecutor" + "yms-daemon/internal/containerengine" + "yms-daemon/internal/deploymentconfig" + "yms-daemon/internal/hostnginx" + "yms-daemon/internal/transaction" +) + +const containerTestImage = "harbor.ymswell.asia/ymswell/glory-ymswell:20260813-184902-a37bf50d-v1.1.8.1" +const containerTestDigest = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + +func TestContainerUpdaterPullsSwitchesAndStopsPreviousSlot(t *testing.T) { + ctx := context.Background() + updater, store, engine, gateway := newContainerUpdaterFixture(t, map[string]containerengine.Container{ + "backend-8080": { + ID: "existing-backend-8080", Name: "backend-8080", Running: true, + }, + }, 0) + + record, err := updater.UpdateContainerImage(ctx, containerTestImage, nil) + if err != nil { + t.Fatalf("update container backend: %v", err) + } + if record.State != transaction.StateCommitted || gateway.snapshot.ActivePort != 8081 { + t.Fatalf("unexpected committed container update: record=%+v gateway=%+v", record, gateway.snapshot) + } + if engine.containers["backend-8080"].Running { + t.Fatal("previous backend container is still running") + } + if len(engine.stopped) != 1 || engine.stopped[0] != "backend-8080" { + t.Fatalf("unexpected stopped containers: %+v", engine.stopped) + } + assertCreatedContainerSpec(t, engine.lastCreateSpec) + assertCommittedContainerDeployment(t, store, record.ID, 8081, "backend-8081", "container-id-backend-8081") +} + +func TestContainerUpdaterFirstInstallCreatesInactiveSlotBeforeSwitch(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + updater, store, engine, gateway := newContainerUpdaterFixture(t, map[string]containerengine.Container{}, time.Hour) + var progress []Progress + gateway.beforeApply = func(snapshot hostnginx.Snapshot) error { + target, found := engine.containers["backend-8081"] + if !found || !target.Running || engine.healthChecks == 0 { + return errors.New("gateway switch occurred before the target container passed health checking") + } + return nil + } + + record, err := updater.UpdateContainerImage(ctx, containerTestImage, func(item Progress) { + progress = append(progress, item) + }) + if err != nil { + t.Fatalf("first install container backend: %v", err) + } + if record.State != transaction.StateCommitted { + t.Fatalf("unexpected committed first install: record=%+v", record) + } + if gateway.snapshot.ActivePort != 8081 || gateway.applyCount != 1 { + t.Fatalf("first install did not switch once to the healthy inactive slot: gateway=%+v applies=%d", gateway.snapshot, gateway.applyCount) + } + target, found := engine.containers["backend-8081"] + if !found || !target.Running || target.Dead { + t.Fatalf("first install target container is not running: %+v", target) + } + if len(engine.stopped) != 0 { + t.Fatalf("first install must not stop a previous container: %+v", engine.stopped) + } + if engine.healthChecks != 1 { + t.Fatalf("unexpected first-install health check count: %d", engine.healthChecks) + } + for _, item := range progress { + if strings.HasPrefix(item.Message, "Draining previous backend container") { + t.Fatalf("first install entered previous-container drain: %+v", progress) + } + } + assertCreatedContainerSpec(t, engine.lastCreateSpec) + assertCommittedContainerDeployment(t, store, record.ID, 8081, "backend-8081", "container-id-backend-8081") +} + +func TestContainerUpdaterRejectsMissingActiveWithPresentInactive(t *testing.T) { + updater, _, _, _ := newContainerUpdaterFixture(t, map[string]containerengine.Container{ + "backend-8081": { + ID: "unexpected-backend-8081", Name: "backend-8081", Running: true, + }, + }, 0) + + _, err := updater.UpdateContainerImage(context.Background(), containerTestImage, nil) + if err == nil || !strings.Contains(err.Error(), "active backend container backend-8080 is missing but inactive container backend-8081 is running") { + t.Fatalf("unexpected missing-active result: %v", err) + } +} + +func TestContainerUpdaterRecoversFailedRollbackThenRetriesSameImage(t *testing.T) { + ctx := context.Background() + updater, _, engine, gateway := newContainerUpdaterFixture(t, map[string]containerengine.Container{}, 0) + failGateway := true + gateway.beforeApply = func(hostnginx.Snapshot) error { + if failGateway { + return errors.New("host Nginx is unavailable") + } + return nil + } + + rollingBack, err := updater.UpdateContainerImage(ctx, containerTestImage, nil) + if err == nil || rollingBack.State != transaction.StateRollingBack { + t.Fatalf("unexpected failed rollback result: record=%+v err=%v", rollingBack, err) + } + target, found := engine.containers["backend-8081"] + if !found || target.Running { + t.Fatalf("failed update target was not stopped: %+v", target) + } + + failGateway = false + rolledBack, err := updater.UpdateContainerImage(ctx, containerTestImage, nil) + if err == nil || rolledBack.State != transaction.StateRolledBack { + t.Fatalf("unexpected resumed rollback result: record=%+v err=%v", rolledBack, err) + } + + committed, err := updater.UpdateContainerImage(ctx, containerTestImage, nil) + if err != nil || committed.State != transaction.StateCommitted { + t.Fatalf("retry same image after rollback: record=%+v err=%v", committed, err) + } + if gateway.snapshot.ActivePort != 8081 { + t.Fatalf("retry did not switch to the healthy container: %+v", gateway.snapshot) + } +} + +func TestContainerUpdaterRejectsMissingCommittedContainer(t *testing.T) { + ctx := context.Background() + updater, store, _, _ := newContainerUpdaterFixture(t, map[string]containerengine.Container{}, 0) + transactionID := "committed-container-transaction" + _, _, err := store.CreateTransaction(ctx, transaction.CreateRequest{ + ID: transactionID, IdempotencyKey: "backend:container:previous", Source: sourceLocalCLI, + Service: serviceBackend, Request: []byte(`{"inputType":"container-image"}`), + }) + if err != nil { + t.Fatalf("create previous 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 previous backend transaction to %s: %v", state, err) + } + } + if _, err := store.CommitBackendContainerDeployment(ctx, transactionID, transaction.BackendContainerDeployment{ + ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: containerTestDigest, ContainerID: "missing-backend-8080", + }, "seed committed backend container deployment"); err != nil { + t.Fatalf("commit previous backend deployment: %v", err) + } + + _, err = updater.UpdateContainerImage(ctx, containerTestImage, nil) + if err == nil || !strings.Contains(err.Error(), "committed active backend container backend-8080 is missing") { + t.Fatalf("unexpected committed-container drift result: %v", err) + } +} + +func newContainerUpdaterFixture( + t *testing.T, + containers map[string]containerengine.Container, + drain time.Duration, +) (*Updater, *transaction.Store, *containerUpdateEngine, *memoryGateway) { + t.Helper() + ctx := context.Background() + root := t.TempDir() + store, err := transaction.OpenStore(ctx, filepath.Join(root, "transactions.db")) + if err != nil { + t.Fatalf("open transaction store: %v", err) + } + t.Cleanup(func() { + if err := store.Close(); err != nil { + t.Errorf("close transaction store: %v", err) + } + }) + coordinator, err := transaction.NewCoordinator(store, slog.New(slog.NewTextHandler(io.Discard, nil))) + if err != nil { + t.Fatalf("create transaction coordinator: %v", err) + } + engine := &containerUpdateEngine{ + image: containerengine.Image{ + ID: "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + RepoDigests: []string{"harbor.ymswell.asia/ymswell/glory-ymswell@" + containerTestDigest}, + Platform: containerengine.Platform{OS: "linux", Architecture: "amd64"}, + }, + containers: containers, + } + configSource := filepath.Join(root, "yms.yaml") + if err := os.WriteFile(configSource, []byte("server: {}\n"), 0o600); err != nil { + t.Fatalf("write backend configuration: %v", err) + } + tmpSource := filepath.Join(root, "tmp") + if err := os.Mkdir(tmpSource, 0o755); err != nil { + t.Fatalf("create backend temporary directory: %v", err) + } + client := &http.Client{Transport: containerUpdateRoundTripFunc(func(request *http.Request) (*http.Response, error) { + engine.healthChecks++ + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(bytes.NewBufferString(`{"status":"UP","components":{"ping":{"status":"UP"}}}`)), + Request: request, + }, nil + })} + executor, err := backendexecutor.New(store, coordinator, engine, client) + if err != nil { + t.Fatalf("create backend container executor: %v", err) + } + gateway := &memoryGateway{snapshot: hostnginx.Snapshot{Content: []byte(serverConfiguration8080), ActivePort: 8080}} + updater := &Updater{ + config: deploymentconfig.Config{ + Daemon: deploymentconfig.Daemon{Environment: deploymentconfig.EnvironmentDev}, + Backend: deploymentconfig.Backend{Type: deploymentconfig.BackendTypeContainer, 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"}, + }}, + }, + workRoot: filepath.Join(root, "work"), store: store, coordinator: coordinator, + gateway: gateway, containerExecutor: executor, engine: engine, drain: drain, + containerConfigSource: configSource, containerConfigTarget: deploymentconfig.ContainerConfigTarget, + containerTmpSource: tmpSource, containerTmpTarget: deploymentconfig.ContainerTmpTarget, + } + return updater, store, engine, gateway +} + +func assertCreatedContainerSpec(t *testing.T, request containerengine.ContainerSpec) { + t.Helper() + if request.Name != "backend-8081" || request.ImageReference != "harbor.ymswell.asia/ymswell/glory-ymswell@"+containerTestDigest { + t.Fatalf("unexpected target container identity: %+v", request) + } + if request.NetworkMode != "host" || request.RestartPolicy.Name != "no" || request.User != "0:0" { + t.Fatalf("unexpected target container runtime contract: %+v", request) + } + if len(request.Environment) != 2 || request.Environment[0] != "SERVER_PORT=8081" || request.Environment[1] != "SPRING_CONFIG_LOCATION=file:/app/config/yms.yaml" { + t.Fatalf("unexpected target container environment: %+v", request.Environment) + } + if len(request.Mounts) != 2 || request.Mounts[0].Target != deploymentconfig.ContainerConfigTarget || !request.Mounts[0].ReadOnly || request.Mounts[1].Target != deploymentconfig.ContainerTmpTarget { + t.Fatalf("unexpected target container mounts: %+v", request.Mounts) + } +} + +func assertCommittedContainerDeployment( + t *testing.T, + store *transaction.Store, + transactionID string, + port int, + containerName string, + containerID string, +) { + t.Helper() + deployment, err := store.BackendContainerDeployment(context.Background()) + if err != nil { + t.Fatalf("read committed backend container deployment: %v", err) + } + if deployment.ActivePort != port || deployment.ContainerName != containerName || deployment.ContainerID != containerID || deployment.ImageDigest != containerTestDigest || deployment.TransactionID != transactionID { + t.Fatalf("unexpected committed backend container deployment: %+v", deployment) + } +} + +const serverConfiguration8080 = `http { + upstream yms-server { + # yms-update managed upstream begin + server 10.11.1.117:8080 max_fails=1 fail_timeout=2s; + # server 10.11.1.117:8081 max_fails=1 fail_timeout=2s; + # yms-update managed upstream end + } +} +` + +type containerUpdateRoundTripFunc func(*http.Request) (*http.Response, error) + +func (function containerUpdateRoundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) { + return function(request) +} + +type containerUpdateEngine struct { + image containerengine.Image + containers map[string]containerengine.Container + stopped []string + lastCreateSpec containerengine.ContainerSpec + healthChecks int +} + +func (e *containerUpdateEngine) Ping(context.Context) error { return nil } +func (e *containerUpdateEngine) PullImage(context.Context, string) error { return nil } +func (e *containerUpdateEngine) LoadImage(context.Context, io.Reader) error { return nil } +func (e *containerUpdateEngine) InspectImage(context.Context, string) (containerengine.Image, error) { + return e.image, nil +} +func (e *containerUpdateEngine) CreateContainer(_ context.Context, spec containerengine.ContainerSpec) (containerengine.Container, error) { + e.lastCreateSpec = spec + record := containerengine.Container{ + ID: "container-id-" + spec.Name, + Name: spec.Name, + ImageID: e.image.ID, + ImageReference: spec.ImageReference, + Platform: spec.Platform.OS + "/" + spec.Platform.Architecture, + Status: "created", + Environment: append([]string(nil), spec.Environment...), + Labels: spec.Labels, + NetworkMode: spec.NetworkMode, + RestartPolicy: spec.RestartPolicy, + Mounts: append([]containerengine.Mount(nil), spec.Mounts...), + User: spec.User, + StopTimeoutSeconds: spec.StopTimeoutSeconds, + } + e.containers[spec.Name] = record + return record, nil +} +func (e *containerUpdateEngine) StartContainer(_ context.Context, name string) error { + record, found := e.containers[name] + if !found { + return containerengine.ErrNotFound + } + record.Running = true + record.Status = "running" + e.containers[name] = record + return nil +} +func (e *containerUpdateEngine) StopContainer(_ context.Context, name string) error { + e.stopped = append(e.stopped, name) + record, found := e.containers[name] + if !found { + return containerengine.ErrNotFound + } + record.Running = false + record.Status = "exited" + e.containers[name] = record + return nil +} +func (e *containerUpdateEngine) 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 *containerUpdateEngine) RemoveContainer(_ context.Context, name string, _ bool) error { + if _, found := e.containers[name]; !found { + return containerengine.ErrNotFound + } + delete(e.containers, name) + return nil +} +func (e *containerUpdateEngine) Close() error { return nil } + +var _ containerengine.Engine = (*containerUpdateEngine)(nil) diff --git a/internal/backendupdate/restart.go b/internal/backendupdate/restart.go index 07a8a54..7f98633 100644 --- a/internal/backendupdate/restart.go +++ b/internal/backendupdate/restart.go @@ -8,6 +8,7 @@ import ( "os" "path/filepath" + "yms-daemon/internal/deploymentconfig" "yms-daemon/internal/transaction" "yms-daemon/internal/updatepackage" ) @@ -15,6 +16,9 @@ import ( // Restart performs a zero-downtime rotation with the exact release currently // exposed by the compatibility JAR path. 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") + } reportProgress(report, Progress{Message: "Resolving the current native backend release"}) active, err := u.store.ActiveTransaction(ctx) if err == nil { diff --git a/internal/backendupdate/updater.go b/internal/backendupdate/updater.go index 8516ce4..4c40eec 100644 --- a/internal/backendupdate/updater.go +++ b/internal/backendupdate/updater.go @@ -14,6 +14,8 @@ import ( "path/filepath" "time" + "yms-daemon/internal/backendexecutor" + "yms-daemon/internal/containerengine" "yms-daemon/internal/daemonapi" "yms-daemon/internal/deploymentconfig" "yms-daemon/internal/filestore" @@ -36,16 +38,22 @@ const ( // Updater executes the current native backend contract on one server. type Updater struct { - config deploymentconfig.Config - workRoot string - store *transaction.Store - coordinator *transaction.Coordinator - releaseStore *filestore.Store - units systemd.Manager - gateway gatewayController - executor nativeExecutor - logger *slog.Logger - drain time.Duration + config deploymentconfig.Config + workRoot string + store *transaction.Store + coordinator *transaction.Coordinator + releaseStore *filestore.Store + units systemd.Manager + gateway gatewayController + executor nativeExecutor + containerExecutor containerExecutor + engine containerengine.Engine + containerConfigSource string + containerConfigTarget string + containerTmpSource string + containerTmpTarget string + logger *slog.Logger + drain time.Duration } type gatewayController interface { @@ -57,6 +65,10 @@ type nativeExecutor interface { Run(context.Context, string, nativebackendexecutor.Request) error } +type containerExecutor interface { + Run(context.Context, string, backendexecutor.Request) error +} + // New creates the complete native backend update orchestrator. func New( config deploymentconfig.Config, @@ -72,6 +84,9 @@ func New( if err := config.Validate(); err != nil { return nil, err } + if config.Backend.Type != deploymentconfig.BackendTypeNative { + return nil, errors.New("backend.type must be native") + } if !filepath.IsAbs(workRoot) || filepath.Clean(workRoot) != workRoot { return nil, errors.New("backend update work root must be a clean absolute path") } @@ -99,8 +114,54 @@ func New( }, nil } +// NewContainer creates the Docker standalone backend update orchestrator. +func NewContainer( + config deploymentconfig.Config, + workRoot string, + store *transaction.Store, + coordinator *transaction.Coordinator, + engine containerengine.Engine, + gateway gatewayController, + httpClient *http.Client, + logger *slog.Logger, +) (*Updater, error) { + if err := config.Validate(); err != nil { + return nil, err + } + if config.Backend.Type != deploymentconfig.BackendTypeContainer { + return nil, errors.New("backend.type must be container") + } + if config.Daemon.Environment != deploymentconfig.EnvironmentDev { + return nil, errors.New("--container-image requires daemon.environment = dev") + } + if !filepath.IsAbs(workRoot) || filepath.Clean(workRoot) != workRoot { + return nil, errors.New("container backend update work root must be a clean absolute path") + } + if store == nil || coordinator == nil || engine == nil || gateway == nil { + return nil, errors.New("container backend update dependencies are required") + } + if logger == nil { + logger = slog.Default() + } + executor, err := backendexecutor.New(store, coordinator, engine, httpClient) + if err != nil { + return nil, err + } + 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, + }, nil +} + // UpdateRepack applies one repack ZIP selected by an absolute local path. func (u *Updater) UpdateRepack(ctx context.Context, packagePath string, report ProgressReporter) (transaction.Transaction, error) { + if u.config.Backend.Type == deploymentconfig.BackendTypeContainer { + return transaction.Transaction{}, errors.New("repack ZIP update is not implemented for container backend") + } reportProgress(report, Progress{Message: "Validating repack ZIP"}) updatePackage, err := updatepackage.OpenBackendNative(packagePath) if err != nil { @@ -125,6 +186,9 @@ func (u *Updater) UpdateRepack(ctx context.Context, packagePath string, report P // UpdateNativeJAR applies one JAR copied directly to the server. func (u *Updater) UpdateNativeJAR(ctx context.Context, jarPath string, report ProgressReporter) (transaction.Transaction, error) { + if u.config.Backend.Type == deploymentconfig.BackendTypeContainer { + return transaction.Transaction{}, errors.New("--native-jar requires backend.type = native") + } reportProgress(report, Progress{Message: "Validating direct native backend JAR and computing SHA-256"}) jar, err := updatepackage.OpenDirectNativeJAR(jarPath) if err != nil { diff --git a/internal/backendupdate/updater_test.go b/internal/backendupdate/updater_test.go index 4fd54e8..f7cd97e 100644 --- a/internal/backendupdate/updater_test.go +++ b/internal/backendupdate/updater_test.go @@ -175,7 +175,9 @@ const serverConfiguration8081 = `http { ` type memoryGateway struct { - snapshot hostnginx.Snapshot + snapshot hostnginx.Snapshot + applyCount int + beforeApply func(hostnginx.Snapshot) error } func (g *memoryGateway) Read() (hostnginx.Snapshot, error) { @@ -183,7 +185,13 @@ func (g *memoryGateway) Read() (hostnginx.Snapshot, error) { } func (g *memoryGateway) Apply(_ context.Context, snapshot hostnginx.Snapshot) error { + if g.beforeApply != nil { + if err := g.beforeApply(snapshot); err != nil { + return err + } + } g.snapshot = hostnginx.Snapshot{Content: append([]byte(nil), snapshot.Content...), ActivePort: snapshot.ActivePort} + g.applyCount++ return nil } diff --git a/internal/containerengine/engine.go b/internal/containerengine/engine.go index 0be70bf..3cb720d 100644 --- a/internal/containerengine/engine.go +++ b/internal/containerengine/engine.go @@ -40,40 +40,46 @@ type Mount struct { // ContainerSpec contains every property controlled by the backend executor. type ContainerSpec struct { - Name string - ImageReference string - Platform Platform - Environment []string - Labels map[string]string - NetworkMode string - RestartPolicy RestartPolicy - Mounts []Mount + Name string + ImageReference string + Platform Platform + Environment []string + Labels map[string]string + NetworkMode string + RestartPolicy RestartPolicy + Mounts []Mount + User string + StopTimeoutSeconds int } // Container is the runtime state required for idempotent inspection. type Container struct { - ID string - Name string - ImageID string - ImageReference string - Platform string - Running bool - Dead bool - Status string - Environment []string - Labels map[string]string - NetworkMode string - RestartPolicy RestartPolicy - Mounts []Mount + ID string + Name string + ImageID string + ImageReference string + Platform string + Running bool + Dead bool + Status string + Environment []string + Labels map[string]string + NetworkMode string + RestartPolicy RestartPolicy + Mounts []Mount + User string + StopTimeoutSeconds int } // Engine is the smallest container runtime API required by an update executor. type Engine interface { Ping(context.Context) error + PullImage(context.Context, string) error LoadImage(context.Context, io.Reader) error InspectImage(context.Context, string) (Image, error) CreateContainer(context.Context, ContainerSpec) (Container, error) StartContainer(context.Context, string) error + StopContainer(context.Context, string) error InspectContainer(context.Context, string) (Container, error) RemoveContainer(context.Context, string, bool) error Close() error diff --git a/internal/containerengine/moby.go b/internal/containerengine/moby.go index 4f562ef..dd2866f 100644 --- a/internal/containerengine/moby.go +++ b/internal/containerengine/moby.go @@ -37,6 +37,18 @@ func (e *MobyEngine) Ping(ctx context.Context) error { return nil } +func (e *MobyEngine) PullImage(ctx context.Context, imageReference string) error { + response, err := e.client.ImagePull(ctx, imageReference, client.ImagePullOptions{}) + if err != nil { + return fmt.Errorf("pull image %s: %w", imageReference, err) + } + defer response.Close() + if err := decodeImageLoadResponse(response); err != nil { + return fmt.Errorf("pull image %s response: %w", imageReference, err) + } + return nil +} + func (e *MobyEngine) LoadImage(ctx context.Context, input io.Reader) error { if input == nil { return errors.New("image archive reader is required") @@ -99,10 +111,13 @@ func (e *MobyEngine) CreateContainer(ctx context.Context, spec ContainerSpec) (C ReadOnly: item.ReadOnly, }) } + stopTimeout := spec.StopTimeoutSeconds result, err := e.client.ContainerCreate(ctx, client.ContainerCreateOptions{ Config: &container.Config{ - Env: append([]string(nil), spec.Environment...), - Labels: cloneMap(spec.Labels), + Env: append([]string(nil), spec.Environment...), + Labels: cloneMap(spec.Labels), + User: spec.User, + StopTimeout: &stopTimeout, }, HostConfig: &container.HostConfig{ NetworkMode: container.NetworkMode(spec.NetworkMode), @@ -133,6 +148,13 @@ func (e *MobyEngine) StartContainer(ctx context.Context, idOrName string) error return nil } +func (e *MobyEngine) StopContainer(ctx context.Context, idOrName string) error { + if _, err := e.client.ContainerStop(ctx, idOrName, client.ContainerStopOptions{}); err != nil { + return engineError("stop container", err) + } + return nil +} + func (e *MobyEngine) InspectContainer(ctx context.Context, idOrName string) (Container, error) { result, err := e.client.ContainerInspect(ctx, idOrName, client.ContainerInspectOptions{}) if err != nil { @@ -154,6 +176,10 @@ func (e *MobyEngine) InspectContainer(ctx context.Context, idOrName string) (Con record.ImageReference = response.Config.Image record.Environment = append([]string(nil), response.Config.Env...) record.Labels = cloneMap(response.Config.Labels) + record.User = response.Config.User + if response.Config.StopTimeout != nil { + record.StopTimeoutSeconds = *response.Config.StopTimeout + } } if response.HostConfig != nil { record.NetworkMode = string(response.HostConfig.NetworkMode) diff --git a/internal/daemonapi/protocol.go b/internal/daemonapi/protocol.go index f424560..fc9d207 100644 --- a/internal/daemonapi/protocol.go +++ b/internal/daemonapi/protocol.go @@ -2,19 +2,21 @@ package daemonapi const ( - OperationUpdate = "update" - OperationRestart = "restart" - InputTypeRepackZIP = "repack-zip" - InputTypeNativeJAR = "native-jar" - ResponseProgress = "progress" - ResponseResult = "result" + OperationUpdate = "update" + OperationRestart = "restart" + InputTypeRepackZIP = "repack-zip" + InputTypeNativeJAR = "native-jar" + InputTypeContainerImage = "container-image" + ResponseProgress = "progress" + ResponseResult = "result" ) type Request struct { - Operation string `json:"operation"` - Service string `json:"service"` - InputType string `json:"inputType"` - File string `json:"file"` + Operation string `json:"operation"` + Service string `json:"service"` + InputType string `json:"inputType"` + File string `json:"file"` + ImageReference string `json:"imageReference"` } type Response struct { diff --git a/internal/daemonclient/client.go b/internal/daemonclient/client.go index c221d66..22fd2dd 100644 --- a/internal/daemonclient/client.go +++ b/internal/daemonclient/client.go @@ -20,6 +20,17 @@ func Update(ctx context.Context, socketPath string, service string, inputType st return submit(ctx, socketPath, request, progress) } +func UpdateContainerImage(ctx context.Context, socketPath string, service string, imageReference string, progress func(daemonapi.Response)) (daemonapi.Response, error) { + if !filepath.IsAbs(socketPath) { + return daemonapi.Response{}, errors.New("daemon socket path must be absolute") + } + if imageReference == "" { + return daemonapi.Response{}, errors.New("container image reference is required") + } + request := daemonapi.Request{Operation: daemonapi.OperationUpdate, Service: service, InputType: daemonapi.InputTypeContainerImage, ImageReference: imageReference} + return submit(ctx, socketPath, request, progress) +} + func Restart(ctx context.Context, socketPath string, service string, progress func(daemonapi.Response)) (daemonapi.Response, error) { if !filepath.IsAbs(socketPath) { return daemonapi.Response{}, errors.New("daemon socket path must be absolute") diff --git a/internal/daemonserver/server.go b/internal/daemonserver/server.go index 0e450d4..1b1c8e6 100644 --- a/internal/daemonserver/server.go +++ b/internal/daemonserver/server.go @@ -25,6 +25,7 @@ const maximumRequestBytes = 1 << 20 type backendUpdater interface { UpdateRepack(context.Context, string, backendupdate.ProgressReporter) (transaction.Transaction, error) UpdateNativeJAR(context.Context, string, backendupdate.ProgressReporter) (transaction.Transaction, error) + UpdateContainerImage(context.Context, string, backendupdate.ProgressReporter) (transaction.Transaction, error) Restart(context.Context, backendupdate.ProgressReporter) (transaction.Transaction, error) } @@ -124,15 +125,29 @@ func (s *Server) handle(ctx context.Context, connection net.Conn) { case daemonapi.OperationUpdate: switch request.InputType { case daemonapi.InputTypeRepackZIP: + if request.File == "" || request.ImageReference != "" { + _ = s.writeResponse(connection, daemonapi.Response{Kind: daemonapi.ResponseResult, Error: "repack-zip requires file and does not accept imageReference"}) + return + } record, updateErr = s.updater.UpdateRepack(ctx, request.File, report) case daemonapi.InputTypeNativeJAR: + if request.File == "" || request.ImageReference != "" { + _ = s.writeResponse(connection, daemonapi.Response{Kind: daemonapi.ResponseResult, Error: "native-jar requires file and does not accept imageReference"}) + return + } record, updateErr = s.updater.UpdateNativeJAR(ctx, request.File, report) + case daemonapi.InputTypeContainerImage: + if request.File != "" || request.ImageReference == "" { + _ = s.writeResponse(connection, daemonapi.Response{Kind: daemonapi.ResponseResult, Error: "container-image requires imageReference and does not accept file"}) + return + } + record, updateErr = s.updater.UpdateContainerImage(ctx, request.ImageReference, report) default: - _ = s.writeResponse(connection, daemonapi.Response{Kind: daemonapi.ResponseResult, Error: "inputType must be repack-zip or native-jar"}) + _ = s.writeResponse(connection, daemonapi.Response{Kind: daemonapi.ResponseResult, Error: "inputType must be repack-zip, native-jar, or container-image"}) return } case daemonapi.OperationRestart: - if request.InputType != "" || request.File != "" { + if request.InputType != "" || request.File != "" || request.ImageReference != "" { _ = s.writeResponse(connection, daemonapi.Response{Kind: daemonapi.ResponseResult, Error: "restart does not accept inputType or file"}) return } @@ -172,7 +187,7 @@ func decodeRequest(reader io.Reader) (daemonapi.Request, error) { } return daemonapi.Request{}, fmt.Errorf("decode daemon request trailing content: %w", err) } - if strings.TrimSpace(request.Operation) != request.Operation || strings.TrimSpace(request.Service) != request.Service || strings.TrimSpace(request.InputType) != request.InputType || strings.TrimSpace(request.File) != request.File { + if strings.TrimSpace(request.Operation) != request.Operation || strings.TrimSpace(request.Service) != request.Service || strings.TrimSpace(request.InputType) != request.InputType || strings.TrimSpace(request.File) != request.File || strings.TrimSpace(request.ImageReference) != request.ImageReference { return daemonapi.Request{}, errors.New("daemon request fields must not contain surrounding whitespace") } return request, nil diff --git a/internal/daemonserver/server_test.go b/internal/daemonserver/server_test.go index 82b3fd9..b3ce3da 100644 --- a/internal/daemonserver/server_test.go +++ b/internal/daemonserver/server_test.go @@ -84,6 +84,29 @@ 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))) + 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) + + imageReference := "harbor.ymswell.asia/ymswell/glory-ymswell:20260813-184902-a37bf50d-v1.1.8.1" + response, err := daemonclient.UpdateContainerImage(context.Background(), socketPath, "backend", imageReference, nil) + if err != nil { + t.Fatalf("submit container backend image: %v", err) + } + if response.TransactionID != updater.record.ID || updater.file != imageReference || updater.inputType != daemonapi.InputTypeContainerImage { + t.Fatalf("unexpected container update dispatch: response=%+v image=%s inputType=%s", response, updater.file, updater.inputType) + } +} + func TestServerReturnsTransactionFailure(t *testing.T) { socketPath := shortSocketPath(t) updater := &fakeUpdater{ @@ -194,6 +217,16 @@ func (u *fakeUpdater) UpdateNativeJAR(_ context.Context, file string, report bac return u.record, u.err } +func (u *fakeUpdater) UpdateContainerImage(_ context.Context, imageReference string, report backendupdate.ProgressReporter) (transaction.Transaction, error) { + u.file = imageReference + u.inputType = daemonapi.InputTypeContainerImage + u.operation = daemonapi.OperationUpdate + if report != nil { + report(backendupdate.Progress{TransactionID: u.record.ID, State: transaction.StateStarting, Message: "test update progress"}) + } + return u.record, u.err +} + func (u *fakeUpdater) Restart(_ context.Context, report backendupdate.ProgressReporter) (transaction.Transaction, error) { u.operation = daemonapi.OperationRestart if report != nil { diff --git a/internal/deploymentconfig/config.go b/internal/deploymentconfig/config.go index df8cd38..8475c5b 100644 --- a/internal/deploymentconfig/config.go +++ b/internal/deploymentconfig/config.go @@ -19,7 +19,11 @@ const ( // DefaultPath is the only default location used by the daemon entrypoint. DefaultPath = "/etc/yms-daemon/yms-daemon.toml" - BackendTypeNative = "native" + EnvironmentDev = "dev" + EnvironmentProd = "prod" + + BackendTypeNative = "native" + BackendTypeContainer = "container" BackendPort8080 = 8080 BackendPort8081 = 8081 @@ -34,14 +38,32 @@ const ( nativeSlotJAR8081 = "/home/yms/lib/glory-soft-yms-8081.jar" nativeHealthURL8080 = "http://127.0.0.1:8080/yms/actuator/health" nativeHealthURL8081 = "http://127.0.0.1:8081/yms/actuator/health" + containerName8080 = "backend-8080" + containerName8081 = "backend-8081" +) + +const ( + ContainerConfigSource = "/home/yms/conf/yms.yaml" + ContainerConfigTarget = "/app/config/yms.yaml" + ContainerTmpSource = "/home/yms/tmp" + ContainerTmpTarget = "/home/yms/tmp" + ContainerPortEnvironment = "SERVER_PORT" + ContainerConfigEnvironment = "SPRING_CONFIG_LOCATION" + ContainerConfigLocation = "file:/app/config/yms.yaml" ) // Config is the complete local deployment configuration currently understood by the daemon. type Config struct { + Daemon Daemon `toml:"daemon"` Backend Backend `toml:"backend"` } -// Backend describes the explicitly selected backend runtime and its native blue/green slots. +// Daemon contains machine-wide behavior that is independent of a component runtime. +type Daemon struct { + Environment string `toml:"environment"` +} + +// Backend describes the explicitly selected backend runtime and its blue/green slots. type Backend struct { Type string `toml:"type"` ReleaseDir string `toml:"release_dir"` @@ -50,20 +72,21 @@ type Backend struct { Slot BackendSlots `toml:"slot"` } -// BackendSlots lists the only native backend ports supported by the current deployment contract. +// BackendSlots lists the only backend ports supported by the current deployment contract. type BackendSlots struct { Port8080 BackendSlot `toml:"8080"` Port8081 BackendSlot `toml:"8081"` } -// BackendSlot contains values that are passed to the native backend executor without derivation. +// BackendSlot contains exact values for one native or container backend slot. type BackendSlot struct { Unit string `toml:"unit"` JAR string `toml:"jar"` + ContainerName string `toml:"container_name"` HealthEndpoint string `toml:"health_endpoint"` } -// Load opens path, performs strict TOML decoding, and validates the native backend contract. +// Load opens path, performs strict TOML decoding, and validates the selected backend contract. func Load(path string) (Config, error) { if err := validateAbsolutePath("deployment configuration", path); err != nil { return Config{}, err @@ -106,7 +129,14 @@ func validateExactDocumentKeys(document []byte) error { if err := toml.Unmarshal(document, &root); err != nil { return err } - if err := rejectUnknownKeys(root, "", "backend"); err != nil { + if err := rejectUnknownKeys(root, "", "daemon", "backend"); err != nil { + return err + } + daemon, err := exactTable(root, "", "daemon") + if err != nil { + return err + } + if err := rejectUnknownKeys(daemon, "daemon", "environment"); err != nil { return err } backend, err := exactTable(root, "", "backend") @@ -128,7 +158,7 @@ func validateExactDocumentKeys(document []byte) error { if err != nil { return err } - if err := rejectUnknownKeys(slot, "backend.slot."+port, "unit", "jar", "health_endpoint"); err != nil { + if err := rejectUnknownKeys(slot, "backend.slot."+port, "unit", "jar", "container_name", "health_endpoint"); err != nil { return err } } @@ -176,23 +206,46 @@ func rejectUnknownKeys(table map[string]any, parent string, allowed ...string) e // Validate rejects incomplete or altered local deployment identifiers. func (c Config) Validate() error { - if c.Backend.Type != BackendTypeNative { - return fmt.Errorf("backend.type must be %q", BackendTypeNative) - } - if c.Backend.ReleaseDir != nativeReleaseDir { - return fmt.Errorf("backend.release_dir must be %q", nativeReleaseDir) - } - if c.Backend.ActiveJAR != nativeActiveJAR { - return fmt.Errorf("backend.active_jar must be %q", nativeActiveJAR) + switch c.Daemon.Environment { + case EnvironmentDev, EnvironmentProd: + default: + return fmt.Errorf("daemon.environment must be %q or %q", EnvironmentDev, EnvironmentProd) } if err := validateAbsolutePath("backend.systemctl_path", c.Backend.SystemctlPath); err != nil { return err } - if err := validateSlot("backend.slot.8080", c.Backend.Slot.Port8080, nativeUnit8080, nativeSlotJAR8080, nativeHealthURL8080); err != nil { - return err - } - if err := validateSlot("backend.slot.8081", c.Backend.Slot.Port8081, nativeUnit8081, nativeSlotJAR8081, nativeHealthURL8081); err != nil { - return err + switch c.Backend.Type { + case BackendTypeNative: + if c.Backend.Slot.Port8080.ContainerName != "" || c.Backend.Slot.Port8081.ContainerName != "" { + return errors.New("native backend slots do not accept container_name") + } + if c.Backend.ReleaseDir != nativeReleaseDir { + return fmt.Errorf("backend.release_dir must be %q", nativeReleaseDir) + } + if c.Backend.ActiveJAR != nativeActiveJAR { + return fmt.Errorf("backend.active_jar must be %q", nativeActiveJAR) + } + if err := validateNativeSlot("backend.slot.8080", c.Backend.Slot.Port8080, nativeUnit8080, nativeSlotJAR8080, nativeHealthURL8080); err != nil { + return err + } + if err := validateNativeSlot("backend.slot.8081", c.Backend.Slot.Port8081, nativeUnit8081, nativeSlotJAR8081, nativeHealthURL8081); err != nil { + return err + } + case BackendTypeContainer: + if c.Backend.ReleaseDir != "" || c.Backend.ActiveJAR != "" { + return errors.New("container backend does not accept release_dir or active_jar") + } + if c.Backend.Slot.Port8080.Unit != "" || c.Backend.Slot.Port8080.JAR != "" || c.Backend.Slot.Port8081.Unit != "" || c.Backend.Slot.Port8081.JAR != "" { + return errors.New("container backend slots do not accept unit or jar") + } + if err := validateContainerSlot("backend.slot.8080", c.Backend.Slot.Port8080, containerName8080, nativeHealthURL8080); err != nil { + return err + } + if err := validateContainerSlot("backend.slot.8081", c.Backend.Slot.Port8081, containerName8081, nativeHealthURL8081); err != nil { + return err + } + default: + return fmt.Errorf("backend.type must be %q or %q", BackendTypeNative, BackendTypeContainer) } return nil } @@ -209,7 +262,7 @@ func (b Backend) SlotForPort(port int) (BackendSlot, error) { } } -func validateSlot(field string, slot BackendSlot, unit string, jar string, endpoint string) error { +func validateNativeSlot(field string, slot BackendSlot, unit string, jar string, endpoint string) error { if slot.Unit != unit { return fmt.Errorf("%s.unit must be %q", field, unit) } @@ -222,6 +275,16 @@ func validateSlot(field string, slot BackendSlot, unit string, jar string, endpo return nil } +func validateContainerSlot(field string, slot BackendSlot, containerName string, endpoint string) error { + if slot.ContainerName != containerName { + return fmt.Errorf("%s.container_name must be %q", field, containerName) + } + if slot.HealthEndpoint != endpoint { + return fmt.Errorf("%s.health_endpoint must be %q", field, endpoint) + } + return nil +} + func validateAbsolutePath(field string, value string) error { if value == "" { return fmt.Errorf("%s is required", field) diff --git a/internal/deploymentconfig/config_test.go b/internal/deploymentconfig/config_test.go index bf02564..dea0925 100644 --- a/internal/deploymentconfig/config_test.go +++ b/internal/deploymentconfig/config_test.go @@ -7,7 +7,10 @@ import ( "testing" ) -const validNativeConfig = `[backend] +const validNativeConfig = `[daemon] +environment = "dev" + +[backend] type = "native" release_dir = "/home/yms/lib/releases" active_jar = "/home/yms/lib/glory-soft-yms.jar" @@ -24,6 +27,22 @@ jar = "/home/yms/lib/glory-soft-yms-8081.jar" health_endpoint = "http://127.0.0.1:8081/yms/actuator/health" ` +const validContainerConfig = `[daemon] +environment = "dev" + +[backend] +type = "container" +systemctl_path = "/bin/systemctl" + +[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" +` + func TestLoadValidNativeConfiguration(t *testing.T) { path := writeConfig(t, validNativeConfig) @@ -31,6 +50,9 @@ func TestLoadValidNativeConfiguration(t *testing.T) { if err != nil { t.Fatalf("load native deployment configuration: %v", err) } + if config.Daemon.Environment != EnvironmentDev { + t.Fatalf("unexpected daemon configuration: %+v", config.Daemon) + } if config.Backend.Type != BackendTypeNative || config.Backend.SystemctlPath != "/bin/systemctl" { t.Fatalf("unexpected backend configuration: %+v", config.Backend) } @@ -50,6 +72,27 @@ func TestLoadValidNativeConfiguration(t *testing.T) { } } +func TestLoadValidContainerConfiguration(t *testing.T) { + config, err := Load(writeConfig(t, validContainerConfig)) + if err != nil { + t.Fatalf("load container deployment configuration: %v", err) + } + if config.Backend.Type != BackendTypeContainer { + t.Fatalf("unexpected backend type: %q", config.Backend.Type) + } + if config.Backend.Slot.Port8080.ContainerName != "backend-8080" || config.Backend.Slot.Port8081.ContainerName != "backend-8081" { + t.Fatalf("unexpected container slots: %+v", config.Backend.Slot) + } +} + +func TestLoadRejectsChangedContainerName(t *testing.T) { + content := strings.Replace(validContainerConfig, `container_name = "backend-8081"`, `container_name = "backend"`, 1) + _, err := Load(writeConfig(t, content)) + if err == nil || !strings.Contains(err.Error(), "backend.slot.8081.container_name") { + t.Fatalf("expected exact container name rejection, got %v", err) + } +} + func TestPackagedNativeConfigurationMatchesContract(t *testing.T) { path, err := filepath.Abs(filepath.Join("..", "..", "packaging", "etc", "yms-daemon", "yms-daemon.toml")) if err != nil { @@ -68,6 +111,41 @@ func TestLoadRejectsUnknownField(t *testing.T) { } } +func TestLoadAcceptsProductionEnvironment(t *testing.T) { + content := strings.Replace(validNativeConfig, `environment = "dev"`, `environment = "prod"`, 1) + config, err := Load(writeConfig(t, content)) + if err != nil { + t.Fatalf("load production environment: %v", err) + } + if config.Daemon.Environment != EnvironmentProd { + t.Fatalf("unexpected daemon environment: %q", config.Daemon.Environment) + } +} + +func TestLoadRejectsMissingDaemonTable(t *testing.T) { + content := strings.Replace(validNativeConfig, "[daemon]\nenvironment = \"dev\"\n\n", "", 1) + _, err := Load(writeConfig(t, content)) + if err == nil || !strings.Contains(err.Error(), "daemon table is required") { + t.Fatalf("expected missing daemon table rejection, got %v", err) + } +} + +func TestLoadRejectsChangedEnvironment(t *testing.T) { + content := strings.Replace(validNativeConfig, `environment = "dev"`, `environment = "development"`, 1) + _, err := Load(writeConfig(t, content)) + if err == nil || !strings.Contains(err.Error(), "daemon.environment") { + t.Fatalf("expected exact daemon environment rejection, got %v", err) + } +} + +func TestLoadRejectsUnknownDaemonField(t *testing.T) { + content := strings.Replace(validNativeConfig, `environment = "dev"`, "environment = \"dev\"\nEnvironment = \"dev\"", 1) + _, err := Load(writeConfig(t, content)) + if err == nil || !strings.Contains(err.Error(), "daemon.Environment") { + t.Fatalf("expected exact daemon field rejection, got %v", err) + } +} + func TestLoadRejectsWrongTableCase(t *testing.T) { content := strings.Replace(validNativeConfig, "[backend]", "[Backend]", 1) _, err := Load(writeConfig(t, content)) diff --git a/internal/transaction/backend_container_deployment.go b/internal/transaction/backend_container_deployment.go new file mode 100644 index 0000000..8a4580a --- /dev/null +++ b/internal/transaction/backend_container_deployment.go @@ -0,0 +1,185 @@ +package transaction + +import ( + "context" + "database/sql" + "errors" + "fmt" + "strings" + "time" +) + +const backendService = "backend" + +// BackendContainerDeployment is the last container slot committed for backend. +// Runtime inspection remains authoritative for the current external state; this +// record distinguishes a fresh installation from loss or drift after deployment. +type BackendContainerDeployment struct { + ActivePort int + ContainerName string + ImageDigest string + ContainerID string + TransactionID string + UpdatedAt time.Time +} + +// BackendContainerDeployment returns the single committed backend container state. +func (s *Store) BackendContainerDeployment(ctx context.Context) (BackendContainerDeployment, error) { + return scanBackendContainerDeployment(s.db.QueryRowContext(ctx, ` + SELECT active_port, container_name, image_digest, container_id, + transaction_id, updated_at + FROM backend_container_deployment + WHERE singleton_id = 1`)) +} + +// HasCommittedBackendContainerTransactionHistory reports whether this store +// contains a committed container backend transaction. This is the legacy +// deployment evidence used when a database predates the deployment table. +// Failed and rolled-back first-install attempts do not mark a machine deployed. +func (s *Store) HasCommittedBackendContainerTransactionHistory(ctx context.Context) (bool, error) { + var found int + if err := s.db.QueryRowContext(ctx, ` + SELECT EXISTS ( + SELECT 1 + FROM transactions + WHERE service = ? + AND idempotency_key GLOB 'backend:container:*' + AND state = ? + )`, backendService, StateCommitted).Scan(&found); err != nil { + return false, fmt.Errorf("query committed backend container transaction history: %w", err) + } + return found == 1, nil +} + +// CommitBackendContainerDeployment atomically records the active container and +// commits its transaction. A crash cannot leave COMMITTED without the matching +// deployment row, or publish a deployment row for an unfinished transaction. +func (s *Store) CommitBackendContainerDeployment( + ctx context.Context, + transactionID string, + deployment BackendContainerDeployment, + message string, +) (Transaction, error) { + if transactionID == "" || strings.TrimSpace(transactionID) != transactionID { + return Transaction{}, errors.New("exact transaction ID is required") + } + if err := validateBackendContainerDeployment(deployment); err != nil { + return Transaction{}, err + } + + tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable}) + if err != nil { + return Transaction{}, fmt.Errorf("begin backend container deployment commit: %w", err) + } + defer tx.Rollback() + + record, err := getTransactionByID(ctx, tx, transactionID) + if err != nil { + return Transaction{}, err + } + if record.Service != backendService { + return Transaction{}, fmt.Errorf("transaction %s service must be %s", transactionID, backendService) + } + if record.State != StateDraining { + return Transaction{}, &TransitionError{From: record.State, To: StateCommitted} + } + + now := s.now().UTC() + _, err = tx.ExecContext(ctx, ` + INSERT INTO backend_container_deployment ( + singleton_id, active_port, container_name, image_digest, + container_id, transaction_id, updated_at + ) VALUES (1, ?, ?, ?, ?, ?, ?) + ON CONFLICT(singleton_id) DO UPDATE SET + active_port = excluded.active_port, + container_name = excluded.container_name, + image_digest = excluded.image_digest, + container_id = excluded.container_id, + transaction_id = excluded.transaction_id, + updated_at = excluded.updated_at`, + deployment.ActivePort, + deployment.ContainerName, + deployment.ImageDigest, + deployment.ContainerID, + transactionID, + formatTime(now), + ) + if err != nil { + return Transaction{}, fmt.Errorf("write backend container deployment: %w", err) + } + + result, err := tx.ExecContext(ctx, ` + UPDATE transactions + SET state = ?, version = version + 1, updated_at = ? + WHERE id = ? AND version = ? AND state = ?`, + StateCommitted, + formatTime(now), + transactionID, + record.Version, + StateDraining, + ) + if err != nil { + return Transaction{}, fmt.Errorf("commit backend container transaction state: %w", err) + } + rows, err := result.RowsAffected() + if err != nil { + return Transaction{}, fmt.Errorf("read committed backend container transaction rows: %w", err) + } + if rows != 1 { + return Transaction{}, errors.New("backend container transaction changed concurrently") + } + if err := insertEvent(ctx, tx, transactionID, "", "TRANSACTION_STATE_CHANGED", record.State, StateCommitted, message, now); err != nil { + return Transaction{}, err + } + if err := tx.Commit(); err != nil { + return Transaction{}, fmt.Errorf("commit backend container deployment: %w", err) + } + + record.State = StateCommitted + record.Version++ + record.UpdatedAt = now + return record, nil +} + +func validateBackendContainerDeployment(deployment BackendContainerDeployment) error { + if deployment.ActivePort != 8080 && deployment.ActivePort != 8081 { + return fmt.Errorf("backend container deployment port must be 8080 or 8081: %d", deployment.ActivePort) + } + for _, field := range []struct { + name string + value string + }{ + {"container name", deployment.ContainerName}, + {"image digest", deployment.ImageDigest}, + {"container ID", deployment.ContainerID}, + } { + if field.value == "" || strings.TrimSpace(field.value) != field.value { + return fmt.Errorf("exact backend container %s is required", field.name) + } + } + return nil +} + +func scanBackendContainerDeployment(row rowScanner) (BackendContainerDeployment, error) { + var deployment BackendContainerDeployment + var updatedAt string + if err := row.Scan( + &deployment.ActivePort, + &deployment.ContainerName, + &deployment.ImageDigest, + &deployment.ContainerID, + &deployment.TransactionID, + &updatedAt, + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return BackendContainerDeployment{}, ErrNotFound + } + return BackendContainerDeployment{}, fmt.Errorf("scan backend container deployment: %w", err) + } + parsed, err := parseTime(updatedAt) + if err != nil { + return BackendContainerDeployment{}, fmt.Errorf("parse backend container deployment update time: %w", err) + } + deployment.UpdatedAt = parsed + return deployment, nil +} diff --git a/internal/transaction/coordinator.go b/internal/transaction/coordinator.go index 67f5948..524a1a0 100644 --- a/internal/transaction/coordinator.go +++ b/internal/transaction/coordinator.go @@ -93,7 +93,28 @@ func (c *Coordinator) ExecuteStep(ctx context.Context, transactionID string, int return step, nil } if step.Status == StepFailed { - return step, fmt.Errorf("external step already failed: %s", step.Error) + inspection, inspectErr := operation.Inspect(ctx) + if inspectErr != nil { + return step, c.uncertain(ctx, transactionID, intent.Key, fmt.Errorf("inspect failed step: %w", inspectErr)) + } + switch inspection.Status { + case InspectionApplied: + if _, err := c.store.ReopenFailedStep(ctx, transactionID, intent.Key, "manual resume found failed step applied"); err != nil { + return step, err + } + return c.completeApplied(ctx, transactionID, intent.Key, inspection) + case InspectionNotApplied: + reopened, err := c.store.ReopenFailedStep(ctx, transactionID, intent.Key, "manual resume found failed step not applied") + if err != nil { + return step, err + } + step = reopened + created = true + case InspectionUnknown: + return step, c.uncertain(ctx, transactionID, intent.Key, errors.New("inspect failed step returned UNKNOWN")) + default: + return step, c.uncertain(ctx, transactionID, intent.Key, fmt.Errorf("inspect failed step returned invalid status %q", inspection.Status)) + } } if !created { diff --git a/internal/transaction/coordinator_test.go b/internal/transaction/coordinator_test.go index 5a00dc4..41ad37b 100644 --- a/internal/transaction/coordinator_test.go +++ b/internal/transaction/coordinator_test.go @@ -72,6 +72,56 @@ func TestCoordinatorLeavesIntentPendingWhenInspectionIsUnknown(t *testing.T) { } } +func TestCoordinatorRetriesFailedStepOnlyAfterManualResumeInspection(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := openTestStore(t) + record := createTestTransaction(t, store, "retry-failed") + intent := StepIntent{Key: "restore-gateway", Name: "restore gateway", Intent: json.RawMessage(`{}`)} + operation := &fakeOperation{applyErr: errors.New("gateway executable unavailable"), result: json.RawMessage(`{"restored":true}`)} + coordinator := newTestCoordinator(t, store) + + failed, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation) + if err == nil || failed.Status != StepFailed { + t.Fatalf("unexpected initial failed step: step=%+v err=%v", failed, err) + } + if operation.applyCalls.Load() != 1 || operation.inspectCalls.Load() != 1 { + t.Fatalf("initial call retried unexpectedly: apply=%d inspect=%d", operation.applyCalls.Load(), operation.inspectCalls.Load()) + } + + operation.applyErr = nil + recovered, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation) + if err != nil || recovered.Status != StepSucceeded { + t.Fatalf("recover failed step: step=%+v err=%v", recovered, err) + } + if operation.applyCalls.Load() != 2 || operation.inspectCalls.Load() != 3 { + t.Fatalf("unexpected manual recovery calls: apply=%d inspect=%d", operation.applyCalls.Load(), operation.inspectCalls.Load()) + } +} + +func TestCoordinatorCompletesFailedStepAlreadyAppliedBeforeManualResume(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := openTestStore(t) + record := createTestTransaction(t, store, "recover-failed-applied") + intent := StepIntent{Key: "restore-gateway", Name: "restore gateway", Intent: json.RawMessage(`{}`)} + operation := &fakeOperation{applyErr: errors.New("gateway reload result unavailable"), result: json.RawMessage(`{"restored":true}`)} + coordinator := newTestCoordinator(t, store) + + if _, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation); err == nil { + t.Fatal("expected initial external step failure") + } + operation.applyErr = nil + operation.applied = true + recovered, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation) + if err != nil || recovered.Status != StepSucceeded { + t.Fatalf("complete externally applied failed step: step=%+v err=%v", recovered, err) + } + if operation.applyCalls.Load() != 1 || operation.inspectCalls.Load() != 2 { + t.Fatalf("externally applied step was repeated: apply=%d inspect=%d", operation.applyCalls.Load(), operation.inspectCalls.Load()) + } +} + func TestCoordinatorExclusiveExecutionHonorsContext(t *testing.T) { t.Parallel() store := openTestStore(t) diff --git a/internal/transaction/store.go b/internal/transaction/store.go index 080d9f0..60b724d 100644 --- a/internal/transaction/store.go +++ b/internal/transaction/store.go @@ -17,7 +17,7 @@ import ( "github.com/ncruces/go-sqlite3/driver" ) -const schemaVersion = 1 +const schemaVersion = 2 const schemaV1 = ` CREATE TABLE transactions ( @@ -70,13 +70,27 @@ CREATE INDEX transaction_events_by_transaction PRAGMA user_version = 1; ` +const schemaV2 = ` +CREATE TABLE backend_container_deployment ( + singleton_id INTEGER PRIMARY KEY CHECK (singleton_id = 1), + active_port INTEGER NOT NULL CHECK (active_port IN (8080, 8081)), + container_name TEXT NOT NULL, + image_digest TEXT NOT NULL, + container_id TEXT NOT NULL, + transaction_id TEXT NOT NULL REFERENCES transactions(id), + updated_at TEXT NOT NULL +) STRICT; + +PRAGMA user_version = 2; +` + // Store 是服务端 SQLite 事务记录。一个进程只应创建一个 Store。 type Store struct { db *sql.DB now func() time.Time } -// OpenStore 打开本地 SQLite，并强制校验第一版持久化参数。 +// OpenStore 打开本地 SQLite，并强制校验持久化参数和 schema 版本。 func OpenStore(ctx context.Context, path string) (*Store, error) { if path == "" { return nil, errors.New("sqlite path is required") @@ -168,8 +182,21 @@ func migrate(ctx context.Context, db *sql.DB) error { return fmt.Errorf("begin sqlite migration: %w", err) } defer tx.Rollback() - if _, err := tx.ExecContext(ctx, schemaV1); err != nil { - return fmt.Errorf("apply sqlite schema version 1: %w", err) + for version < schemaVersion { + nextVersion := version + 1 + var script string + switch nextVersion { + case 1: + script = schemaV1 + case 2: + script = schemaV2 + default: + return fmt.Errorf("sqlite migration script is missing for version %d", nextVersion) + } + if _, err := tx.ExecContext(ctx, script); err != nil { + return fmt.Errorf("apply sqlite schema version %d: %w", nextVersion, err) + } + version = nextVersion } if err := tx.Commit(); err != nil { return fmt.Errorf("commit sqlite migration: %w", err) @@ -199,7 +226,37 @@ func (s *Store) CreateTransaction(ctx context.Context, request CreateRequest) (r existing, err := getTransactionByIdempotencyKey(ctx, tx, request.IdempotencyKey) if err == nil { - return existing, false, nil + if existing.State != StateFailed && existing.State != StateRolledBack { + return existing, false, nil + } + archivedKey := existing.IdempotencyKey + ":terminal:" + existing.ID + result, updateErr := tx.ExecContext(ctx, ` + UPDATE transactions + SET idempotency_key = ?, version = version + 1, updated_at = ? + WHERE id = ? AND version = ? AND idempotency_key = ? + AND state IN (?, ?)`, + archivedKey, + formatTime(now), + existing.ID, + existing.Version, + existing.IdempotencyKey, + StateFailed, + StateRolledBack, + ) + if updateErr != nil { + return Transaction{}, false, fmt.Errorf("archive terminal transaction idempotency key: %w", updateErr) + } + rows, rowsErr := result.RowsAffected() + if rowsErr != nil { + return Transaction{}, false, fmt.Errorf("read archived transaction rows: %w", rowsErr) + } + if rows != 1 { + return Transaction{}, false, errors.New("terminal transaction changed concurrently") + } + if eventErr := insertEvent(ctx, tx, existing.ID, "", "TRANSACTION_RETRY_RELEASED", existing.State, existing.State, "terminal transaction idempotency key archived for manual retry", now); eventErr != nil { + return Transaction{}, false, eventErr + } + err = ErrNotFound } if !errors.Is(err, ErrNotFound) { return Transaction{}, false, err @@ -469,6 +526,59 @@ func (s *Store) CompleteStep(ctx context.Context, transactionID, stepKey string, return record, nil } +// ReopenFailedStep makes one manually resumed external step pending again. +// The coordinator calls this only after Inspect has established an exact +// APPLIED or NOT_APPLIED state; UNKNOWN never reopens a failed step. +func (s *Store) ReopenFailedStep(ctx context.Context, transactionID, stepKey, message string) (Step, error) { + tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable}) + if err != nil { + return Step{}, fmt.Errorf("begin reopen failed step: %w", err) + } + defer tx.Rollback() + record, err := getStep(ctx, tx, transactionID, stepKey) + if err != nil { + return Step{}, err + } + if record.Status == StepIntentRecorded { + return record, nil + } + if record.Status != StepFailed { + return Step{}, ErrStepNotPending + } + now := s.now().UTC() + result, err := tx.ExecContext(ctx, ` + UPDATE transaction_steps + SET status = ?, result_json = NULL, error_message = '', updated_at = ? + WHERE transaction_id = ? AND step_key = ? AND status = ?`, + StepIntentRecorded, + formatTime(now), + transactionID, + stepKey, + StepFailed, + ) + if err != nil { + return Step{}, fmt.Errorf("reopen failed step: %w", err) + } + rows, err := result.RowsAffected() + if err != nil { + return Step{}, fmt.Errorf("read reopened step rows: %w", err) + } + if rows != 1 { + return Step{}, errors.New("failed step changed concurrently") + } + if err := insertEvent(ctx, tx, transactionID, stepKey, "STEP_REOPENED", "", "", message, now); err != nil { + return Step{}, err + } + if err := tx.Commit(); err != nil { + return Step{}, fmt.Errorf("commit reopened step: %w", err) + } + record.Status = StepIntentRecorded + record.Result = nil + record.Error = "" + record.UpdatedAt = now + return record, nil +} + // PendingSteps 返回重启后必须先核对实际外部状态的步骤。 func (s *Store) PendingSteps(ctx context.Context, transactionID string) ([]Step, error) { rows, err := s.db.QueryContext(ctx, ` diff --git a/internal/transaction/store_test.go b/internal/transaction/store_test.go index 29d88d8..9c67aa9 100644 --- a/internal/transaction/store_test.go +++ b/internal/transaction/store_test.go @@ -4,12 +4,135 @@ import ( "context" "encoding/json" "errors" + "net/url" "path/filepath" "sync" "testing" "time" + + "github.com/ncruces/go-sqlite3/driver" ) +func TestStoreMigratesVersionOneAndCommitsBackendContainerDeployment(t *testing.T) { + t.Parallel() + ctx := context.Background() + databasePath := filepath.Join(t.TempDir(), "transaction.db") + dsn := (&url.URL{Scheme: "file", Path: databasePath}).String() + database, err := driver.Open(dsn) + if err != nil { + t.Fatalf("open version one database: %v", err) + } + if _, err := database.ExecContext(ctx, schemaV1); err != nil { + _ = database.Close() + t.Fatalf("create version one database: %v", err) + } + if err := database.Close(); err != nil { + t.Fatalf("close version one database: %v", err) + } + + store, err := OpenStore(ctx, databasePath) + if err != nil { + t.Fatalf("migrate transaction store: %v", err) + } + t.Cleanup(func() { _ = store.Close() }) + + if _, err := store.BackendContainerDeployment(ctx); !errors.Is(err, ErrNotFound) { + t.Fatalf("unexpected deployment before commit: %v", err) + } + hasHistory, err := store.HasCommittedBackendContainerTransactionHistory(ctx) + if err != nil || hasHistory { + t.Fatalf("unexpected empty backend history: found=%t err=%v", hasHistory, err) + } + nativeRecord, _, err := store.CreateTransaction(ctx, CreateRequest{ + ID: "native-backend-transaction", + IdempotencyKey: "backend:native-request", + Source: "test", + Service: "backend", + }) + if err != nil { + t.Fatalf("create native backend transaction: %v", err) + } + if _, err := store.Transition(ctx, nativeRecord.ID, StateFailed, "finish native backend transaction"); err != nil { + t.Fatalf("finish native backend transaction: %v", err) + } + hasHistory, err = store.HasCommittedBackendContainerTransactionHistory(ctx) + if err != nil || hasHistory { + t.Fatalf("native backend history was treated as container history: found=%t err=%v", hasHistory, err) + } + + record, created, err := store.CreateTransaction(ctx, CreateRequest{ + ID: "container-deployment-transaction", + IdempotencyKey: "backend:container:deployment-request", + Source: "test", + Service: "backend", + }) + if err != nil || !created { + t.Fatalf("create backend container transaction: record=%+v created=%t err=%v", record, created, err) + } + for _, state := range []State{StateValidating, StatePrepared, StateStarting, StateSwitching, StateVerifying, StateDraining} { + if _, err := store.Transition(ctx, record.ID, state, "test transition"); err != nil { + t.Fatalf("transition backend container transaction to %s: %v", state, err) + } + } + hasHistory, err = store.HasCommittedBackendContainerTransactionHistory(ctx) + if err != nil || hasHistory { + t.Fatalf("unfinished backend container transaction was treated as committed history: found=%t err=%v", hasHistory, err) + } + + committed, err := store.CommitBackendContainerDeployment(ctx, record.ID, BackendContainerDeployment{ + ActivePort: 8081, + ContainerName: "backend-8081", + ImageDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + ContainerID: "container-backend-8081", + }, "container deployment committed") + if err != nil || committed.State != StateCommitted { + t.Fatalf("commit backend container deployment: record=%+v err=%v", committed, err) + } + hasHistory, err = store.HasCommittedBackendContainerTransactionHistory(ctx) + if err != nil || !hasHistory { + t.Fatalf("committed backend container history was not recorded: found=%t err=%v", hasHistory, err) + } + deployment, err := store.BackendContainerDeployment(ctx) + if err != nil { + t.Fatalf("read backend container deployment: %v", err) + } + if deployment.ActivePort != 8081 || deployment.ContainerName != "backend-8081" || deployment.ContainerID != "container-backend-8081" || deployment.TransactionID != record.ID { + t.Fatalf("unexpected backend container deployment: %+v", deployment) + } + var version int + if err := store.db.QueryRowContext(ctx, "PRAGMA user_version").Scan(&version); err != nil || version != schemaVersion { + t.Fatalf("unexpected migrated schema version: version=%d err=%v", version, err) + } +} + +func TestBackendContainerDeploymentCommitRequiresDrainingTransaction(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := openTestStore(t) + record, _, err := store.CreateTransaction(ctx, CreateRequest{ + ID: "container-deployment-wrong-state", + IdempotencyKey: "container-deployment-wrong-state-request", + Source: "test", + Service: "backend", + }) + if err != nil { + t.Fatalf("create backend transaction: %v", err) + } + _, err = store.CommitBackendContainerDeployment(ctx, record.ID, BackendContainerDeployment{ + ActivePort: 8080, + ContainerName: "backend-8080", + ImageDigest: "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + ContainerID: "container-backend-8080", + }, "must fail") + var transitionErr *TransitionError + if !errors.As(err, &transitionErr) { + t.Fatalf("expected deployment commit transition error, got %v", err) + } + if _, err := store.BackendContainerDeployment(ctx); !errors.Is(err, ErrNotFound) { + t.Fatalf("failed commit wrote backend deployment: %v", err) + } +} + func TestStoreCreateIsIdempotentAndAllowsOnlyOneActiveTransaction(t *testing.T) { t.Parallel() ctx := context.Background() @@ -51,6 +174,26 @@ func TestStoreCreateIsIdempotentAndAllowsOnlyOneActiveTransaction(t *testing.T) if _, err := store.Transition(ctx, created.ID, StateFailed, "test terminal state"); err != nil { t.Fatalf("finish first transaction: %v", err) } + retriedAfterFailure, isNew, err := store.CreateTransaction(ctx, CreateRequest{ + ID: "transaction-1-retry", + IdempotencyKey: request.IdempotencyKey, + Source: request.Source, + Service: request.Service, + Request: request.Request, + }) + if err != nil || !isNew || retriedAfterFailure.ID != "transaction-1-retry" { + t.Fatalf("retry failed transaction: record=%+v new=%v err=%v", retriedAfterFailure, isNew, err) + } + archived, err := store.Transaction(ctx, created.ID) + if err != nil { + t.Fatalf("read archived failed transaction: %v", err) + } + if archived.IdempotencyKey != request.IdempotencyKey+":terminal:"+created.ID { + t.Fatalf("unexpected archived idempotency key: %q", archived.IdempotencyKey) + } + if _, err := store.Transition(ctx, retriedAfterFailure.ID, StateFailed, "finish retried transaction"); err != nil { + t.Fatalf("finish retried transaction: %v", err) + } second, isNew, err := store.CreateTransaction(ctx, CreateRequest{ ID: "transaction-2", IdempotencyKey: "request-2", diff --git a/main.go b/main.go index c9ac85b..6e9d7f8 100644 --- a/main.go +++ b/main.go @@ -14,6 +14,7 @@ import ( "time" "yms-daemon/internal/backendupdate" + "yms-daemon/internal/containerengine" "yms-daemon/internal/daemonapi" "yms-daemon/internal/daemonclient" "yms-daemon/internal/daemonserver" @@ -63,7 +64,12 @@ func run(ctx context.Context, arguments []string, stdout io.Writer, stderr io.Wr writeUpdateProgress(stdout, event) } } - response, err := daemonclient.Update(ctx, runtimepaths.Socket, request.service, request.inputType, request.file, progress) + var response daemonapi.Response + if request.inputType == daemonapi.InputTypeContainerImage { + response, err = daemonclient.UpdateContainerImage(ctx, runtimepaths.Socket, request.service, request.imageReference, progress) + } else { + response, err = daemonclient.Update(ctx, runtimepaths.Socket, request.service, request.inputType, request.file, progress) + } if err != nil { if response.TransactionID != "" { fmt.Fprintf(stderr, "transaction=%s state=%s error=%v\n", response.TransactionID, response.State, err) @@ -112,10 +118,11 @@ func run(ctx context.Context, arguments []string, stdout io.Writer, stderr io.Wr } type updateArguments struct { - service string - inputType string - file string - quite bool + service string + inputType string + file string + imageReference string + quite bool } type restartArguments struct { @@ -129,6 +136,7 @@ func parseUpdateArgs(arguments []string, output io.Writer) (updateArguments, err service := flags.String("service", "", "service to update") file := flags.String("f", "", "repack ZIP path") nativeJAR := flags.String("native-jar", "", "direct native backend JAR path") + containerImage := flags.String("container-image", "", "development backend container image reference") quite := flags.Bool("quite", false, "suppress progress and successful result output") if err := flags.Parse(arguments); err != nil { return updateArguments{}, err @@ -139,8 +147,17 @@ func parseUpdateArgs(arguments []string, output io.Writer) (updateArguments, err if *service != serviceBackend { return updateArguments{}, errors.New("--service currently accepts only backend") } - if (*file == "") == (*nativeJAR == "") { - return updateArguments{}, errors.New("exactly one of -f and --native-jar is required") + inputCount := 0 + for _, value := range []string{*file, *nativeJAR, *containerImage} { + if value != "" { + inputCount++ + } + } + if inputCount != 1 { + return updateArguments{}, errors.New("exactly one of -f, --native-jar, and --container-image is required; update inputs are mutually exclusive") + } + if *containerImage != "" { + return updateArguments{service: *service, inputType: daemonapi.InputTypeContainerImage, imageReference: *containerImage, quite: *quite}, nil } inputType := daemonapi.InputTypeRepackZIP inputFile := *file @@ -205,14 +222,6 @@ func runServe(ctx context.Context) (result error) { if err != nil { return err } - releaseStore, err := filestore.New(config.Backend.ReleaseDir) - if err != nil { - return err - } - units, err := systemd.NewSystemctl(config.Backend.SystemctlPath) - if err != nil { - return err - } gateway, err := hostnginx.NewController( runtimepaths.HostNginxConfig, runtimepaths.HostNginxExecutable, @@ -223,7 +232,28 @@ func runServe(ctx context.Context) (result error) { return err } httpClient := &http.Client{Timeout: 5 * time.Second} - updater, err := backendupdate.New(config, runtimepaths.WorkRoot, store, coordinator, releaseStore, units, gateway, httpClient, logger) + var updater *backendupdate.Updater + switch config.Backend.Type { + case deploymentconfig.BackendTypeNative: + releaseStore, err := filestore.New(config.Backend.ReleaseDir) + if err != nil { + return err + } + units, err := systemd.NewSystemctl(config.Backend.SystemctlPath) + if err != nil { + return err + } + updater, err = backendupdate.New(config, runtimepaths.WorkRoot, store, coordinator, releaseStore, units, gateway, httpClient, logger) + case deploymentconfig.BackendTypeContainer: + var engine *containerengine.MobyEngine + engine, err = containerengine.NewMobyEngine() + if err == nil { + defer func() { result = errors.Join(result, engine.Close()) }() + updater, err = backendupdate.NewContainer(config, runtimepaths.WorkRoot, store, coordinator, engine, gateway, httpClient, logger) + } + default: + err = fmt.Errorf("unsupported backend.type %q", config.Backend.Type) + } if err != nil { return err } @@ -239,6 +269,7 @@ func writeUsage(output io.Writer) { fmt.Fprintln(output, " yms-daemon serve") fmt.Fprintln(output, " yms-daemon update --service backend -f [--quite]") fmt.Fprintln(output, " yms-daemon update --service backend --native-jar [--quite]") + fmt.Fprintln(output, " yms-daemon update --service backend --container-image [--quite]") fmt.Fprintln(output, " yms-daemon restart --service backend [--quite]") } diff --git a/main_test.go b/main_test.go index 8cc28c9..c651697 100644 --- a/main_test.go +++ b/main_test.go @@ -5,6 +5,7 @@ import ( "context" "os" "path/filepath" + "strings" "testing" "yms-daemon/internal/daemonapi" @@ -38,6 +39,30 @@ func TestParseUpdateArgsAcceptsDirectNativeJAR(t *testing.T) { } } +func TestParseUpdateArgsAcceptsContainerImage(t *testing.T) { + imageReference := "harbor.ymswell.asia/ymswell/glory-ymswell:20260813-184902-a37bf50d-v1.1.8.1" + request, err := parseUpdateArgs([]string{"--service", "backend", "--container-image", imageReference}, &bytes.Buffer{}) + if err != nil { + t.Fatalf("parse container backend update arguments: %v", err) + } + if request.service != "backend" || request.inputType != daemonapi.InputTypeContainerImage || request.imageReference != imageReference || request.file != "" { + t.Fatalf("unexpected container backend update arguments: %+v", request) + } +} + +func TestParseUpdateArgsRejectsNativeJARAndContainerImageTogether(t *testing.T) { + jarPath := filepath.Join(t.TempDir(), "glory-soft-yms.jar") + arguments := []string{ + "--service", "backend", + "--native-jar", jarPath, + "--container-image", "harbor.ymswell.asia/ymswell/glory-ymswell:20260813-184902-a37bf50d-v1.1.8.1", + } + _, err := parseUpdateArgs(arguments, &bytes.Buffer{}) + if err == nil || !strings.Contains(err.Error(), "exactly one") { + t.Fatalf("expected mutually exclusive backend inputs, got %v", err) + } +} + func TestParseUpdateArgsAcceptsQuite(t *testing.T) { file := filepath.Join(t.TempDir(), "glory-soft-yms.jar") if err := os.WriteFile(file, []byte("jar"), 0o600); err != nil { @@ -86,11 +111,12 @@ func TestParseRestartArgsRejectsUnsupportedInput(t *testing.T) { func TestParseUpdateArgsRejectsIncompleteOrUnsupportedInput(t *testing.T) { for name, arguments := range map[string][]string{ - "missing service": {"-f", "/tmp/package.zip"}, - "missing file": {"--service", "backend"}, - "other service": {"--service", "frontend", "-f", "/tmp/package.zip"}, - "positional": {"--service", "backend", "-f", "/tmp/package.zip", "extra"}, - "both inputs": {"--service", "backend", "-f", "/tmp/package.zip", "--native-jar", "/tmp/backend.jar"}, + "missing service": {"-f", "/tmp/package.zip"}, + "missing file": {"--service", "backend"}, + "other service": {"--service", "frontend", "-f", "/tmp/package.zip"}, + "positional": {"--service", "backend", "-f", "/tmp/package.zip", "extra"}, + "both inputs": {"--service", "backend", "-f", "/tmp/package.zip", "--native-jar", "/tmp/backend.jar"}, + "native and container": {"--service", "backend", "--native-jar", "/tmp/backend.jar", "--container-image", "repository:tag"}, } { t.Run(name, func(t *testing.T) { if _, err := parseUpdateArgs(arguments, &bytes.Buffer{}); err == nil { diff --git a/packaging/etc/yms-daemon/yms-daemon.toml b/packaging/etc/yms-daemon/yms-daemon.toml index 25fe831..32b48bb 100644 --- a/packaging/etc/yms-daemon/yms-daemon.toml +++ b/packaging/etc/yms-daemon/yms-daemon.toml @@ -1,3 +1,6 @@ +[daemon] +environment = "prod" + [backend] type = "native" release_dir = "/home/yms/lib/releases" diff --git a/packaging/rpm/build-rpm.sh b/packaging/rpm/build-rpm.sh new file mode 100755 index 0000000..5a7d221 --- /dev/null +++ b/packaging/rpm/build-rpm.sh @@ -0,0 +1,102 @@ +#!/bin/sh +set -eu + +script_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd) +project_dir=$(CDPATH= cd -- "$script_dir/../.." && pwd) + +version=0.1.0 +release="$(date -u +%Y%m%d%H%M%S).$(git -C "$project_dir" rev-parse --short=8 HEAD)" +go_arch=$(go env GOARCH) + +while [ "$#" -gt 0 ]; do + case "$1" in + --version) + [ "$#" -ge 2 ] || { echo "--version requires a value" >&2; exit 2; } + version=$2 + shift 2 + ;; + --release) + [ "$#" -ge 2 ] || { echo "--release requires a value" >&2; exit 2; } + release=$2 + shift 2 + ;; + --goarch) + [ "$#" -ge 2 ] || { echo "--goarch requires a value" >&2; exit 2; } + go_arch=$2 + shift 2 + ;; + -h|--help) + echo "usage: packaging/rpm/build-rpm.sh [--version ] [--release ] [--goarch amd64|arm64]" + exit 0 + ;; + *) + echo "unknown argument: $1" >&2 + exit 2 + ;; + esac +done + +case "$version" in + ""|*[!A-Za-z0-9._+]*|*-*) + echo "RPM version contains an unsupported character: $version" >&2 + exit 2 + ;; +esac +case "$release" in + ""|*[!A-Za-z0-9._+]*|*-*) + echo "RPM release contains an unsupported character: $release" >&2 + exit 2 + ;; +esac + +case "$go_arch" in + amd64) + rpm_arch=x86_64 + ;; + arm64) + rpm_arch=aarch64 + ;; + *) + echo "--goarch accepts only amd64 or arm64" >&2 + exit 2 + ;; +esac + +mkdir -p "$project_dir/.build" "$project_dir/dist" +build_dir=$(mktemp -d "$project_dir/.build/rpm.XXXXXX") +trap 'rm -rf "$build_dir"' EXIT HUP INT TERM + +top_dir="$build_dir/rpmbuild" +sources_dir="$top_dir/SOURCES" +temporary_dir="$build_dir/tmp" +mkdir -p "$sources_dir" "$top_dir/BUILD" "$top_dir/BUILDROOT" "$top_dir/RPMS" "$top_dir/SRPMS" "$temporary_dir" + +echo "Building static linux/$go_arch yms-daemon" +( + cd "$project_dir" + GOCACHE="$build_dir/go-cache" CGO_ENABLED=0 GOOS=linux GOARCH="$go_arch" \ + go build -trimpath -ldflags="-s -w" -o "$sources_dir/yms-daemon" . +) + +install -m 0640 "$project_dir/packaging/etc/yms-daemon/yms-daemon.toml" "$sources_dir/yms-daemon.toml" +install -m 0644 "$project_dir/packaging/systemd/yms-daemon.service" "$sources_dir/yms-daemon.service" +install -m 0644 "$project_dir/packaging/systemd/yms-backend@.service" "$sources_dir/yms-backend@.service" +install -m 0644 "$project_dir/packaging/tmpfiles.d/yms-daemon.conf" "$sources_dir/yms-daemon-tmpfiles.conf" + +echo "Building $rpm_arch RPM version=$version release=$release" +TMPDIR="$temporary_dir" rpmbuild -bb "$project_dir/packaging/rpm/yms-daemon.spec" \ + --target "$rpm_arch" \ + --define "_topdir $top_dir" \ + --define "_tmppath $temporary_dir" \ + --define "package_version $version" \ + --define "package_release $release" + +rpm_path="$top_dir/RPMS/$rpm_arch/yms-daemon-$version-$release.$rpm_arch.rpm" +if [ ! -f "$rpm_path" ]; then + echo "rpmbuild did not create the expected RPM: $rpm_path" >&2 + exit 1 +fi + +output_path="$project_dir/dist/$(basename "$rpm_path")" +install -m 0644 "$rpm_path" "$output_path" +echo "$output_path" diff --git a/packaging/rpm/yms-daemon.spec b/packaging/rpm/yms-daemon.spec new file mode 100644 index 0000000..d169f96 --- /dev/null +++ b/packaging/rpm/yms-daemon.spec @@ -0,0 +1,79 @@ +%global debug_package %{nil} +%global __strip /bin/true +%global _binary_payload w9.gzdio + +%{!?package_version:%global package_version 0.1.0} +%{!?package_release:%global package_release 1} + +Name: yms-daemon +Version: %{package_version} +Release: %{package_release} +Summary: YMS update daemon +License: Proprietary +Requires: systemd + +Source0: yms-daemon +Source1: yms-daemon.toml +Source2: yms-daemon.service +Source3: yms-backend@.service +Source4: yms-daemon-tmpfiles.conf + +%description +Transactional update daemon and native backend systemd units for YMS installations. + +%prep + +%build + +%install +install -d -m 0755 %{buildroot}/usr/bin +install -m 0755 %{SOURCE0} %{buildroot}/usr/bin/yms-daemon + +install -d -m 0755 %{buildroot}/etc/yms-daemon +install -m 0640 %{SOURCE1} %{buildroot}/etc/yms-daemon/yms-daemon.toml + +install -d -m 0755 %{buildroot}/etc/systemd/system +install -m 0644 %{SOURCE2} %{buildroot}/etc/systemd/system/yms-daemon.service +install -m 0644 %{SOURCE3} %{buildroot}/etc/systemd/system/yms-backend@.service + +install -d -m 0755 %{buildroot}/etc/tmpfiles.d +install -m 0644 %{SOURCE4} %{buildroot}/etc/tmpfiles.d/yms-daemon.conf + +install -d -m 0755 %{buildroot}/home/yms/dump + +%post +tmpfiles_path="$(command -v systemd-tmpfiles 2>/dev/null || :)" +if [ -n "$tmpfiles_path" ]; then + "$tmpfiles_path" --create /etc/tmpfiles.d/yms-daemon.conf >/dev/null 2>&1 || : +fi +systemctl_path="$(command -v systemctl 2>/dev/null || :)" +if [ -n "$systemctl_path" ]; then + "$systemctl_path" daemon-reload >/dev/null 2>&1 || : +fi + +%preun +if [ "$1" -eq 0 ]; then + systemctl_path="$(command -v systemctl 2>/dev/null || :)" + if [ -n "$systemctl_path" ]; then + "$systemctl_path" disable --now yms-daemon.service >/dev/null 2>&1 || : + fi +fi + +%postun +systemctl_path="$(command -v systemctl 2>/dev/null || :)" +if [ -n "$systemctl_path" ]; then + "$systemctl_path" daemon-reload >/dev/null 2>&1 || : +fi + +%files +%dir %attr(0755,root,root) /etc/yms-daemon +%config(noreplace) %attr(0640,root,root) /etc/yms-daemon/yms-daemon.toml +%attr(0644,root,root) /etc/systemd/system/yms-daemon.service +%attr(0644,root,root) /etc/systemd/system/yms-backend@.service +%attr(0644,root,root) /etc/tmpfiles.d/yms-daemon.conf +%attr(0755,root,root) /usr/bin/yms-daemon +%dir %attr(0755,root,root) /home/yms/dump + +%changelog +* Sun Aug 16 2026 YMS Engineering +- Add the initial native backend update daemon package.