mirror of
https://github.com/Awuqing/BackupX.git
synced 2026-08-25 10:10:02 +08:00
Compare commits
6 Commits
dependabot
...
codex/reso
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b44050ef9b | ||
|
|
755632e19b | ||
|
|
c2805da6df | ||
|
|
f8deafcb00 | ||
|
|
cc50637b4b | ||
|
|
05dd1baa61 |
@@ -60,11 +60,15 @@ docker run -d --name backupx -p 8340:8340 -v backupx-data:/app/data awuqing/back
|
||||
# Or prebuilt archive
|
||||
curl -LO https://github.com/Awuqing/BackupX/releases/latest/download/backupx-linux-amd64.tar.gz
|
||||
tar xzf backupx-*.tar.gz && cd backupx-* && sudo ./install.sh
|
||||
|
||||
# Or build and install on bare metal without Docker
|
||||
git clone https://github.com/Awuqing/BackupX.git && cd BackupX
|
||||
make build && sudo ./deploy/install.sh
|
||||
```
|
||||
|
||||
For ARM64 hosts, use `backupx-linux-arm64.tar.gz`. The archive contains `backupx`, `web/`, `config.example.yaml`, and `install.sh`; run `install.sh` from the extracted directory.
|
||||
|
||||
Open `http://your-server:8340`, create the admin account, then follow the [5-minute Quick Start](https://awuqing.github.io/BackupX/docs/getting-started/quick-start).
|
||||
Open `http://your-server:8340`, choose English or Chinese on the setup screen, create the first administrator account, then follow the [5-minute Quick Start](https://awuqing.github.io/BackupX/docs/getting-started/quick-start).
|
||||
|
||||
## Documentation
|
||||
|
||||
|
||||
@@ -60,11 +60,15 @@ docker run -d --name backupx -p 8340:8340 -v backupx-data:/app/data awuqing/back
|
||||
# 或使用预编译包
|
||||
curl -LO https://github.com/Awuqing/BackupX/releases/latest/download/backupx-linux-amd64.tar.gz
|
||||
tar xzf backupx-*.tar.gz && cd backupx-* && sudo ./install.sh
|
||||
|
||||
# 或从源码构建并裸机安装(无需 Docker)
|
||||
git clone https://github.com/Awuqing/BackupX.git && cd BackupX
|
||||
make build && sudo ./deploy/install.sh
|
||||
```
|
||||
|
||||
ARM64 主机请下载 `backupx-linux-arm64.tar.gz`。预编译包内包含 `backupx`、`web/`、`config.example.yaml` 和 `install.sh`,请在解压后的目录内执行 `install.sh`。
|
||||
|
||||
打开 `http://your-server:8340`,创建管理员账户,按 [5 分钟快速开始](https://awuqing.github.io/BackupX/zh-Hans/docs/getting-started/quick-start) 完成首次备份。
|
||||
打开 `http://your-server:8340`,在初始化页选择中文或 English 并创建首个管理员账户,按 [5 分钟快速开始](https://awuqing.github.io/BackupX/zh-Hans/docs/getting-started/quick-start) 完成首次备份。
|
||||
|
||||
## 文档
|
||||
|
||||
|
||||
@@ -14,7 +14,13 @@ if [ -f "$SCRIPT_DIR/backupx" ] && [ -d "$SCRIPT_DIR/web" ]; then
|
||||
CONFIG_TEMPLATE="${CONFIG_TEMPLATE:-$SCRIPT_DIR/config.example.yaml}"
|
||||
NGINX_SOURCE="${NGINX_SOURCE:-$SCRIPT_DIR/nginx.conf}"
|
||||
else
|
||||
BIN_SOURCE="${BIN_SOURCE:-$PROJECT_ROOT/server/backupx}"
|
||||
SOURCE_BIN_DEFAULT="$PROJECT_ROOT/server/bin/backupx"
|
||||
# Keep compatibility with contributors who built the historical path by
|
||||
# hand, while matching the canonical `make build` output first.
|
||||
if [ ! -f "$SOURCE_BIN_DEFAULT" ] && [ -f "$PROJECT_ROOT/server/backupx" ]; then
|
||||
SOURCE_BIN_DEFAULT="$PROJECT_ROOT/server/backupx"
|
||||
fi
|
||||
BIN_SOURCE="${BIN_SOURCE:-$SOURCE_BIN_DEFAULT}"
|
||||
WEB_SOURCE="${WEB_SOURCE:-$PROJECT_ROOT/web/dist}"
|
||||
CONFIG_TEMPLATE="${CONFIG_TEMPLATE:-$PROJECT_ROOT/server/config.example.yaml}"
|
||||
NGINX_SOURCE="${NGINX_SOURCE:-$PROJECT_ROOT/deploy/nginx.conf}"
|
||||
@@ -27,8 +33,9 @@ if [ "$(id -u)" -ne 0 ]; then
|
||||
fi
|
||||
|
||||
if [ ! -f "$BIN_SOURCE" ]; then
|
||||
echo "未找到后端二进制:$BIN_SOURCE" >&2
|
||||
echo "源码树安装请先执行:cd \"$PROJECT_ROOT/server\" && go build -o backupx ./cmd/backupx" >&2
|
||||
echo "Backend binary not found / 未找到后端二进制:$BIN_SOURCE" >&2
|
||||
echo "源码树安装请先在仓库根目录执行 make build(产物:server/bin/backupx)。" >&2
|
||||
echo "For a source install, run 'make build' in the repository root first." >&2
|
||||
echo "发布包安装请确认当前目录包含 ./backupx、./web 和 ./install.sh。" >&2
|
||||
exit 1
|
||||
fi
|
||||
@@ -92,6 +99,41 @@ fi
|
||||
systemctl daemon-reload
|
||||
systemctl enable --now "$SERVICE_NAME"
|
||||
|
||||
# systemctl may return before the process has opened its HTTP listener. Verify
|
||||
# the same unauthenticated endpoint used by the first-administrator screen so a
|
||||
# broken bare-metal install cannot print a false success message.
|
||||
HEALTH_URL="${HEALTH_URL:-http://127.0.0.1:8340/api/auth/setup/status}"
|
||||
READY=0
|
||||
ATTEMPT=1
|
||||
while [ "$ATTEMPT" -le 30 ]; do
|
||||
if systemctl is-active --quiet "$SERVICE_NAME"; then
|
||||
if command -v curl >/dev/null 2>&1; then
|
||||
if curl -fsS --max-time 2 "$HEALTH_URL" >/dev/null 2>&1; then
|
||||
READY=1
|
||||
break
|
||||
fi
|
||||
elif command -v wget >/dev/null 2>&1; then
|
||||
if wget -q -T 2 -O /dev/null "$HEALTH_URL"; then
|
||||
READY=1
|
||||
break
|
||||
fi
|
||||
else
|
||||
echo "Warning / 警告:未找到 curl 或 wget,仅验证 systemd 服务状态。" >&2
|
||||
READY=1
|
||||
break
|
||||
fi
|
||||
fi
|
||||
ATTEMPT=$((ATTEMPT + 1))
|
||||
sleep 1
|
||||
done
|
||||
|
||||
if [ "$READY" -ne 1 ]; then
|
||||
echo "BackupX did not become ready at $HEALTH_URL / 服务未通过就绪检查。" >&2
|
||||
systemctl status "$SERVICE_NAME" --no-pager >&2 || true
|
||||
journalctl -u "$SERVICE_NAME" -n 50 --no-pager >&2 || true
|
||||
exit 1
|
||||
fi
|
||||
|
||||
if [ -d "/etc/nginx/conf.d" ] && [ -f "$NGINX_SOURCE" ]; then
|
||||
install -m 0644 "$NGINX_SOURCE" "/etc/nginx/conf.d/$SERVICE_NAME.conf"
|
||||
if command -v nginx >/dev/null 2>&1; then
|
||||
@@ -111,6 +153,11 @@ cat <<MESSAGE
|
||||
Web 控制台已由后端直接托管,无需额外的 nginx 反向代理即可访问:
|
||||
http://<本机IP>:8340
|
||||
|
||||
首次访问 / First sign-in:
|
||||
1. 打开上面的地址,并可在登录页右上角选择 中文 或 English。
|
||||
2. 页面显示“系统初始化 / System setup”时,创建首个管理员用户名和密码。
|
||||
3. 如果未显示初始化表单,请先检查:$HEALTH_URL
|
||||
|
||||
(如已安装 nginx,脚本会自动写入反向代理配置,可继续用 80 端口访问。)
|
||||
|
||||
排查:若服务未监听端口,请查看日志:
|
||||
|
||||
@@ -10,20 +10,21 @@ description: systemd + Nginx deployment from the prebuilt release tarball or sou
|
||||
|
||||
```bash
|
||||
# Download the matching tarball
|
||||
curl -LO https://github.com/Awuqing/BackupX/releases/latest/download/backupx-v1.6.0-linux-amd64.tar.gz
|
||||
curl -LO https://github.com/Awuqing/BackupX/releases/latest/download/backupx-linux-amd64.tar.gz
|
||||
|
||||
# Extract and install
|
||||
tar xzf backupx-v*-linux-amd64.tar.gz && cd backupx-*
|
||||
tar xzf backupx-linux-amd64.tar.gz && cd backupx-*-linux-amd64
|
||||
sudo ./install.sh
|
||||
```
|
||||
|
||||
The installer performs these steps automatically:
|
||||
|
||||
1. Creates a system user `backupx`
|
||||
2. Copies the binary to `/opt/backupx/`
|
||||
3. Generates a default `config.yaml` with safe JWT/encryption secrets
|
||||
2. Copies the binary to `/opt/backupx/bin/backupx` and the web console to `/opt/backupx/web`
|
||||
3. Installs the default configuration at `/etc/backupx/config.yaml`
|
||||
4. Installs `backupx.service` (systemd), enabled at boot
|
||||
5. (Optional) installs an Nginx site file — see [Nginx Reverse Proxy](./nginx)
|
||||
6. Verifies the first-setup API before reporting success
|
||||
|
||||
For multi-node clusters, edit `/etc/backupx/config.yaml` after installation and set the Master URL that remote Agents can reach:
|
||||
|
||||
@@ -63,11 +64,13 @@ After=network.target
|
||||
[Service]
|
||||
Type=simple
|
||||
User=backupx
|
||||
Group=backupx
|
||||
WorkingDirectory=/opt/backupx
|
||||
ExecStart=/opt/backupx/backupx --config /opt/backupx/config.yaml
|
||||
ExecStart=/opt/backupx/bin/backupx -config /etc/backupx/config.yaml
|
||||
Restart=on-failure
|
||||
RestartSec=5s
|
||||
LimitNOFILE=65536
|
||||
RestartSec=5
|
||||
NoNewPrivileges=true
|
||||
LimitNOFILE=65535
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
@@ -79,17 +82,20 @@ Typical operations:
|
||||
sudo systemctl status backupx
|
||||
sudo journalctl -u backupx -f # live logs
|
||||
sudo systemctl restart backupx
|
||||
curl -fsS http://127.0.0.1:8340/api/auth/setup/status
|
||||
```
|
||||
|
||||
Open `http://your-server:8340`, switch to English if desired, and create the first administrator on the **System setup** screen. For a custom listen port, run the installer with a matching `HEALTH_URL`.
|
||||
|
||||
## Password reset
|
||||
|
||||
If the admin password is lost:
|
||||
|
||||
```bash
|
||||
/opt/backupx/backupx reset-password \
|
||||
/opt/backupx/bin/backupx reset-password \
|
||||
--username admin \
|
||||
--password 'newpass123' \
|
||||
--config /opt/backupx/config.yaml
|
||||
--config /etc/backupx/config.yaml
|
||||
```
|
||||
|
||||
Docker equivalent:
|
||||
|
||||
@@ -12,12 +12,22 @@ When a task is routed to a remote Agent, the source tools and paths are resolved
|
||||
|
||||
## File / Directory
|
||||
|
||||
Tars (and optionally gzips) one or more filesystem paths.
|
||||
File tasks offer three backup modes:
|
||||
|
||||
- **Full archive** — writes a self-contained tar artifact on every run
|
||||
- **Differential archive** — writes only changes since the current full baseline and periodically refreshes that baseline
|
||||
- **CDC repository** — splits content with stable 512 KiB / 1 MiB / 4 MiB boundaries, stores new chunks in immutable 32 MiB packs, and writes a small snapshot manifest for each run
|
||||
|
||||
The CDC repository deduplicates identical content across files and snapshots. Restore, selective restore, verification, download-as-tar, retention, and garbage collection all resolve data through the repository index. Compression and encryption are applied per chunk; encrypted repositories use keyed chunk IDs so plaintext hashes are not exposed.
|
||||
|
||||
Repository mode currently uses a single-writer index and therefore runs on the Master only. To keep repository copies on multiple backends, select multiple primary storage targets on the task. Object-level replication is intentionally disabled because a snapshot manifest without its shared packs and indexes is not a complete backup.
|
||||
|
||||
Common file-task options:
|
||||
|
||||
- **Source** accepts multiple paths — one per line in the UI
|
||||
- **Exclude patterns** accept gitignore-style globs
|
||||
- Supports following symlinks, preserving permissions
|
||||
- Output is a single `.tar` or `.tar.gz` artifact
|
||||
- Full and differential modes output `.tar`, `.tar.gz`, or `.tar.zst` artifacts
|
||||
|
||||
## MySQL
|
||||
|
||||
|
||||
@@ -26,6 +26,27 @@ BackupX supports Master-Agent mode: backup tasks can be routed to specific nodes
|
||||
- **Execution** — Agent reuses the same BackupRunner (file / mysql / postgresql / sqlite / saphana) and uploads directly to storage
|
||||
- **Security** — Each node has its own token; the Agent never holds the Master's JWT secret or AES-256 key
|
||||
|
||||
## Centralize backups from servers B/C/D into storage M
|
||||
|
||||
Use the Master as the control plane and register every source server as an Agent. A task's **Source server** determines where paths and database tools are resolved; its **Storage targets** determine where the resulting artifact is retained.
|
||||
|
||||
BackupX chooses the data path per target:
|
||||
|
||||
| Destination | Data path |
|
||||
| --- | --- |
|
||||
| S3, WebDAV, FTP, cloud drive, or another network backend | Agent streams directly to the destination |
|
||||
| `local_disk` with **Relay remote backups through Master** enabled (for example storage server M mounted through NFS) | Agent streams through the authenticated Master API; Master writes to its configured local path |
|
||||
|
||||
The relay is streaming: the Master does not create a second temporary copy of the entire artifact. The reverse path is used when restoring a Master-local artifact back to its source Agent. Use HTTPS whenever Agent traffic crosses an untrusted network.
|
||||
|
||||
To configure the common `A → {B,C,D} → M` topology:
|
||||
|
||||
1. Run BackupX Master on A and mount M on A if M is exposed as NFS or another filesystem.
|
||||
2. Create a `local_disk` target for that mount and keep **Relay remote backups through Master** enabled, or create an S3/WebDAV target exposed by M. Existing local-disk targets keep their prior Agent-local behavior until this switch is enabled.
|
||||
3. Install one Agent on B, C, and D from **Node Management**.
|
||||
4. Create a backup task for each source, choose B/C/D under **Source server**, browse that server's paths, and select M as the storage target. A source-server pool label can route identical tasks dynamically.
|
||||
5. Verify the per-target result in the backup record. For a Master-local target, the record reports transfer mode `master_relay`; network backends remain `direct`.
|
||||
|
||||
## Walkthrough
|
||||
|
||||
### 0. Set the Master URL for production clusters
|
||||
@@ -78,7 +99,7 @@ In Step 1 choose "Batch" and paste node names (one per line, max 50). Step 3 sho
|
||||
|
||||
### 5. Route a task to the node
|
||||
|
||||
In the **Backup Tasks** page, pick the target node when creating the task. When the task runs:
|
||||
In the **Backup Tasks** page, pick the source server when creating the task. When the task runs:
|
||||
|
||||
- Local (`nodeId=0`) → Master executes in-process
|
||||
- Remote node → Master enqueues the command → Agent claims → Agent runs locally → uploads → reports back
|
||||
|
||||
@@ -19,7 +19,9 @@ BackupX aims to accept any place you'd want to drop a backup file.
|
||||
| **Google Drive** | Client ID/Secret + OAuth authorization |
|
||||
| **WebDAV** | URL + username/password |
|
||||
| **FTP / FTPS** | Host + port + username/password |
|
||||
| **Local disk** | Target directory (absolute path) |
|
||||
| **Local disk** | Target directory (absolute path) + optional Master relay for remote Agents |
|
||||
|
||||
New local-disk targets enable **Relay remote backups through Master** by default. This makes the configured path belong to the Master, so a storage server mounted there can collect backups from many source Agents. Turn the switch off when the path intentionally belongs to each Agent. Existing targets retain their previous Agent-local behavior until explicitly changed.
|
||||
|
||||
## Rclone backends
|
||||
|
||||
|
||||
@@ -55,10 +55,11 @@ sudo ./install.sh # creates system user, installs to /opt/backupx, sets u
|
||||
The installer:
|
||||
|
||||
1. Creates a `backupx` system user
|
||||
2. Installs binary to `/opt/backupx/backupx`
|
||||
3. Creates `/opt/backupx/config.yaml` with safe defaults
|
||||
2. Installs the binary to `/opt/backupx/bin/backupx` and the web console to `/opt/backupx/web`
|
||||
3. Creates `/etc/backupx/config.yaml` with safe defaults
|
||||
4. Installs and enables the `backupx.service` systemd unit
|
||||
5. (Optional) Configures an Nginx reverse proxy
|
||||
6. Waits for `/api/auth/setup/status`; if startup fails, prints systemd diagnostics and exits non-zero
|
||||
|
||||
## From source
|
||||
|
||||
@@ -67,16 +68,17 @@ Requires Go ≥ 1.25 and Node.js ≥ 20.
|
||||
```bash
|
||||
git clone https://github.com/Awuqing/BackupX.git && cd BackupX
|
||||
make build
|
||||
# or, for builds behind the great firewall
|
||||
make docker-cn
|
||||
sudo ./deploy/install.sh
|
||||
```
|
||||
|
||||
After `make build`, the binary is at `server/bin/backupx` and the built web UI is at `web/dist/`.
|
||||
The installer consumes those exact paths, so no Docker runtime is required. If an existing configuration uses a non-default port, set `HEALTH_URL` for the readiness check, for example `sudo HEALTH_URL=http://127.0.0.1:9000/api/auth/setup/status ./deploy/install.sh`.
|
||||
|
||||
## Verify the install
|
||||
|
||||
```bash
|
||||
backupx --version # e.g. v1.6.0
|
||||
/opt/backupx/bin/backupx --version
|
||||
curl -fsS http://127.0.0.1:8340/api/auth/setup/status
|
||||
```
|
||||
|
||||
Then open `http://your-server:8340` to see the initial admin setup screen.
|
||||
Then open `http://your-server:8340`. Choose **English** or **中文** in the upper-right corner. A fresh database shows **System setup**, where you create the first administrator username and password. If that form does not appear, retry the status request above before attempting to sign in.
|
||||
|
||||
@@ -10,20 +10,21 @@ description: 从预编译包或源码部署 BackupX(systemd + Nginx)。
|
||||
|
||||
```bash
|
||||
# 下载对应平台的压缩包
|
||||
curl -LO https://github.com/Awuqing/BackupX/releases/latest/download/backupx-v1.6.0-linux-amd64.tar.gz
|
||||
curl -LO https://github.com/Awuqing/BackupX/releases/latest/download/backupx-linux-amd64.tar.gz
|
||||
|
||||
# 解压并安装
|
||||
tar xzf backupx-v*-linux-amd64.tar.gz && cd backupx-*
|
||||
tar xzf backupx-linux-amd64.tar.gz && cd backupx-*-linux-amd64
|
||||
sudo ./install.sh
|
||||
```
|
||||
|
||||
安装脚本自动完成以下步骤:
|
||||
|
||||
1. 创建系统用户 `backupx`
|
||||
2. 复制二进制到 `/opt/backupx/`
|
||||
3. 生成默认 `config.yaml`(含安全的 JWT/加密密钥)
|
||||
2. 复制二进制到 `/opt/backupx/bin/backupx`,并把 Web 控制台复制到 `/opt/backupx/web`
|
||||
3. 把默认配置安装到 `/etc/backupx/config.yaml`
|
||||
4. 安装并启用 `backupx.service` systemd 单元
|
||||
5. (可选)生成 Nginx 站点配置 — 参见 [Nginx 反向代理](./nginx)
|
||||
6. 验证首次初始化接口就绪后才报告安装成功
|
||||
|
||||
如果要部署多节点集群,安装后请编辑 `/etc/backupx/config.yaml`,设置远程 Agent 可访问到的 Master URL:
|
||||
|
||||
@@ -63,11 +64,13 @@ After=network.target
|
||||
[Service]
|
||||
Type=simple
|
||||
User=backupx
|
||||
Group=backupx
|
||||
WorkingDirectory=/opt/backupx
|
||||
ExecStart=/opt/backupx/backupx --config /opt/backupx/config.yaml
|
||||
ExecStart=/opt/backupx/bin/backupx -config /etc/backupx/config.yaml
|
||||
Restart=on-failure
|
||||
RestartSec=5s
|
||||
LimitNOFILE=65536
|
||||
RestartSec=5
|
||||
NoNewPrivileges=true
|
||||
LimitNOFILE=65535
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
@@ -79,17 +82,20 @@ WantedBy=multi-user.target
|
||||
sudo systemctl status backupx
|
||||
sudo journalctl -u backupx -f # 实时日志
|
||||
sudo systemctl restart backupx
|
||||
curl -fsS http://127.0.0.1:8340/api/auth/setup/status
|
||||
```
|
||||
|
||||
访问 `http://your-server:8340`,可按需切换到 English,然后在“系统初始化 / System setup”页面创建首个管理员。若监听端口不是默认值,请为安装脚本传入对应的 `HEALTH_URL`。
|
||||
|
||||
## 密码重置
|
||||
|
||||
忘记管理员密码时:
|
||||
|
||||
```bash
|
||||
/opt/backupx/backupx reset-password \
|
||||
/opt/backupx/bin/backupx reset-password \
|
||||
--username admin \
|
||||
--password 'newpass123' \
|
||||
--config /opt/backupx/config.yaml
|
||||
--config /etc/backupx/config.yaml
|
||||
```
|
||||
|
||||
Docker 等效命令:
|
||||
|
||||
@@ -12,12 +12,22 @@ BackupX 支持五种内置备份类型,类型决定了用哪个 runner 执行
|
||||
|
||||
## 文件 / 目录
|
||||
|
||||
打包(可选 gzip)一个或多个文件系统路径。
|
||||
文件任务提供三种备份模式:
|
||||
|
||||
- **全量归档** — 每次生成一份可独立使用的 tar 产物
|
||||
- **差异归档** — 只保存相对当前全量基线的变化,并按周期刷新全量基线
|
||||
- **CDC 去重仓库** — 按稳定的 512 KiB / 1 MiB / 4 MiB 内容边界切块,将新块合并到不可变的 32 MiB pack,每次运行只新增一份小型快照清单
|
||||
|
||||
CDC 仓库会在不同文件、不同快照之间复用相同内容。完整恢复、选择性恢复、完整性校验、下载为 tar、保留策略和垃圾回收都通过仓库索引定位分块。压缩与加密按块执行;启用加密时使用带密钥的块 ID,不暴露明文哈希。
|
||||
|
||||
当前仓库索引采用单写者模型,因此 CDC 模式仅在 Master 本机执行。如需保存多份完整仓库,请在任务中直接多选主存储目标。对象级副本复制会被禁用,因为只有快照清单、没有共享 pack 与索引并不是完整备份。
|
||||
|
||||
文件任务的通用选项:
|
||||
|
||||
- **源路径** 支持多个(UI 中每行一个)
|
||||
- **排除模式** 支持 gitignore 风格的通配符
|
||||
- 可选跟随符号链接、保留权限
|
||||
- 输出单个 `.tar` 或 `.tar.gz`
|
||||
- 全量与差异模式输出 `.tar`、`.tar.gz` 或 `.tar.zst`
|
||||
|
||||
## MySQL
|
||||
|
||||
|
||||
@@ -26,6 +26,27 @@ BackupX 支持 Master-Agent 模式:备份任务可以指定在哪个节点执
|
||||
- **执行** — Agent 复用 BackupRunner(file / mysql / postgresql / sqlite / saphana)并直接上传到存储
|
||||
- **安全** — 每个节点独立 Token;Agent 不持有 Master 的 JWT 密钥或 AES-256 加密密钥
|
||||
|
||||
## 把 B/C/D 服务器集中备份到 M
|
||||
|
||||
Master 作为控制面,每台源服务器安装一个 Agent。任务里的 **源服务器** 决定源路径和数据库工具在哪台机器解析,**存储目标** 决定备份产物最终保留在哪里。
|
||||
|
||||
BackupX 会根据目标类型选择数据路径:
|
||||
|
||||
| 目标 | 数据路径 |
|
||||
| --- | --- |
|
||||
| S3、WebDAV、FTP、云盘或其他网络后端 | Agent 直接流式上传到目标 |
|
||||
| 启用 **远程备份经 Master 中转** 的 `local_disk`(例如通过 NFS 挂载的存储服务器 M) | Agent 通过认证后的 Master API 流式中转,由 Master 写入配置目录 |
|
||||
|
||||
中转过程不会在 Master 上额外落一份完整临时文件。把 Master 本地磁盘中的备份恢复到源 Agent 时会走反向流式通道。Agent 与 Master 之间跨越不可信网络时必须配置 HTTPS。
|
||||
|
||||
典型的 `A → {B,C,D} → M` 拓扑按以下步骤配置:
|
||||
|
||||
1. 在 A 运行 BackupX Master;如果 M 以 NFS 等文件系统提供存储,先把 M 挂载到 A。
|
||||
2. 为该挂载点创建 `local_disk` 目标并保持 **远程备份经 Master 中转** 开启;如果 M 提供 S3/WebDAV,也可直接创建对应网络目标。升级前已有的本地磁盘目标会继续沿用 Agent 本机落盘,手动开启该选项后才切换到中央目录。
|
||||
3. 从 **节点管理** 分别在 B、C、D 安装 Agent。
|
||||
4. 为每台源服务器创建任务,在 **源服务器** 选择 B/C/D,浏览该服务器的路径,再把 M 选为存储目标。相同任务也可用源服务器池标签动态调度。
|
||||
5. 在备份记录中检查逐目标结果。Master 本地磁盘目标会记录 `master_relay` 中转模式,网络后端仍为 `direct` 直传。
|
||||
|
||||
## 一键部署步骤
|
||||
|
||||
### 0. 为生产集群设置 Master 对外 URL
|
||||
@@ -78,7 +99,7 @@ Docker 模式使用同一组环境变量约定:`BACKUPX_AGENT_MASTER`、`BACKU
|
||||
|
||||
### 5. 把任务路由到该节点
|
||||
|
||||
在 **备份任务** 页面新建任务时选择对应节点。任务触发时:
|
||||
在 **备份任务** 页面新建任务时选择对应源服务器。任务触发时:
|
||||
|
||||
- 本机 / 未指定(`nodeId=0`):Master 进程内直接执行
|
||||
- 远程节点:Master 写入命令队列 → Agent 拉取 → Agent 本地执行 → 上传 → 回报
|
||||
|
||||
@@ -19,7 +19,9 @@ BackupX 的目标是接入任何你想放置备份文件的地方。
|
||||
| **Google Drive** | Client ID/Secret + OAuth 授权 |
|
||||
| **WebDAV** | 地址 + 用户名/密码 |
|
||||
| **FTP / FTPS** | 主机 + 端口 + 用户名/密码 |
|
||||
| **本地磁盘** | 目标目录(绝对路径) |
|
||||
| **本地磁盘** | 目标目录(绝对路径)+ 可选的远程 Agent 经 Master 中转 |
|
||||
|
||||
新建本地磁盘目标默认开启 **远程备份经 Master 中转**。开启时,配置目录属于 Master,挂载到 Master 的存储服务器可集中接收多台源 Agent 的备份;如果该路径本就属于各 Agent,请关闭此选项。升级前已有目标保持原来的 Agent 本机落盘行为,只有显式开启后才会切换。
|
||||
|
||||
## Rclone 后端
|
||||
|
||||
|
||||
@@ -55,10 +55,11 @@ sudo ./install.sh # 创建系统用户、安装到 /opt/backupx、配置
|
||||
安装脚本会自动:
|
||||
|
||||
1. 创建 `backupx` 系统用户
|
||||
2. 安装二进制到 `/opt/backupx/backupx`
|
||||
3. 生成 `/opt/backupx/config.yaml`(含安全默认值)
|
||||
2. 安装二进制到 `/opt/backupx/bin/backupx`,并把 Web 控制台安装到 `/opt/backupx/web`
|
||||
3. 生成 `/etc/backupx/config.yaml`(含安全默认值)
|
||||
4. 注册并启用 `backupx.service` systemd 单元
|
||||
5. (可选)配置 Nginx 反向代理
|
||||
6. 等待 `/api/auth/setup/status` 就绪;启动失败时输出 systemd 诊断并返回非零状态
|
||||
|
||||
## 从源码构建
|
||||
|
||||
@@ -67,16 +68,17 @@ sudo ./install.sh # 创建系统用户、安装到 /opt/backupx、配置
|
||||
```bash
|
||||
git clone https://github.com/Awuqing/BackupX.git && cd BackupX
|
||||
make build
|
||||
# 或使用国内镜像加速构建 Docker
|
||||
make docker-cn
|
||||
sudo ./deploy/install.sh
|
||||
```
|
||||
|
||||
`make build` 完成后,二进制位于 `server/bin/backupx`,构建好的 Web UI 位于 `web/dist/`。
|
||||
安装脚本会直接使用这两个路径,不需要 Docker 运行时。如果已有配置修改了默认端口,可覆盖就绪检查地址,例如:`sudo HEALTH_URL=http://127.0.0.1:9000/api/auth/setup/status ./deploy/install.sh`。
|
||||
|
||||
## 验证安装
|
||||
|
||||
```bash
|
||||
backupx --version # 输出如 v1.6.0
|
||||
/opt/backupx/bin/backupx --version
|
||||
curl -fsS http://127.0.0.1:8340/api/auth/setup/status
|
||||
```
|
||||
|
||||
打开浏览器访问 `http://your-server:8340`,会进入初始化管理员账户页面。
|
||||
打开浏览器访问 `http://your-server:8340`,可在右上角选择 **中文** 或 **English**。全新数据库会显示“系统初始化 / System setup”,在这里创建首个管理员用户名和密码。如果没有出现初始化表单,请先重试上面的状态接口,不要直接尝试登录。
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
@@ -125,10 +126,11 @@ type TaskSpec struct {
|
||||
|
||||
// StorageTargetConfig 与 service.AgentStorageTargetConfig 对齐
|
||||
type StorageTargetConfig struct {
|
||||
ID uint `json:"id"`
|
||||
Type string `json:"type"`
|
||||
Name string `json:"name"`
|
||||
Config json.RawMessage `json:"config"`
|
||||
ID uint `json:"id"`
|
||||
Type string `json:"type"`
|
||||
Name string `json:"name"`
|
||||
Config json.RawMessage `json:"config"`
|
||||
TransferMode string `json:"transferMode"`
|
||||
}
|
||||
|
||||
// GetTaskSpec 拉取任务规格
|
||||
@@ -149,6 +151,7 @@ type RecordUpdate struct {
|
||||
Checksum string `json:"checksum,omitempty"`
|
||||
StoragePath string `json:"storagePath,omitempty"`
|
||||
StorageTargetID uint `json:"storageTargetId,omitempty"`
|
||||
StorageTransferMode string `json:"storageTransferMode,omitempty"`
|
||||
StorageUploadResults []StorageResultItem `json:"storageUploadResults,omitempty"`
|
||||
ErrorMessage string `json:"errorMessage,omitempty"`
|
||||
LogAppend string `json:"logAppend,omitempty"`
|
||||
@@ -160,6 +163,7 @@ type StorageResultItem struct {
|
||||
Status string `json:"status"`
|
||||
StoragePath string `json:"storagePath,omitempty"`
|
||||
FileSize int64 `json:"fileSize,omitempty"`
|
||||
TransferMode string `json:"transferMode,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
@@ -169,6 +173,39 @@ func (c *MasterClient) UpdateRecord(ctx context.Context, recordID uint, update R
|
||||
return c.do(ctx, http.MethodPost, path, update, nil)
|
||||
}
|
||||
|
||||
// UploadArtifact streams an artifact through the Master for storage targets
|
||||
// that are not directly reachable from the Agent.
|
||||
func (c *MasterClient) UploadArtifact(ctx context.Context, recordID, targetID uint, objectKey string, size int64, checksum string, reader io.Reader) error {
|
||||
path := fmt.Sprintf("/api/agent/records/%d/artifacts/%d", recordID, targetID)
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPut, c.baseURL+path, reader)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// The executor owns and closes the artifact file. Prevent net/http from
|
||||
// closing that underlying reader when it finishes the request body.
|
||||
req.Body = io.NopCloser(reader)
|
||||
req.ContentLength = size
|
||||
req.Header.Set("Content-Type", "application/octet-stream")
|
||||
req.Header.Set("X-Agent-Token", c.token)
|
||||
req.Header.Set("X-BackupX-Object-Key", objectKey)
|
||||
req.Header.Set("X-BackupX-SHA256", checksum)
|
||||
client := *c.httpClient
|
||||
client.Timeout = 0
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("relay artifact to Master: %w", err)
|
||||
}
|
||||
data, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
||||
closeErr := resp.Body.Close()
|
||||
if readErr != nil || closeErr != nil {
|
||||
return errors.Join(readErr, closeErr)
|
||||
}
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
return fmt.Errorf("relay artifact to Master: http %d: %s", resp.StatusCode, string(data))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// RestoreSpec 与 service.AgentRestoreSpec 对齐
|
||||
type RestoreSpec struct {
|
||||
RestoreRecordID uint `json:"restoreRecordId"`
|
||||
@@ -210,6 +247,27 @@ func (c *MasterClient) GetRestoreSpec(ctx context.Context, restoreRecordID uint)
|
||||
return &spec, nil
|
||||
}
|
||||
|
||||
func (c *MasterClient) DownloadRestoreArtifact(ctx context.Context, restoreRecordID uint) (io.ReadCloser, error) {
|
||||
path := fmt.Sprintf("/api/agent/restores/%d/artifact", restoreRecordID)
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+path, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.Header.Set("X-Agent-Token", c.token)
|
||||
client := *c.httpClient
|
||||
client.Timeout = 0
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("download relayed artifact from Master: %w", err)
|
||||
}
|
||||
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
|
||||
return resp.Body, nil
|
||||
}
|
||||
data, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
||||
closeErr := resp.Body.Close()
|
||||
return nil, errors.Join(fmt.Errorf("download relayed artifact from Master: http %d: %s", resp.StatusCode, string(data)), readErr, closeErr)
|
||||
}
|
||||
|
||||
// UpdateRestore 上报恢复记录的状态/日志
|
||||
func (c *MasterClient) UpdateRestore(ctx context.Context, restoreRecordID uint, update RestoreUpdate) error {
|
||||
path := fmt.Sprintf("/api/agent/restores/%d", restoreRecordID)
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
@@ -123,7 +124,7 @@ func (e *Executor) ExecuteRunTask(ctx context.Context, taskID, recordID uint) er
|
||||
}
|
||||
fileName := filepath.Base(finalPath)
|
||||
fileSize := info.Size()
|
||||
storagePath := backup.BuildStorageKey(spec.Type, startedAt, fileName)
|
||||
storagePath := backup.BuildRecordStorageKey(spec.Type, startedAt, recordID, fileName)
|
||||
|
||||
// 5) 计算 checksum(一次读一次)并上传到所有目标
|
||||
checksum, err := computeFileSHA256(finalPath)
|
||||
@@ -137,13 +138,15 @@ func (e *Executor) ExecuteRunTask(ctx context.Context, taskID, recordID uint) er
|
||||
}
|
||||
uploadResults := make([]StorageResultItem, 0, len(spec.StorageTargets))
|
||||
selectedStorageTargetID := uint(0)
|
||||
selectedStorageTransferMode := ""
|
||||
var uploadErrors []string
|
||||
for _, target := range spec.StorageTargets {
|
||||
if err := e.uploadToTarget(ctx, recordID, target, finalPath, storagePath, fileSize, spec.TaskID); err != nil {
|
||||
if err := e.uploadToTarget(ctx, recordID, target, finalPath, storagePath, fileSize, checksum, spec.TaskID); err != nil {
|
||||
uploadResults = append(uploadResults, StorageResultItem{
|
||||
StorageTargetID: target.ID,
|
||||
StorageTargetName: target.Name,
|
||||
Status: "failed",
|
||||
TransferMode: target.TransferMode,
|
||||
Error: err.Error(),
|
||||
})
|
||||
uploadErrors = append(uploadErrors, fmt.Sprintf("%s: %v", target.Name, err))
|
||||
@@ -152,6 +155,7 @@ func (e *Executor) ExecuteRunTask(ctx context.Context, taskID, recordID uint) er
|
||||
}
|
||||
if selectedStorageTargetID == 0 {
|
||||
selectedStorageTargetID = target.ID
|
||||
selectedStorageTransferMode = target.TransferMode
|
||||
}
|
||||
uploadResults = append(uploadResults, StorageResultItem{
|
||||
StorageTargetID: target.ID,
|
||||
@@ -159,6 +163,7 @@ func (e *Executor) ExecuteRunTask(ctx context.Context, taskID, recordID uint) er
|
||||
Status: "success",
|
||||
StoragePath: storagePath,
|
||||
FileSize: fileSize,
|
||||
TransferMode: target.TransferMode,
|
||||
})
|
||||
e.appendLog(ctx, recordID, fmt.Sprintf("[agent] 已上传到存储目标 %s\n", target.Name))
|
||||
}
|
||||
@@ -179,34 +184,40 @@ func (e *Executor) ExecuteRunTask(ctx context.Context, taskID, recordID uint) er
|
||||
Checksum: checksum,
|
||||
StoragePath: storagePath,
|
||||
StorageTargetID: selectedStorageTargetID,
|
||||
StorageTransferMode: selectedStorageTransferMode,
|
||||
StorageUploadResults: uploadResults,
|
||||
LogAppend: fmt.Sprintf("[agent] 任务完成,总计 %d 字节\n", fileSize),
|
||||
})
|
||||
}
|
||||
|
||||
// uploadToTarget 上传单个目标。为保持简化不做上传级重试(rclone 本身已有 low-level 重试)。
|
||||
func (e *Executor) uploadToTarget(ctx context.Context, recordID uint, target StorageTargetConfig, filePath, objectKey string, fileSize int64, taskID uint) error {
|
||||
var rawConfig map[string]any
|
||||
if len(target.Config) > 0 {
|
||||
// DecodeRawConfig 通过 json 解析
|
||||
if err := jsonUnmarshalMap(target.Config, &rawConfig); err != nil {
|
||||
return fmt.Errorf("parse storage config: %w", err)
|
||||
}
|
||||
}
|
||||
provider, err := e.storageRegistry.Create(ctx, target.Type, rawConfig)
|
||||
if err != nil {
|
||||
return fmt.Errorf("create provider: %w", err)
|
||||
}
|
||||
func (e *Executor) uploadToTarget(ctx context.Context, recordID uint, target StorageTargetConfig, filePath, objectKey string, fileSize int64, checksum string, taskID uint) error {
|
||||
f, err := os.Open(filePath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("open artifact: %w", err)
|
||||
}
|
||||
defer f.Close()
|
||||
if target.TransferMode == storage.TransferModeMasterRelay {
|
||||
uploadErr := e.client.UploadArtifact(ctx, recordID, target.ID, objectKey, fileSize, checksum, f)
|
||||
return errors.Join(uploadErr, f.Close())
|
||||
}
|
||||
var rawConfig map[string]any
|
||||
if len(target.Config) > 0 {
|
||||
// DecodeRawConfig 通过 json 解析
|
||||
if err := jsonUnmarshalMap(target.Config, &rawConfig); err != nil {
|
||||
return errors.Join(fmt.Errorf("parse storage config: %w", err), f.Close())
|
||||
}
|
||||
}
|
||||
provider, err := e.storageRegistry.Create(ctx, target.Type, rawConfig)
|
||||
if err != nil {
|
||||
closeErr := f.Close()
|
||||
return errors.Join(fmt.Errorf("create provider: %w", err), closeErr)
|
||||
}
|
||||
meta := map[string]string{
|
||||
"taskId": fmt.Sprintf("%d", taskID),
|
||||
"recordId": fmt.Sprintf("%d", recordID),
|
||||
}
|
||||
return provider.Upload(ctx, objectKey, f, fileSize, meta)
|
||||
uploadErr := provider.Upload(ctx, objectKey, f, fileSize, meta)
|
||||
return errors.Join(uploadErr, f.Close())
|
||||
}
|
||||
|
||||
// appendLog 追加日志到 Master 记录(尽力而为,失败不中断主流程)
|
||||
@@ -328,7 +339,7 @@ func (e *Executor) DeleteStorageObject(ctx context.Context, targetType string, t
|
||||
// ExecuteRestore 处理 restore_record 命令:拉规格 → 下载 → 解压 → 执行 runner.Restore → 上报结果。
|
||||
//
|
||||
// 与 ExecuteRunTask 对称,但方向相反:
|
||||
// - 下载:通过 spec.Storage 创建 provider → Download(spec.StoragePath)
|
||||
// - 下载:直连共享存储,或通过 Master 中转其本地磁盘对象
|
||||
// - 解密:当前 Agent 不支持加密恢复(密钥未下发),spec.Encrypt=true 会直接失败
|
||||
// - 执行:backup.Registry.Runner(spec.Type).Restore
|
||||
// - 上报:通过 UpdateRestore(status/logAppend)
|
||||
@@ -357,28 +368,31 @@ func (e *Executor) ExecuteRestore(ctx context.Context, restoreRecordID uint) err
|
||||
}
|
||||
defer os.RemoveAll(tmpDir)
|
||||
|
||||
// 1) 创建 storage provider
|
||||
var rawConfig map[string]any
|
||||
if len(spec.Storage.Config) > 0 {
|
||||
if err := jsonUnmarshalMap(spec.Storage.Config, &rawConfig); err != nil {
|
||||
e.reportRestoreFailure(ctx, restoreRecordID, fmt.Sprintf("解析存储配置失败: %v", err))
|
||||
return err
|
||||
}
|
||||
}
|
||||
provider, err := e.storageRegistry.Create(ctx, spec.Storage.Type, rawConfig)
|
||||
if err != nil {
|
||||
e.reportRestoreFailure(ctx, restoreRecordID, fmt.Sprintf("创建存储客户端失败: %v", err))
|
||||
return err
|
||||
}
|
||||
|
||||
// 2) 下载
|
||||
// 1) 下载
|
||||
fileName := spec.FileName
|
||||
if strings.TrimSpace(fileName) == "" {
|
||||
fileName = filepath.Base(spec.StoragePath)
|
||||
}
|
||||
artifactPath := filepath.Join(tmpDir, filepath.Base(fileName))
|
||||
e.appendRestoreLog(ctx, restoreRecordID, fmt.Sprintf("[agent] 下载备份文件 %s\n", spec.StoragePath))
|
||||
reader, err := provider.Download(ctx, spec.StoragePath)
|
||||
var reader io.ReadCloser
|
||||
if spec.Storage.TransferMode == storage.TransferModeMasterRelay {
|
||||
reader, err = e.client.DownloadRestoreArtifact(ctx, restoreRecordID)
|
||||
} else {
|
||||
var rawConfig map[string]any
|
||||
if len(spec.Storage.Config) > 0 {
|
||||
if err := jsonUnmarshalMap(spec.Storage.Config, &rawConfig); err != nil {
|
||||
e.reportRestoreFailure(ctx, restoreRecordID, fmt.Sprintf("解析存储配置失败: %v", err))
|
||||
return err
|
||||
}
|
||||
}
|
||||
provider, providerErr := e.storageRegistry.Create(ctx, spec.Storage.Type, rawConfig)
|
||||
if providerErr != nil {
|
||||
e.reportRestoreFailure(ctx, restoreRecordID, fmt.Sprintf("创建存储客户端失败: %v", providerErr))
|
||||
return providerErr
|
||||
}
|
||||
reader, err = provider.Download(ctx, spec.StoragePath)
|
||||
}
|
||||
if err != nil {
|
||||
e.reportRestoreFailure(ctx, restoreRecordID, fmt.Sprintf("下载备份失败: %v", err))
|
||||
return err
|
||||
@@ -489,8 +503,10 @@ func buildRestoreBackupTaskSpec(spec *RestoreSpec, startedAt time.Time, tempDir
|
||||
}
|
||||
|
||||
// writeReaderToLocal 把 reader 写到本地文件(Agent 侧工具函数)。
|
||||
func writeReaderToLocal(targetPath string, reader io.ReadCloser) error {
|
||||
defer reader.Close()
|
||||
func writeReaderToLocal(targetPath string, reader io.ReadCloser) (err error) {
|
||||
defer func() {
|
||||
err = errors.Join(err, reader.Close())
|
||||
}()
|
||||
if err := os.MkdirAll(filepath.Dir(targetPath), 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -498,9 +514,8 @@ func writeReaderToLocal(targetPath string, reader io.ReadCloser) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer file.Close()
|
||||
_, err = io.Copy(file, reader)
|
||||
return err
|
||||
_, copyErr := io.Copy(file, reader)
|
||||
return errors.Join(copyErr, file.Close())
|
||||
}
|
||||
|
||||
// 辅助函数
|
||||
|
||||
@@ -1,7 +1,10 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
@@ -109,6 +112,150 @@ func TestExecuteRunTaskRecordsPerTargetUploadResults(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteRunTaskRelaysMasterLocalDiskTarget(t *testing.T) {
|
||||
sourceDir := t.TempDir()
|
||||
if err := os.WriteFile(filepath.Join(sourceDir, "index.html"), []byte("centralize me"), 0o644); err != nil {
|
||||
t.Fatalf("WriteFile returned error: %v", err)
|
||||
}
|
||||
var relayed []byte
|
||||
var finalUpdate RecordUpdate
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch {
|
||||
case r.Method == http.MethodGet && r.URL.Path == "/api/agent/tasks/1":
|
||||
writeAgentEnvelope(t, w, TaskSpec{
|
||||
TaskID: 1,
|
||||
Name: "remote-source",
|
||||
Type: "file",
|
||||
SourcePath: sourceDir,
|
||||
Compression: "gzip",
|
||||
StorageTargets: []StorageTargetConfig{{
|
||||
ID: 11, Name: "master-disk", Type: storage.TypeLocalDisk, TransferMode: storage.TransferModeMasterRelay,
|
||||
}},
|
||||
})
|
||||
case r.Method == http.MethodPut && r.URL.Path == "/api/agent/records/99/artifacts/11":
|
||||
body, err := io.ReadAll(r.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("ReadAll relayed body: %v", err)
|
||||
}
|
||||
digest := sha256.Sum256(body)
|
||||
if got := r.Header.Get("X-BackupX-SHA256"); got != fmt.Sprintf("%x", digest[:]) {
|
||||
t.Fatalf("relay checksum header = %q", got)
|
||||
}
|
||||
objectKey := r.Header.Get("X-BackupX-Object-Key")
|
||||
if !strings.Contains(objectKey, "/records/99/") || r.ContentLength != int64(len(body)) {
|
||||
t.Fatalf("invalid relay metadata: key=%q length=%d body=%d", objectKey, r.ContentLength, len(body))
|
||||
}
|
||||
relayed = append([]byte(nil), body...)
|
||||
writeAgentEnvelope(t, w, map[string]string{"status": "ok"})
|
||||
case r.Method == http.MethodPost && r.URL.Path == "/api/agent/records/99":
|
||||
var update RecordUpdate
|
||||
if err := json.NewDecoder(r.Body).Decode(&update); err != nil {
|
||||
t.Fatalf("Decode update returned error: %v", err)
|
||||
}
|
||||
if update.Status != "" {
|
||||
finalUpdate = update
|
||||
}
|
||||
writeAgentEnvelope(t, w, map[string]string{"status": "ok"})
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
executor := NewExecutor(NewMasterClient(server.URL, "token", false), filepath.Join(t.TempDir(), "tmp"))
|
||||
if err := executor.ExecuteRunTask(context.Background(), 1, 99); err != nil {
|
||||
t.Fatalf("ExecuteRunTask returned error: %v", err)
|
||||
}
|
||||
if len(relayed) == 0 {
|
||||
t.Fatal("expected artifact bytes to be streamed through Master")
|
||||
}
|
||||
if finalUpdate.Status != "success" || finalUpdate.StorageTransferMode != storage.TransferModeMasterRelay {
|
||||
t.Fatalf("unexpected final relay update: %#v", finalUpdate)
|
||||
}
|
||||
if len(finalUpdate.StorageUploadResults) != 1 || finalUpdate.StorageUploadResults[0].TransferMode != storage.TransferModeMasterRelay {
|
||||
t.Fatalf("unexpected relay target result: %#v", finalUpdate.StorageUploadResults)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteRestoreDownloadsMasterRelayedArtifact(t *testing.T) {
|
||||
var archive bytes.Buffer
|
||||
tarWriter := tar.NewWriter(&archive)
|
||||
content := []byte("restored through Master")
|
||||
header := &tar.Header{Name: "site/index.html", Mode: 0o644, Size: int64(len(content)), Typeflag: tar.TypeReg}
|
||||
if err := tarWriter.WriteHeader(header); err != nil {
|
||||
t.Fatalf("WriteHeader returned error: %v", err)
|
||||
}
|
||||
if _, err := tarWriter.Write(content); err != nil {
|
||||
t.Fatalf("Write returned error: %v", err)
|
||||
}
|
||||
if err := tarWriter.Close(); err != nil {
|
||||
t.Fatalf("Close returned error: %v", err)
|
||||
}
|
||||
artifact := append([]byte(nil), archive.Bytes()...)
|
||||
digest := sha256.Sum256(artifact)
|
||||
restoreRoot := t.TempDir()
|
||||
restoreSource := filepath.Join(restoreRoot, "site")
|
||||
artifactRequests := 0
|
||||
var finalUpdate RestoreUpdate
|
||||
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch {
|
||||
case r.Method == http.MethodGet && r.URL.Path == "/api/agent/restores/77/spec":
|
||||
writeAgentEnvelope(t, w, RestoreSpec{
|
||||
RestoreRecordID: 77,
|
||||
BackupRecordID: 99,
|
||||
TaskID: 1,
|
||||
TaskName: "remote-source",
|
||||
Type: "file",
|
||||
SourcePath: restoreSource,
|
||||
Storage: StorageTargetConfig{
|
||||
ID: 11, Name: "master-disk", Type: storage.TypeLocalDisk, TransferMode: storage.TransferModeMasterRelay,
|
||||
},
|
||||
StoragePath: "BackupX/file/site.tar",
|
||||
FileName: "site.tar",
|
||||
Checksum: fmt.Sprintf("%x", digest[:]),
|
||||
})
|
||||
case r.Method == http.MethodGet && r.URL.Path == "/api/agent/restores/77/artifact":
|
||||
artifactRequests++
|
||||
if r.Header.Get("X-Agent-Token") != "token" {
|
||||
t.Fatalf("missing Agent token on relay download")
|
||||
}
|
||||
w.Header().Set("Content-Length", fmt.Sprintf("%d", len(artifact)))
|
||||
_, _ = w.Write(artifact)
|
||||
case r.Method == http.MethodPost && r.URL.Path == "/api/agent/restores/77":
|
||||
var update RestoreUpdate
|
||||
if err := json.NewDecoder(r.Body).Decode(&update); err != nil {
|
||||
t.Fatalf("Decode update returned error: %v", err)
|
||||
}
|
||||
if update.Status != "" {
|
||||
finalUpdate = update
|
||||
}
|
||||
writeAgentEnvelope(t, w, map[string]string{"status": "ok"})
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
executor := NewExecutor(NewMasterClient(server.URL, "token", false), filepath.Join(t.TempDir(), "tmp"))
|
||||
if err := executor.ExecuteRestore(context.Background(), 77); err != nil {
|
||||
t.Fatalf("ExecuteRestore returned error: %v", err)
|
||||
}
|
||||
if artifactRequests != 1 {
|
||||
t.Fatalf("expected one relay artifact request, got %d", artifactRequests)
|
||||
}
|
||||
restored, err := os.ReadFile(filepath.Join(restoreRoot, "site", "index.html"))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFile returned error: %v", err)
|
||||
}
|
||||
if string(restored) != string(content) {
|
||||
t.Fatalf("restored content = %q, want %q", restored, content)
|
||||
}
|
||||
if finalUpdate.Status != "success" {
|
||||
t.Fatalf("unexpected final restore update: %#v", finalUpdate)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteRunTaskReportsPerTargetUploadResultsWhenAllTargetsFail(t *testing.T) {
|
||||
sourceDir := t.TempDir()
|
||||
if err := os.WriteFile(filepath.Join(sourceDir, "index.html"), []byte("hello"), 0o644); err != nil {
|
||||
|
||||
@@ -84,7 +84,7 @@ func New(ctx context.Context, cfg config.Config, version string) (*Application,
|
||||
// nodeRepo 在下方 Cluster 节点管理区块才实例化,这里延后注入
|
||||
backupRunnerRegistry := backup.NewRegistry(backup.NewFileRunner(), backup.NewSQLiteRunner(), backup.NewMySQLRunner(nil), backup.NewPostgreSQLRunner(nil), backup.NewSAPHANARunner(nil), backup.NewMongoDBRunner(nil))
|
||||
logHub := backup.NewLogHub()
|
||||
retentionService := backupretention.NewService(backupRecordRepo)
|
||||
retentionService := backupretention.NewService(backupRecordRepo, configCipher.Key())
|
||||
notifyRegistry := notify.NewRegistry(notify.NewEmailNotifier(), notify.NewWebhookNotifier(), notify.NewTelegramNotifier())
|
||||
notificationService := service.NewNotificationService(notificationRepo, notifyRegistry, configCipher)
|
||||
authService.SetNotificationService(notificationService)
|
||||
@@ -135,7 +135,7 @@ func New(ctx context.Context, cfg config.Config, version string) (*Application,
|
||||
// Agent 协议服务:命令队列 + 任务下发 + 记录上报
|
||||
agentCmdRepo := repository.NewAgentCommandRepository(db)
|
||||
nodeService.SetAgentCommandRepository(agentCmdRepo)
|
||||
agentService := service.NewAgentService(nodeRepo, backupTaskRepo, backupRecordRepo, storageTargetRepo, agentCmdRepo, configCipher)
|
||||
agentService := service.NewAgentService(nodeRepo, backupTaskRepo, backupRecordRepo, storageTargetRepo, agentCmdRepo, configCipher, storageRegistry)
|
||||
agentService.SetRestoreRepository(restoreRecordRepo)
|
||||
agentService.StartCommandTimeoutMonitor(ctx, 30*time.Second, 10*time.Minute)
|
||||
|
||||
|
||||
1356
server/internal/backup/repository.go
Normal file
1356
server/internal/backup/repository.go
Normal file
File diff suppressed because it is too large
Load Diff
131
server/internal/backup/repository_chunker.go
Normal file
131
server/internal/backup/repository_chunker.go
Normal file
@@ -0,0 +1,131 @@
|
||||
package backup
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
)
|
||||
|
||||
const (
|
||||
repositoryChunkMin = 512 << 10
|
||||
repositoryChunkAvg = 1 << 20
|
||||
repositoryChunkMax = 4 << 20
|
||||
)
|
||||
|
||||
// contentDefinedChunker implements the normalized FastCDC cut-point strategy.
|
||||
// The rolling Gear hash only retains the latest 64 bytes through uint64
|
||||
// overflow, so chunk boundaries re-synchronize after insertions or deletions.
|
||||
type contentDefinedChunker struct {
|
||||
minSize int
|
||||
avgSize int
|
||||
maxSize int
|
||||
smallMask uint64
|
||||
largeMask uint64
|
||||
gear [256]uint64
|
||||
}
|
||||
|
||||
func newContentDefinedChunker() *contentDefinedChunker {
|
||||
chunker := &contentDefinedChunker{
|
||||
minSize: repositoryChunkMin,
|
||||
avgSize: repositoryChunkAvg,
|
||||
maxSize: repositoryChunkMax,
|
||||
smallMask: (1 << 21) - 1,
|
||||
largeMask: (1 << 19) - 1,
|
||||
}
|
||||
|
||||
// SplitMix64 produces a stable, well-distributed Gear table. The seed and
|
||||
// generation algorithm are part of repository format v1 and must not change.
|
||||
seed := uint64(0x6a09e667f3bcc909)
|
||||
for i := range chunker.gear {
|
||||
seed += 0x9e3779b97f4a7c15
|
||||
value := seed
|
||||
value = (value ^ (value >> 30)) * 0xbf58476d1ce4e5b9
|
||||
value = (value ^ (value >> 27)) * 0x94d049bb133111eb
|
||||
chunker.gear[i] = value ^ (value >> 31)
|
||||
}
|
||||
return chunker
|
||||
}
|
||||
|
||||
func (c *contentDefinedChunker) Split(ctx context.Context, reader io.Reader, emit func([]byte) error) error {
|
||||
if reader == nil || emit == nil {
|
||||
return fmt.Errorf("chunk reader and emitter are required")
|
||||
}
|
||||
|
||||
pending := make([]byte, 0, c.maxSize+(256<<10))
|
||||
readBuffer := make([]byte, 256<<10)
|
||||
eof := false
|
||||
for {
|
||||
if !eof {
|
||||
readCount, readErr := reader.Read(readBuffer)
|
||||
if readCount > 0 {
|
||||
pending = append(pending, readBuffer[:readCount]...)
|
||||
}
|
||||
switch readErr {
|
||||
case nil:
|
||||
case io.EOF:
|
||||
eof = true
|
||||
default:
|
||||
return fmt.Errorf("read source for chunking: %w", readErr)
|
||||
}
|
||||
if readCount == 0 && readErr == nil {
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
for len(pending) > 0 {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
cut := c.findCutPoint(pending, eof)
|
||||
if cut == 0 {
|
||||
break
|
||||
}
|
||||
chunk := make([]byte, cut)
|
||||
copy(chunk, pending[:cut])
|
||||
if err := emit(chunk); err != nil {
|
||||
return err
|
||||
}
|
||||
copy(pending, pending[cut:])
|
||||
pending = pending[:len(pending)-cut]
|
||||
}
|
||||
|
||||
if eof {
|
||||
if len(pending) != 0 {
|
||||
return fmt.Errorf("chunker stopped with %d buffered bytes", len(pending))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *contentDefinedChunker) findCutPoint(data []byte, eof bool) int {
|
||||
if len(data) < c.minSize {
|
||||
if eof {
|
||||
return len(data)
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
limit := len(data)
|
||||
if limit > c.maxSize {
|
||||
limit = c.maxSize
|
||||
}
|
||||
var hash uint64
|
||||
for index := c.minSize; index < limit; index++ {
|
||||
hash = (hash << 1) + c.gear[data[index]]
|
||||
mask := c.largeMask
|
||||
if index < c.avgSize {
|
||||
mask = c.smallMask
|
||||
}
|
||||
if hash&mask == 0 {
|
||||
return index + 1
|
||||
}
|
||||
}
|
||||
if len(data) >= c.maxSize {
|
||||
return c.maxSize
|
||||
}
|
||||
if eof {
|
||||
return len(data)
|
||||
}
|
||||
return 0
|
||||
}
|
||||
414
server/internal/backup/repository_test.go
Normal file
414
server/internal/backup/repository_test.go
Normal file
@@ -0,0 +1,414 @@
|
||||
package backup
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"math/rand"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"backupx/server/internal/storage"
|
||||
)
|
||||
|
||||
func TestContentDefinedChunkerResynchronizesAfterInsertion(t *testing.T) {
|
||||
source := make([]byte, 8<<20)
|
||||
if _, err := rand.New(rand.NewSource(42)).Read(source); err != nil {
|
||||
t.Fatalf("generate source: %v", err)
|
||||
}
|
||||
modified := make([]byte, 0, len(source)+4096)
|
||||
modified = append(modified, source[:2<<20]...)
|
||||
modified = append(modified, bytes.Repeat([]byte("inserted"), 512)...)
|
||||
modified = append(modified, source[2<<20:]...)
|
||||
|
||||
chunker := newContentDefinedChunker()
|
||||
collect := func(data []byte) map[string]struct{} {
|
||||
t.Helper()
|
||||
ids := make(map[string]struct{})
|
||||
err := chunker.Split(context.Background(), bytes.NewReader(data), func(chunk []byte) error {
|
||||
digest := sha256.Sum256(chunk)
|
||||
ids[fmt.Sprintf("%x", digest[:])] = struct{}{}
|
||||
if len(chunk) > repositoryChunkMax {
|
||||
return fmt.Errorf("chunk exceeds maximum: %d", len(chunk))
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("split chunks: %v", err)
|
||||
}
|
||||
return ids
|
||||
}
|
||||
originalChunks := collect(source)
|
||||
modifiedChunks := collect(modified)
|
||||
shared := 0
|
||||
for chunkID := range originalChunks {
|
||||
if _, ok := modifiedChunks[chunkID]; ok {
|
||||
shared++
|
||||
}
|
||||
}
|
||||
if shared < len(originalChunks)/2 {
|
||||
t.Fatalf("content-defined boundaries did not resynchronize: shared=%d original=%d", shared, len(originalChunks))
|
||||
}
|
||||
}
|
||||
|
||||
func TestRepositoryRoundTripDedupAndPrune(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
tempDir := t.TempDir()
|
||||
sourceDir := filepath.Join(tempDir, "dataset")
|
||||
if err := os.MkdirAll(filepath.Join(sourceDir, "empty"), 0o755); err != nil {
|
||||
t.Fatalf("create source: %v", err)
|
||||
}
|
||||
original := make([]byte, 6<<20)
|
||||
if _, err := rand.New(rand.NewSource(7)).Read(original); err != nil {
|
||||
t.Fatalf("generate fixture: %v", err)
|
||||
}
|
||||
primaryPath := filepath.Join(sourceDir, "primary.bin")
|
||||
duplicatePath := filepath.Join(sourceDir, "duplicate.bin")
|
||||
if err := os.WriteFile(primaryPath, original, 0o640); err != nil {
|
||||
t.Fatalf("write primary: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(duplicatePath, original, 0o640); err != nil {
|
||||
t.Fatalf("write duplicate: %v", err)
|
||||
}
|
||||
|
||||
key := sha256.Sum256([]byte("repository-test-key"))
|
||||
store := NewRepositoryStore(key[:])
|
||||
provider := newMemoryRepositoryProvider()
|
||||
task := TaskSpec{
|
||||
ID: 12,
|
||||
Name: "repository-test",
|
||||
Type: "file",
|
||||
SourcePaths: []string{sourceDir},
|
||||
Compression: "zstd",
|
||||
Encrypt: true,
|
||||
StartedAt: time.Date(2026, 8, 6, 1, 2, 3, 0, time.UTC),
|
||||
TempDir: tempDir,
|
||||
}
|
||||
|
||||
firstPlan, err := store.BuildPlan(ctx, task, NopLogWriter{})
|
||||
if err != nil {
|
||||
t.Fatalf("build first plan: %v", err)
|
||||
}
|
||||
firstKey := store.SnapshotKey(task.ID, 1, task.StartedAt)
|
||||
firstResult, err := store.Upload(ctx, provider, firstPlan, firstKey)
|
||||
if closeErr := firstPlan.Close(); closeErr != nil {
|
||||
t.Fatalf("close first plan: %v", closeErr)
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatalf("upload first snapshot: %v", err)
|
||||
}
|
||||
if firstResult.NewChunks == 0 || firstResult.UniqueChunks == 0 {
|
||||
t.Fatalf("first upload did not create chunks: %+v", firstResult)
|
||||
}
|
||||
if firstPlan.UniqueSize >= firstPlan.LogicalSize {
|
||||
t.Fatalf("duplicate file was not deduplicated within plan: unique=%d logical=%d", firstPlan.UniqueSize, firstPlan.LogicalSize)
|
||||
}
|
||||
|
||||
modified := make([]byte, 0, len(original)+4096)
|
||||
modified = append(modified, original[:2<<20]...)
|
||||
modified = append(modified, bytes.Repeat([]byte("changed!"), 512)...)
|
||||
modified = append(modified, original[2<<20:]...)
|
||||
if err := os.WriteFile(primaryPath, modified, 0o640); err != nil {
|
||||
t.Fatalf("modify primary: %v", err)
|
||||
}
|
||||
task.StartedAt = task.StartedAt.Add(time.Hour)
|
||||
secondPlan, err := store.BuildPlan(ctx, task, NopLogWriter{})
|
||||
if err != nil {
|
||||
t.Fatalf("build second plan: %v", err)
|
||||
}
|
||||
secondKey := store.SnapshotKey(task.ID, 2, task.StartedAt)
|
||||
secondResult, err := store.Upload(ctx, provider, secondPlan, secondKey)
|
||||
if closeErr := secondPlan.Close(); closeErr != nil {
|
||||
t.Fatalf("close second plan: %v", closeErr)
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatalf("upload second snapshot: %v", err)
|
||||
}
|
||||
if secondResult.ReusedBytes <= secondResult.LogicalSize/3 {
|
||||
t.Fatalf("second snapshot reused too little data: %+v", secondResult)
|
||||
}
|
||||
if secondResult.UploadedBytes >= secondResult.LogicalSize {
|
||||
t.Fatalf("incremental upload was not smaller than logical data: %+v", secondResult)
|
||||
}
|
||||
|
||||
verify, err := store.Verify(ctx, provider, secondKey, secondResult.Checksum)
|
||||
if err != nil {
|
||||
t.Fatalf("verify repository: %v", err)
|
||||
}
|
||||
if verify.Chunks == 0 || verify.Bytes == 0 {
|
||||
t.Fatalf("empty verification result: %+v", verify)
|
||||
}
|
||||
|
||||
restoreRoot := filepath.Join(tempDir, "restore")
|
||||
restoreTask := task
|
||||
restoreTask.RestoreTargetPath = restoreRoot
|
||||
if err := store.Restore(ctx, provider, secondKey, strings.Repeat("0", sha256.Size*2), restoreTask, NopLogWriter{}); err == nil {
|
||||
t.Fatal("restore accepted a mismatched snapshot checksum")
|
||||
}
|
||||
if err := store.Restore(ctx, provider, secondKey, "", restoreTask, NopLogWriter{}); err == nil {
|
||||
t.Fatal("restore accepted a missing snapshot checksum")
|
||||
}
|
||||
if err := store.Restore(ctx, provider, secondKey, secondResult.Checksum, restoreTask, NopLogWriter{}); err != nil {
|
||||
t.Fatalf("restore snapshot: %v", err)
|
||||
}
|
||||
restored, err := os.ReadFile(filepath.Join(restoreRoot, filepath.Base(sourceDir), "primary.bin"))
|
||||
if err != nil {
|
||||
t.Fatalf("read restored primary: %v", err)
|
||||
}
|
||||
if !bytes.Equal(restored, modified) {
|
||||
t.Fatalf("restored primary differs from source")
|
||||
}
|
||||
if info, err := os.Stat(filepath.Join(restoreRoot, filepath.Base(sourceDir), "empty")); err != nil || !info.IsDir() {
|
||||
t.Fatalf("empty directory was not restored: info=%v err=%v", info, err)
|
||||
}
|
||||
|
||||
if err := provider.Delete(ctx, firstKey); err != nil {
|
||||
t.Fatalf("delete first snapshot: %v", err)
|
||||
}
|
||||
if _, err := store.Prune(ctx, provider); err != nil {
|
||||
t.Fatalf("prune with live snapshot: %v", err)
|
||||
}
|
||||
if err := provider.Delete(ctx, secondKey); err != nil {
|
||||
t.Fatalf("delete second snapshot: %v", err)
|
||||
}
|
||||
pruned, err := store.Prune(ctx, provider)
|
||||
if err != nil {
|
||||
t.Fatalf("prune empty repository: %v", err)
|
||||
}
|
||||
if pruned.DeletedPacks == 0 || pruned.DeletedIndexes == 0 {
|
||||
t.Fatalf("prune did not reclaim repository data: %+v", pruned)
|
||||
}
|
||||
if objects, err := provider.List(ctx, repositoryPackPrefix); err != nil || len(objects) != 0 {
|
||||
t.Fatalf("packs remain after prune: objects=%v err=%v", objects, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRepositoryRestoreRejectsUnsafeSnapshotMetadata(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
entries []repositoryEntry
|
||||
}{
|
||||
{
|
||||
name: "path traversal",
|
||||
entries: []repositoryEntry{{Path: "../escape", Kind: "directory", Mode: 0o755}},
|
||||
},
|
||||
{
|
||||
name: "entry below symlink",
|
||||
entries: []repositoryEntry{
|
||||
{Path: "link", Kind: "symlink", Mode: 0o777, LinkTarget: "inside"},
|
||||
{Path: "link/payload", Kind: "file", Mode: 0o600, Size: 1, Chunks: []string{"p-" + strings.Repeat("0", sha256.Size*2)}},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "escaping symlink target",
|
||||
entries: []repositoryEntry{{Path: "escape", Kind: "symlink", Mode: 0o777, LinkTarget: "../outside"}},
|
||||
},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := NewRepositoryStore(nil)
|
||||
provider := newMemoryRepositoryProvider()
|
||||
snapshot := repositorySnapshot{
|
||||
Version: repositoryFormatVersion,
|
||||
TaskID: 1,
|
||||
CreatedAt: time.Now().UTC(),
|
||||
Compression: "none",
|
||||
Entries: tc.entries,
|
||||
}
|
||||
data, err := store.encodeSnapshot(snapshot)
|
||||
if err != nil {
|
||||
t.Fatalf("encodeSnapshot returned error: %v", err)
|
||||
}
|
||||
key := store.SnapshotKey(1, 1, snapshot.CreatedAt)
|
||||
if err := provider.Upload(ctx, key, bytes.NewReader(data), int64(len(data)), nil); err != nil {
|
||||
t.Fatalf("Upload snapshot returned error: %v", err)
|
||||
}
|
||||
digest := sha256.Sum256(data)
|
||||
task := TaskSpec{SourcePath: filepath.Join(t.TempDir(), "source"), RestoreTargetPath: filepath.Join(t.TempDir(), "restore")}
|
||||
if err := store.Restore(ctx, provider, key, fmt.Sprintf("%x", digest[:]), task, NopLogWriter{}); err == nil {
|
||||
t.Fatal("restore accepted unsafe snapshot metadata")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRepositoryRestorePreservesDirectoryWhenSnapshotContainsSymlink(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := NewRepositoryStore(nil)
|
||||
provider := newMemoryRepositoryProvider()
|
||||
snapshot := repositorySnapshot{
|
||||
Version: repositoryFormatVersion,
|
||||
TaskID: 1,
|
||||
CreatedAt: time.Now().UTC(),
|
||||
Compression: "none",
|
||||
Entries: []repositoryEntry{{Path: "link", Kind: "symlink", Mode: 0o777, LinkTarget: "inside"}},
|
||||
}
|
||||
data, err := store.encodeSnapshot(snapshot)
|
||||
if err != nil {
|
||||
t.Fatalf("encodeSnapshot returned error: %v", err)
|
||||
}
|
||||
key := store.SnapshotKey(1, 1, snapshot.CreatedAt)
|
||||
if err := provider.Upload(ctx, key, bytes.NewReader(data), int64(len(data)), nil); err != nil {
|
||||
t.Fatalf("Upload snapshot returned error: %v", err)
|
||||
}
|
||||
digest := sha256.Sum256(data)
|
||||
restoreRoot := filepath.Join(t.TempDir(), "restore")
|
||||
markerPath := filepath.Join(restoreRoot, "link", "keep.txt")
|
||||
if err := os.MkdirAll(filepath.Dir(markerPath), 0o755); err != nil {
|
||||
t.Fatalf("MkdirAll marker parent: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(markerPath, []byte("keep"), 0o600); err != nil {
|
||||
t.Fatalf("WriteFile marker: %v", err)
|
||||
}
|
||||
task := TaskSpec{SourcePath: filepath.Join(t.TempDir(), "source"), RestoreTargetPath: restoreRoot}
|
||||
if err := store.Restore(ctx, provider, key, fmt.Sprintf("%x", digest[:]), task, NopLogWriter{}); err == nil {
|
||||
t.Fatal("restore replaced an existing directory with a symlink")
|
||||
}
|
||||
if data, err := os.ReadFile(markerPath); err != nil || string(data) != "keep" {
|
||||
t.Fatalf("existing directory content changed: data=%q err=%v", data, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRepositoryRejectsOversizedChunkLocation(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := NewRepositoryStore(nil)
|
||||
provider := newMemoryRepositoryProvider()
|
||||
chunkID := "p-" + strings.Repeat("0", sha256.Size*2)
|
||||
packID := strings.Repeat("a", sha256.Size*2)
|
||||
segment := repositoryIndexSegment{
|
||||
Version: repositoryFormatVersion,
|
||||
Pack: fmt.Sprintf("%s/%s/%s.pack", repositoryPackPrefix, packID[:2], packID),
|
||||
Chunks: map[string]repositoryChunkLocation{
|
||||
chunkID: {Pack: fmt.Sprintf("%s/%s/%s.pack", repositoryPackPrefix, packID[:2], packID), Offset: 0, Length: repositoryMaxEncoded + 1, PlainSize: 1, Compression: "none"},
|
||||
},
|
||||
}
|
||||
data, err := json.Marshal(segment)
|
||||
if err != nil {
|
||||
t.Fatalf("Marshal index returned error: %v", err)
|
||||
}
|
||||
indexKey := fmt.Sprintf("%s/%s.json", repositoryIndexPrefix, packID)
|
||||
if err := provider.Upload(ctx, indexKey, bytes.NewReader(data), int64(len(data)), nil); err != nil {
|
||||
t.Fatalf("Upload index returned error: %v", err)
|
||||
}
|
||||
if _, err := store.loadIndex(ctx, provider); err == nil {
|
||||
t.Fatal("loadIndex accepted an oversized encoded chunk")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRepositoryRejectsIndexPackMismatch(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := NewRepositoryStore(nil)
|
||||
provider := newMemoryRepositoryProvider()
|
||||
chunkID := "p-" + strings.Repeat("0", sha256.Size*2)
|
||||
indexID := strings.Repeat("a", sha256.Size*2)
|
||||
otherPackID := strings.Repeat("b", sha256.Size*2)
|
||||
expectedPack := fmt.Sprintf("%s/%s/%s.pack", repositoryPackPrefix, indexID[:2], indexID)
|
||||
segment := repositoryIndexSegment{
|
||||
Version: repositoryFormatVersion,
|
||||
Pack: expectedPack,
|
||||
Chunks: map[string]repositoryChunkLocation{
|
||||
chunkID: {
|
||||
Pack: fmt.Sprintf("%s/%s/%s.pack", repositoryPackPrefix, otherPackID[:2], otherPackID),
|
||||
Offset: 0,
|
||||
Length: 1,
|
||||
PlainSize: 1,
|
||||
Compression: "none",
|
||||
},
|
||||
},
|
||||
}
|
||||
data, err := json.Marshal(segment)
|
||||
if err != nil {
|
||||
t.Fatalf("Marshal index returned error: %v", err)
|
||||
}
|
||||
indexKey := fmt.Sprintf("%s/%s.json", repositoryIndexPrefix, indexID)
|
||||
if err := provider.Upload(ctx, indexKey, bytes.NewReader(data), int64(len(data)), nil); err != nil {
|
||||
t.Fatalf("Upload index returned error: %v", err)
|
||||
}
|
||||
if _, err := store.loadIndex(ctx, provider); err == nil {
|
||||
t.Fatal("loadIndex accepted a chunk location pointing to a different pack")
|
||||
}
|
||||
}
|
||||
|
||||
type memoryRepositoryProvider struct {
|
||||
mu sync.RWMutex
|
||||
objects map[string][]byte
|
||||
times map[string]time.Time
|
||||
}
|
||||
|
||||
func newMemoryRepositoryProvider() *memoryRepositoryProvider {
|
||||
return &memoryRepositoryProvider{objects: make(map[string][]byte), times: make(map[string]time.Time)}
|
||||
}
|
||||
|
||||
func (p *memoryRepositoryProvider) Type() storage.ProviderType { return "memory" }
|
||||
func (p *memoryRepositoryProvider) TestConnection(context.Context) error { return nil }
|
||||
|
||||
func (p *memoryRepositoryProvider) Upload(_ context.Context, key string, reader io.Reader, size int64, _ map[string]string) error {
|
||||
data, err := io.ReadAll(reader)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if int64(len(data)) != size {
|
||||
return fmt.Errorf("size mismatch for %s: %d != %d", key, len(data), size)
|
||||
}
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
p.objects[key] = append([]byte(nil), data...)
|
||||
p.times[key] = time.Now().UTC()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *memoryRepositoryProvider) Download(_ context.Context, key string) (io.ReadCloser, error) {
|
||||
p.mu.RLock()
|
||||
defer p.mu.RUnlock()
|
||||
data, ok := p.objects[key]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("object %s not found", key)
|
||||
}
|
||||
return io.NopCloser(bytes.NewReader(append([]byte(nil), data...))), nil
|
||||
}
|
||||
|
||||
func (p *memoryRepositoryProvider) DownloadRange(_ context.Context, key string, offset, length int64) (io.ReadCloser, error) {
|
||||
p.mu.RLock()
|
||||
defer p.mu.RUnlock()
|
||||
data, ok := p.objects[key]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("object %s not found", key)
|
||||
}
|
||||
if offset < 0 || length <= 0 || offset+length > int64(len(data)) {
|
||||
return nil, fmt.Errorf("invalid range %d:%d for %s", offset, length, key)
|
||||
}
|
||||
return io.NopCloser(bytes.NewReader(append([]byte(nil), data[offset:offset+length]...))), nil
|
||||
}
|
||||
|
||||
func (p *memoryRepositoryProvider) Delete(_ context.Context, key string) error {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
if _, ok := p.objects[key]; !ok {
|
||||
return fmt.Errorf("object %s not found", key)
|
||||
}
|
||||
delete(p.objects, key)
|
||||
delete(p.times, key)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *memoryRepositoryProvider) List(_ context.Context, prefix string) ([]storage.ObjectInfo, error) {
|
||||
p.mu.RLock()
|
||||
defer p.mu.RUnlock()
|
||||
result := make([]storage.ObjectInfo, 0)
|
||||
for key, data := range p.objects {
|
||||
if strings.HasPrefix(key, prefix) {
|
||||
result = append(result, storage.ObjectInfo{Key: key, Size: int64(len(data)), UpdatedAt: p.times[key]})
|
||||
}
|
||||
}
|
||||
sort.Slice(result, func(i, j int) bool { return result[i].Key < result[j].Key })
|
||||
return result, nil
|
||||
}
|
||||
@@ -2,11 +2,13 @@ package retention
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"backupx/server/internal/backup"
|
||||
"backupx/server/internal/model"
|
||||
"backupx/server/internal/repository"
|
||||
"backupx/server/internal/storage"
|
||||
@@ -40,16 +42,48 @@ type CleanupResult struct {
|
||||
Warnings []string
|
||||
}
|
||||
|
||||
type Service struct {
|
||||
records repository.BackupRecordRepository
|
||||
now func() time.Time
|
||||
type cleanupObject struct {
|
||||
targetID uint
|
||||
path string
|
||||
}
|
||||
|
||||
func NewService(records repository.BackupRecordRepository) *Service {
|
||||
return &Service{records: records, now: func() time.Time { return time.Now().UTC() }}
|
||||
type storedUploadResult struct {
|
||||
StorageTargetID uint `json:"storageTargetId"`
|
||||
Status string `json:"status"`
|
||||
StoragePath string `json:"storagePath"`
|
||||
}
|
||||
|
||||
type Service struct {
|
||||
records repository.BackupRecordRepository
|
||||
repositoryKey []byte
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
func NewService(records repository.BackupRecordRepository, repositoryKey ...[]byte) *Service {
|
||||
var key []byte
|
||||
if len(repositoryKey) > 0 {
|
||||
key = append([]byte(nil), repositoryKey[0]...)
|
||||
}
|
||||
return &Service{records: records, repositoryKey: key, now: func() time.Time { return time.Now().UTC() }}
|
||||
}
|
||||
|
||||
func (s *Service) Cleanup(ctx context.Context, task *model.BackupTask, provider storage.StorageProvider) (*CleanupResult, error) {
|
||||
return s.cleanup(ctx, task, func(uint) (storage.StorageProvider, bool) {
|
||||
return provider, provider != nil
|
||||
})
|
||||
}
|
||||
|
||||
// CleanupProviders applies one retention decision to every successful copy of
|
||||
// a record before deleting its database row. This prevents multi-target tasks
|
||||
// from leaving stale objects after the first target removes the shared record.
|
||||
func (s *Service) CleanupProviders(ctx context.Context, task *model.BackupTask, providers map[uint]storage.StorageProvider) (*CleanupResult, error) {
|
||||
return s.cleanup(ctx, task, func(targetID uint) (storage.StorageProvider, bool) {
|
||||
provider, ok := providers[targetID]
|
||||
return provider, ok && provider != nil
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Service) cleanup(ctx context.Context, task *model.BackupTask, resolveProvider func(uint) (storage.StorageProvider, bool)) (*CleanupResult, error) {
|
||||
if task == nil {
|
||||
return nil, fmt.Errorf("backup task is required")
|
||||
}
|
||||
@@ -67,17 +101,35 @@ func (s *Service) Cleanup(ctx context.Context, task *model.BackupTask, provider
|
||||
// 差异链保护:保留仍被存活差异依赖的全量,避免删除基线后差异无法恢复。
|
||||
candidates = protectDifferentialBases(records, candidates)
|
||||
result := &CleanupResult{}
|
||||
repositoryProviders := make(map[uint]storage.StorageProvider)
|
||||
touchedProviders := make(map[uint]storage.StorageProvider)
|
||||
for _, record := range candidates {
|
||||
if strings.TrimSpace(record.StoragePath) != "" {
|
||||
if provider == nil {
|
||||
result.Warnings = append(result.Warnings, fmt.Sprintf("record %d missing storage provider for cleanup", record.ID))
|
||||
objects, objectErr := cleanupObjectsForRecord(record)
|
||||
if objectErr != nil {
|
||||
result.Warnings = append(result.Warnings, fmt.Sprintf("decode storage copies for record %d failed: %v", record.ID, objectErr))
|
||||
continue
|
||||
}
|
||||
allObjectsDeleted := true
|
||||
for _, object := range objects {
|
||||
provider, ok := resolveProvider(object.targetID)
|
||||
if !ok {
|
||||
result.Warnings = append(result.Warnings, fmt.Sprintf("record %d missing storage provider %d for cleanup", record.ID, object.targetID))
|
||||
allObjectsDeleted = false
|
||||
continue
|
||||
}
|
||||
if err := provider.Delete(ctx, record.StoragePath); err != nil {
|
||||
result.Warnings = append(result.Warnings, fmt.Sprintf("delete storage object %s failed: %v", record.StoragePath, err))
|
||||
if err := provider.Delete(ctx, object.path); err != nil {
|
||||
result.Warnings = append(result.Warnings, fmt.Sprintf("delete storage object %s from target %d failed: %v", object.path, object.targetID, err))
|
||||
allObjectsDeleted = false
|
||||
continue
|
||||
}
|
||||
result.DeletedObjects++
|
||||
touchedProviders[object.targetID] = provider
|
||||
if record.BackupKind == model.BackupKindRepository {
|
||||
repositoryProviders[object.targetID] = provider
|
||||
}
|
||||
}
|
||||
if !allObjectsDeleted {
|
||||
continue
|
||||
}
|
||||
if err := s.records.Delete(ctx, record.ID); err != nil {
|
||||
result.Warnings = append(result.Warnings, fmt.Sprintf("delete backup record %d failed: %v", record.ID, err))
|
||||
@@ -85,13 +137,25 @@ func (s *Service) Cleanup(ctx context.Context, task *model.BackupTask, provider
|
||||
}
|
||||
result.DeletedRecords++
|
||||
}
|
||||
for targetID, provider := range repositoryProviders {
|
||||
pruned, pruneErr := backup.NewRepositoryStore(s.repositoryKey).Prune(ctx, provider)
|
||||
if pruneErr != nil {
|
||||
result.Warnings = append(result.Warnings, fmt.Sprintf("prune CDC repository on target %d failed: %v", targetID, pruneErr))
|
||||
} else {
|
||||
result.DeletedObjects += pruned.DeletedPacks + pruned.DeletedIndexes
|
||||
}
|
||||
}
|
||||
|
||||
// 清理空目录:收集被删除文件的父目录,尝试移除空目录
|
||||
if dirCleaner, ok := provider.(storage.StorageDirCleaner); ok && result.DeletedObjects > 0 {
|
||||
for targetID, provider := range touchedProviders {
|
||||
dirCleaner, ok := provider.(storage.StorageDirCleaner)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
prefixes := collectDirPrefixes(candidates)
|
||||
for _, prefix := range prefixes {
|
||||
if err := dirCleaner.RemoveEmptyDirs(ctx, prefix); err != nil {
|
||||
result.Warnings = append(result.Warnings, fmt.Sprintf("cleanup empty dirs for %s: %v", prefix, err))
|
||||
result.Warnings = append(result.Warnings, fmt.Sprintf("cleanup empty dirs for %s on target %d: %v", prefix, targetID, err))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -99,6 +163,43 @@ func (s *Service) Cleanup(ctx context.Context, task *model.BackupTask, provider
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func cleanupObjectsForRecord(record model.BackupRecord) ([]cleanupObject, error) {
|
||||
defaultPath := strings.TrimSpace(record.StoragePath)
|
||||
if strings.TrimSpace(record.StorageUploadResults) == "" {
|
||||
if defaultPath == "" {
|
||||
return nil, nil
|
||||
}
|
||||
return []cleanupObject{{targetID: record.StorageTargetID, path: defaultPath}}, nil
|
||||
}
|
||||
var results []storedUploadResult
|
||||
if err := json.Unmarshal([]byte(record.StorageUploadResults), &results); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
objects := make([]cleanupObject, 0, len(results))
|
||||
seen := make(map[uint]struct{}, len(results))
|
||||
for _, result := range results {
|
||||
if !strings.EqualFold(strings.TrimSpace(result.Status), model.BackupRecordStatusSuccess) {
|
||||
continue
|
||||
}
|
||||
objectPath := strings.TrimSpace(result.StoragePath)
|
||||
if objectPath == "" {
|
||||
objectPath = defaultPath
|
||||
}
|
||||
if objectPath == "" {
|
||||
continue
|
||||
}
|
||||
if _, exists := seen[result.StorageTargetID]; exists {
|
||||
continue
|
||||
}
|
||||
seen[result.StorageTargetID] = struct{}{}
|
||||
objects = append(objects, cleanupObject{targetID: result.StorageTargetID, path: objectPath})
|
||||
}
|
||||
if len(objects) == 0 && defaultPath != "" {
|
||||
return nil, fmt.Errorf("successful record has no successful storage copy")
|
||||
}
|
||||
return objects, nil
|
||||
}
|
||||
|
||||
// protectDifferentialBases 从删除候选中剔除「仍被存活差异依赖的全量」,
|
||||
// 避免删除基线后其差异备份失去依据、无法恢复。全量仅当其全部差异都已过期/删除时才会被清理。
|
||||
func protectDifferentialBases(all []model.BackupRecord, candidates []model.BackupRecord) []model.BackupRecord {
|
||||
|
||||
@@ -221,3 +221,66 @@ func TestCleanupDeletesExpiredRecords(t *testing.T) {
|
||||
t.Fatalf("unexpected deleted objects: %#v", provider.deleted)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCleanupProvidersDeletesEverySuccessfulCopyBeforeRecord(t *testing.T) {
|
||||
now := time.Date(2026, 3, 7, 16, 0, 0, 0, time.UTC)
|
||||
completedNew := now.Add(-time.Hour)
|
||||
completedOld := now.Add(-24 * time.Hour)
|
||||
repo := &fakeRecordRepository{records: []model.BackupRecord{
|
||||
{ID: 2, TaskID: 1, StoragePath: "records/2", Status: model.BackupRecordStatusSuccess, CompletedAt: &completedNew},
|
||||
{
|
||||
ID: 1, TaskID: 1, StoragePath: "records/1", Status: model.BackupRecordStatusSuccess, CompletedAt: &completedOld,
|
||||
StorageUploadResults: `[{"storageTargetId":11,"status":"success","storagePath":"first/1"},{"storageTargetId":12,"status":"success","storagePath":"second/1"},{"storageTargetId":13,"status":"failed"}]`,
|
||||
},
|
||||
}}
|
||||
first := &fakeProvider{}
|
||||
second := &fakeProvider{}
|
||||
service := NewService(repo)
|
||||
service.now = func() time.Time { return now }
|
||||
|
||||
result, err := service.CleanupProviders(context.Background(), &model.BackupTask{ID: 1, MaxBackups: 1}, map[uint]storage.StorageProvider{
|
||||
11: first,
|
||||
12: second,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("CleanupProviders returned error: %v", err)
|
||||
}
|
||||
if result.DeletedRecords != 1 || result.DeletedObjects != 2 || len(result.Warnings) != 0 {
|
||||
t.Fatalf("unexpected cleanup result: %#v", result)
|
||||
}
|
||||
if len(repo.deleted) != 1 || repo.deleted[0] != 1 {
|
||||
t.Fatalf("unexpected deleted records: %#v", repo.deleted)
|
||||
}
|
||||
if len(first.deleted) != 1 || first.deleted[0] != "first/1" {
|
||||
t.Fatalf("unexpected first-target deletes: %#v", first.deleted)
|
||||
}
|
||||
if len(second.deleted) != 1 || second.deleted[0] != "second/1" {
|
||||
t.Fatalf("unexpected second-target deletes: %#v", second.deleted)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCleanupProvidersKeepsRecordWhenCopyProviderIsUnavailable(t *testing.T) {
|
||||
now := time.Date(2026, 3, 7, 16, 0, 0, 0, time.UTC)
|
||||
completedNew := now.Add(-time.Hour)
|
||||
completedOld := now.Add(-24 * time.Hour)
|
||||
repo := &fakeRecordRepository{records: []model.BackupRecord{
|
||||
{ID: 2, TaskID: 1, StoragePath: "records/2", Status: model.BackupRecordStatusSuccess, CompletedAt: &completedNew},
|
||||
{
|
||||
ID: 1, TaskID: 1, StoragePath: "records/1", Status: model.BackupRecordStatusSuccess, CompletedAt: &completedOld,
|
||||
StorageUploadResults: `[{"storageTargetId":11,"status":"success","storagePath":"first/1"},{"storageTargetId":12,"status":"success","storagePath":"second/1"}]`,
|
||||
},
|
||||
}}
|
||||
first := &fakeProvider{}
|
||||
service := NewService(repo)
|
||||
|
||||
result, err := service.CleanupProviders(context.Background(), &model.BackupTask{ID: 1, MaxBackups: 1}, map[uint]storage.StorageProvider{11: first})
|
||||
if err != nil {
|
||||
t.Fatalf("CleanupProviders returned error: %v", err)
|
||||
}
|
||||
if result.DeletedRecords != 0 || result.DeletedObjects != 1 || len(result.Warnings) != 1 {
|
||||
t.Fatalf("unexpected safe partial-cleanup result: %#v", result)
|
||||
}
|
||||
if len(repo.deleted) != 0 {
|
||||
t.Fatalf("record must remain until all copies are deleted: %#v", repo.deleted)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -55,6 +55,17 @@ func BuildStorageKey(backupType string, startedAt time.Time, fileName string) st
|
||||
return filepath.ToSlash(filepath.Join("BackupX", typeName, startedAt.UTC().Format("060102"), fileName))
|
||||
}
|
||||
|
||||
// BuildRecordStorageKey gives remote-Agent artifacts a record-owned namespace.
|
||||
// The Master validates this namespace before accepting a relayed upload, so one
|
||||
// Agent cannot overwrite another record's object on centrally mounted storage.
|
||||
func BuildRecordStorageKey(backupType string, startedAt time.Time, recordID uint, fileName string) string {
|
||||
typeName := strings.TrimSpace(strings.ToLower(backupType))
|
||||
if typeName == "" {
|
||||
typeName = "file"
|
||||
}
|
||||
return filepath.ToSlash(filepath.Join("BackupX", typeName, startedAt.UTC().Format("060102"), "records", fmt.Sprintf("%d", recordID), fileName))
|
||||
}
|
||||
|
||||
func sanitizeTaskName(value string) string {
|
||||
trimmed := strings.TrimSpace(strings.ToLower(value))
|
||||
trimmed = strings.ReplaceAll(trimmed, " ", "-")
|
||||
|
||||
@@ -156,6 +156,44 @@ func (h *AgentHandler) UpdateRecord(c *gin.Context) {
|
||||
response.Success(c, gin.H{"status": "ok"})
|
||||
}
|
||||
|
||||
// UploadArtifact streams a remote source artifact into storage mounted only on
|
||||
// the Master. The request body is never buffered as a whole in memory or disk.
|
||||
func (h *AgentHandler) UploadArtifact(c *gin.Context) {
|
||||
node, err := h.agentService.AuthenticatedNode(c.Request.Context(), extractToken(c))
|
||||
if err != nil {
|
||||
response.Error(c, err)
|
||||
return
|
||||
}
|
||||
recordID, err := strconv.ParseUint(c.Param("id"), 10, 32)
|
||||
if err != nil {
|
||||
response.Error(c, err)
|
||||
return
|
||||
}
|
||||
targetID, err := strconv.ParseUint(c.Param("targetId"), 10, 32)
|
||||
if err != nil {
|
||||
response.Error(c, err)
|
||||
return
|
||||
}
|
||||
if c.Request.ContentLength < 0 {
|
||||
c.JSON(stdhttp.StatusLengthRequired, gin.H{"code": "CONTENT_LENGTH_REQUIRED", "message": "artifact content length is required"})
|
||||
return
|
||||
}
|
||||
if err := h.agentService.UploadArtifact(
|
||||
c.Request.Context(),
|
||||
node,
|
||||
uint(recordID),
|
||||
uint(targetID),
|
||||
c.GetHeader("X-BackupX-Object-Key"),
|
||||
c.Request.ContentLength,
|
||||
c.GetHeader("X-BackupX-SHA256"),
|
||||
c.Request.Body,
|
||||
); err != nil {
|
||||
response.Error(c, err)
|
||||
return
|
||||
}
|
||||
response.Success(c, gin.H{"status": "ok"})
|
||||
}
|
||||
|
||||
// GetRestoreSpec Agent 拉取恢复规格。
|
||||
func (h *AgentHandler) GetRestoreSpec(c *gin.Context) {
|
||||
if h.restoreService == nil {
|
||||
@@ -208,6 +246,34 @@ func (h *AgentHandler) UpdateRestore(c *gin.Context) {
|
||||
response.Success(c, gin.H{"status": "ok"})
|
||||
}
|
||||
|
||||
// DownloadRestoreArtifact streams a Master-local backup back to its source
|
||||
// Agent for restore without exposing the local storage configuration.
|
||||
func (h *AgentHandler) DownloadRestoreArtifact(c *gin.Context) {
|
||||
if h.restoreService == nil {
|
||||
c.JSON(stdhttp.StatusServiceUnavailable, gin.H{"code": "RESTORE_SERVICE_DISABLED", "message": "restore service is not enabled"})
|
||||
return
|
||||
}
|
||||
node, err := h.agentService.AuthenticatedNode(c.Request.Context(), extractToken(c))
|
||||
if err != nil {
|
||||
response.Error(c, err)
|
||||
return
|
||||
}
|
||||
restoreID, err := strconv.ParseUint(c.Param("id"), 10, 32)
|
||||
if err != nil {
|
||||
response.Error(c, err)
|
||||
return
|
||||
}
|
||||
artifact, err := h.restoreService.DownloadAgentArtifact(c.Request.Context(), node, uint(restoreID))
|
||||
if err != nil {
|
||||
response.Error(c, err)
|
||||
return
|
||||
}
|
||||
c.DataFromReader(stdhttp.StatusOK, artifact.Size, "application/octet-stream", artifact.Reader, nil)
|
||||
if err := artifact.Reader.Close(); err != nil {
|
||||
_ = c.Error(err)
|
||||
}
|
||||
}
|
||||
|
||||
// Self 返回当前 Agent token 所属节点的状态,供安装脚本末尾探活。
|
||||
func (h *AgentHandler) Self(c *gin.Context) {
|
||||
node, err := h.agentService.AuthenticatedNode(c.Request.Context(), extractToken(c))
|
||||
|
||||
@@ -322,7 +322,9 @@ func NewRouter(deps RouterDependencies) *gin.Engine {
|
||||
agent.POST("/commands/:id/result", agentHandler.SubmitCommandResult)
|
||||
agent.GET("/tasks/:id", agentHandler.GetTaskSpec)
|
||||
agent.POST("/records/:id", agentHandler.UpdateRecord)
|
||||
agent.PUT("/records/:id/artifacts/:targetId", agentHandler.UploadArtifact)
|
||||
agent.GET("/restores/:id/spec", agentHandler.GetRestoreSpec)
|
||||
agent.GET("/restores/:id/artifact", agentHandler.DownloadRestoreArtifact)
|
||||
agent.POST("/restores/:id", agentHandler.UpdateRestore)
|
||||
|
||||
// Agent v1(安装脚本探活用),仅 Self 端点
|
||||
|
||||
@@ -39,3 +39,23 @@ func TestDeployInstallScriptSupportsReleasePackageLayout(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeployInstallScriptSupportsSourceBuildAndVerifiesFirstSetup(t *testing.T) {
|
||||
scriptPath := filepath.Join("..", "..", "..", "deploy", "install.sh")
|
||||
data, err := os.ReadFile(scriptPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
script := string(data)
|
||||
for _, want := range []string{
|
||||
`SOURCE_BIN_DEFAULT="$PROJECT_ROOT/server/bin/backupx"`,
|
||||
`For a source install, run 'make build' in the repository root first.`,
|
||||
`HEALTH_URL="${HEALTH_URL:-http://127.0.0.1:8340/api/auth/setup/status}"`,
|
||||
`systemctl is-active --quiet "$SERVICE_NAME"`,
|
||||
`System setup`,
|
||||
} {
|
||||
if !strings.Contains(script, want) {
|
||||
t.Fatalf("install.sh missing %q", want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,9 +9,11 @@ const (
|
||||
)
|
||||
|
||||
const (
|
||||
// BackupKindFull 全量备份;BackupKindDifferential 差异备份(仅含自基线全量以来的变更)。
|
||||
// BackupKindFull 全量归档;BackupKindDifferential 差异归档;
|
||||
// BackupKindRepository 为可独立恢复的 CDC 内容寻址快照。
|
||||
BackupKindFull = "full"
|
||||
BackupKindDifferential = "differential"
|
||||
BackupKindRepository = "repository"
|
||||
)
|
||||
|
||||
type BackupRecord struct {
|
||||
@@ -20,20 +22,22 @@ type BackupRecord struct {
|
||||
Task BackupTask `json:"task,omitempty"`
|
||||
StorageTargetID uint `gorm:"column:storage_target_id;index;not null" json:"storageTargetId"`
|
||||
StorageTarget StorageTarget `json:"storageTarget,omitempty"`
|
||||
// NodeID 执行该次备份的节点(0 = 本机 Master)。用于集群中识别 local_disk 类型
|
||||
// 存储的归属节点,避免 Master 端试图跨节点访问远程 Agent 的本地存储。
|
||||
NodeID uint `gorm:"column:node_id;index;default:0" json:"nodeId"`
|
||||
Status string `gorm:"size:20;index;not null" json:"status"`
|
||||
FileName string `gorm:"column:file_name;size:255" json:"fileName"`
|
||||
FileSize int64 `gorm:"column:file_size;not null;default:0" json:"fileSize"`
|
||||
Checksum string `gorm:"column:checksum;size:64" json:"checksum"`
|
||||
StoragePath string `gorm:"column:storage_path;size:500" json:"storagePath"`
|
||||
// NodeID 执行该次备份的节点(0 = 本机 Master)。StorageTransferMode 进一步
|
||||
// 区分远程 Agent 直写与 Master 中转,避免在错误节点访问 local_disk。
|
||||
NodeID uint `gorm:"column:node_id;index;default:0" json:"nodeId"`
|
||||
Status string `gorm:"size:20;index;not null" json:"status"`
|
||||
FileName string `gorm:"column:file_name;size:255" json:"fileName"`
|
||||
FileSize int64 `gorm:"column:file_size;not null;default:0" json:"fileSize"`
|
||||
Checksum string `gorm:"column:checksum;size:64" json:"checksum"`
|
||||
StoragePath string `gorm:"column:storage_path;size:500" json:"storagePath"`
|
||||
// 空值表示旧版 Agent 直写;direct / master_relay 记录新协议的实际数据路径。
|
||||
StorageTransferMode string `gorm:"column:storage_transfer_mode;size:20" json:"storageTransferMode,omitempty"`
|
||||
StorageUploadResults string `gorm:"column:storage_upload_results;type:text" json:"-"`
|
||||
DurationSeconds int `gorm:"column:duration_seconds;not null;default:0" json:"durationSeconds"`
|
||||
// Locked 保留锁定(法律保留):为 true 时该备份不参与保留期/数量自动清理,
|
||||
// 且禁止手动删除,直到显式解锁。用于保护合规快照、迁移前基线等关键备份。
|
||||
Locked bool `gorm:"column:locked;not null;default:false;index" json:"locked"`
|
||||
// BackupKind 备份类型:full(全量)/ differential(差异)。
|
||||
// BackupKind 备份类型:full(全量)/ differential(差异)/ repository(CDC 快照)。
|
||||
BackupKind string `gorm:"column:backup_kind;size:16;not null;default:'full';index" json:"backupKind"`
|
||||
// BaseRecordID 差异备份所基于的全量备份记录 ID(全量记录为 0)。
|
||||
BaseRecordID uint `gorm:"column:base_record_id;index;not null;default:0" json:"baseRecordId"`
|
||||
|
||||
@@ -12,9 +12,11 @@ const (
|
||||
)
|
||||
|
||||
const (
|
||||
// BackupModeFull 全量模式(默认);BackupModeDifferential 差异模式(仅文件类型本机任务)。
|
||||
// BackupModeFull 全量模式(默认);BackupModeDifferential 差异归档;
|
||||
// BackupModeRepository 为 CDC 内容寻址仓库模式(仅文件类型本机任务)。
|
||||
BackupModeFull = "full"
|
||||
BackupModeDifferential = "differential"
|
||||
BackupModeRepository = "repository"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -55,7 +57,8 @@ type BackupTask struct {
|
||||
Compression string `gorm:"size:10;not null;default:'gzip'" json:"compression"`
|
||||
Encrypt bool `gorm:"not null;default:false" json:"encrypt"`
|
||||
MaxBackups int `gorm:"column:max_backups;not null;default:10" json:"maxBackups"`
|
||||
// BackupMode 备份模式:full(全量,默认)/ differential(差异)。差异仅支持本机文件任务。
|
||||
// BackupMode 备份模式:full(全量,默认)/ differential(差异归档)/
|
||||
// repository(FastCDC 分块、全局去重快照)。后两者仅支持本机文件任务。
|
||||
BackupMode string `gorm:"column:backup_mode;size:16;not null;default:'full'" json:"backupMode"`
|
||||
// DiffFullIntervalDays 差异模式下强制全量的间隔(天):最近全量超过该天数则本次自动改为全量,
|
||||
// 限制差异链跨度与单个差异体积。默认 7。
|
||||
|
||||
@@ -2,15 +2,22 @@ package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"path"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"backupx/server/internal/apperror"
|
||||
"backupx/server/internal/backup"
|
||||
"backupx/server/internal/model"
|
||||
"backupx/server/internal/repository"
|
||||
"backupx/server/internal/storage"
|
||||
"backupx/server/internal/storage/codec"
|
||||
)
|
||||
|
||||
@@ -23,6 +30,7 @@ type AgentService struct {
|
||||
storageRepo repository.StorageTargetRepository
|
||||
cmdRepo repository.AgentCommandRepository
|
||||
restoreRepo repository.RestoreRecordRepository
|
||||
registry *storage.Registry
|
||||
cipher *codec.ConfigCipher
|
||||
}
|
||||
|
||||
@@ -33,6 +41,7 @@ func NewAgentService(
|
||||
storageRepo repository.StorageTargetRepository,
|
||||
cmdRepo repository.AgentCommandRepository,
|
||||
cipher *codec.ConfigCipher,
|
||||
registry *storage.Registry,
|
||||
) *AgentService {
|
||||
return &AgentService{
|
||||
nodeRepo: nodeRepo,
|
||||
@@ -40,6 +49,7 @@ func NewAgentService(
|
||||
recordRepo: recordRepo,
|
||||
storageRepo: storageRepo,
|
||||
cmdRepo: cmdRepo,
|
||||
registry: registry,
|
||||
cipher: cipher,
|
||||
}
|
||||
}
|
||||
@@ -145,10 +155,11 @@ type AgentTaskSpec struct {
|
||||
|
||||
// AgentStorageTargetConfig 存储目标配置(已解密)
|
||||
type AgentStorageTargetConfig struct {
|
||||
ID uint `json:"id"`
|
||||
Type string `json:"type"`
|
||||
Name string `json:"name"`
|
||||
Config json.RawMessage `json:"config"`
|
||||
ID uint `json:"id"`
|
||||
Type string `json:"type"`
|
||||
Name string `json:"name"`
|
||||
Config json.RawMessage `json:"config"`
|
||||
TransferMode string `json:"transferMode"`
|
||||
}
|
||||
|
||||
// GetTaskSpec 返回 Agent 执行任务所需的完整规格。
|
||||
@@ -187,11 +198,22 @@ func (s *AgentService) GetTaskSpec(ctx context.Context, node *model.Node, taskID
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("decrypt storage config: %w", err)
|
||||
}
|
||||
transferMode := storage.TransferModeDirect
|
||||
if strings.EqualFold(target.Type, storage.TypeLocalDisk) {
|
||||
var localConfig storage.LocalDiskConfig
|
||||
if err := json.Unmarshal(configRaw, &localConfig); err != nil {
|
||||
return nil, fmt.Errorf("decode local disk config: %w", err)
|
||||
}
|
||||
if localConfig.MasterRelay {
|
||||
transferMode = storage.TransferModeMasterRelay
|
||||
}
|
||||
}
|
||||
storageTargets = append(storageTargets, AgentStorageTargetConfig{
|
||||
ID: target.ID,
|
||||
Type: target.Type,
|
||||
Name: target.Name,
|
||||
Config: json.RawMessage(configRaw),
|
||||
ID: target.ID,
|
||||
Type: target.Type,
|
||||
Name: target.Name,
|
||||
Config: json.RawMessage(configRaw),
|
||||
TransferMode: transferMode,
|
||||
})
|
||||
}
|
||||
return &AgentTaskSpec{
|
||||
@@ -214,6 +236,102 @@ func (s *AgentService) GetTaskSpec(ctx context.Context, node *model.Node, taskID
|
||||
}, nil
|
||||
}
|
||||
|
||||
// UploadArtifact receives a remote Agent artifact as a stream and writes it
|
||||
// with a provider created on the Master. The first supported use is local_disk,
|
||||
// whose configured path belongs to the Master rather than the source Agent.
|
||||
func (s *AgentService) UploadArtifact(ctx context.Context, node *model.Node, recordID, targetID uint, objectKey string, size int64, checksum string, reader io.Reader) error {
|
||||
if node == nil || reader == nil || s.registry == nil {
|
||||
return apperror.BadRequest("AGENT_ARTIFACT_INVALID", "中转上传参数不完整", nil)
|
||||
}
|
||||
record, err := s.recordRepo.FindByID(ctx, recordID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if record == nil {
|
||||
return apperror.New(404, "BACKUP_RECORD_NOT_FOUND", "记录不存在", nil)
|
||||
}
|
||||
task, err := s.taskRepo.FindByID(ctx, record.TaskID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if task == nil || !recordBelongsToNode(record, task, node.ID) {
|
||||
return apperror.Unauthorized("BACKUP_RECORD_FORBIDDEN", "记录不属于当前节点", nil)
|
||||
}
|
||||
if isBackupRecordTerminal(record.Status) {
|
||||
return apperror.BadRequest("BACKUP_RECORD_TERMINAL", "备份记录已结束,不能继续上传产物", nil)
|
||||
}
|
||||
allowedTarget := false
|
||||
for _, configuredTargetID := range collectTargetIDs(task) {
|
||||
if configuredTargetID == targetID {
|
||||
allowedTarget = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !allowedTarget {
|
||||
return apperror.Unauthorized("BACKUP_STORAGE_TARGET_FORBIDDEN", "存储目标不属于该任务", nil)
|
||||
}
|
||||
target, err := s.storageRepo.FindByID(ctx, targetID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if target == nil || !strings.EqualFold(target.Type, storage.TypeLocalDisk) {
|
||||
return apperror.BadRequest("AGENT_ARTIFACT_RELAY_UNSUPPORTED", "仅 Master 本地磁盘目标需要中转上传", nil)
|
||||
}
|
||||
configMap := map[string]any{}
|
||||
if err := s.cipher.DecryptJSON(target.ConfigCiphertext, &configMap); err != nil {
|
||||
return fmt.Errorf("decrypt storage config: %w", err)
|
||||
}
|
||||
masterRelay, _ := configMap["masterRelay"].(bool)
|
||||
if !masterRelay {
|
||||
return apperror.BadRequest("AGENT_ARTIFACT_RELAY_UNSUPPORTED", "该本地磁盘目标配置为 Agent 直接写入", nil)
|
||||
}
|
||||
cleanKey, keyErr := s.validateArtifactKey(record, task, objectKey, true)
|
||||
if keyErr != nil {
|
||||
return keyErr
|
||||
}
|
||||
checksum = strings.TrimSpace(checksum)
|
||||
checksumBytes, checksumErr := hex.DecodeString(checksum)
|
||||
if size < 0 || size == math.MaxInt64 || checksumErr != nil || len(checksumBytes) != sha256.Size {
|
||||
return apperror.BadRequest("AGENT_ARTIFACT_INVALID", "中转上传需要有效的大小和 SHA-256", checksumErr)
|
||||
}
|
||||
if target.QuotaBytes > 0 {
|
||||
usage, usageErr := s.recordRepo.StorageUsage(ctx)
|
||||
if usageErr != nil {
|
||||
return fmt.Errorf("read storage usage: %w", usageErr)
|
||||
}
|
||||
currentUsed := int64(0)
|
||||
for _, item := range usage {
|
||||
if item.StorageTargetID == targetID {
|
||||
currentUsed = item.TotalSize
|
||||
break
|
||||
}
|
||||
}
|
||||
if currentUsed >= target.QuotaBytes || size > target.QuotaBytes-currentUsed {
|
||||
return apperror.BadRequest("BACKUP_STORAGE_QUOTA_EXCEEDED", fmt.Sprintf("超出存储目标配额(当前 %d,新增 %d,配额 %d)", currentUsed, size, target.QuotaBytes), nil)
|
||||
}
|
||||
}
|
||||
provider, err := s.registry.Create(ctx, target.Type, configMap)
|
||||
if err != nil {
|
||||
return fmt.Errorf("create master relay provider: %w", err)
|
||||
}
|
||||
limited := io.LimitReader(reader, size+1)
|
||||
hashed := newHashingReader(limited)
|
||||
metadata := map[string]string{
|
||||
"taskId": fmt.Sprintf("%d", task.ID),
|
||||
"recordId": fmt.Sprintf("%d", record.ID),
|
||||
"sourceNodeId": fmt.Sprintf("%d", node.ID),
|
||||
"transferMode": storage.TransferModeMasterRelay,
|
||||
}
|
||||
if err := provider.Upload(ctx, cleanKey, hashed, size, metadata); err != nil {
|
||||
return errors.Join(fmt.Errorf("relay artifact to master storage: %w", err), provider.Delete(ctx, cleanKey))
|
||||
}
|
||||
if hashed.n != size || !strings.EqualFold(hashed.Sum(), checksum) {
|
||||
deleteErr := provider.Delete(ctx, cleanKey)
|
||||
return errors.Join(fmt.Errorf("relayed artifact integrity mismatch: received %d of %d bytes", hashed.n, size), deleteErr)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *AgentService) ensureTaskSpecAccess(ctx context.Context, node *model.Node, task *model.BackupTask) error {
|
||||
if task.NodeID == node.ID {
|
||||
return nil
|
||||
@@ -236,6 +354,7 @@ type AgentRecordUpdate struct {
|
||||
Checksum string `json:"checksum,omitempty"`
|
||||
StoragePath string `json:"storagePath,omitempty"`
|
||||
StorageTargetID uint `json:"storageTargetId,omitempty"`
|
||||
StorageTransferMode string `json:"storageTransferMode,omitempty"`
|
||||
StorageUploadResults []StorageUploadResultItem `json:"storageUploadResults,omitempty"`
|
||||
ErrorMessage string `json:"errorMessage,omitempty"`
|
||||
LogAppend string `json:"logAppend,omitempty"` // 增量日志,追加到 record.log_content
|
||||
@@ -260,6 +379,99 @@ func (s *AgentService) UpdateRecord(ctx context.Context, node *model.Node, recor
|
||||
if isBackupRecordTerminal(record.Status) {
|
||||
return nil
|
||||
}
|
||||
allowedTargets := make(map[uint]struct{})
|
||||
for _, targetID := range collectTargetIDs(task) {
|
||||
allowedTargets[targetID] = struct{}{}
|
||||
}
|
||||
targetCache := make(map[uint]*model.StorageTarget)
|
||||
validateTransferMode := func(targetID uint, transferMode string) (string, error) {
|
||||
if _, ok := allowedTargets[targetID]; !ok {
|
||||
return "", apperror.Unauthorized("BACKUP_STORAGE_TARGET_FORBIDDEN", "存储目标不属于该任务", nil)
|
||||
}
|
||||
target := targetCache[targetID]
|
||||
if target == nil {
|
||||
var findErr error
|
||||
target, findErr = s.storageRepo.FindByID(ctx, targetID)
|
||||
if findErr != nil {
|
||||
return "", findErr
|
||||
}
|
||||
if target == nil {
|
||||
return "", apperror.BadRequest("BACKUP_STORAGE_TARGET_INVALID", "存储目标不存在", nil)
|
||||
}
|
||||
targetCache[targetID] = target
|
||||
}
|
||||
expectedMode := storage.TransferModeDirect
|
||||
if strings.EqualFold(target.Type, storage.TypeLocalDisk) {
|
||||
var localConfig storage.LocalDiskConfig
|
||||
if err := s.cipher.DecryptJSON(target.ConfigCiphertext, &localConfig); err != nil {
|
||||
return "", fmt.Errorf("decrypt storage config: %w", err)
|
||||
}
|
||||
if localConfig.MasterRelay {
|
||||
expectedMode = storage.TransferModeMasterRelay
|
||||
}
|
||||
}
|
||||
if transferMode != "" && transferMode != expectedMode {
|
||||
return "", apperror.BadRequest("AGENT_STORAGE_TRANSFER_MODE_INVALID", "Agent 上报的存储传输模式与目标配置不一致", nil)
|
||||
}
|
||||
return expectedMode, nil
|
||||
}
|
||||
selectedTransferMode := ""
|
||||
if update.StorageTargetID > 0 {
|
||||
if _, ok := allowedTargets[update.StorageTargetID]; !ok {
|
||||
return apperror.Unauthorized("BACKUP_STORAGE_TARGET_FORBIDDEN", "存储目标不属于该任务", nil)
|
||||
}
|
||||
var modeErr error
|
||||
selectedTransferMode, modeErr = validateTransferMode(update.StorageTargetID, update.StorageTransferMode)
|
||||
if modeErr != nil {
|
||||
return modeErr
|
||||
}
|
||||
} else if update.StorageTransferMode != "" {
|
||||
return apperror.BadRequest("AGENT_STORAGE_TRANSFER_MODE_INVALID", "传输模式缺少对应的存储目标", nil)
|
||||
}
|
||||
for index := range update.StorageUploadResults {
|
||||
result := &update.StorageUploadResults[index]
|
||||
expectedMode, err := validateTransferMode(result.StorageTargetID, result.TransferMode)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if result.FileSize < 0 || (result.Status != "" && result.Status != "success" && result.Status != "failed") {
|
||||
return apperror.BadRequest("AGENT_ARTIFACT_INVALID", "Agent 上报的存储结果无效", nil)
|
||||
}
|
||||
if result.StoragePath != "" {
|
||||
normalizedPath, err := s.validateArtifactKey(record, task, result.StoragePath, expectedMode == storage.TransferModeMasterRelay)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
result.StoragePath = normalizedPath
|
||||
}
|
||||
result.TransferMode = expectedMode
|
||||
}
|
||||
if update.StoragePath != "" {
|
||||
if update.StorageTargetID == 0 {
|
||||
return apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "存储路径缺少对应的存储目标", nil)
|
||||
}
|
||||
cleanStoragePath, err := s.validateArtifactKey(record, task, update.StoragePath, selectedTransferMode == storage.TransferModeMasterRelay)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if update.FileName != "" && path.Base(cleanStoragePath) != update.FileName {
|
||||
return apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "Agent 上报的文件名与存储路径不一致", nil)
|
||||
}
|
||||
update.StoragePath = cleanStoragePath
|
||||
}
|
||||
if update.Status != "" && update.Status != model.BackupRecordStatusRunning && update.Status != model.BackupRecordStatusSuccess && update.Status != model.BackupRecordStatusFailed {
|
||||
return apperror.BadRequest("BACKUP_RECORD_STATUS_INVALID", "Agent 上报的备份状态无效", nil)
|
||||
}
|
||||
if update.FileSize < 0 || (update.FileName != "" && (path.Base(update.FileName) != update.FileName || strings.Contains(update.FileName, "\\"))) {
|
||||
return apperror.BadRequest("AGENT_ARTIFACT_INVALID", "Agent 上报的备份文件信息无效", nil)
|
||||
}
|
||||
if update.Checksum != "" {
|
||||
checksumBytes, checksumErr := hex.DecodeString(strings.TrimSpace(update.Checksum))
|
||||
if checksumErr != nil || len(checksumBytes) != sha256.Size {
|
||||
return apperror.BadRequest("AGENT_ARTIFACT_INVALID", "Agent 上报的 SHA-256 无效", checksumErr)
|
||||
}
|
||||
update.Checksum = strings.ToLower(strings.TrimSpace(update.Checksum))
|
||||
}
|
||||
if update.Status != "" {
|
||||
record.Status = update.Status
|
||||
}
|
||||
@@ -277,6 +489,7 @@ func (s *AgentService) UpdateRecord(ctx context.Context, node *model.Node, recor
|
||||
}
|
||||
if update.StorageTargetID > 0 {
|
||||
record.StorageTargetID = update.StorageTargetID
|
||||
record.StorageTransferMode = selectedTransferMode
|
||||
}
|
||||
if len(update.StorageUploadResults) > 0 {
|
||||
if resultsJSON, marshalErr := json.Marshal(update.StorageUploadResults); marshalErr == nil {
|
||||
@@ -312,6 +525,27 @@ func (s *AgentService) UpdateRecord(ctx context.Context, node *model.Node, recor
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *AgentService) validateArtifactKey(record *model.BackupRecord, task *model.BackupTask, objectKey string, requireRecordNamespace bool) (string, error) {
|
||||
if record == nil || task == nil {
|
||||
return "", apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "无法确认中转对象归属", nil)
|
||||
}
|
||||
rawKey := objectKey
|
||||
cleanKey := path.Clean(rawKey)
|
||||
fileName := path.Base(cleanKey)
|
||||
if rawKey == "" || strings.TrimSpace(rawKey) != rawKey || cleanKey == "." || path.IsAbs(cleanKey) || strings.HasPrefix(cleanKey, "../") || cleanKey != rawKey || strings.Contains(rawKey, "\\") || fileName == "." || fileName == "/" {
|
||||
return "", apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "中转上传对象路径不安全", nil)
|
||||
}
|
||||
expectedKey := backup.BuildRecordStorageKey(task.Type, record.StartedAt, record.ID, fileName)
|
||||
if requireRecordNamespace {
|
||||
legacyKey := backup.BuildStorageKey(task.Type, record.StartedAt, fileName)
|
||||
if cleanKey != expectedKey && cleanKey != legacyKey {
|
||||
return "", apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "中转上传对象不属于当前备份记录", nil)
|
||||
}
|
||||
return expectedKey, nil
|
||||
}
|
||||
return cleanKey, nil
|
||||
}
|
||||
|
||||
func recordBelongsToNode(record *model.BackupRecord, task *model.BackupTask, nodeID uint) bool {
|
||||
if record.NodeID != 0 {
|
||||
return record.NodeID == nodeID
|
||||
|
||||
@@ -1,19 +1,26 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"backupx/server/internal/backup"
|
||||
"backupx/server/internal/config"
|
||||
"backupx/server/internal/database"
|
||||
"backupx/server/internal/logger"
|
||||
"backupx/server/internal/model"
|
||||
"backupx/server/internal/repository"
|
||||
"backupx/server/internal/storage"
|
||||
"backupx/server/internal/storage/codec"
|
||||
storageRclone "backupx/server/internal/storage/rclone"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
@@ -42,7 +49,7 @@ func newAgentServicePoolTestHarness(t *testing.T) (*AgentService, *gorm.DB, repo
|
||||
if err := nodeRepo.Create(context.Background(), other); err != nil {
|
||||
t.Fatalf("create other node: %v", err)
|
||||
}
|
||||
targetConfig, err := cipher.EncryptJSON(map[string]any{"basePath": t.TempDir()})
|
||||
targetConfig, err := cipher.EncryptJSON(map[string]any{"basePath": t.TempDir(), "masterRelay": true})
|
||||
if err != nil {
|
||||
t.Fatalf("EncryptJSON returned error: %v", err)
|
||||
}
|
||||
@@ -76,7 +83,8 @@ func newAgentServicePoolTestHarness(t *testing.T) (*AgentService, *gorm.DB, repo
|
||||
if err := recordRepo.Create(context.Background(), record); err != nil {
|
||||
t.Fatalf("create record: %v", err)
|
||||
}
|
||||
return NewAgentService(nodeRepo, taskRepo, recordRepo, storageRepo, cmdRepo, cipher), db, recordRepo, cmdRepo, owner, other
|
||||
storageRegistry := storage.NewRegistry(storageRclone.NewLocalDiskFactory())
|
||||
return NewAgentService(nodeRepo, taskRepo, recordRepo, storageRepo, cmdRepo, cipher, storageRegistry), db, recordRepo, cmdRepo, owner, other
|
||||
}
|
||||
|
||||
func TestAgentServicePooledTaskUsesRecordNodeForSpecAndRecordUpdates(t *testing.T) {
|
||||
@@ -90,19 +98,27 @@ func TestAgentServicePooledTaskUsesRecordNodeForSpecAndRecordUpdates(t *testing.
|
||||
if spec.TaskID != 1 || len(spec.StorageTargets) != 1 {
|
||||
t.Fatalf("unexpected spec: %#v", spec)
|
||||
}
|
||||
if spec.StorageTargets[0].TransferMode != storage.TransferModeMasterRelay {
|
||||
t.Fatalf("expected local disk to use Master relay, got %#v", spec.StorageTargets[0])
|
||||
}
|
||||
if _, err := svc.GetTaskSpec(ctx, other, 1); err == nil {
|
||||
t.Fatal("expected non-owner node to be forbidden from pooled task spec")
|
||||
}
|
||||
record, err := records.FindByID(ctx, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("FindByID record returned error: %v", err)
|
||||
}
|
||||
storagePath := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID, "backup.tar.gz")
|
||||
|
||||
if err := svc.UpdateRecord(ctx, owner, 1, AgentRecordUpdate{
|
||||
Status: model.BackupRecordStatusSuccess,
|
||||
FileName: "backup.tar.gz",
|
||||
FileSize: 123,
|
||||
StoragePath: "tasks/1/backup.tar.gz",
|
||||
StorageTargetID: 2,
|
||||
Status: model.BackupRecordStatusSuccess,
|
||||
FileName: "backup.tar.gz",
|
||||
FileSize: 123,
|
||||
StoragePath: storagePath,
|
||||
StorageTargetID: 1,
|
||||
StorageTransferMode: storage.TransferModeMasterRelay,
|
||||
StorageUploadResults: []StorageUploadResultItem{
|
||||
{StorageTargetID: 1, StorageTargetName: "first", Status: "failed", Error: "boom"},
|
||||
{StorageTargetID: 2, StorageTargetName: "second", Status: "success", StoragePath: "tasks/1/backup.tar.gz", FileSize: 123},
|
||||
{StorageTargetID: 1, StorageTargetName: "local", Status: "success", StoragePath: storagePath, FileSize: 123, TransferMode: storage.TransferModeMasterRelay},
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatalf("owner UpdateRecord returned error: %v", err)
|
||||
@@ -114,10 +130,13 @@ func TestAgentServicePooledTaskUsesRecordNodeForSpecAndRecordUpdates(t *testing.
|
||||
if updated.Status != model.BackupRecordStatusSuccess || updated.NodeID != owner.ID {
|
||||
t.Fatalf("unexpected updated record: %#v", updated)
|
||||
}
|
||||
if updated.StorageTargetID != 2 {
|
||||
t.Fatalf("expected successful storage target id 2, got %d", updated.StorageTargetID)
|
||||
if updated.StorageTargetID != 1 {
|
||||
t.Fatalf("expected successful storage target id 1, got %d", updated.StorageTargetID)
|
||||
}
|
||||
if !strings.Contains(updated.StorageUploadResults, `"storageTargetName":"second"`) {
|
||||
if updated.StorageTransferMode != storage.TransferModeMasterRelay {
|
||||
t.Fatalf("expected Master relay transfer mode, got %q", updated.StorageTransferMode)
|
||||
}
|
||||
if !strings.Contains(updated.StorageUploadResults, `"storageTargetName":"local"`) {
|
||||
t.Fatalf("expected upload results to be persisted, got %q", updated.StorageUploadResults)
|
||||
}
|
||||
if err := svc.UpdateRecord(ctx, other, 1, AgentRecordUpdate{LogAppend: "bad"}); err == nil {
|
||||
@@ -125,6 +144,157 @@ func TestAgentServicePooledTaskUsesRecordNodeForSpecAndRecordUpdates(t *testing.
|
||||
}
|
||||
}
|
||||
|
||||
func TestAgentServiceRelaysRemoteArtifactToMasterLocalDisk(t *testing.T) {
|
||||
svc, _, records, _, owner, other := newAgentServicePoolTestHarness(t)
|
||||
ctx := context.Background()
|
||||
payload := []byte("artifact from remote source server")
|
||||
digest := sha256.Sum256(payload)
|
||||
checksum := fmt.Sprintf("%x", digest[:])
|
||||
record, err := records.FindByID(ctx, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("FindByID record returned error: %v", err)
|
||||
}
|
||||
objectKey := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID, "remote-source.tar")
|
||||
|
||||
if err := svc.UploadArtifact(ctx, owner, 1, 1, objectKey, int64(len(payload)), checksum, bytes.NewReader(payload)); err != nil {
|
||||
t.Fatalf("UploadArtifact returned error: %v", err)
|
||||
}
|
||||
target, err := svc.storageRepo.FindByID(ctx, 1)
|
||||
if err != nil || target == nil {
|
||||
t.Fatalf("FindByID target: target=%#v err=%v", target, err)
|
||||
}
|
||||
config := map[string]any{}
|
||||
if err := svc.cipher.DecryptJSON(target.ConfigCiphertext, &config); err != nil {
|
||||
t.Fatalf("DecryptJSON target config: %v", err)
|
||||
}
|
||||
basePath, _ := config["basePath"].(string)
|
||||
stored, err := os.ReadFile(filepath.Join(basePath, filepath.FromSlash(objectKey)))
|
||||
if err != nil {
|
||||
t.Fatalf("read relayed artifact: %v", err)
|
||||
}
|
||||
if !bytes.Equal(stored, payload) {
|
||||
t.Fatalf("relayed artifact differs: got %q", stored)
|
||||
}
|
||||
if err := svc.UploadArtifact(ctx, other, 1, 1, objectKey, int64(len(payload)), checksum, bytes.NewReader(payload)); err == nil {
|
||||
t.Fatal("expected a different node to be forbidden from relaying the artifact")
|
||||
}
|
||||
|
||||
legacyPayload := []byte("artifact from an older Agent")
|
||||
legacyDigest := sha256.Sum256(legacyPayload)
|
||||
legacyKey := backup.BuildStorageKey("file", record.StartedAt, "legacy-agent.tar")
|
||||
canonicalKey := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID, "legacy-agent.tar")
|
||||
if err := svc.UploadArtifact(ctx, owner, record.ID, target.ID, legacyKey, int64(len(legacyPayload)), fmt.Sprintf("%x", legacyDigest[:]), bytes.NewReader(legacyPayload)); err != nil {
|
||||
t.Fatalf("UploadArtifact legacy key returned error: %v", err)
|
||||
}
|
||||
stored, err = os.ReadFile(filepath.Join(basePath, filepath.FromSlash(canonicalKey)))
|
||||
if err != nil || !bytes.Equal(stored, legacyPayload) {
|
||||
t.Fatalf("legacy Agent artifact was not normalized: data=%q err=%v", stored, err)
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(basePath, filepath.FromSlash(legacyKey))); !errors.Is(err, os.ErrNotExist) {
|
||||
t.Fatalf("legacy object key should not be written directly: %v", err)
|
||||
}
|
||||
if err := svc.UpdateRecord(ctx, owner, record.ID, AgentRecordUpdate{
|
||||
Status: model.BackupRecordStatusSuccess,
|
||||
FileName: "legacy-agent.tar",
|
||||
FileSize: int64(len(legacyPayload)),
|
||||
Checksum: fmt.Sprintf("%x", legacyDigest[:]),
|
||||
StoragePath: legacyKey,
|
||||
StorageTargetID: target.ID,
|
||||
StorageUploadResults: []StorageUploadResultItem{{
|
||||
StorageTargetID: target.ID,
|
||||
Status: "success",
|
||||
StoragePath: legacyKey,
|
||||
FileSize: int64(len(legacyPayload)),
|
||||
}},
|
||||
}); err != nil {
|
||||
t.Fatalf("UpdateRecord legacy key returned error: %v", err)
|
||||
}
|
||||
updated, err := records.FindByID(ctx, record.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("FindByID updated record returned error: %v", err)
|
||||
}
|
||||
if updated.StoragePath != canonicalKey || updated.StorageTransferMode != storage.TransferModeMasterRelay || !strings.Contains(updated.StorageUploadResults, canonicalKey) {
|
||||
t.Fatalf("legacy Agent record was not normalized: %#v", updated)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAgentServiceRejectsArtifactOutsideRecordNamespace(t *testing.T) {
|
||||
svc, _, records, _, owner, _ := newAgentServicePoolTestHarness(t)
|
||||
ctx := context.Background()
|
||||
record, err := records.FindByID(ctx, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("FindByID record returned error: %v", err)
|
||||
}
|
||||
target, err := svc.storageRepo.FindByID(ctx, 1)
|
||||
if err != nil || target == nil {
|
||||
t.Fatalf("FindByID target: target=%#v err=%v", target, err)
|
||||
}
|
||||
config := map[string]any{}
|
||||
if err := svc.cipher.DecryptJSON(target.ConfigCiphertext, &config); err != nil {
|
||||
t.Fatalf("DecryptJSON target config: %v", err)
|
||||
}
|
||||
basePath, _ := config["basePath"].(string)
|
||||
victimKey := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID+1, "victim.tar")
|
||||
victimPath := filepath.Join(basePath, filepath.FromSlash(victimKey))
|
||||
if err := os.MkdirAll(filepath.Dir(victimPath), 0o755); err != nil {
|
||||
t.Fatalf("MkdirAll victim parent: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(victimPath, []byte("keep me"), 0o600); err != nil {
|
||||
t.Fatalf("WriteFile victim: %v", err)
|
||||
}
|
||||
payload := []byte("overwrite")
|
||||
digest := sha256.Sum256(payload)
|
||||
if err := svc.UploadArtifact(ctx, owner, record.ID, target.ID, victimKey, int64(len(payload)), fmt.Sprintf("%x", digest[:]), bytes.NewReader(payload)); err == nil {
|
||||
t.Fatal("expected another record namespace to be rejected")
|
||||
}
|
||||
stored, err := os.ReadFile(victimPath)
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFile victim: %v", err)
|
||||
}
|
||||
if string(stored) != "keep me" {
|
||||
t.Fatalf("victim object changed: %q", stored)
|
||||
}
|
||||
if err := svc.UpdateRecord(ctx, owner, record.ID, AgentRecordUpdate{StoragePath: victimKey, StorageTargetID: target.ID, StorageTransferMode: storage.TransferModeMasterRelay}); err == nil {
|
||||
t.Fatal("expected another record namespace in status update to be rejected")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAgentServiceKeepsExistingLocalDiskTargetsAgentLocal(t *testing.T) {
|
||||
svc, _, records, _, owner, _ := newAgentServicePoolTestHarness(t)
|
||||
ctx := context.Background()
|
||||
target, err := svc.storageRepo.FindByID(ctx, 1)
|
||||
if err != nil || target == nil {
|
||||
t.Fatalf("FindByID target: target=%#v err=%v", target, err)
|
||||
}
|
||||
legacyConfig, err := svc.cipher.EncryptJSON(map[string]any{"basePath": t.TempDir()})
|
||||
if err != nil {
|
||||
t.Fatalf("EncryptJSON legacy target: %v", err)
|
||||
}
|
||||
target.ConfigCiphertext = legacyConfig
|
||||
if err := svc.storageRepo.Update(ctx, target); err != nil {
|
||||
t.Fatalf("Update legacy target: %v", err)
|
||||
}
|
||||
|
||||
spec, err := svc.GetTaskSpec(ctx, owner, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("GetTaskSpec returned error: %v", err)
|
||||
}
|
||||
if len(spec.StorageTargets) != 1 || spec.StorageTargets[0].TransferMode != storage.TransferModeDirect {
|
||||
t.Fatalf("expected legacy local disk to stay Agent-local, got %#v", spec.StorageTargets)
|
||||
}
|
||||
payload := []byte("must not be relayed")
|
||||
digest := sha256.Sum256(payload)
|
||||
record, findErr := records.FindByID(ctx, 1)
|
||||
if findErr != nil {
|
||||
t.Fatalf("FindByID record returned error: %v", findErr)
|
||||
}
|
||||
objectKey := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID, "legacy.tar")
|
||||
err = svc.UploadArtifact(ctx, owner, 1, 1, objectKey, int64(len(payload)), fmt.Sprintf("%x", digest[:]), bytes.NewReader(payload))
|
||||
if err == nil {
|
||||
t.Fatal("expected relay upload to be rejected for an Agent-local target")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAgentServiceUpdateRecordRefreshesTaskSummaryOnTerminalStatus(t *testing.T) {
|
||||
for _, status := range []string{model.BackupRecordStatusSuccess, model.BackupRecordStatusFailed} {
|
||||
t.Run(status, func(t *testing.T) {
|
||||
|
||||
@@ -5,11 +5,13 @@ import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"hash"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -49,6 +51,7 @@ type StorageUploadResultItem struct {
|
||||
Status string `json:"status"`
|
||||
StoragePath string `json:"storagePath,omitempty"`
|
||||
FileSize int64 `json:"fileSize,omitempty"`
|
||||
TransferMode string `json:"transferMode,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
@@ -62,6 +65,17 @@ type DownloadedArtifact struct {
|
||||
Reader io.ReadCloser
|
||||
}
|
||||
|
||||
type temporaryArtifactReader struct {
|
||||
*os.File
|
||||
directory string
|
||||
}
|
||||
|
||||
func (r *temporaryArtifactReader) Close() error {
|
||||
closeErr := r.File.Close()
|
||||
removeErr := os.RemoveAll(r.directory)
|
||||
return errors.Join(closeErr, removeErr)
|
||||
}
|
||||
|
||||
// collectTargetIDs 获取任务关联的所有存储目标 ID
|
||||
func collectTargetIDs(task *model.BackupTask) []uint {
|
||||
if len(task.StorageTargets) > 0 {
|
||||
@@ -102,6 +116,10 @@ type BackupExecutionService struct {
|
||||
bandwidthLimit string // rclone 带宽限制(全局默认,节点配置可覆盖)
|
||||
metrics *metrics.Metrics
|
||||
taskLocks sync.Map
|
||||
// repositoryLocks serializes immutable index updates per storage target.
|
||||
// Repository mode is intentionally single-writer in v1 to avoid orphaned
|
||||
// duplicate packs when two local tasks discover the same missing chunk.
|
||||
repositoryLocks sync.Map
|
||||
}
|
||||
|
||||
// SetMetrics 注入 Prometheus 采集器。nil 时所有埋点退化为 no-op。
|
||||
@@ -211,6 +229,25 @@ func (s *BackupExecutionService) DownloadRecord(ctx context.Context, recordID ui
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if record.BackupKind == model.BackupKindRepository {
|
||||
tempDir, err := os.MkdirTemp(s.tempDir, "repository-download-*")
|
||||
if err != nil {
|
||||
return nil, apperror.Internal("BACKUP_RECORD_DOWNLOAD_FAILED", "无法创建 CDC 导出目录", err)
|
||||
}
|
||||
exportName := fmt.Sprintf("backupx-record-%d.tar", record.ID)
|
||||
exportPath := filepath.Join(tempDir, exportName)
|
||||
store := backup.NewRepositoryStore(s.cipher.Key())
|
||||
if err := store.ExportTar(ctx, provider, record.StoragePath, record.Checksum, exportPath); err != nil {
|
||||
cleanupErr := os.RemoveAll(tempDir)
|
||||
return nil, apperror.Internal("BACKUP_RECORD_DOWNLOAD_FAILED", "无法从 CDC 仓库导出归档", errors.Join(err, cleanupErr))
|
||||
}
|
||||
file, err := os.Open(exportPath)
|
||||
if err != nil {
|
||||
cleanupErr := os.RemoveAll(tempDir)
|
||||
return nil, apperror.Internal("BACKUP_RECORD_DOWNLOAD_FAILED", "无法打开 CDC 导出归档", errors.Join(err, cleanupErr))
|
||||
}
|
||||
return &DownloadedArtifact{FileName: exportName, Reader: &temporaryArtifactReader{File: file, directory: tempDir}}, nil
|
||||
}
|
||||
reader, err := provider.Download(ctx, record.StoragePath)
|
||||
if err != nil {
|
||||
return nil, apperror.Internal("BACKUP_RECORD_DOWNLOAD_FAILED", "无法下载备份文件", err)
|
||||
@@ -234,6 +271,16 @@ func (s *BackupExecutionService) RestoreRecord(ctx context.Context, recordID uin
|
||||
if task == nil {
|
||||
return apperror.New(404, "BACKUP_TASK_NOT_FOUND", "关联的备份任务不存在,无法执行恢复", fmt.Errorf("backup task %d not found", record.TaskID))
|
||||
}
|
||||
if record.BackupKind == model.BackupKindRepository {
|
||||
spec, specErr := s.buildTaskSpec(task, record.StartedAt)
|
||||
if specErr != nil {
|
||||
return specErr
|
||||
}
|
||||
if err := backup.NewRepositoryStore(s.cipher.Key()).Restore(ctx, provider, record.StoragePath, record.Checksum, spec, backup.NopLogWriter{}); err != nil {
|
||||
return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "从 CDC 仓库恢复备份失败", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
tempDir, err := os.MkdirTemp("", "backupx-restore-*")
|
||||
if err != nil {
|
||||
return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "无法创建恢复目录", err)
|
||||
@@ -291,6 +338,58 @@ func (s *BackupExecutionService) DeleteRecord(ctx context.Context, recordID uint
|
||||
fmt.Sprintf("该全量备份仍有 %d 个差异备份依赖它,删除会导致这些差异无法恢复。请先删除相关差异备份或等待其过期。", deps), nil)
|
||||
}
|
||||
}
|
||||
if record.BackupKind == model.BackupKindRepository {
|
||||
copies := []StorageUploadResultItem{{
|
||||
StorageTargetID: record.StorageTargetID,
|
||||
Status: model.BackupRecordStatusSuccess,
|
||||
StoragePath: record.StoragePath,
|
||||
}}
|
||||
if strings.TrimSpace(record.StorageUploadResults) != "" {
|
||||
if err := json.Unmarshal([]byte(record.StorageUploadResults), &copies); err != nil {
|
||||
return apperror.Internal("BACKUP_RECORD_DELETE_FAILED", "无法解析 CDC 仓库副本信息,已停止删除以避免遗留数据", err)
|
||||
}
|
||||
}
|
||||
copyPaths := make(map[uint]string, len(copies))
|
||||
for _, copy := range copies {
|
||||
if strings.EqualFold(copy.Status, model.BackupRecordStatusSuccess) && strings.TrimSpace(copy.StoragePath) != "" {
|
||||
copyPaths[copy.StorageTargetID] = copy.StoragePath
|
||||
}
|
||||
}
|
||||
targetIDs := make([]uint, 0, len(copyPaths))
|
||||
for targetID := range copyPaths {
|
||||
targetIDs = append(targetIDs, targetID)
|
||||
}
|
||||
sort.Slice(targetIDs, func(i, j int) bool { return targetIDs[i] < targetIDs[j] })
|
||||
unlocks := make([]func(), 0, len(targetIDs))
|
||||
for _, targetID := range targetIDs {
|
||||
unlocks = append(unlocks, s.acquireRepositoryLock(targetID))
|
||||
}
|
||||
defer func() {
|
||||
for index := len(unlocks) - 1; index >= 0; index-- {
|
||||
unlocks[index]()
|
||||
}
|
||||
}()
|
||||
providers := make(map[uint]storage.StorageProvider, len(targetIDs))
|
||||
for _, targetID := range targetIDs {
|
||||
provider, resolveErr := s.resolveProvider(ctx, targetID)
|
||||
if resolveErr != nil {
|
||||
return resolveErr
|
||||
}
|
||||
if deleteErr := provider.Delete(ctx, copyPaths[targetID]); deleteErr != nil {
|
||||
return apperror.Internal("BACKUP_RECORD_DELETE_FAILED", "无法删除 CDC 仓库快照", deleteErr)
|
||||
}
|
||||
providers[targetID] = provider
|
||||
}
|
||||
for targetID, provider := range providers {
|
||||
if _, pruneErr := backup.NewRepositoryStore(s.cipher.Key()).Prune(ctx, provider); pruneErr != nil {
|
||||
return apperror.Internal("BACKUP_REPOSITORY_PRUNE_FAILED", fmt.Sprintf("无法清理存储目标 %d 的 CDC 仓库;记录暂时保留以便重试", targetID), pruneErr)
|
||||
}
|
||||
}
|
||||
if err := s.records.Delete(ctx, recordID); err != nil {
|
||||
return apperror.Internal("BACKUP_RECORD_DELETE_FAILED", "无法删除备份记录", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if remote, err := s.deleteRemoteLocalDiskObject(ctx, record); err != nil {
|
||||
return err
|
||||
} else if !remote && strings.TrimSpace(record.StoragePath) != "" {
|
||||
@@ -312,6 +411,9 @@ func (s *BackupExecutionService) deleteRemoteLocalDiskObject(ctx context.Context
|
||||
if strings.TrimSpace(record.StoragePath) == "" || s.nodeRepo == nil {
|
||||
return false, nil
|
||||
}
|
||||
if record.StorageTransferMode == storage.TransferModeMasterRelay {
|
||||
return false, nil
|
||||
}
|
||||
node, err := s.nodeRepo.FindByID(ctx, record.NodeID)
|
||||
if err != nil || node == nil || node.IsLocal {
|
||||
return false, nil
|
||||
@@ -384,6 +486,9 @@ func (s *BackupExecutionService) startTask(ctx context.Context, id uint, async b
|
||||
return nil, perr
|
||||
}
|
||||
}
|
||||
if strings.EqualFold(task.BackupMode, model.BackupModeRepository) && s.resolveRemoteNode(ctx, resolvedNodeID) != nil {
|
||||
return nil, apperror.BadRequest("BACKUP_TASK_REPOSITORY_REMOTE_UNSUPPORTED", "CDC 仓库模式当前仅支持 Master 本机单写者执行", nil)
|
||||
}
|
||||
startedAt := s.now()
|
||||
// 取第一个存储目标 ID 做兼容
|
||||
primaryTargetID := task.StorageTargetID
|
||||
@@ -630,6 +735,141 @@ func (s *BackupExecutionService) resolveDifferentialBase(ctx context.Context, ta
|
||||
return 0, backup.Manifest{}, false
|
||||
}
|
||||
|
||||
type repositoryTaskResult struct {
|
||||
fileName string
|
||||
logicalSize int64
|
||||
checksum string
|
||||
storagePath string
|
||||
storageTargetID uint
|
||||
manifestJSON string
|
||||
uploadResults []StorageUploadResultItem
|
||||
providers map[uint]storage.StorageProvider
|
||||
}
|
||||
|
||||
func (s *BackupExecutionService) executeRepositoryTask(ctx context.Context, task *model.BackupTask, recordID uint, startedAt time.Time, spec backup.TaskSpec, logger *backup.ExecutionLogger) (*repositoryTaskResult, error) {
|
||||
store := backup.NewRepositoryStore(s.cipher.Key())
|
||||
plan, err := store.BuildPlan(ctx, spec, logger)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
if closeErr := plan.Close(); closeErr != nil {
|
||||
logger.Warnf("清理 CDC 临时计划失败:%v", closeErr)
|
||||
}
|
||||
}()
|
||||
manifestBytes, err := backup.EncodeManifest(plan.Manifest)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("encode repository manifest: %w", err)
|
||||
}
|
||||
targetIDs := collectTargetIDs(task)
|
||||
if len(targetIDs) == 0 {
|
||||
return nil, fmt.Errorf("没有关联的存储目标")
|
||||
}
|
||||
storageUsage, usageErr := s.storageUsageSnapshot(ctx)
|
||||
if usageErr != nil {
|
||||
logger.Warnf("读取存储目标用量失败,跳过本次软配额校验:%v", usageErr)
|
||||
storageUsage = map[uint]int64{}
|
||||
}
|
||||
|
||||
snapshotKey := store.SnapshotKey(task.ID, recordID, startedAt)
|
||||
result := &repositoryTaskResult{
|
||||
fileName: filepath.Base(snapshotKey),
|
||||
logicalSize: plan.LogicalSize,
|
||||
storagePath: snapshotKey,
|
||||
manifestJSON: string(manifestBytes),
|
||||
uploadResults: make([]StorageUploadResultItem, 0, len(targetIDs)),
|
||||
providers: make(map[uint]storage.StorageProvider),
|
||||
}
|
||||
var failures []string
|
||||
for _, targetID := range targetIDs {
|
||||
target, findErr := s.targets.FindByID(ctx, targetID)
|
||||
targetName := fmt.Sprintf("target-%d", targetID)
|
||||
if findErr == nil && target != nil {
|
||||
targetName = target.Name
|
||||
}
|
||||
if findErr != nil || target == nil {
|
||||
message := "存储目标不存在"
|
||||
if findErr != nil {
|
||||
message = findErr.Error()
|
||||
}
|
||||
result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: message})
|
||||
failures = append(failures, fmt.Sprintf("%s: %s", targetName, message))
|
||||
continue
|
||||
}
|
||||
provider, resolveErr := s.resolveProviderForNode(ctx, targetID, task.NodeID)
|
||||
if resolveErr != nil {
|
||||
result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: resolveErr.Error()})
|
||||
failures = append(failures, fmt.Sprintf("%s: %v", targetName, resolveErr))
|
||||
continue
|
||||
}
|
||||
logger.Infof("同步 CDC 仓库到存储目标:%s", targetName)
|
||||
unlock := s.acquireRepositoryLock(targetID)
|
||||
estimatedSize, estimateErr := store.EstimateUploadSize(ctx, provider, plan)
|
||||
if estimateErr != nil {
|
||||
unlock()
|
||||
result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: estimateErr.Error()})
|
||||
failures = append(failures, fmt.Sprintf("%s: %v", targetName, estimateErr))
|
||||
continue
|
||||
}
|
||||
if target.QuotaBytes > 0 && storageUsage[targetID]+estimatedSize > target.QuotaBytes {
|
||||
unlock()
|
||||
message := fmt.Sprintf("超出存储目标配额(%d + 预计 %d > %d)", storageUsage[targetID], estimatedSize, target.QuotaBytes)
|
||||
result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: message})
|
||||
failures = append(failures, fmt.Sprintf("%s: %s", targetName, message))
|
||||
continue
|
||||
}
|
||||
upload, uploadErr := store.Upload(ctx, provider, plan, snapshotKey)
|
||||
unlock()
|
||||
if uploadErr != nil {
|
||||
result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: uploadErr.Error()})
|
||||
failures = append(failures, fmt.Sprintf("%s: %v", targetName, uploadErr))
|
||||
logger.Warnf("存储目标 %s CDC 仓库同步失败:%v", targetName, uploadErr)
|
||||
continue
|
||||
}
|
||||
result.uploadResults = append(result.uploadResults, StorageUploadResultItem{
|
||||
StorageTargetID: targetID, StorageTargetName: targetName, Status: "success",
|
||||
StoragePath: upload.SnapshotKey, FileSize: upload.UploadedBytes,
|
||||
})
|
||||
result.providers[targetID] = provider
|
||||
if result.storageTargetID == 0 {
|
||||
result.storageTargetID = targetID
|
||||
result.checksum = upload.Checksum
|
||||
}
|
||||
logger.Infof("存储目标 %s CDC 同步完成:新块 %d/%d,复用 %d bytes,实际上传 %d bytes", targetName, upload.NewChunks, upload.UniqueChunks, upload.ReusedBytes, upload.UploadedBytes)
|
||||
}
|
||||
if result.storageTargetID == 0 {
|
||||
return nil, fmt.Errorf("所有存储目标 CDC 仓库同步均失败:%s", strings.Join(failures, "; "))
|
||||
}
|
||||
if len(failures) > 0 {
|
||||
logger.Warnf("部分存储目标 CDC 仓库同步失败:%s", strings.Join(failures, "; "))
|
||||
}
|
||||
if s.dependentsResolver != nil {
|
||||
go func(upstreamID uint, upstreamName string) {
|
||||
dependents, resolveErr := s.dependentsResolver.TriggerDependents(context.Background(), upstreamID)
|
||||
if resolveErr != nil {
|
||||
logger.Warnf("解析任务 %s 的下游依赖失败:%v", upstreamName, resolveErr)
|
||||
return
|
||||
}
|
||||
for _, dependentID := range dependents {
|
||||
if _, runErr := s.RunTaskByID(context.Background(), dependentID); runErr != nil {
|
||||
logger.Warnf("触发下游任务 #%d 失败(上游: %s):%v", dependentID, upstreamName, runErr)
|
||||
} else {
|
||||
logger.Infof("已触发下游任务 #%d(上游: %s)", dependentID, upstreamName)
|
||||
}
|
||||
}
|
||||
}(task.ID, task.Name)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (s *BackupExecutionService) acquireRepositoryLock(targetID uint) func() {
|
||||
created := &sync.Mutex{}
|
||||
actual, _ := s.repositoryLocks.LoadOrStore(targetID, created)
|
||||
lock := actual.(*sync.Mutex)
|
||||
lock.Lock()
|
||||
return lock.Unlock
|
||||
}
|
||||
|
||||
func (s *BackupExecutionService) executeTask(ctx context.Context, task *model.BackupTask, recordID uint, startedAt time.Time) {
|
||||
// 节点级并发限流:当任务绑定节点且节点配置了 MaxConcurrent>0,
|
||||
// 该节点上所有任务共享一个节点专属 semaphore,互相排队
|
||||
@@ -658,18 +898,33 @@ func (s *BackupExecutionService) executeTask(ctx context.Context, task *model.Ba
|
||||
backupKind := model.BackupKindFull
|
||||
var baseRecordID uint
|
||||
var manifestJSON string
|
||||
var repositoryProviders map[uint]storage.StorageProvider
|
||||
completeRecord := func() {
|
||||
readyForRepositoryRetention := status == model.BackupRecordStatusSuccess
|
||||
if finalizeErr := s.finalizeRecord(ctx, task, recordID, startedAt, status, errMessage, logger.String(), fileName, fileSize, checksum, storagePath, selectedStorageTargetID); finalizeErr != nil {
|
||||
logger.Errorf("写回备份记录失败:%v", finalizeErr)
|
||||
readyForRepositoryRetention = false
|
||||
}
|
||||
// 采集任务执行结果到 Prometheus(耗时 + 产出字节 + 状态计数)
|
||||
s.metrics.ObserveTaskRun(task.Type, status, time.Since(startedAt).Seconds(), fileSize)
|
||||
// 写入多目标上传结果
|
||||
if len(uploadResults) > 0 {
|
||||
if resultsJSON, marshalErr := json.Marshal(uploadResults); marshalErr == nil {
|
||||
if record, findErr := s.records.FindByID(ctx, recordID); findErr == nil && record != nil {
|
||||
record.StorageUploadResults = string(resultsJSON)
|
||||
_ = s.records.Update(ctx, record)
|
||||
resultsJSON, marshalErr := json.Marshal(uploadResults)
|
||||
if marshalErr != nil {
|
||||
logger.Warnf("序列化多目标上传结果失败:%v", marshalErr)
|
||||
readyForRepositoryRetention = false
|
||||
} else if record, findErr := s.records.FindByID(ctx, recordID); findErr != nil || record == nil {
|
||||
if findErr != nil {
|
||||
logger.Warnf("读取备份记录以写回多目标结果失败:%v", findErr)
|
||||
} else {
|
||||
logger.Warnf("备份记录 #%d 不存在,无法写回多目标结果", recordID)
|
||||
}
|
||||
readyForRepositoryRetention = false
|
||||
} else {
|
||||
record.StorageUploadResults = string(resultsJSON)
|
||||
if updateErr := s.records.Update(ctx, record); updateErr != nil {
|
||||
logger.Warnf("写回多目标上传结果失败:%v", updateErr)
|
||||
readyForRepositoryRetention = false
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -681,6 +936,36 @@ func (s *BackupExecutionService) executeTask(ctx context.Context, task *model.Ba
|
||||
record.Manifest = manifestJSON
|
||||
if updErr := s.records.Update(ctx, record); updErr != nil {
|
||||
logger.Warnf("写回差异链信息失败:%v", updErr)
|
||||
readyForRepositoryRetention = false
|
||||
}
|
||||
} else {
|
||||
if findErr != nil {
|
||||
logger.Warnf("读取备份记录以写回备份类型失败:%v", findErr)
|
||||
} else {
|
||||
logger.Warnf("备份记录 #%d 不存在,无法写回备份类型", recordID)
|
||||
}
|
||||
readyForRepositoryRetention = false
|
||||
}
|
||||
}
|
||||
if readyForRepositoryRetention && backupKind == model.BackupKindRepository && s.retention != nil && len(repositoryProviders) > 0 {
|
||||
targetIDs := make([]uint, 0, len(repositoryProviders))
|
||||
for targetID := range repositoryProviders {
|
||||
targetIDs = append(targetIDs, targetID)
|
||||
}
|
||||
sort.Slice(targetIDs, func(i, j int) bool { return targetIDs[i] < targetIDs[j] })
|
||||
unlocks := make([]func(), 0, len(targetIDs))
|
||||
for _, targetID := range targetIDs {
|
||||
unlocks = append(unlocks, s.acquireRepositoryLock(targetID))
|
||||
}
|
||||
cleanupResult, cleanupErr := s.retention.CleanupProviders(ctx, task, repositoryProviders)
|
||||
for index := len(unlocks) - 1; index >= 0; index-- {
|
||||
unlocks[index]()
|
||||
}
|
||||
if cleanupErr != nil {
|
||||
logger.Warnf("执行 CDC 仓库保留策略失败:%v", cleanupErr)
|
||||
} else {
|
||||
for _, warning := range cleanupResult.Warnings {
|
||||
logger.Warnf("CDC 仓库保留策略警告:%s", warning)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -701,6 +986,26 @@ func (s *BackupExecutionService) executeTask(ctx context.Context, task *model.Ba
|
||||
logger.Errorf("构建任务运行时配置失败:%v", err)
|
||||
return
|
||||
}
|
||||
if task.Type == model.BackupTaskTypeFile && strings.EqualFold(task.BackupMode, model.BackupModeRepository) {
|
||||
backupKind = model.BackupKindRepository
|
||||
repositoryResult, repositoryErr := s.executeRepositoryTask(ctx, task, recordID, startedAt, spec, logger)
|
||||
if repositoryErr != nil {
|
||||
errMessage = repositoryErr.Error()
|
||||
logger.Errorf("执行 CDC 仓库备份失败:%v", repositoryErr)
|
||||
return
|
||||
}
|
||||
fileName = repositoryResult.fileName
|
||||
fileSize = repositoryResult.logicalSize
|
||||
checksum = repositoryResult.checksum
|
||||
storagePath = repositoryResult.storagePath
|
||||
selectedStorageTargetID = repositoryResult.storageTargetID
|
||||
uploadResults = repositoryResult.uploadResults
|
||||
repositoryProviders = repositoryResult.providers
|
||||
manifestJSON = repositoryResult.manifestJSON
|
||||
status = model.BackupRecordStatusSuccess
|
||||
logger.Infof("CDC 仓库备份执行完成")
|
||||
return
|
||||
}
|
||||
// 差异备份:解析基线全量,命中则切换为差异模式(仅本机文件任务)。
|
||||
if baseID, baseManifest, ok := s.resolveDifferentialBase(ctx, task); ok {
|
||||
spec.Differential = true
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
@@ -155,6 +156,108 @@ func TestBackupExecutionServiceRunTaskByIDSync(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackupExecutionServiceRepositoryModeRoundTrip(t *testing.T) {
|
||||
executionService, recordService, tasks, _, records, sourceDir, storageDir := newExecutionTestServices(t)
|
||||
ctx := context.Background()
|
||||
task, err := tasks.FindByID(ctx, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("FindByID task returned error: %v", err)
|
||||
}
|
||||
task.BackupMode = model.BackupModeRepository
|
||||
task.Compression = "zstd"
|
||||
if err := tasks.Update(ctx, task); err != nil {
|
||||
t.Fatalf("Update repository task returned error: %v", err)
|
||||
}
|
||||
large := make([]byte, 4<<20)
|
||||
for index := range large {
|
||||
large[index] = byte((index * 31) % 251)
|
||||
}
|
||||
largePath := filepath.Join(sourceDir, "large.bin")
|
||||
if err := os.WriteFile(largePath, large, 0o640); err != nil {
|
||||
t.Fatalf("write large fixture: %v", err)
|
||||
}
|
||||
|
||||
first, err := executionService.RunTaskByIDSync(ctx, task.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("first repository backup returned error: %v", err)
|
||||
}
|
||||
if first.Status != model.BackupRecordStatusSuccess || first.BackupKind != model.BackupKindRepository {
|
||||
t.Fatalf("unexpected first repository record: %#v", first)
|
||||
}
|
||||
if !strings.HasPrefix(first.StoragePath, ".backupx/repository/v1/snapshots/") {
|
||||
t.Fatalf("unexpected repository snapshot path: %s", first.StoragePath)
|
||||
}
|
||||
|
||||
large[2<<20] ^= 0xff
|
||||
if err := os.WriteFile(largePath, large, 0o640); err != nil {
|
||||
t.Fatalf("modify large fixture: %v", err)
|
||||
}
|
||||
second, err := executionService.RunTaskByIDSync(ctx, task.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("second repository backup returned error: %v", err)
|
||||
}
|
||||
stored, err := records.FindByID(ctx, second.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("FindByID repository record returned error: %v", err)
|
||||
}
|
||||
if stored == nil || stored.BackupKind != model.BackupKindRepository || stored.Manifest == "" {
|
||||
t.Fatalf("repository metadata was not persisted: %#v", stored)
|
||||
}
|
||||
|
||||
download, err := recordService.Download(ctx, second.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("export repository snapshot returned error: %v", err)
|
||||
}
|
||||
exported, readErr := io.ReadAll(download.Reader)
|
||||
closeErr := download.Reader.Close()
|
||||
if readErr != nil || closeErr != nil {
|
||||
t.Fatalf("read repository export: read=%v close=%v", readErr, closeErr)
|
||||
}
|
||||
if len(exported) == 0 || !strings.HasSuffix(download.FileName, ".tar") {
|
||||
t.Fatalf("unexpected repository export: name=%s size=%d", download.FileName, len(exported))
|
||||
}
|
||||
|
||||
if err := os.WriteFile(largePath, bytes.Repeat([]byte{0}, len(large)), 0o640); err != nil {
|
||||
t.Fatalf("damage source before restore: %v", err)
|
||||
}
|
||||
if err := executionService.RestoreRecord(ctx, second.ID); err != nil {
|
||||
t.Fatalf("restore repository record returned error: %v", err)
|
||||
}
|
||||
restored, err := os.ReadFile(largePath)
|
||||
if err != nil {
|
||||
t.Fatalf("read restored source: %v", err)
|
||||
}
|
||||
if !bytes.Equal(restored, large) {
|
||||
t.Fatalf("repository restore did not reproduce the source")
|
||||
}
|
||||
|
||||
if err := recordService.Delete(ctx, first.ID); err != nil {
|
||||
t.Fatalf("delete first repository record: %v", err)
|
||||
}
|
||||
if err := recordService.Delete(ctx, second.ID); err != nil {
|
||||
t.Fatalf("delete second repository record: %v", err)
|
||||
}
|
||||
packRoot := filepath.Join(storageDir, filepath.FromSlash(".backupx/repository/v1/packs"))
|
||||
remainingPacks := 0
|
||||
if err := filepath.Walk(packRoot, func(_ string, info os.FileInfo, walkErr error) error {
|
||||
if walkErr != nil {
|
||||
if os.IsNotExist(walkErr) {
|
||||
return nil
|
||||
}
|
||||
return walkErr
|
||||
}
|
||||
if info != nil && !info.IsDir() {
|
||||
remainingPacks++
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
t.Fatalf("inspect repository packs: %v", err)
|
||||
}
|
||||
if remainingPacks != 0 {
|
||||
t.Fatalf("repository prune left %d packs after deleting all snapshots", remainingPacks)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackupExecutionServiceNodePoolSelectionDoesNotPersistTaskNodeID(t *testing.T) {
|
||||
executionService, _, tasks, _, records, _, _ := newExecutionTestServices(t)
|
||||
ctx := context.Background()
|
||||
@@ -326,6 +429,56 @@ func TestBackupExecutionServiceRestoreRecordRejectsRemoteLocalDisk(t *testing.T)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackupExecutionServiceDownloadsMasterRelayedLocalDiskRecord(t *testing.T) {
|
||||
executionService, _, tasks, _, records, _, storageDir := newExecutionTestServices(t)
|
||||
ctx := context.Background()
|
||||
executionService.SetClusterDependencies(&nodeRepoStub{nodes: []model.Node{
|
||||
{ID: 10, Name: "edge-a", Token: "edge-a-token", Status: model.NodeStatusOnline},
|
||||
}}, &fakeDispatcher{})
|
||||
task, err := tasks.FindByID(ctx, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("FindByID task returned error: %v", err)
|
||||
}
|
||||
storagePath := "file/2026/05/09/relayed.tar"
|
||||
artifactPath := filepath.Join(storageDir, filepath.FromSlash(storagePath))
|
||||
if err := os.MkdirAll(filepath.Dir(artifactPath), 0o755); err != nil {
|
||||
t.Fatalf("MkdirAll artifact parent returned error: %v", err)
|
||||
}
|
||||
content := []byte("stored on Master")
|
||||
if err := os.WriteFile(artifactPath, content, 0o600); err != nil {
|
||||
t.Fatalf("WriteFile artifact returned error: %v", err)
|
||||
}
|
||||
completedAt := time.Now().UTC()
|
||||
record := &model.BackupRecord{
|
||||
TaskID: task.ID,
|
||||
StorageTargetID: task.StorageTargetID,
|
||||
NodeID: 10,
|
||||
Status: model.BackupRecordStatusSuccess,
|
||||
FileName: "relayed.tar",
|
||||
FileSize: int64(len(content)),
|
||||
StoragePath: storagePath,
|
||||
StorageTransferMode: storage.TransferModeMasterRelay,
|
||||
StartedAt: completedAt.Add(-time.Second),
|
||||
CompletedAt: &completedAt,
|
||||
}
|
||||
if err := records.Create(ctx, record); err != nil {
|
||||
t.Fatalf("Create record returned error: %v", err)
|
||||
}
|
||||
|
||||
download, err := executionService.DownloadRecord(ctx, record.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("DownloadRecord returned error: %v", err)
|
||||
}
|
||||
got, readErr := io.ReadAll(download.Reader)
|
||||
closeErr := download.Reader.Close()
|
||||
if readErr != nil || closeErr != nil {
|
||||
t.Fatalf("read relayed artifact: read=%v close=%v", readErr, closeErr)
|
||||
}
|
||||
if !bytes.Equal(got, content) {
|
||||
t.Fatalf("downloaded content = %q, want %q", got, content)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackupExecutionServiceRecordsFirstSuccessfulStorageTarget(t *testing.T) {
|
||||
executionService, _, tasks, targets, records, _, _ := newExecutionTestServices(t)
|
||||
ctx := context.Background()
|
||||
|
||||
@@ -23,22 +23,23 @@ type BackupRecordListInput struct {
|
||||
}
|
||||
|
||||
type BackupRecordSummary struct {
|
||||
ID uint `json:"id"`
|
||||
TaskID uint `json:"taskId"`
|
||||
TaskName string `json:"taskName"`
|
||||
StorageTargetID uint `json:"storageTargetId"`
|
||||
StorageTargetName string `json:"storageTargetName"`
|
||||
Status string `json:"status"`
|
||||
FileName string `json:"fileName"`
|
||||
FileSize int64 `json:"fileSize"`
|
||||
Checksum string `json:"checksum"`
|
||||
StoragePath string `json:"storagePath"`
|
||||
DurationSeconds int `json:"durationSeconds"`
|
||||
ErrorMessage string `json:"errorMessage"`
|
||||
StartedAt time.Time `json:"startedAt"`
|
||||
CompletedAt *time.Time `json:"completedAt,omitempty"`
|
||||
Locked bool `json:"locked"`
|
||||
BackupKind string `json:"backupKind"`
|
||||
ID uint `json:"id"`
|
||||
TaskID uint `json:"taskId"`
|
||||
TaskName string `json:"taskName"`
|
||||
StorageTargetID uint `json:"storageTargetId"`
|
||||
StorageTargetName string `json:"storageTargetName"`
|
||||
Status string `json:"status"`
|
||||
FileName string `json:"fileName"`
|
||||
FileSize int64 `json:"fileSize"`
|
||||
Checksum string `json:"checksum"`
|
||||
StoragePath string `json:"storagePath"`
|
||||
StorageTransferMode string `json:"storageTransferMode,omitempty"`
|
||||
DurationSeconds int `json:"durationSeconds"`
|
||||
ErrorMessage string `json:"errorMessage"`
|
||||
StartedAt time.Time `json:"startedAt"`
|
||||
CompletedAt *time.Time `json:"completedAt,omitempty"`
|
||||
Locked bool `json:"locked"`
|
||||
BackupKind string `json:"backupKind"`
|
||||
}
|
||||
|
||||
type BackupRecordDetail struct {
|
||||
@@ -184,22 +185,23 @@ func (s *BackupRecordService) SetLock(ctx context.Context, id uint, locked bool)
|
||||
|
||||
func toBackupRecordSummary(item *model.BackupRecord) BackupRecordSummary {
|
||||
return BackupRecordSummary{
|
||||
ID: item.ID,
|
||||
TaskID: item.TaskID,
|
||||
TaskName: item.Task.Name,
|
||||
StorageTargetID: item.StorageTargetID,
|
||||
StorageTargetName: item.StorageTarget.Name,
|
||||
Status: item.Status,
|
||||
FileName: item.FileName,
|
||||
FileSize: item.FileSize,
|
||||
Checksum: item.Checksum,
|
||||
StoragePath: item.StoragePath,
|
||||
DurationSeconds: item.DurationSeconds,
|
||||
ErrorMessage: item.ErrorMessage,
|
||||
StartedAt: item.StartedAt,
|
||||
CompletedAt: item.CompletedAt,
|
||||
Locked: item.Locked,
|
||||
BackupKind: item.BackupKind,
|
||||
ID: item.ID,
|
||||
TaskID: item.TaskID,
|
||||
TaskName: item.Task.Name,
|
||||
StorageTargetID: item.StorageTargetID,
|
||||
StorageTargetName: item.StorageTarget.Name,
|
||||
Status: item.Status,
|
||||
FileName: item.FileName,
|
||||
FileSize: item.FileSize,
|
||||
Checksum: item.Checksum,
|
||||
StoragePath: item.StoragePath,
|
||||
StorageTransferMode: item.StorageTransferMode,
|
||||
DurationSeconds: item.DurationSeconds,
|
||||
ErrorMessage: item.ErrorMessage,
|
||||
StartedAt: item.StartedAt,
|
||||
CompletedAt: item.CompletedAt,
|
||||
Locked: item.Locked,
|
||||
BackupKind: item.BackupKind,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -57,8 +57,8 @@ type BackupTaskUpsertInput struct {
|
||||
KeepWeekly int `json:"keepWeekly"`
|
||||
KeepMonthly int `json:"keepMonthly"`
|
||||
KeepYearly int `json:"keepYearly"`
|
||||
// BackupMode 备份模式:full(默认)/ differential(差异,仅文件类型本机任务)
|
||||
BackupMode string `json:"backupMode" binding:"omitempty,oneof=full differential"`
|
||||
// BackupMode 备份模式:full(默认)/ differential(差异归档)/ repository(CDC 去重仓库)
|
||||
BackupMode string `json:"backupMode" binding:"omitempty,oneof=full differential repository"`
|
||||
DiffFullIntervalDays int `json:"diffFullIntervalDays"`
|
||||
// 备份复制目标存储 ID 列表(3-2-1 规则)
|
||||
ReplicationTargetIDs []uint `json:"replicationTargetIds"`
|
||||
@@ -414,21 +414,50 @@ func (s *BackupTaskService) cleanupRemoteFiles(ctx context.Context, taskID uint)
|
||||
recordCount = len(records)
|
||||
// 缓存 provider 避免同一存储目标重复创建连接
|
||||
providerCache := make(map[uint]storage.StorageProvider)
|
||||
repositoryProviders := make(map[uint]storage.StorageProvider)
|
||||
for _, record := range records {
|
||||
if strings.TrimSpace(record.StoragePath) == "" {
|
||||
continue
|
||||
copies := []StorageUploadResultItem{{
|
||||
StorageTargetID: record.StorageTargetID,
|
||||
Status: model.BackupRecordStatusSuccess,
|
||||
StoragePath: record.StoragePath,
|
||||
}}
|
||||
if strings.TrimSpace(record.StorageUploadResults) != "" {
|
||||
var storedCopies []StorageUploadResultItem
|
||||
if unmarshalErr := json.Unmarshal([]byte(record.StorageUploadResults), &storedCopies); unmarshalErr == nil {
|
||||
copies = storedCopies
|
||||
}
|
||||
}
|
||||
provider, ok := providerCache[record.StorageTargetID]
|
||||
if !ok {
|
||||
provider, err = s.resolveStorageProvider(ctx, record.StorageTargetID)
|
||||
if err != nil {
|
||||
seenTargets := make(map[uint]struct{}, len(copies))
|
||||
for _, copy := range copies {
|
||||
if !strings.EqualFold(copy.Status, model.BackupRecordStatusSuccess) || strings.TrimSpace(copy.StoragePath) == "" {
|
||||
continue
|
||||
}
|
||||
providerCache[record.StorageTargetID] = provider
|
||||
if _, seen := seenTargets[copy.StorageTargetID]; seen {
|
||||
continue
|
||||
}
|
||||
seenTargets[copy.StorageTargetID] = struct{}{}
|
||||
provider, ok := providerCache[copy.StorageTargetID]
|
||||
if !ok {
|
||||
provider, err = s.resolveStorageProvider(ctx, copy.StorageTargetID)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
providerCache[copy.StorageTargetID] = provider
|
||||
}
|
||||
if err := provider.Delete(ctx, copy.StoragePath); err == nil {
|
||||
cleanedFiles++
|
||||
if record.BackupKind == model.BackupKindRepository {
|
||||
repositoryProviders[copy.StorageTargetID] = provider
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := provider.Delete(ctx, record.StoragePath); err == nil {
|
||||
cleanedFiles++
|
||||
}
|
||||
for _, provider := range repositoryProviders {
|
||||
pruned, pruneErr := backup.NewRepositoryStore(s.cipher.Key()).Prune(ctx, provider)
|
||||
if pruneErr != nil {
|
||||
continue
|
||||
}
|
||||
cleanedFiles += pruned.DeletedIndexes + pruned.DeletedPacks
|
||||
}
|
||||
return recordCount, cleanedFiles
|
||||
}
|
||||
@@ -530,6 +559,17 @@ func (s *BackupTaskService) validateInput(ctx context.Context, existing *model.B
|
||||
return apperror.BadRequest("BACKUP_TASK_DIFF_REMOTE_UNSUPPORTED", "差异备份当前仅支持本机 Master 执行,请将任务固定在本机或改用全量备份。", nil)
|
||||
}
|
||||
}
|
||||
if strings.EqualFold(strings.TrimSpace(input.BackupMode), model.BackupModeRepository) {
|
||||
if input.Type != model.BackupTaskTypeFile {
|
||||
return apperror.BadRequest("BACKUP_TASK_REPOSITORY_UNSUPPORTED", "CDC 仓库模式仅支持文件目录类型任务", nil)
|
||||
}
|
||||
if strings.TrimSpace(input.NodePoolTag) != "" || (fixedNode != nil && !fixedNode.IsLocal) {
|
||||
return apperror.BadRequest("BACKUP_TASK_REPOSITORY_REMOTE_UNSUPPORTED", "CDC 仓库模式当前采用单写者索引,仅支持 Master 本机执行。远程服务器备份请暂用全量模式。", nil)
|
||||
}
|
||||
if len(input.ReplicationTargetIDs) > 0 {
|
||||
return apperror.BadRequest("BACKUP_TASK_REPOSITORY_REPLICATION_UNSUPPORTED", "CDC 仓库快照不能使用对象级复制;请直接为任务选择多个存储目标以生成完整仓库副本。", nil)
|
||||
}
|
||||
}
|
||||
if input.RetentionDays < 0 {
|
||||
return apperror.BadRequest("BACKUP_TASK_INVALID", "保留天数不能小于 0", nil)
|
||||
}
|
||||
@@ -935,10 +975,17 @@ func decodeExtraConfig(value string) (map[string]any, error) {
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// normalizeBackupMode 归一化备份模式:仅文件类型可启用差异,其余一律全量(双保险,防绕过校验)。
|
||||
// normalizeBackupMode 归一化备份模式:仅文件类型可启用差异或 CDC 仓库,
|
||||
// 其余一律全量(双保险,防绕过校验)。
|
||||
func normalizeBackupMode(mode, taskType string) string {
|
||||
if strings.EqualFold(strings.TrimSpace(mode), model.BackupModeDifferential) && normalizeBackupTaskType(taskType) == model.BackupTaskTypeFile {
|
||||
if normalizeBackupTaskType(taskType) != model.BackupTaskTypeFile {
|
||||
return model.BackupModeFull
|
||||
}
|
||||
switch strings.ToLower(strings.TrimSpace(mode)) {
|
||||
case model.BackupModeDifferential:
|
||||
return model.BackupModeDifferential
|
||||
case model.BackupModeRepository:
|
||||
return model.BackupModeRepository
|
||||
}
|
||||
return model.BackupModeFull
|
||||
}
|
||||
|
||||
@@ -140,6 +140,11 @@ func validateCrossNodeLocalDisk(ctx context.Context, nodeRepo repository.NodeRep
|
||||
if record == nil || record.NodeID == 0 || nodeRepo == nil {
|
||||
return nil
|
||||
}
|
||||
// 中转模式的对象实际落在 Master 配置的本地磁盘,Master 可以安全访问。
|
||||
// 空值和 direct 均按旧版 Agent 本地落盘处理,保持升级兼容。
|
||||
if record.StorageTransferMode == storage.TransferModeMasterRelay {
|
||||
return nil
|
||||
}
|
||||
node, err := nodeRepo.FindByID(ctx, record.NodeID)
|
||||
if err != nil || node == nil || node.IsLocal {
|
||||
return nil
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
@@ -321,6 +322,13 @@ func (s *RestoreService) restoreArtifact(ctx context.Context, record *model.Back
|
||||
if err != nil {
|
||||
return fmt.Errorf("创建存储客户端失败:%w", err)
|
||||
}
|
||||
if record.BackupKind == model.BackupKindRepository {
|
||||
logger.Infof("读取 CDC 仓库快照:%s", record.StoragePath)
|
||||
if err := backup.NewRepositoryStore(s.cipher.Key()).Restore(ctx, provider, record.StoragePath, record.Checksum, spec, logger); err != nil {
|
||||
return fmt.Errorf("恢复 CDC 仓库快照失败:%w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
recDir, err := os.MkdirTemp(parentTempDir, fmt.Sprintf("rec-%d-*", record.ID))
|
||||
if err != nil {
|
||||
return fmt.Errorf("创建恢复子目录失败:%w", err)
|
||||
@@ -368,6 +376,9 @@ func (s *RestoreService) buildRestoreChain(ctx context.Context, record *model.Ba
|
||||
}
|
||||
|
||||
func backupKindLabel(kind string) string {
|
||||
if kind == model.BackupKindRepository {
|
||||
return "CDC 仓库快照"
|
||||
}
|
||||
if kind == model.BackupKindDifferential {
|
||||
return "差异"
|
||||
}
|
||||
@@ -591,15 +602,22 @@ func (s *RestoreService) GetAgentRestoreSpec(ctx context.Context, node *model.No
|
||||
if target == nil {
|
||||
return nil, apperror.BadRequest("BACKUP_STORAGE_TARGET_INVALID", "存储目标不存在", nil)
|
||||
}
|
||||
configRaw, err := s.cipher.Decrypt(target.ConfigCiphertext)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("decrypt storage config: %w", err)
|
||||
}
|
||||
// 拆开 sourcePaths
|
||||
sourcePaths := []string{}
|
||||
if strings.TrimSpace(task.SourcePaths) != "" {
|
||||
_ = json.Unmarshal([]byte(task.SourcePaths), &sourcePaths)
|
||||
}
|
||||
transferMode := storage.TransferModeDirect
|
||||
if backupRecord.StorageTransferMode == storage.TransferModeMasterRelay {
|
||||
transferMode = storage.TransferModeMasterRelay
|
||||
}
|
||||
var configRaw []byte
|
||||
if transferMode == storage.TransferModeDirect {
|
||||
configRaw, err = s.cipher.Decrypt(target.ConfigCiphertext)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("decrypt storage config: %w", err)
|
||||
}
|
||||
}
|
||||
return &AgentRestoreSpec{
|
||||
RestoreRecordID: restore.ID,
|
||||
BackupRecordID: backupRecord.ID,
|
||||
@@ -618,10 +636,11 @@ func (s *RestoreService) GetAgentRestoreSpec(ctx context.Context, node *model.No
|
||||
Compression: task.Compression,
|
||||
Encrypt: task.Encrypt,
|
||||
Storage: AgentStorageTargetConfig{
|
||||
ID: target.ID,
|
||||
Type: target.Type,
|
||||
Name: target.Name,
|
||||
Config: json.RawMessage(configRaw),
|
||||
ID: target.ID,
|
||||
Type: target.Type,
|
||||
Name: target.Name,
|
||||
Config: json.RawMessage(configRaw),
|
||||
TransferMode: transferMode,
|
||||
},
|
||||
StoragePath: backupRecord.StoragePath,
|
||||
FileName: backupRecord.FileName,
|
||||
@@ -629,6 +648,63 @@ func (s *RestoreService) GetAgentRestoreSpec(ctx context.Context, node *model.No
|
||||
}, nil
|
||||
}
|
||||
|
||||
type AgentArtifactDownload struct {
|
||||
Reader io.ReadCloser
|
||||
Size int64
|
||||
}
|
||||
|
||||
// DownloadAgentArtifact opens a Master-local object for authenticated streaming
|
||||
// back to the Agent that owns the restore record.
|
||||
func (s *RestoreService) DownloadAgentArtifact(ctx context.Context, node *model.Node, restoreID uint) (*AgentArtifactDownload, error) {
|
||||
if node == nil {
|
||||
return nil, apperror.Unauthorized("RESTORE_RECORD_FORBIDDEN", "恢复记录不属于当前节点", nil)
|
||||
}
|
||||
restore, err := s.restores.FindByID(ctx, restoreID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if restore == nil {
|
||||
return nil, apperror.New(404, "RESTORE_RECORD_NOT_FOUND", "恢复记录不存在", nil)
|
||||
}
|
||||
if restore.NodeID != node.ID {
|
||||
return nil, apperror.Unauthorized("RESTORE_RECORD_FORBIDDEN", "恢复记录不属于当前节点", nil)
|
||||
}
|
||||
if isRestoreRecordTerminal(restore.Status) {
|
||||
return nil, apperror.BadRequest("RESTORE_RECORD_TERMINAL", "恢复记录已结束,不能继续下载产物", nil)
|
||||
}
|
||||
record, err := s.records.FindByID(ctx, restore.BackupRecordID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if record == nil {
|
||||
return nil, apperror.New(404, "BACKUP_RECORD_NOT_FOUND", "源备份记录不存在", nil)
|
||||
}
|
||||
target, err := s.targets.FindByID(ctx, record.StorageTargetID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if target == nil || !strings.EqualFold(target.Type, storage.TypeLocalDisk) || record.StorageTransferMode != storage.TransferModeMasterRelay {
|
||||
return nil, apperror.BadRequest("AGENT_ARTIFACT_RELAY_UNSUPPORTED", "该存储目标应由 Agent 直接下载", nil)
|
||||
}
|
||||
configMap := map[string]any{}
|
||||
if err := s.cipher.DecryptJSON(target.ConfigCiphertext, &configMap); err != nil {
|
||||
return nil, fmt.Errorf("decrypt storage config: %w", err)
|
||||
}
|
||||
provider, err := s.storageRegistry.Create(ctx, target.Type, configMap)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create master relay provider: %w", err)
|
||||
}
|
||||
reader, err := provider.Download(ctx, record.StoragePath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("open master relay artifact: %w", err)
|
||||
}
|
||||
size := record.FileSize
|
||||
if size <= 0 {
|
||||
size = -1
|
||||
}
|
||||
return &AgentArtifactDownload{Reader: reader, Size: size}, nil
|
||||
}
|
||||
|
||||
// UpdateAgentRestore Agent 回传状态/日志。
|
||||
func (s *RestoreService) UpdateAgentRestore(ctx context.Context, node *model.Node, restoreID uint, update AgentRestoreUpdate) error {
|
||||
restore, err := s.restores.FindByID(ctx, restoreID)
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
@@ -427,16 +429,27 @@ func TestRestoreServiceAgentRestoreAccessUsesRestoreRecordNode(t *testing.T) {
|
||||
}
|
||||
startedAt := time.Now().UTC()
|
||||
completedAt := startedAt.Add(time.Second)
|
||||
artifact := []byte("central backup artifact")
|
||||
storagePath := "file/2026/05/09/remote.tar.gz"
|
||||
artifactPath := filepath.Join(h.storageDir, filepath.FromSlash(storagePath))
|
||||
if err := os.MkdirAll(filepath.Dir(artifactPath), 0o755); err != nil {
|
||||
t.Fatalf("MkdirAll artifact parent: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(artifactPath, artifact, 0o600); err != nil {
|
||||
t.Fatalf("WriteFile artifact: %v", err)
|
||||
}
|
||||
backupRecord := &model.BackupRecord{
|
||||
TaskID: task.ID,
|
||||
StorageTargetID: task.StorageTargetID,
|
||||
NodeID: owner.ID,
|
||||
Status: model.BackupRecordStatusSuccess,
|
||||
FileName: "remote.tar.gz",
|
||||
StoragePath: "file/2026/05/09/remote.tar.gz",
|
||||
Checksum: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
|
||||
StartedAt: startedAt,
|
||||
CompletedAt: &completedAt,
|
||||
TaskID: task.ID,
|
||||
StorageTargetID: task.StorageTargetID,
|
||||
NodeID: owner.ID,
|
||||
Status: model.BackupRecordStatusSuccess,
|
||||
FileName: "remote.tar.gz",
|
||||
StoragePath: storagePath,
|
||||
FileSize: int64(len(artifact)),
|
||||
StorageTransferMode: storage.TransferModeMasterRelay,
|
||||
Checksum: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
|
||||
StartedAt: startedAt,
|
||||
CompletedAt: &completedAt,
|
||||
}
|
||||
if err := h.records.Create(ctx, backupRecord); err != nil {
|
||||
t.Fatalf("Create backup record: %v", err)
|
||||
@@ -464,6 +477,21 @@ func TestRestoreServiceAgentRestoreAccessUsesRestoreRecordNode(t *testing.T) {
|
||||
if spec.Checksum != backupRecord.Checksum {
|
||||
t.Fatalf("expected spec.Checksum=%q, got %q", backupRecord.Checksum, spec.Checksum)
|
||||
}
|
||||
if spec.Storage.TransferMode != storage.TransferModeMasterRelay {
|
||||
t.Fatalf("expected Master relay restore, got %#v", spec.Storage)
|
||||
}
|
||||
download, err := h.service.DownloadAgentArtifact(ctx, owner, restore.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("DownloadAgentArtifact returned error: %v", err)
|
||||
}
|
||||
downloaded, readErr := io.ReadAll(download.Reader)
|
||||
closeErr := download.Reader.Close()
|
||||
if readErr != nil || closeErr != nil {
|
||||
t.Fatalf("read relayed restore artifact: read=%v close=%v", readErr, closeErr)
|
||||
}
|
||||
if !bytes.Equal(downloaded, artifact) {
|
||||
t.Fatalf("relayed restore artifact differs: %q", downloaded)
|
||||
}
|
||||
if _, err := h.service.GetAgentRestoreSpec(ctx, other, restore.ID); err == nil {
|
||||
t.Fatal("expected non-owner node to be forbidden from restore spec")
|
||||
}
|
||||
|
||||
@@ -299,6 +299,20 @@ func (s *VerificationService) executeLocally(ctx context.Context, verID uint, ta
|
||||
logger.Errorf("创建存储客户端失败:%v", err)
|
||||
return
|
||||
}
|
||||
if backupRecord.BackupKind == model.BackupKindRepository {
|
||||
logger.Infof("验证 CDC 仓库快照及全部引用块:%s", backupRecord.StoragePath)
|
||||
report, verifyErr := backup.NewRepositoryStore(s.cipher.Key()).Verify(ctx, provider, backupRecord.StoragePath, backupRecord.Checksum)
|
||||
if verifyErr != nil {
|
||||
errMessage = verifyErr.Error()
|
||||
summary = "CDC 仓库完整性校验失败"
|
||||
logger.Errorf("验证未通过:%v", verifyErr)
|
||||
return
|
||||
}
|
||||
status = model.VerificationRecordStatusSuccess
|
||||
summary = fmt.Sprintf("CDC 仓库完整性校验通过:%d 个条目、%d 个唯一块、%d bytes", report.Entries, report.Chunks, report.Bytes)
|
||||
logger.Infof("%s", summary)
|
||||
return
|
||||
}
|
||||
fileName := backupRecord.FileName
|
||||
if strings.TrimSpace(fileName) == "" {
|
||||
fileName = filepath.Base(backupRecord.StoragePath)
|
||||
|
||||
@@ -2,6 +2,7 @@ package rclone
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"sort"
|
||||
@@ -68,13 +69,56 @@ func (p *Provider) Download(ctx context.Context, objectKey string) (io.ReadClose
|
||||
return reader, nil
|
||||
}
|
||||
|
||||
// DownloadRange reads one slice from an object. Most object-storage backends
|
||||
// map this to a native HTTP Range request. Backends that reject ranged reads
|
||||
// fall back to a full stream while preserving the same interface contract.
|
||||
func (p *Provider) DownloadRange(ctx context.Context, objectKey string, offset, length int64) (io.ReadCloser, error) {
|
||||
if offset < 0 || length <= 0 {
|
||||
return nil, fmt.Errorf("rclone download range %s: invalid offset=%d length=%d", objectKey, offset, length)
|
||||
}
|
||||
obj, err := p.rfs.NewObject(ctx, objectKey)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("rclone find object %s: %w", objectKey, err)
|
||||
}
|
||||
reader, rangeErr := obj.Open(ctx, &fs.RangeOption{Start: offset, End: offset + length - 1})
|
||||
if rangeErr == nil {
|
||||
return reader, nil
|
||||
}
|
||||
reader, err = obj.Open(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("rclone download range %s (range: %v; fallback: %w)", objectKey, rangeErr, err)
|
||||
}
|
||||
if offset > 0 {
|
||||
if _, err := io.CopyN(io.Discard, reader, offset); err != nil {
|
||||
closeErr := reader.Close()
|
||||
return nil, errors.Join(fmt.Errorf("rclone seek object %s: %w", objectKey, err), closeErr)
|
||||
}
|
||||
}
|
||||
return &limitedReadCloser{Reader: io.LimitReader(reader, length), closer: reader}, nil
|
||||
}
|
||||
|
||||
type limitedReadCloser struct {
|
||||
io.Reader
|
||||
closer io.Closer
|
||||
}
|
||||
|
||||
func (r *limitedReadCloser) Close() error {
|
||||
return r.closer.Close()
|
||||
}
|
||||
|
||||
// Delete 通过 rclone 删除远端对象。
|
||||
func (p *Provider) Delete(ctx context.Context, objectKey string) error {
|
||||
obj, err := p.rfs.NewObject(ctx, objectKey)
|
||||
if err != nil {
|
||||
if errors.Is(err, fs.ErrorObjectNotFound) || errors.Is(err, fs.ErrorDirNotFound) {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("rclone find object %s: %w", objectKey, err)
|
||||
}
|
||||
if err := obj.Remove(ctx); err != nil {
|
||||
if errors.Is(err, fs.ErrorObjectNotFound) || errors.Is(err, fs.ErrorDirNotFound) {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("rclone delete %s: %w", objectKey, err)
|
||||
}
|
||||
return nil
|
||||
@@ -102,6 +146,9 @@ func (p *Provider) List(ctx context.Context, prefix string) ([]storage.ObjectInf
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
if errors.Is(err, fs.ErrorDirNotFound) || errors.Is(err, fs.ErrorObjectNotFound) {
|
||||
return []storage.ObjectInfo{}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("rclone list %s: %w", prefix, err)
|
||||
}
|
||||
return items, nil
|
||||
|
||||
@@ -34,6 +34,14 @@ const (
|
||||
TypeFTP = string(ProviderTypeFTP)
|
||||
)
|
||||
|
||||
const (
|
||||
// TransferModeDirect lets an Agent write to a network-accessible backend.
|
||||
TransferModeDirect = "direct"
|
||||
// TransferModeMasterRelay streams an artifact through the authenticated
|
||||
// Agent API so a remote source can use storage mounted only on the Master.
|
||||
TransferModeMasterRelay = "master_relay"
|
||||
)
|
||||
|
||||
type ObjectInfo struct {
|
||||
Key string `json:"key"`
|
||||
Size int64 `json:"size"`
|
||||
@@ -49,6 +57,13 @@ type StorageProvider interface {
|
||||
List(ctx context.Context, prefix string) ([]ObjectInfo, error)
|
||||
}
|
||||
|
||||
// StorageRangeDownloader is an optional capability used by packed repository
|
||||
// backups. Implementations return exactly the requested byte range when the
|
||||
// backend supports ranged reads and may transparently fall back to a full read.
|
||||
type StorageRangeDownloader interface {
|
||||
DownloadRange(ctx context.Context, objectKey string, offset, length int64) (io.ReadCloser, error)
|
||||
}
|
||||
|
||||
type ProviderFactory interface {
|
||||
Type() ProviderType
|
||||
}
|
||||
@@ -92,7 +107,8 @@ func ParseProviderType(value string) ProviderType {
|
||||
}
|
||||
|
||||
type LocalDiskConfig struct {
|
||||
BasePath string `json:"basePath"`
|
||||
BasePath string `json:"basePath"`
|
||||
MasterRelay bool `json:"masterRelay"`
|
||||
}
|
||||
|
||||
type S3Config struct {
|
||||
@@ -151,4 +167,3 @@ type FTPConfig struct {
|
||||
type StorageDirCleaner interface {
|
||||
RemoveEmptyDirs(ctx context.Context, prefix string) error
|
||||
}
|
||||
|
||||
|
||||
@@ -280,6 +280,10 @@ export function BackupRecordLogDrawer({ visible, recordId, onCancel, onChanged }
|
||||
{ label: '文件名', value: record.fileName || '-' },
|
||||
{ label: '文件大小', value: formatBytes(record.fileSize) },
|
||||
{ label: '存储路径', value: record.storagePath || '-' },
|
||||
...(record.storageTransferMode ? [{
|
||||
label: '传输路径',
|
||||
value: record.storageTransferMode === 'master_relay' ? 'Master 流式中转' : 'Agent 直传',
|
||||
}] : []),
|
||||
{ label: '开始时间', value: formatDateTime(record.startedAt) },
|
||||
{ label: '完成时间', value: formatDateTime(record.completedAt) },
|
||||
{ label: '耗时', value: formatDuration(record.durationSeconds) },
|
||||
@@ -316,14 +320,16 @@ export function BackupRecordLogDrawer({ visible, recordId, onCancel, onChanged }
|
||||
</Button>
|
||||
)}
|
||||
</Space>
|
||||
{record.storageUploadResults && record.storageUploadResults.length > 1 && (
|
||||
{record.storageUploadResults && (record.storageUploadResults.length > 1 || record.storageUploadResults.some((result) => result.transferMode)) && (
|
||||
<div>
|
||||
<Typography.Title heading={6}>存储目标上传结果</Typography.Title>
|
||||
<Descriptions
|
||||
column={1}
|
||||
data={record.storageUploadResults.map((r: StorageUploadResultItem) => ({
|
||||
label: r.storageTargetName,
|
||||
value: r.status === 'success' ? '上传成功' : `上传失败: ${r.error || '未知错误'}`,
|
||||
value: r.status === 'success'
|
||||
? `上传成功${r.transferMode === 'master_relay' ? ' · Master 流式中转' : r.transferMode === 'direct' ? ' · Agent 直传' : ''}`
|
||||
: `上传失败: ${r.error || '未知错误'}`,
|
||||
}))}
|
||||
/>
|
||||
</div>
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { Alert, Button, Divider, Drawer, Input, InputNumber, Select, Space, Steps, Switch, Typography, Grid } from '@arco-design/web-react'
|
||||
import { IconDelete, IconPlus } from '@arco-design/web-react/icon'
|
||||
import { IconDelete, IconPlus } from '../icons'
|
||||
import { useEffect, useMemo, useState } from 'react'
|
||||
import { CronInput } from '../CronInput'
|
||||
import type { StorageTargetDetail, StorageTargetPayload, StorageTargetSummary } from '../../types/storage-targets'
|
||||
@@ -9,6 +9,8 @@ import type { NodeSummary } from '../../types/nodes'
|
||||
import { DatabasePicker } from '../common/DatabasePicker'
|
||||
import { DirectoryPicker } from '../common/DirectoryPicker'
|
||||
import { StorageTargetFormDrawer } from '../storage-targets/StorageTargetFormDrawer'
|
||||
import { StorageTargetName } from '../storage-targets/StorageTargetName'
|
||||
import { SourceServerSelector } from './SourceServerSelector'
|
||||
import {
|
||||
backupCompressionOptions,
|
||||
backupTaskTypeOptions,
|
||||
@@ -168,7 +170,7 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa
|
||||
return 0
|
||||
})
|
||||
return sorted.map((item) => ({
|
||||
label: item.starred ? `★ ${item.name}` : item.name,
|
||||
label: <StorageTargetName name={item.name} starred={item.starred} />,
|
||||
value: item.id,
|
||||
disabled: !item.enabled,
|
||||
}))
|
||||
@@ -176,21 +178,6 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa
|
||||
[storageTargets],
|
||||
)
|
||||
|
||||
// 执行节点选项:本地节点显示 "本机 (local)",远程节点带状态后缀
|
||||
const nodeOptions = useMemo(() => {
|
||||
const list = nodes ?? []
|
||||
return [
|
||||
{ label: '本机 (Master)', value: 0 },
|
||||
...list
|
||||
.filter((item) => !item.isLocal)
|
||||
.map((item) => ({
|
||||
label: `${item.name}${item.status === 'online' ? '' : '(离线)'}`,
|
||||
value: item.id,
|
||||
disabled: item.status !== 'online',
|
||||
})),
|
||||
]
|
||||
}, [nodes])
|
||||
|
||||
function updateDraft(patch: Partial<BackupTaskPayload>) {
|
||||
setDraft((current) => ({ ...current, ...patch }))
|
||||
}
|
||||
@@ -251,6 +238,12 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa
|
||||
if (validPaths.length === 0 && !value.sourcePath.trim()) {
|
||||
return '请输入至少一个源路径'
|
||||
}
|
||||
if (value.backupMode === 'repository' && (((value.nodeId ?? 0) > 0 && value.nodeId !== localNodeId) || value.nodePoolTag?.trim())) {
|
||||
return 'CDC 仓库模式当前仅支持 Master 本机执行'
|
||||
}
|
||||
if (value.backupMode === 'repository' && value.replicationTargetIds.length > 0) {
|
||||
return 'CDC 仓库模式请直接多选存储目标,不能使用对象级副本复制'
|
||||
}
|
||||
}
|
||||
if (isSQLiteBackupTask(value.type) && !value.dbPath.trim()) {
|
||||
return '请输入 SQLite 数据库路径'
|
||||
@@ -306,33 +299,26 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa
|
||||
<Typography.Text>备份类型</Typography.Text>
|
||||
<Select value={draft.type} options={backupTaskTypeOptions as unknown as { label: string; value: string }[]} onChange={(value) => updateTaskType(value as BackupTaskType)} />
|
||||
</div>
|
||||
<div>
|
||||
<Typography.Text>执行节点</Typography.Text>
|
||||
<Select
|
||||
value={draft.nodeId ?? 0}
|
||||
options={nodeOptions}
|
||||
onChange={(value) => {
|
||||
const nodeId = Number(value ?? 0)
|
||||
// 固定节点与节点池互斥:切到固定节点时清空 NodePoolTag
|
||||
updateDraft(nodeId > 0 ? { nodeId, nodePoolTag: '' } : { nodeId })
|
||||
}}
|
||||
/>
|
||||
<Typography.Paragraph type="secondary" style={{ marginBottom: 0, marginTop: 4 }}>
|
||||
任务在所选节点上执行备份与恢复;源路径/数据库以该节点视角解析。远程节点需先在"节点管理"中安装 Agent。
|
||||
</Typography.Paragraph>
|
||||
</div>
|
||||
<div>
|
||||
<Typography.Text>节点池标签(可选)</Typography.Text>
|
||||
<Input
|
||||
placeholder="填写标签后从节点池动态调度(与固定节点互斥)"
|
||||
value={draft.nodePoolTag ?? ''}
|
||||
disabled={(draft.nodeId ?? 0) > 0}
|
||||
onChange={(value) => updateDraft({ nodePoolTag: value })}
|
||||
/>
|
||||
<Typography.Paragraph type="secondary" style={{ marginBottom: 0, marginTop: 4 }}>
|
||||
执行节点选"本机 / 未指定"时可启用;从节点 Labels 命中此 tag 的在线节点中按当前运行任务数最少的挑选一台执行。
|
||||
</Typography.Paragraph>
|
||||
</div>
|
||||
<SourceServerSelector
|
||||
nodeId={draft.nodeId ?? 0}
|
||||
nodePoolTag={draft.nodePoolTag ?? ''}
|
||||
localNodeId={localNodeId}
|
||||
nodes={nodes}
|
||||
onNodeChange={(nodeId) => {
|
||||
// 固定源服务器与服务器池互斥;CDC 仓库仍固定在 Master 单写者。
|
||||
updateDraft(nodeId > 0
|
||||
? {
|
||||
nodeId,
|
||||
nodePoolTag: '',
|
||||
backupMode: nodeId !== localNodeId && draft.backupMode === 'repository' ? 'full' : draft.backupMode,
|
||||
}
|
||||
: { nodeId })
|
||||
}}
|
||||
onNodePoolTagChange={(value) => updateDraft({
|
||||
nodePoolTag: value,
|
||||
backupMode: value.trim() && draft.backupMode === 'repository' ? 'full' : draft.backupMode,
|
||||
})}
|
||||
/>
|
||||
<div>
|
||||
<Typography.Text>Cron 表达式</Typography.Text>
|
||||
<CronInput value={draft.cronExpr} onChange={(value) => updateDraft({ cronExpr: value })} />
|
||||
@@ -587,6 +573,11 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa
|
||||
</Button>
|
||||
)}
|
||||
</Space>
|
||||
{((draft.nodeId ?? 0) > 0 && draft.nodeId !== localNodeId) || draft.nodePoolTag?.trim() ? (
|
||||
<Typography.Paragraph type="secondary" style={{ marginBottom: 0, marginTop: 4 }}>
|
||||
远程源服务器会直传 S3、WebDAV 等网络存储;本地磁盘目标启用 Master 中转后,文件经 Agent 认证 API 流式写入中央目录。跨公网部署请为 Master 配置 HTTPS。
|
||||
</Typography.Paragraph>
|
||||
) : null}
|
||||
</div>
|
||||
<div>
|
||||
<Typography.Text>压缩策略</Typography.Text>
|
||||
@@ -600,8 +591,11 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa
|
||||
options={[
|
||||
{ label: '全量备份', value: 'full' },
|
||||
{ label: '差异备份(仅文件、本机)', value: 'differential' },
|
||||
{ label: 'CDC 去重仓库(仅文件、本机)', value: 'repository' },
|
||||
]}
|
||||
onChange={(value) => updateDraft({ backupMode: value as BackupMode })}
|
||||
onChange={(value) => updateDraft(value === 'repository'
|
||||
? { backupMode: value as BackupMode, nodeId: 0, nodePoolTag: '', replicationTargetIds: [] }
|
||||
: { backupMode: value as BackupMode })}
|
||||
/>
|
||||
{draft.backupMode === 'differential' && (
|
||||
<div style={{ marginTop: 8 }}>
|
||||
@@ -618,6 +612,11 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa
|
||||
/>
|
||||
</div>
|
||||
)}
|
||||
{draft.backupMode === 'repository' && (
|
||||
<Typography.Paragraph type="secondary" style={{ marginBottom: 0, marginTop: 8 }}>
|
||||
文件按内容边界切块并写入全局分块池;相同数据跨文件、跨快照只上传一次。新块会合并为 pack,恢复时通过索引按需读取。当前版本采用单写者索引,因此固定在 Master 本机执行;需要多副本时请直接多选上方存储目标。
|
||||
</Typography.Paragraph>
|
||||
)}
|
||||
</div>
|
||||
)}
|
||||
<div>
|
||||
@@ -736,10 +735,13 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa
|
||||
value={draft.replicationTargetIds}
|
||||
placeholder="选择副本目标(不选 = 不启用复制)"
|
||||
options={storageTargetOptions.filter((opt) => !(draft.storageTargetIds ?? []).includes(opt.value as number))}
|
||||
disabled={draft.backupMode === 'repository'}
|
||||
onChange={(values: number[]) => updateDraft({ replicationTargetIds: values })}
|
||||
/>
|
||||
<Typography.Paragraph type="secondary" style={{ marginBottom: 0, marginTop: 4 }}>
|
||||
备份成功后自动镜像到副本存储。满足 3-2-1 规则:至少 2 份副本、至少 1 份异地。建议选不同 provider 的目标。
|
||||
{draft.backupMode === 'repository'
|
||||
? 'CDC 仓库包含共享 pack 与索引,不能只复制单个快照对象;请在“存储目标”中直接多选以生成完整仓库副本。'
|
||||
: '备份成功后自动镜像到副本存储。满足 3-2-1 规则:至少 2 份副本、至少 1 份异地。建议选不同 provider 的目标。'}
|
||||
</Typography.Paragraph>
|
||||
</div>
|
||||
|
||||
|
||||
35
web/src/components/backup-tasks/SourceServerSelector.test.ts
Normal file
35
web/src/components/backup-tasks/SourceServerSelector.test.ts
Normal file
@@ -0,0 +1,35 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import type { NodeSummary } from '../../types/nodes'
|
||||
import { buildSourceServerOptions } from './SourceServerSelector'
|
||||
|
||||
function node(id: number, name: string, status: NodeSummary['status'], isLocal = false): NodeSummary {
|
||||
return {
|
||||
id,
|
||||
name,
|
||||
status,
|
||||
isLocal,
|
||||
hostname: '',
|
||||
ipAddress: '',
|
||||
os: '',
|
||||
arch: '',
|
||||
agentVersion: '',
|
||||
lastSeen: '',
|
||||
createdAt: '',
|
||||
}
|
||||
}
|
||||
|
||||
describe('buildSourceServerOptions', () => {
|
||||
it('keeps Master first and disables offline remote sources', () => {
|
||||
const options = buildSourceServerOptions([
|
||||
node(1, 'local', 'online', true),
|
||||
node(2, 'source-b', 'online'),
|
||||
node(3, 'source-c', 'offline'),
|
||||
])
|
||||
|
||||
expect(options).toEqual([
|
||||
{ label: 'Master 本机', value: 0, disabled: false },
|
||||
{ label: 'source-b', value: 2, disabled: false },
|
||||
{ label: 'source-c(离线)', value: 3, disabled: true },
|
||||
])
|
||||
})
|
||||
})
|
||||
57
web/src/components/backup-tasks/SourceServerSelector.tsx
Normal file
57
web/src/components/backup-tasks/SourceServerSelector.tsx
Normal file
@@ -0,0 +1,57 @@
|
||||
import { Input, Select, Typography } from '@arco-design/web-react'
|
||||
import { useMemo } from 'react'
|
||||
import type { NodeSummary } from '../../types/nodes'
|
||||
|
||||
interface SourceServerSelectorProps {
|
||||
nodeId: number
|
||||
nodePoolTag: string
|
||||
localNodeId?: number
|
||||
nodes?: NodeSummary[]
|
||||
onNodeChange: (nodeId: number) => void
|
||||
onNodePoolTagChange: (tag: string) => void
|
||||
}
|
||||
|
||||
export function buildSourceServerOptions(nodes: NodeSummary[] = []) {
|
||||
return [
|
||||
{ label: 'Master 本机', value: 0, disabled: false },
|
||||
...nodes
|
||||
.filter((node) => !node.isLocal)
|
||||
.map((node) => ({
|
||||
label: `${node.name}${node.status === 'online' ? '' : '(离线)'}`,
|
||||
value: node.id,
|
||||
disabled: node.status !== 'online',
|
||||
})),
|
||||
]
|
||||
}
|
||||
|
||||
export function SourceServerSelector({ nodeId, nodePoolTag, localNodeId, nodes, onNodeChange, onNodePoolTagChange }: SourceServerSelectorProps) {
|
||||
const options = useMemo(() => buildSourceServerOptions(nodes), [nodes])
|
||||
const selectedNode = nodes?.find((node) => node.id === nodeId)
|
||||
const isRemote = nodeId > 0 && nodeId !== localNodeId
|
||||
|
||||
return (
|
||||
<>
|
||||
<div>
|
||||
<Typography.Text>源服务器</Typography.Text>
|
||||
<Select value={nodeId} options={options} onChange={(value) => onNodeChange(Number(value ?? 0))} />
|
||||
<Typography.Paragraph type="secondary" style={{ marginBottom: 0, marginTop: 4 }}>
|
||||
{isRemote
|
||||
? `源路径与数据库在 ${selectedNode?.name ?? '远程服务器'} 上解析,由 Agent 就地生成备份。网络存储由 Agent 直传;启用 Master 中转的本地磁盘目标会通过认证连接写入中央目录。`
|
||||
: '源路径与数据库在 Master 本机解析。要集中备份其他服务器,请先在“节点管理”安装 Agent,再在这里选择对应源服务器。'}
|
||||
</Typography.Paragraph>
|
||||
</div>
|
||||
<div>
|
||||
<Typography.Text>源服务器池标签(可选)</Typography.Text>
|
||||
<Input
|
||||
placeholder="按标签从在线源服务器中动态选择(与固定源服务器互斥)"
|
||||
value={nodePoolTag}
|
||||
disabled={nodeId > 0}
|
||||
onChange={onNodePoolTagChange}
|
||||
/>
|
||||
<Typography.Paragraph type="secondary" style={{ marginBottom: 0, marginTop: 4 }}>
|
||||
仅在选择 Master 本机时可填写;系统从 Labels 命中该标签的在线 Agent 中选择当前运行任务最少的一台。
|
||||
</Typography.Paragraph>
|
||||
</div>
|
||||
</>
|
||||
)
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
import { Button, Input, Message, Modal, Space, Spin, Tree, Typography, Empty } from '@arco-design/web-react'
|
||||
import { IconFolder, IconFile, IconFolderAdd } from '@arco-design/web-react/icon'
|
||||
import { IconFolder, IconFile, IconFolderAdd } from '../icons'
|
||||
import { useCallback, useState } from 'react'
|
||||
import { listNodeDirectory } from '../../services/nodes'
|
||||
import type { DirEntry } from '../../types/nodes'
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { Badge, Button, Drawer, Empty, Notification, Space, Tag, Typography } from '@arco-design/web-react'
|
||||
import { IconNotification } from '@arco-design/web-react/icon'
|
||||
import { IconNotification } from '../icons'
|
||||
import { useEffect, useState } from 'react'
|
||||
import { useEventStream, type SystemEvent } from '../../hooks/useEventStream'
|
||||
import { useEventStore } from '../../stores/events'
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { Empty, Input, Modal, Space, Spin, Tag, Typography } from '@arco-design/web-react'
|
||||
import { IconSearch } from '@arco-design/web-react/icon'
|
||||
import { IconSearch } from '../icons'
|
||||
import { useCallback, useEffect, useRef, useState } from 'react'
|
||||
import { useNavigate } from 'react-router-dom'
|
||||
import { globalSearch, type SearchKind, type SearchResult, type SearchResultItem } from '../../services/search'
|
||||
|
||||
18
web/src/components/common/LanguageSwitcher.tsx
Normal file
18
web/src/components/common/LanguageSwitcher.tsx
Normal file
@@ -0,0 +1,18 @@
|
||||
import { Select } from '@arco-design/web-react'
|
||||
import { useTranslation } from 'react-i18next'
|
||||
import { languageOptions, normalizeLanguage, setApplicationLanguage, type SupportedLanguage } from '../../i18n'
|
||||
|
||||
export function LanguageSwitcher() {
|
||||
const { t, i18n } = useTranslation()
|
||||
const currentLanguage = normalizeLanguage(i18n.resolvedLanguage)
|
||||
|
||||
return (
|
||||
<Select
|
||||
aria-label={t('auth.language')}
|
||||
value={currentLanguage}
|
||||
options={languageOptions}
|
||||
style={{ width: 120 }}
|
||||
onChange={(value) => void setApplicationLanguage(value as SupportedLanguage)}
|
||||
/>
|
||||
)
|
||||
}
|
||||
50
web/src/components/icons/BackupServerIllustration.tsx
Normal file
50
web/src/components/icons/BackupServerIllustration.tsx
Normal file
@@ -0,0 +1,50 @@
|
||||
import type { SVGProps } from 'react'
|
||||
|
||||
export function BackupServerIllustration(props: SVGProps<SVGSVGElement>) {
|
||||
return (
|
||||
<svg
|
||||
width="320"
|
||||
height="320"
|
||||
viewBox="0 0 320 320"
|
||||
fill="none"
|
||||
xmlns="http://www.w3.org/2000/svg"
|
||||
aria-hidden="true"
|
||||
focusable="false"
|
||||
{...props}
|
||||
>
|
||||
<circle cx="160" cy="160" r="120" fill="white" fillOpacity="0.05">
|
||||
<animate attributeName="r" values="115;125;115" dur="4s" repeatCount="indefinite" />
|
||||
<animate attributeName="fill-opacity" values="0.03;0.08;0.03" dur="4s" repeatCount="indefinite" />
|
||||
</circle>
|
||||
<circle cx="160" cy="160" r="80" fill="white" fillOpacity="0.1">
|
||||
<animate attributeName="r" values="75;85;75" dur="3s" repeatCount="indefinite" />
|
||||
</circle>
|
||||
|
||||
<g>
|
||||
<animateTransform attributeName="transform" type="translate" values="0,0; 0,-8; 0,0" dur="5s" repeatCount="indefinite" />
|
||||
<path d="M120 120C120 111.163 137.909 104 160 104C182.091 104 200 111.163 200 120V144C200 152.837 182.091 160 160 160C137.909 160 120 152.837 120 144V120Z" fill="white" fillOpacity="0.95" />
|
||||
<ellipse cx="160" cy="120" rx="40" ry="16" fill="white" />
|
||||
<path d="M120 152C120 143.163 137.909 136 160 136C182.091 136 200 143.163 200 152V176C200 184.837 182.091 192 160 192C137.909 192 120 184.837 120 176V152Z" fill="white" fillOpacity="0.75" />
|
||||
<ellipse cx="160" cy="152" rx="40" ry="16" fill="white" fillOpacity="0.9" />
|
||||
<path d="M120 184C120 175.163 137.909 168 160 168C182.091 168 200 175.163 200 184V208C200 216.837 182.091 224 160 224C137.909 224 120 216.837 120 208V184Z" fill="white" fillOpacity="0.5" />
|
||||
<ellipse cx="160" cy="184" rx="40" ry="16" fill="white" fillOpacity="0.6" />
|
||||
|
||||
<g fill="var(--color-primary-6, #165dff)">
|
||||
<circle cx="140" cy="120" r="4">
|
||||
<animate attributeName="opacity" values="0.3;1;0.3" dur="2s" begin="0s" repeatCount="indefinite" />
|
||||
</circle>
|
||||
<circle cx="140" cy="152" r="4">
|
||||
<animate attributeName="opacity" values="0.3;1;0.3" dur="2s" begin="0.6s" repeatCount="indefinite" />
|
||||
</circle>
|
||||
<circle cx="140" cy="184" r="4">
|
||||
<animate attributeName="opacity" values="0.3;1;0.3" dur="2s" begin="1.2s" repeatCount="indefinite" />
|
||||
</circle>
|
||||
</g>
|
||||
|
||||
<path d="M160 120V152V184" stroke="var(--color-primary-6, #165dff)" strokeWidth="2" strokeDasharray="4 4" opacity="0.6">
|
||||
<animate attributeName="stroke-dashoffset" from="16" to="0" dur="1s" repeatCount="indefinite" />
|
||||
</path>
|
||||
</g>
|
||||
</svg>
|
||||
)
|
||||
}
|
||||
38
web/src/components/icons/index.ts
Normal file
38
web/src/components/icons/index.ts
Normal file
@@ -0,0 +1,38 @@
|
||||
export {
|
||||
IconBook,
|
||||
IconCheckCircle,
|
||||
IconCloud,
|
||||
IconCloudDownload,
|
||||
IconCommand,
|
||||
IconCopy,
|
||||
IconDashboard,
|
||||
IconDelete,
|
||||
IconDesktop,
|
||||
IconDown,
|
||||
IconDownload,
|
||||
IconEdit,
|
||||
IconFile,
|
||||
IconFilePdf,
|
||||
IconFolder,
|
||||
IconFolderAdd,
|
||||
IconHistory,
|
||||
IconInfoCircle,
|
||||
IconList,
|
||||
IconLock,
|
||||
IconMenuFold,
|
||||
IconMenuUnfold,
|
||||
IconMore,
|
||||
IconNotification,
|
||||
IconPlus,
|
||||
IconPoweroff,
|
||||
IconRefresh,
|
||||
IconSafe,
|
||||
IconSave,
|
||||
IconSearch,
|
||||
IconSettings,
|
||||
IconStarFill,
|
||||
IconStorage,
|
||||
IconUser,
|
||||
} from '@arco-design/web-react/icon'
|
||||
|
||||
export { BackupServerIllustration } from './BackupServerIllustration'
|
||||
@@ -16,7 +16,14 @@ interface StorageTargetFormDrawerProps {
|
||||
}
|
||||
|
||||
function createEmptyDraft(type: StorageTargetType = 'local_disk'): StorageTargetPayload {
|
||||
return { name: '', type, description: '', enabled: true, config: {}, quotaBytes: 0 }
|
||||
return {
|
||||
name: '',
|
||||
type,
|
||||
description: '',
|
||||
enabled: true,
|
||||
config: type === 'local_disk' ? { masterRelay: true } : {},
|
||||
quotaBytes: 0,
|
||||
}
|
||||
}
|
||||
|
||||
export function StorageTargetFormDrawer({
|
||||
@@ -207,7 +214,8 @@ export function StorageTargetFormDrawer({
|
||||
return label.toLowerCase().includes(input.toLowerCase())
|
||||
}}
|
||||
onChange={(value) => {
|
||||
setDraft((c) => ({ ...c, type: value as string, config: {} }))
|
||||
const config: StorageTargetPayload['config'] = value === 'local_disk' ? { masterRelay: true } : {}
|
||||
setDraft((c) => ({ ...c, type: value as string, config }))
|
||||
setTestResult(null)
|
||||
}}
|
||||
>
|
||||
|
||||
@@ -0,0 +1,14 @@
|
||||
import { render, screen } from '@testing-library/react'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { StorageTargetName } from './StorageTargetName'
|
||||
|
||||
describe('StorageTargetName', () => {
|
||||
it('uses an SVG icon for starred targets without character symbols', () => {
|
||||
const { container } = render(<StorageTargetName name="Central storage" starred />)
|
||||
|
||||
expect(screen.getByText('Central storage')).toBeInTheDocument()
|
||||
expect(screen.getByLabelText('Central storage,已收藏')).toBeInTheDocument()
|
||||
expect(container.querySelector('svg')).not.toBeNull()
|
||||
expect(container.textContent).not.toContain(String.fromCodePoint(0x2605))
|
||||
})
|
||||
})
|
||||
18
web/src/components/storage-targets/StorageTargetName.tsx
Normal file
18
web/src/components/storage-targets/StorageTargetName.tsx
Normal file
@@ -0,0 +1,18 @@
|
||||
import { IconStarFill } from '../icons'
|
||||
|
||||
interface StorageTargetNameProps {
|
||||
name: string
|
||||
starred?: boolean
|
||||
}
|
||||
|
||||
export function StorageTargetName({ name, starred = false }: StorageTargetNameProps) {
|
||||
return (
|
||||
<span
|
||||
aria-label={starred ? `${name},已收藏` : undefined}
|
||||
style={{ display: 'inline-flex', alignItems: 'center', gap: 4 }}
|
||||
>
|
||||
{starred ? <IconStarFill /> : null}
|
||||
<span>{name}</span>
|
||||
</span>
|
||||
)
|
||||
}
|
||||
@@ -4,8 +4,9 @@ import { getStorageTargetFieldConfigs, getStorageTargetTypeLabel } from './field
|
||||
describe('storage target field config', () => {
|
||||
it('returns local disk field config', () => {
|
||||
const fields = getStorageTargetFieldConfigs('local_disk')
|
||||
expect(fields).toHaveLength(1)
|
||||
expect(fields).toHaveLength(2)
|
||||
expect(fields[0]?.key).toBe('basePath')
|
||||
expect(fields[1]).toMatchObject({ key: 'masterRelay', type: 'switch' })
|
||||
})
|
||||
|
||||
it('returns readable type labels', () => {
|
||||
|
||||
@@ -7,6 +7,7 @@ import type { StorageTargetFieldConfig, StorageTargetType } from '../../types/st
|
||||
const BUILTIN_FIELD_CONFIG: Record<string, StorageTargetFieldConfig[]> = {
|
||||
local_disk: [
|
||||
{ key: 'basePath', label: '基础目录', type: 'input', required: true, placeholder: '/data/backups', description: 'BackupX 将在该目录下创建和管理备份文件。' },
|
||||
{ key: 'masterRelay', label: '远程备份经 Master 中转', type: 'switch', description: '开启后,远程 Agent 会把产物流式传给 Master 并写入上述目录;关闭则沿用 Agent 本机目录。' },
|
||||
],
|
||||
s3: [
|
||||
{ key: 'endpoint', label: 'Endpoint', type: 'input', required: true, placeholder: 'https://s3.amazonaws.com' },
|
||||
|
||||
21
web/src/i18n.test.ts
Normal file
21
web/src/i18n.test.ts
Normal file
@@ -0,0 +1,21 @@
|
||||
import i18n, { normalizeLanguage, setApplicationLanguage } from './i18n'
|
||||
|
||||
describe('application language', () => {
|
||||
afterEach(async () => {
|
||||
await setApplicationLanguage('zh-CN')
|
||||
})
|
||||
|
||||
it('normalizes supported English variants', () => {
|
||||
expect(normalizeLanguage('en')).toBe('en-US')
|
||||
expect(normalizeLanguage('en-GB')).toBe('en-US')
|
||||
expect(normalizeLanguage('zh-CN')).toBe('zh-CN')
|
||||
})
|
||||
|
||||
it('persists the language selected before login', async () => {
|
||||
await setApplicationLanguage('en-US')
|
||||
|
||||
expect(localStorage.getItem('backupx-language')).toBe('en-US')
|
||||
expect(document.documentElement.lang).toBe('en-US')
|
||||
expect(i18n.t('auth.setupTitle')).toBe('System setup')
|
||||
})
|
||||
})
|
||||
@@ -3,7 +3,22 @@ import { initReactI18next } from 'react-i18next'
|
||||
import zhCN from './locales/zh-CN.json'
|
||||
import enUS from './locales/en-US.json'
|
||||
|
||||
const savedLanguage = localStorage.getItem('backupx-language') || 'zh-CN'
|
||||
export type SupportedLanguage = 'zh-CN' | 'en-US'
|
||||
|
||||
export const languageOptions: Array<{ label: string; value: SupportedLanguage }> = [
|
||||
{ label: '中文', value: 'zh-CN' },
|
||||
{ label: 'English', value: 'en-US' },
|
||||
]
|
||||
|
||||
export function normalizeLanguage(value?: string | null): SupportedLanguage {
|
||||
return value?.toLowerCase().startsWith('en') ? 'en-US' : 'zh-CN'
|
||||
}
|
||||
|
||||
const savedLanguage = normalizeLanguage(typeof window === 'undefined' ? null : window.localStorage.getItem('backupx-language'))
|
||||
|
||||
if (typeof document !== 'undefined') {
|
||||
document.documentElement.lang = savedLanguage
|
||||
}
|
||||
|
||||
i18n.use(initReactI18next).init({
|
||||
resources: {
|
||||
@@ -17,4 +32,14 @@ i18n.use(initReactI18next).init({
|
||||
},
|
||||
})
|
||||
|
||||
export async function setApplicationLanguage(language: SupportedLanguage) {
|
||||
if (typeof window !== 'undefined') {
|
||||
window.localStorage.setItem('backupx-language', language)
|
||||
}
|
||||
if (typeof document !== 'undefined') {
|
||||
document.documentElement.lang = language
|
||||
}
|
||||
await i18n.changeLanguage(language)
|
||||
}
|
||||
|
||||
export default i18n
|
||||
|
||||
@@ -21,7 +21,7 @@ import {
|
||||
IconDesktop,
|
||||
IconList,
|
||||
IconFilePdf,
|
||||
} from '@arco-design/web-react/icon'
|
||||
} from '../components/icons'
|
||||
import { useState } from 'react'
|
||||
import { Outlet, useLocation, useNavigate } from 'react-router-dom'
|
||||
import {
|
||||
|
||||
@@ -9,7 +9,7 @@ import {
|
||||
IconDashboard,
|
||||
IconInfoCircle,
|
||||
IconPoweroff,
|
||||
} from '@arco-design/web-react/icon';
|
||||
} from '../components/icons';
|
||||
import { Outlet, useLocation, useNavigate } from 'react-router-dom';
|
||||
|
||||
import { useAuthStore } from '../stores/auth';
|
||||
|
||||
@@ -40,7 +40,62 @@
|
||||
"oldPassword": "Old Password",
|
||||
"newPassword": "New Password",
|
||||
"loginTitle": "Sign in to BackupX",
|
||||
"loginSubtitle": "Linux Server Backup Manager"
|
||||
"loginSubtitle": "Linux Server Backup Manager",
|
||||
"language": "Language",
|
||||
"bannerTitle": "Protect your data",
|
||||
"bannerSubtitle": "Secure and reliable server backup management",
|
||||
"setupStatusTitle": "Connect to BackupX",
|
||||
"checkingStatus": "Checking system initialization status...",
|
||||
"statusErrorTitle": "Unable to check initialization status",
|
||||
"statusErrorDescription": "The web console could not reach the BackupX setup API. Confirm that the service is running, then retry.",
|
||||
"retry": "Retry",
|
||||
"setupTitle": "System setup",
|
||||
"welcomeTitle": "Welcome back",
|
||||
"setupSubtitle": "Create the first administrator account.",
|
||||
"welcomeSubtitle": "Enter an administrator account to open the console.",
|
||||
"displayName": "Display name",
|
||||
"displayNamePlaceholder": "Administrator display name",
|
||||
"usernamePlaceholder": "Administrator username",
|
||||
"passwordPlaceholder": "Password",
|
||||
"setupPasswordPlaceholder": "At least 8 characters",
|
||||
"setupSubmit": "Create administrator and sign in",
|
||||
"setupSuccess": "Setup complete. Opening the console...",
|
||||
"loginSuccess": "Signed in",
|
||||
"credentialsRequired": "Enter your username and password first",
|
||||
"mfaCode": "Verification or recovery code",
|
||||
"mfaCodePlaceholder": "TOTP, recovery, email, or SMS code",
|
||||
"sendEmailCode": "Send email code",
|
||||
"sendSmsCode": "Send SMS code",
|
||||
"emailCodeSent": "Email verification code sent",
|
||||
"smsCodeSent": "SMS verification code sent",
|
||||
"usePasskey": "Use passkey",
|
||||
"trustDevice": "Trust this device for 30 days",
|
||||
"verifyAndLogin": "Verify and sign in",
|
||||
"requestFailed": "The request failed. Please try again.",
|
||||
"validation": {
|
||||
"displayNameRequired": "Enter a display name",
|
||||
"usernameRequired": "Enter a username",
|
||||
"usernameLength": "Username must contain at least 3 characters",
|
||||
"passwordRequired": "Enter a password",
|
||||
"passwordLength": "Password must contain at least 8 characters",
|
||||
"mfaRequired": "Enter a verification or recovery code",
|
||||
"mfaLength": "Code must contain 6 to 32 characters"
|
||||
},
|
||||
"errors": {
|
||||
"AUTH_INVALID_CREDENTIALS": "Invalid username or password",
|
||||
"AUTH_WRONG_PASSWORD": "Invalid username or password",
|
||||
"AUTH_USER_DISABLED": "This account is disabled",
|
||||
"AUTH_RATE_LIMITED": "Too many attempts. Please try again later.",
|
||||
"AUTH_2FA_REQUIRED": "Complete two-factor authentication to continue",
|
||||
"AUTH_2FA_INVALID": "The verification or recovery code is invalid",
|
||||
"AUTH_SETUP_DISABLED": "BackupX is already initialized. Sign in instead.",
|
||||
"AUTH_USERNAME_EXISTS": "This username already exists",
|
||||
"AUTH_EMAIL_OTP_DISABLED": "Email verification is not enabled",
|
||||
"AUTH_SMS_OTP_DISABLED": "SMS verification is not enabled",
|
||||
"AUTH_EMAIL_REQUIRED": "No email address is configured for this account",
|
||||
"AUTH_PHONE_REQUIRED": "No phone number is configured for this account",
|
||||
"AUTH_WEBAUTHN_NOT_ENABLED": "No passkey is configured for this account"
|
||||
}
|
||||
},
|
||||
"dashboard": {
|
||||
"title": "Dashboard",
|
||||
|
||||
@@ -40,7 +40,62 @@
|
||||
"oldPassword": "旧密码",
|
||||
"newPassword": "新密码",
|
||||
"loginTitle": "登录 BackupX",
|
||||
"loginSubtitle": "Linux 服务器备份管理系统"
|
||||
"loginSubtitle": "Linux 服务器备份管理系统",
|
||||
"language": "语言",
|
||||
"bannerTitle": "守护您的数据资产",
|
||||
"bannerSubtitle": "安全、可靠的服务器备份管理平台",
|
||||
"setupStatusTitle": "连接 BackupX",
|
||||
"checkingStatus": "正在检查系统初始化状态...",
|
||||
"statusErrorTitle": "无法检查初始化状态",
|
||||
"statusErrorDescription": "Web 控制台无法访问 BackupX 初始化接口。请确认服务已启动,然后重试。",
|
||||
"retry": "重试",
|
||||
"setupTitle": "系统初始化",
|
||||
"welcomeTitle": "欢迎回来",
|
||||
"setupSubtitle": "请创建首个管理员账户以完成初始化。",
|
||||
"welcomeSubtitle": "请输入管理员账户信息登录控制台。",
|
||||
"displayName": "显示名称",
|
||||
"displayNamePlaceholder": "请输入管理员显示名称",
|
||||
"usernamePlaceholder": "请输入管理员用户名",
|
||||
"passwordPlaceholder": "请输入密码",
|
||||
"setupPasswordPlaceholder": "请输入至少 8 位密码",
|
||||
"setupSubmit": "创建管理员并登录",
|
||||
"setupSuccess": "初始化完成,正在进入控制台...",
|
||||
"loginSuccess": "登录成功",
|
||||
"credentialsRequired": "请先输入用户名和密码",
|
||||
"mfaCode": "验证码或恢复码",
|
||||
"mfaCodePlaceholder": "请输入 TOTP、恢复码、邮件或短信验证码",
|
||||
"sendEmailCode": "发送邮件验证码",
|
||||
"sendSmsCode": "发送短信验证码",
|
||||
"emailCodeSent": "邮件验证码已发送",
|
||||
"smsCodeSent": "短信验证码已发送",
|
||||
"usePasskey": "使用通行密钥",
|
||||
"trustDevice": "信任此设备 30 天",
|
||||
"verifyAndLogin": "验证并登录",
|
||||
"requestFailed": "请求失败,请稍后重试",
|
||||
"validation": {
|
||||
"displayNameRequired": "请输入显示名称",
|
||||
"usernameRequired": "请输入用户名",
|
||||
"usernameLength": "用户名至少需要 3 个字符",
|
||||
"passwordRequired": "请输入密码",
|
||||
"passwordLength": "密码至少需要 8 个字符",
|
||||
"mfaRequired": "请输入验证码或恢复码",
|
||||
"mfaLength": "验证码或恢复码需为 6 至 32 个字符"
|
||||
},
|
||||
"errors": {
|
||||
"AUTH_INVALID_CREDENTIALS": "用户名或密码错误",
|
||||
"AUTH_WRONG_PASSWORD": "用户名或密码错误",
|
||||
"AUTH_USER_DISABLED": "该账户已被停用",
|
||||
"AUTH_RATE_LIMITED": "尝试次数过多,请稍后再试",
|
||||
"AUTH_2FA_REQUIRED": "请完成双因素验证后继续",
|
||||
"AUTH_2FA_INVALID": "验证码或恢复码无效",
|
||||
"AUTH_SETUP_DISABLED": "系统已完成初始化,请直接登录",
|
||||
"AUTH_USERNAME_EXISTS": "该用户名已存在",
|
||||
"AUTH_EMAIL_OTP_DISABLED": "邮件验证码未启用",
|
||||
"AUTH_SMS_OTP_DISABLED": "短信验证码未启用",
|
||||
"AUTH_EMAIL_REQUIRED": "该账户未配置邮箱",
|
||||
"AUTH_PHONE_REQUIRED": "该账户未配置手机号",
|
||||
"AUTH_WEBAUTHN_NOT_ENABLED": "该账户未配置通行密钥"
|
||||
}
|
||||
},
|
||||
"dashboard": {
|
||||
"title": "仪表盘",
|
||||
|
||||
@@ -114,6 +114,7 @@ export function BackupRecordsPage() {
|
||||
<Typography.Text>{record.fileName || '-'}</Typography.Text>
|
||||
{record.locked && <Tag color="orange" size="small" bordered>已锁定</Tag>}
|
||||
{record.backupKind === 'differential' && <Tag color="purple" size="small" bordered>差异</Tag>}
|
||||
{record.backupKind === 'repository' && <Tag color="blue" size="small" bordered>CDC</Tag>}
|
||||
</Space>
|
||||
<Typography.Text type="secondary">{formatBytes(record.fileSize)}</Typography.Text>
|
||||
{record.checksum && (
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { Alert, Avatar, Card, Empty, Grid, PageHeader, Space, Table, Tag, Typography } from '@arco-design/web-react'
|
||||
import { IconCheckCircle, IconDesktop, IconHistory, IconSafe, IconSave, IconStorage } from '@arco-design/web-react/icon'
|
||||
import { IconCheckCircle, IconDesktop, IconHistory, IconSafe, IconSave, IconStorage } from '../../components/icons'
|
||||
import ReactEChartsCore from 'echarts-for-react/lib/core'
|
||||
import * as echarts from 'echarts/core'
|
||||
import { BarChart, LineChart, PieChart } from 'echarts/charts'
|
||||
|
||||
70
web/src/pages/login/LoginPage.test.tsx
Normal file
70
web/src/pages/login/LoginPage.test.tsx
Normal file
@@ -0,0 +1,70 @@
|
||||
import { act, cleanup, render, screen } from '@testing-library/react'
|
||||
import { MemoryRouter } from 'react-router-dom'
|
||||
import { setApplicationLanguage } from '../../i18n'
|
||||
import { LoginPage } from './LoginPage'
|
||||
|
||||
const mocks = vi.hoisted(() => ({
|
||||
fetchSetupStatus: vi.fn(),
|
||||
login: vi.fn(),
|
||||
setup: vi.fn(),
|
||||
}))
|
||||
|
||||
vi.mock('../../services/auth', () => ({
|
||||
beginWebAuthnLogin: vi.fn(),
|
||||
fetchSetupStatus: mocks.fetchSetupStatus,
|
||||
sendLoginOtp: vi.fn(),
|
||||
}))
|
||||
|
||||
vi.mock('../../stores/auth', () => ({
|
||||
useAuthStore: (selector: (state: unknown) => unknown) => selector({
|
||||
status: 'anonymous',
|
||||
login: mocks.login,
|
||||
setup: mocks.setup,
|
||||
}),
|
||||
}))
|
||||
|
||||
vi.mock('../../utils/webauthn', () => ({
|
||||
getWebAuthnAssertion: vi.fn(),
|
||||
}))
|
||||
|
||||
describe('LoginPage initialization', () => {
|
||||
beforeEach(async () => {
|
||||
mocks.fetchSetupStatus.mockReset()
|
||||
mocks.login.mockReset()
|
||||
mocks.setup.mockReset()
|
||||
await act(() => setApplicationLanguage('en-US'))
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
cleanup()
|
||||
await act(() => setApplicationLanguage('zh-CN'))
|
||||
})
|
||||
|
||||
it('shows the first-administrator form in English', async () => {
|
||||
mocks.fetchSetupStatus.mockResolvedValue({ initialized: false })
|
||||
|
||||
render(
|
||||
<MemoryRouter>
|
||||
<LoginPage />
|
||||
</MemoryRouter>,
|
||||
)
|
||||
|
||||
expect(await screen.findByText('System setup')).toBeInTheDocument()
|
||||
expect(screen.getByText('Create the first administrator account.')).toBeInTheDocument()
|
||||
expect(screen.getByRole('button', { name: 'Create administrator and sign in' })).toBeInTheDocument()
|
||||
})
|
||||
|
||||
it('does not mistake an unreachable fresh install for an initialized system', async () => {
|
||||
mocks.fetchSetupStatus.mockRejectedValue(new Error('connection refused'))
|
||||
|
||||
render(
|
||||
<MemoryRouter>
|
||||
<LoginPage />
|
||||
</MemoryRouter>,
|
||||
)
|
||||
|
||||
expect(await screen.findByText('Unable to check initialization status')).toBeInTheDocument()
|
||||
expect(screen.getByRole('button', { name: 'Retry' })).toBeInTheDocument()
|
||||
expect(screen.queryByText('Welcome back')).not.toBeInTheDocument()
|
||||
})
|
||||
})
|
||||
@@ -1,8 +1,10 @@
|
||||
import { Button, Checkbox, Form, Input, Space, Typography, Message } from '@arco-design/web-react'
|
||||
import { IconCloud, IconLock, IconSafe, IconUser } from '@arco-design/web-react/icon'
|
||||
import { useEffect, useState } from 'react'
|
||||
import { BackupServerIllustration, IconCloud, IconLock, IconSafe, IconUser } from '../../components/icons'
|
||||
import { useCallback, useEffect, useRef, useState } from 'react'
|
||||
import { useTranslation } from 'react-i18next'
|
||||
import { useNavigate } from 'react-router-dom'
|
||||
import axios from 'axios'
|
||||
import { LanguageSwitcher } from '../../components/common/LanguageSwitcher'
|
||||
import { beginWebAuthnLogin, fetchSetupStatus, sendLoginOtp } from '../../services/auth'
|
||||
import { useAuthStore } from '../../stores/auth'
|
||||
import { getWebAuthnAssertion } from '../../utils/webauthn'
|
||||
@@ -20,17 +22,8 @@ interface LoginFormValues {
|
||||
rememberDevice?: boolean
|
||||
}
|
||||
|
||||
function resolveErrorMessage(error: unknown) {
|
||||
if (axios.isAxiosError(error)) {
|
||||
return error.response?.data?.message ?? '请求失败,请稍后重试'
|
||||
}
|
||||
if (error instanceof Error) {
|
||||
return error.message
|
||||
}
|
||||
return '请求失败,请稍后重试'
|
||||
}
|
||||
|
||||
export function LoginPage() {
|
||||
const { t, i18n } = useTranslation()
|
||||
const navigate = useNavigate()
|
||||
const authStatus = useAuthStore((state) => state.status)
|
||||
const doLogin = useAuthStore((state) => state.login)
|
||||
@@ -40,6 +33,25 @@ export function LoginPage() {
|
||||
const [loading, setLoading] = useState(false)
|
||||
const [mfaActionLoading, setMfaActionLoading] = useState('')
|
||||
const [twoFactorRequired, setTwoFactorRequired] = useState(false)
|
||||
const [setupStatusFailed, setSetupStatusFailed] = useState(false)
|
||||
const setupStatusRequest = useRef(0)
|
||||
|
||||
function resolveErrorMessage(error: unknown) {
|
||||
if (axios.isAxiosError(error)) {
|
||||
const code = error.response?.data?.code
|
||||
const translationKey = code ? `auth.errors.${code}` : ''
|
||||
if (translationKey && i18n.exists(translationKey)) {
|
||||
return t(translationKey)
|
||||
}
|
||||
if (i18n.resolvedLanguage === 'zh-CN' && error.response?.data?.message) {
|
||||
return error.response.data.message
|
||||
}
|
||||
}
|
||||
if (error instanceof Error && i18n.resolvedLanguage === 'zh-CN') {
|
||||
return error.message
|
||||
}
|
||||
return t('auth.requestFailed')
|
||||
}
|
||||
|
||||
function resetTwoFactorPrompt() {
|
||||
if (!twoFactorRequired) {
|
||||
@@ -56,30 +68,36 @@ export function LoginPage() {
|
||||
}
|
||||
}, [authStatus, navigate])
|
||||
|
||||
useEffect(() => {
|
||||
let mounted = true
|
||||
void (async () => {
|
||||
try {
|
||||
const result = await fetchSetupStatus()
|
||||
if (mounted) {
|
||||
setInitialized(result.initialized)
|
||||
}
|
||||
} catch {
|
||||
if (mounted) {
|
||||
setInitialized(true)
|
||||
}
|
||||
const loadSetupStatus = useCallback(async () => {
|
||||
const requestID = ++setupStatusRequest.current
|
||||
setInitialized(null)
|
||||
setSetupStatusFailed(false)
|
||||
try {
|
||||
const result = await fetchSetupStatus()
|
||||
if (requestID === setupStatusRequest.current) {
|
||||
setInitialized(result.initialized)
|
||||
}
|
||||
} catch {
|
||||
// Do not guess that an unreachable fresh install is initialized. That
|
||||
// would hide the first-administrator form behind an impossible login.
|
||||
if (requestID === setupStatusRequest.current) {
|
||||
setSetupStatusFailed(true)
|
||||
}
|
||||
})()
|
||||
return () => {
|
||||
mounted = false
|
||||
}
|
||||
}, [])
|
||||
|
||||
useEffect(() => {
|
||||
void loadSetupStatus()
|
||||
return () => {
|
||||
setupStatusRequest.current++
|
||||
}
|
||||
}, [loadSetupStatus])
|
||||
|
||||
const handleSetup = async (values: SetupFormValues) => {
|
||||
setLoading(true)
|
||||
try {
|
||||
await doSetup(values)
|
||||
Message.success('初始化完成,正在进入控制台')
|
||||
Message.success(t('auth.setupSuccess'))
|
||||
navigate('/dashboard', { replace: true })
|
||||
} catch (error) {
|
||||
Message.error(resolveErrorMessage(error))
|
||||
@@ -96,7 +114,7 @@ export function LoginPage() {
|
||||
trustedDeviceName: values.rememberDevice ? navigator.userAgent.slice(0, 120) : undefined,
|
||||
})
|
||||
setTwoFactorRequired(false)
|
||||
Message.success('登录成功')
|
||||
Message.success(t('auth.loginSuccess'))
|
||||
navigate('/dashboard', { replace: true })
|
||||
} catch (error) {
|
||||
if (axios.isAxiosError(error)) {
|
||||
@@ -116,7 +134,7 @@ export function LoginPage() {
|
||||
function readLoginCredentials(): (LoginFormValues & { username: string; password: string }) | null {
|
||||
const values = loginForm.getFieldsValue()
|
||||
if (!values.username?.trim() || !values.password?.trim()) {
|
||||
Message.error('请先输入用户名和密码')
|
||||
Message.error(t('auth.credentialsRequired'))
|
||||
return null
|
||||
}
|
||||
return {
|
||||
@@ -132,7 +150,7 @@ export function LoginPage() {
|
||||
setMfaActionLoading(channel)
|
||||
try {
|
||||
await sendLoginOtp({ username: values.username, password: values.password, channel })
|
||||
Message.success(channel === 'email' ? '邮件验证码已发送' : '短信验证码已发送')
|
||||
Message.success(channel === 'email' ? t('auth.emailCodeSent') : t('auth.smsCodeSent'))
|
||||
} catch (error) {
|
||||
Message.error(resolveErrorMessage(error))
|
||||
} finally {
|
||||
@@ -156,7 +174,7 @@ export function LoginPage() {
|
||||
trustedDeviceName: navigator.userAgent.slice(0, 120),
|
||||
})
|
||||
setTwoFactorRequired(false)
|
||||
Message.success('登录成功')
|
||||
Message.success(t('auth.loginSuccess'))
|
||||
navigate('/dashboard', { replace: true })
|
||||
} catch (error) {
|
||||
Message.error(resolveErrorMessage(error))
|
||||
@@ -165,128 +183,124 @@ export function LoginPage() {
|
||||
}
|
||||
}
|
||||
|
||||
const pageTitle = initialized === null
|
||||
? t('auth.setupStatusTitle')
|
||||
: initialized
|
||||
? t('auth.welcomeTitle')
|
||||
: t('auth.setupTitle')
|
||||
const pageSubtitle = initialized === null
|
||||
? setupStatusFailed ? t('auth.statusErrorDescription') : t('auth.checkingStatus')
|
||||
: initialized
|
||||
? t('auth.welcomeSubtitle')
|
||||
: t('auth.setupSubtitle')
|
||||
|
||||
return (
|
||||
<div className="login-shell">
|
||||
<div className="login-bg" />
|
||||
<div className="login-container">
|
||||
<div className="login-banner">
|
||||
{/* Background decorative circles for the banner */}
|
||||
<div style={{ position: 'absolute', width: 400, height: 400, borderRadius: '50%', background: 'rgba(255,255,255,0.05)', top: -100, right: -100 }} />
|
||||
<div style={{ position: 'absolute', width: 300, height: 300, borderRadius: '50%', background: 'rgba(255,255,255,0.05)', bottom: -50, left: -50 }} />
|
||||
|
||||
<div className="login-banner-inner">
|
||||
<svg width="320" height="320" viewBox="0 0 320 320" fill="none" xmlns="http://www.w3.org/2000/svg" style={{ marginBottom: 16 }}>
|
||||
{/* Outer pulsing rings */}
|
||||
<circle cx="160" cy="160" r="120" fill="white" fillOpacity="0.05">
|
||||
<animate attributeName="r" values="115;125;115" dur="4s" repeatCount="indefinite"/>
|
||||
<animate attributeName="fill-opacity" values="0.03;0.08;0.03" dur="4s" repeatCount="indefinite"/>
|
||||
</circle>
|
||||
<circle cx="160" cy="160" r="80" fill="white" fillOpacity="0.1">
|
||||
<animate attributeName="r" values="75;85;75" dur="3s" repeatCount="indefinite"/>
|
||||
</circle>
|
||||
|
||||
<g>
|
||||
<animateTransform attributeName="transform" type="translate" values="0,0; 0,-8; 0,0" dur="5s" repeatCount="indefinite"/>
|
||||
{/* Layer 1 (Top) */}
|
||||
<path d="M120 120C120 111.163 137.909 104 160 104C182.091 104 200 111.163 200 120V144C200 152.837 182.091 160 160 160C137.909 160 120 152.837 120 144V120Z" fill="white" fillOpacity="0.95"/>
|
||||
<ellipse cx="160" cy="120" rx="40" ry="16" fill="white"/>
|
||||
|
||||
{/* Layer 2 (Middle) */}
|
||||
<path d="M120 152C120 143.163 137.909 136 160 136C182.091 136 200 143.163 200 152V176C200 184.837 182.091 192 160 192C137.909 192 120 184.837 120 176V152Z" fill="white" fillOpacity="0.75"/>
|
||||
<ellipse cx="160" cy="152" rx="40" ry="16" fill="white" fillOpacity="0.9"/>
|
||||
<BackupServerIllustration style={{ marginBottom: 16 }} />
|
||||
|
||||
{/* Layer 3 (Bottom) */}
|
||||
<path d="M120 184C120 175.163 137.909 168 160 168C182.091 168 200 175.163 200 184V208C200 216.837 182.091 224 160 224C137.909 224 120 216.837 120 208V184Z" fill="white" fillOpacity="0.5"/>
|
||||
<ellipse cx="160" cy="184" rx="40" ry="16" fill="white" fillOpacity="0.6"/>
|
||||
|
||||
{/* Glowing Dots Output - Animated */}
|
||||
<g fill="var(--color-primary-6, #165dff)">
|
||||
<circle cx="140" cy="120" r="4">
|
||||
<animate attributeName="opacity" values="0.3;1;0.3" dur="2s" begin="0s" repeatCount="indefinite"/>
|
||||
</circle>
|
||||
<circle cx="140" cy="152" r="4">
|
||||
<animate attributeName="opacity" values="0.3;1;0.3" dur="2s" begin="0.6s" repeatCount="indefinite"/>
|
||||
</circle>
|
||||
<circle cx="140" cy="184" r="4">
|
||||
<animate attributeName="opacity" values="0.3;1;0.3" dur="2s" begin="1.2s" repeatCount="indefinite"/>
|
||||
</circle>
|
||||
</g>
|
||||
|
||||
{/* Connecting Data Line */}
|
||||
<path d="M160 120V152V184" stroke="var(--color-primary-6, #165dff)" strokeWidth="2" strokeDasharray="4 4" opacity="0.6">
|
||||
<animate attributeName="stroke-dashoffset" from="16" to="0" dur="1s" repeatCount="indefinite" />
|
||||
</path>
|
||||
</g>
|
||||
</svg>
|
||||
|
||||
<Typography.Title heading={2} style={{ color: 'white', marginTop: 0, marginBottom: 12, fontWeight: 700 }}>
|
||||
守护您的数据资产
|
||||
<Typography.Title heading={2} style={{ color: 'white', marginTop: 0, marginBottom: 12 }}>
|
||||
{t('auth.bannerTitle')}
|
||||
</Typography.Title>
|
||||
<Typography.Text style={{ color: 'rgba(255,255,255,0.75)', fontSize: 16 }}>
|
||||
安全、可靠、高效的企业级服务器备份管理平台
|
||||
{t('auth.bannerSubtitle')}
|
||||
</Typography.Text>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div className="login-form-wrapper">
|
||||
<Space direction="vertical" size="large" style={{ width: '100%' }}>
|
||||
<div style={{ display: 'flex', justifyContent: 'flex-end' }}>
|
||||
<LanguageSwitcher />
|
||||
</div>
|
||||
<div style={{ paddingBottom: 8 }}>
|
||||
<div style={{ display: 'inline-flex', alignItems: 'center', marginBottom: 16 }}>
|
||||
<div style={{ display: 'inline-flex', alignItems: 'center', justifyContent: 'center', width: 36, height: 36, borderRadius: 10, background: 'linear-gradient(135deg, var(--color-primary-5) 0%, var(--color-primary-7) 100%)', marginRight: 12 }}>
|
||||
<div style={{ display: 'inline-flex', alignItems: 'center', justifyContent: 'center', width: 36, height: 36, borderRadius: 4, background: 'var(--color-primary-6)', marginRight: 12 }}>
|
||||
<IconCloud style={{ fontSize: 20, color: 'white' }} />
|
||||
</div>
|
||||
<Typography.Title heading={4} style={{ margin: 0, fontWeight: 700 }}>
|
||||
<Typography.Title heading={4} style={{ margin: 0 }}>
|
||||
BackupX
|
||||
</Typography.Title>
|
||||
</div>
|
||||
<Typography.Title heading={3} style={{ marginTop: 0, marginBottom: 8, fontWeight: 600 }}>
|
||||
{initialized === false ? '系统初始化' : '欢迎回来'}
|
||||
<Typography.Title heading={3} style={{ marginTop: 0, marginBottom: 8 }}>
|
||||
{pageTitle}
|
||||
</Typography.Title>
|
||||
<Typography.Paragraph type="secondary" style={{ marginBottom: 0, fontSize: 14 }}>
|
||||
{initialized === false ? '请设定首个管理员账户以启动系统。' : '请输入管理员账户信息登录控制台。'}
|
||||
{pageSubtitle}
|
||||
</Typography.Paragraph>
|
||||
</div>
|
||||
|
||||
{initialized === false ? (
|
||||
{initialized === null ? (
|
||||
setupStatusFailed ? (
|
||||
<div>
|
||||
<Typography.Text>{t('auth.statusErrorTitle')}</Typography.Text>
|
||||
<div style={{ marginTop: 12 }}>
|
||||
<Button type="primary" loading={loading} onClick={() => void loadSetupStatus()}>
|
||||
{t('auth.retry')}
|
||||
</Button>
|
||||
</div>
|
||||
</div>
|
||||
) : (
|
||||
<Typography.Text type="secondary">{t('auth.checkingStatus')}</Typography.Text>
|
||||
)
|
||||
) : initialized === false ? (
|
||||
<Form<SetupFormValues> layout="vertical" onSubmit={handleSetup}>
|
||||
<Form.Item field="displayName" label="显示名称" rules={[{ required: true, minLength: 1 }]}>
|
||||
<Input placeholder="请输入显示名称" prefix={<IconUser />} size="large" />
|
||||
<Form.Item field="displayName" label={t('auth.displayName')} rules={[{ required: true, minLength: 1, message: t('auth.validation.displayNameRequired') }]}>
|
||||
<Input autoComplete="name" placeholder={t('auth.displayNamePlaceholder')} prefix={<IconUser />} size="large" />
|
||||
</Form.Item>
|
||||
<Form.Item field="username" label="用户名" rules={[{ required: true, minLength: 3 }]}>
|
||||
<Input placeholder="请输入管理员用户名" prefix={<IconUser />} size="large" />
|
||||
<Form.Item field="username" label={t('auth.username')} rules={[
|
||||
{ required: true, message: t('auth.validation.usernameRequired') },
|
||||
{ minLength: 3, message: t('auth.validation.usernameLength') },
|
||||
]}>
|
||||
<Input autoComplete="username" placeholder={t('auth.usernamePlaceholder')} prefix={<IconUser />} size="large" />
|
||||
</Form.Item>
|
||||
<Form.Item field="password" label="密码" rules={[{ required: true, minLength: 8 }]}>
|
||||
<Input.Password placeholder="请输入至少 8 位密码" prefix={<IconLock />} size="large" />
|
||||
<Form.Item field="password" label={t('auth.password')} rules={[
|
||||
{ required: true, message: t('auth.validation.passwordRequired') },
|
||||
{ minLength: 8, message: t('auth.validation.passwordLength') },
|
||||
]}>
|
||||
<Input.Password autoComplete="new-password" placeholder={t('auth.setupPasswordPlaceholder')} prefix={<IconLock />} size="large" />
|
||||
</Form.Item>
|
||||
<Button long type="primary" htmlType="submit" loading={loading} size="large" style={{ borderRadius: 8, height: 44, marginTop: 8 }}>
|
||||
初始化并登录
|
||||
<Button long type="primary" htmlType="submit" loading={loading} size="large" style={{ borderRadius: 4, height: 44, marginTop: 8 }}>
|
||||
{t('auth.setupSubmit')}
|
||||
</Button>
|
||||
</Form>
|
||||
) : (
|
||||
<Form<LoginFormValues> form={loginForm} layout="vertical" onSubmit={handleLogin}>
|
||||
<Form.Item field="username" label="用户名" rules={[{ required: true, minLength: 3 }]}>
|
||||
<Input placeholder="请输入用户名" prefix={<IconUser />} size="large" onChange={resetTwoFactorPrompt} />
|
||||
<Form.Item field="username" label={t('auth.username')} rules={[
|
||||
{ required: true, message: t('auth.validation.usernameRequired') },
|
||||
{ minLength: 3, message: t('auth.validation.usernameLength') },
|
||||
]}>
|
||||
<Input autoComplete="username" placeholder={t('auth.usernamePlaceholder')} prefix={<IconUser />} size="large" onChange={resetTwoFactorPrompt} />
|
||||
</Form.Item>
|
||||
<Form.Item field="password" label="密码" rules={[{ required: true, minLength: 8 }]}>
|
||||
<Input.Password placeholder="请输入密码" prefix={<IconLock />} size="large" onChange={resetTwoFactorPrompt} />
|
||||
<Form.Item field="password" label={t('auth.password')} rules={[
|
||||
{ required: true, message: t('auth.validation.passwordRequired') },
|
||||
{ minLength: 8, message: t('auth.validation.passwordLength') },
|
||||
]}>
|
||||
<Input.Password autoComplete="current-password" placeholder={t('auth.passwordPlaceholder')} prefix={<IconLock />} size="large" onChange={resetTwoFactorPrompt} />
|
||||
</Form.Item>
|
||||
{twoFactorRequired && (
|
||||
<>
|
||||
<Form.Item field="twoFactorCode" label="验证码或恢复码" rules={[{ required: true, minLength: 6, maxLength: 32 }]}>
|
||||
<Input placeholder="请输入 TOTP、恢复码、邮件或短信验证码" prefix={<IconSafe />} size="large" maxLength={32} />
|
||||
<Form.Item field="twoFactorCode" label={t('auth.mfaCode')} rules={[
|
||||
{ required: true, message: t('auth.validation.mfaRequired') },
|
||||
{ minLength: 6, maxLength: 32, message: t('auth.validation.mfaLength') },
|
||||
]}>
|
||||
<Input autoComplete="one-time-code" placeholder={t('auth.mfaCodePlaceholder')} prefix={<IconSafe />} size="large" maxLength={32} />
|
||||
</Form.Item>
|
||||
<Space wrap style={{ marginTop: -8, marginBottom: 8 }}>
|
||||
<Button loading={mfaActionLoading === 'email'} onClick={() => void handleSendOTP('email')}>发送邮件验证码</Button>
|
||||
<Button loading={mfaActionLoading === 'sms'} onClick={() => void handleSendOTP('sms')}>发送短信验证码</Button>
|
||||
<Button loading={mfaActionLoading === 'webauthn'} onClick={() => void handleWebAuthnLogin()}>使用通行密钥</Button>
|
||||
<Button loading={mfaActionLoading === 'email'} onClick={() => void handleSendOTP('email')}>{t('auth.sendEmailCode')}</Button>
|
||||
<Button loading={mfaActionLoading === 'sms'} onClick={() => void handleSendOTP('sms')}>{t('auth.sendSmsCode')}</Button>
|
||||
<Button loading={mfaActionLoading === 'webauthn'} onClick={() => void handleWebAuthnLogin()}>{t('auth.usePasskey')}</Button>
|
||||
</Space>
|
||||
<Form.Item field="rememberDevice" triggerPropName="checked">
|
||||
<Checkbox>信任此设备 30 天</Checkbox>
|
||||
<Checkbox>{t('auth.trustDevice')}</Checkbox>
|
||||
</Form.Item>
|
||||
</>
|
||||
)}
|
||||
<Button long type="primary" htmlType="submit" loading={loading} size="large" style={{ borderRadius: 8, height: 44, marginTop: 16 }}>
|
||||
{twoFactorRequired ? '验证并登录' : '登录'}
|
||||
<Button long type="primary" htmlType="submit" loading={loading} size="large" style={{ borderRadius: 4, height: 44, marginTop: 16 }}>
|
||||
{twoFactorRequired ? t('auth.verifyAndLogin') : t('auth.login')}
|
||||
</Button>
|
||||
</Form>
|
||||
)}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import React, { useEffect, useState } from 'react'
|
||||
import { Table, Button, Space, Message, Typography, Tag } from '@arco-design/web-react'
|
||||
import { IconCopy, IconDownload, IconRefresh } from '@arco-design/web-react/icon'
|
||||
import { IconCopy, IconDownload, IconRefresh } from '../../components/icons'
|
||||
|
||||
const { Text } = Typography
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@ import {
|
||||
} from '@arco-design/web-react'
|
||||
import {
|
||||
IconPlus, IconDelete, IconDesktop, IconCloudDownload, IconEdit, IconMore,
|
||||
} from '@arco-design/web-react/icon'
|
||||
} from '../../components/icons'
|
||||
import type { NodeSummary } from '../../types/nodes'
|
||||
import { listNodes, deleteNode, updateNode, rotateNodeToken } from '../../services/nodes'
|
||||
import { fetchSystemInfo } from '../../services/system'
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import React, { useEffect, useState } from 'react'
|
||||
import { Typography, Button, Space, Collapse, Spin, Message, Tag } from '@arco-design/web-react'
|
||||
import { IconCopy, IconRefresh } from '@arco-design/web-react/icon'
|
||||
import { IconCopy, IconRefresh } from '../../../components/icons'
|
||||
import { fetchScriptPreview } from '../../../services/nodes'
|
||||
import type { InstallTokenResult, InstallMode } from '../../../types/nodes'
|
||||
import { buildAgentDownloadCommand, buildAgentInstallCommand, buildEmbeddedAgentInstallCommand } from '../installCommands'
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { Button, Card, Grid, Message, Select, Space, Statistic, Table, Tag, Typography } from '@arco-design/web-react'
|
||||
import { IconDownload, IconRefresh } from '@arco-design/web-react/icon'
|
||||
import { IconDownload, IconRefresh } from '../../components/icons'
|
||||
import { useCallback, useEffect, useState } from 'react'
|
||||
import { downloadComplianceCSV, fetchComplianceReport } from '../../services/reports'
|
||||
import type { ComplianceReport, ComplianceRisk, ComplianceTaskRow } from '../../types/reports'
|
||||
|
||||
@@ -18,6 +18,7 @@ import { formatBytes } from '../../utils/format'
|
||||
import type { StorageConnectionTestResult, StorageTargetDetail, StorageTargetPayload, StorageTargetSummary } from '../../types/storage-targets'
|
||||
import { getStorageTargetTypeLabel } from '../../components/storage-targets/field-config'
|
||||
import { StorageTargetFormDrawer } from '../../components/storage-targets/StorageTargetFormDrawer'
|
||||
import { StorageTargetName } from '../../components/storage-targets/StorageTargetName'
|
||||
|
||||
function resolveErrorMessage(error: unknown) {
|
||||
if (axios.isAxiosError(error)) {
|
||||
@@ -224,7 +225,7 @@ export function StorageTargetsPage() {
|
||||
<Space size="large" align="start" style={{ marginBottom: 16, width: '100%', justifyContent: 'space-between' }}>
|
||||
<div>
|
||||
<Typography.Title heading={6} style={{ marginBottom: 4 }}>
|
||||
{target.starred ? '★ ' : ''}{target.name}
|
||||
<StorageTargetName name={target.name} starred={target.starred} />
|
||||
</Typography.Title>
|
||||
<Space>
|
||||
{getStorageTargetTypeLabel(target.type) && <Tag color="arcoblue" bordered>{getStorageTargetTypeLabel(target.type)}</Tag>}
|
||||
|
||||
@@ -42,48 +42,25 @@ body {
|
||||
.login-bg {
|
||||
position: fixed;
|
||||
inset: 0;
|
||||
background: linear-gradient(135deg, #111a2c 0%, #1f2d47 100%);
|
||||
background: #111a2c;
|
||||
z-index: 0;
|
||||
}
|
||||
|
||||
.login-bg::before {
|
||||
content: '';
|
||||
position: absolute;
|
||||
width: 800px;
|
||||
height: 800px;
|
||||
border-radius: 50%;
|
||||
background: radial-gradient(circle, rgba(52,145,250,0.08) 0%, transparent 70%);
|
||||
top: -300px;
|
||||
right: -200px;
|
||||
}
|
||||
|
||||
.login-bg::after {
|
||||
content: '';
|
||||
position: absolute;
|
||||
width: 600px;
|
||||
height: 600px;
|
||||
border-radius: 50%;
|
||||
background: radial-gradient(circle, rgba(114,46,209,0.06) 0%, transparent 70%);
|
||||
bottom: -200px;
|
||||
left: -100px;
|
||||
}
|
||||
|
||||
.login-container {
|
||||
display: flex;
|
||||
width: 1000px;
|
||||
max-width: 90vw;
|
||||
min-height: 560px;
|
||||
background: var(--color-bg-2);
|
||||
border-radius: 20px;
|
||||
border-radius: 4px;
|
||||
overflow: hidden;
|
||||
box-shadow: 0 20px 60px rgba(0,0,0,0.4);
|
||||
z-index: 1;
|
||||
animation: slideUp 0.6s cubic-bezier(0.34, 1.56, 0.64, 1);
|
||||
}
|
||||
|
||||
.login-banner {
|
||||
flex: 1;
|
||||
background: linear-gradient(135deg, var(--color-primary-6, #165dff) 0%, var(--color-primary-8, #0e42d2) 100%);
|
||||
background: var(--color-primary-6, #165dff);
|
||||
position: relative;
|
||||
display: flex;
|
||||
align-items: center;
|
||||
|
||||
@@ -24,3 +24,17 @@ Object.defineProperty(window, 'localStorage', {
|
||||
value: storage,
|
||||
configurable: true,
|
||||
})
|
||||
|
||||
Object.defineProperty(window, 'matchMedia', {
|
||||
configurable: true,
|
||||
value: (query: string) => ({
|
||||
matches: false,
|
||||
media: query,
|
||||
onchange: null,
|
||||
addEventListener: () => undefined,
|
||||
removeEventListener: () => undefined,
|
||||
addListener: () => undefined,
|
||||
removeListener: () => undefined,
|
||||
dispatchEvent: () => false,
|
||||
}),
|
||||
})
|
||||
|
||||
@@ -21,12 +21,13 @@ export interface BackupRecordSummary {
|
||||
fileSize: number
|
||||
checksum: string
|
||||
storagePath: string
|
||||
storageTransferMode?: 'direct' | 'master_relay'
|
||||
durationSeconds: number
|
||||
errorMessage: string
|
||||
startedAt: string
|
||||
completedAt?: string
|
||||
locked: boolean
|
||||
backupKind: 'full' | 'differential'
|
||||
backupKind: 'full' | 'differential' | 'repository'
|
||||
}
|
||||
|
||||
export interface BackupRecordContentEntry {
|
||||
@@ -49,6 +50,7 @@ export interface StorageUploadResultItem {
|
||||
status: 'success' | 'failed'
|
||||
storagePath?: string
|
||||
fileSize?: number
|
||||
transferMode?: 'direct' | 'master_relay'
|
||||
error?: string
|
||||
}
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
export type BackupTaskType = 'file' | 'mysql' | 'sqlite' | 'postgresql' | 'saphana' | 'mongodb'
|
||||
export type BackupTaskStatus = 'idle' | 'running' | 'success' | 'failed'
|
||||
export type BackupCompression = 'gzip' | 'zstd' | 'none'
|
||||
export type BackupMode = 'full' | 'differential'
|
||||
export type BackupMode = 'full' | 'differential' | 'repository'
|
||||
|
||||
export interface BackupTaskSummary {
|
||||
id: number
|
||||
|
||||
Reference in New Issue
Block a user