diff --git a/docs-site/docs/features/multi-node.md b/docs-site/docs/features/multi-node.md index 0cfc477..b47346b 100644 --- a/docs-site/docs/features/multi-node.md +++ b/docs-site/docs/features/multi-node.md @@ -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 diff --git a/docs-site/docs/features/storage-backends.md b/docs-site/docs/features/storage-backends.md index 731526b..6e977cb 100644 --- a/docs-site/docs/features/storage-backends.md +++ b/docs-site/docs/features/storage-backends.md @@ -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 diff --git a/docs-site/i18n/zh-CN/docusaurus-plugin-content-docs/current/features/multi-node.md b/docs-site/i18n/zh-CN/docusaurus-plugin-content-docs/current/features/multi-node.md index 9b5a540..7ad1032 100644 --- a/docs-site/i18n/zh-CN/docusaurus-plugin-content-docs/current/features/multi-node.md +++ b/docs-site/i18n/zh-CN/docusaurus-plugin-content-docs/current/features/multi-node.md @@ -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 本地执行 → 上传 → 回报 diff --git a/docs-site/i18n/zh-CN/docusaurus-plugin-content-docs/current/features/storage-backends.md b/docs-site/i18n/zh-CN/docusaurus-plugin-content-docs/current/features/storage-backends.md index cdd25ac..b3f7895 100644 --- a/docs-site/i18n/zh-CN/docusaurus-plugin-content-docs/current/features/storage-backends.md +++ b/docs-site/i18n/zh-CN/docusaurus-plugin-content-docs/current/features/storage-backends.md @@ -19,7 +19,9 @@ BackupX 的目标是接入任何你想放置备份文件的地方。 | **Google Drive** | Client ID/Secret + OAuth 授权 | | **WebDAV** | 地址 + 用户名/密码 | | **FTP / FTPS** | 主机 + 端口 + 用户名/密码 | -| **本地磁盘** | 目标目录（绝对路径） | +| **本地磁盘** | 目标目录（绝对路径）+ 可选的远程 Agent 经 Master 中转 | + +新建本地磁盘目标默认开启 **远程备份经 Master 中转**。开启时，配置目录属于 Master，挂载到 Master 的存储服务器可集中接收多台源 Agent 的备份；如果该路径本就属于各 Agent，请关闭此选项。升级前已有目标保持原来的 Agent 本机落盘行为，只有显式开启后才会切换。 ## Rclone 后端 diff --git a/server/internal/agent/client.go b/server/internal/agent/client.go index 08751ea..64ba166 100644 --- a/server/internal/agent/client.go +++ b/server/internal/agent/client.go @@ -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) diff --git a/server/internal/agent/executor.go b/server/internal/agent/executor.go index b1163b6..f80732a 100644 --- a/server/internal/agent/executor.go +++ b/server/internal/agent/executor.go @@ -5,6 +5,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "io" "os" @@ -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()) } // 辅助函数 diff --git a/server/internal/agent/executor_test.go b/server/internal/agent/executor_test.go index d4f35ff..72189a1 100644 --- a/server/internal/agent/executor_test.go +++ b/server/internal/agent/executor_test.go @@ -1,7 +1,10 @@ package agent import ( + "archive/tar" + "bytes" "context" + "crypto/sha256" "encoding/json" "fmt" "io" @@ -109,6 +112,149 @@ 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) + } + if r.Header.Get("X-BackupX-Object-Key") == "" || r.ContentLength != int64(len(body)) { + t.Fatalf("invalid relay metadata: key=%q length=%d body=%d", r.Header.Get("X-BackupX-Object-Key"), 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 { diff --git a/server/internal/app/app.go b/server/internal/app/app.go index 281b3e9..2cfac10 100644 --- a/server/internal/app/app.go +++ b/server/internal/app/app.go @@ -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) diff --git a/server/internal/http/agent_handler.go b/server/internal/http/agent_handler.go index 8750c8f..23d28c1 100644 --- a/server/internal/http/agent_handler.go +++ b/server/internal/http/agent_handler.go @@ -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)) diff --git a/server/internal/http/router.go b/server/internal/http/router.go index ba5bf9d..f003fb2 100644 --- a/server/internal/http/router.go +++ b/server/internal/http/router.go @@ -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 端点 diff --git a/server/internal/model/backup_record.go b/server/internal/model/backup_record.go index 5f0a965..8734c19 100644 --- a/server/internal/model/backup_record.go +++ b/server/internal/model/backup_record.go @@ -22,14 +22,16 @@ 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 时该备份不参与保留期/数量自动清理， diff --git a/server/internal/service/agent_service.go b/server/internal/service/agent_service.go index 7723325..d4c97b9 100644 --- a/server/internal/service/agent_service.go +++ b/server/internal/service/agent_service.go @@ -2,15 +2,19 @@ package service import ( "context" + "encoding/hex" "encoding/json" "errors" "fmt" + "io" + "path" "strings" "time" "backupx/server/internal/apperror" "backupx/server/internal/model" "backupx/server/internal/repository" + "backupx/server/internal/storage" "backupx/server/internal/storage/codec" ) @@ -23,6 +27,7 @@ type AgentService struct { storageRepo repository.StorageTargetRepository cmdRepo repository.AgentCommandRepository restoreRepo repository.RestoreRecordRepository + registry *storage.Registry cipher *codec.ConfigCipher } @@ -33,6 +38,7 @@ func NewAgentService( storageRepo repository.StorageTargetRepository, cmdRepo repository.AgentCommandRepository, cipher *codec.ConfigCipher, + registry *storage.Registry, ) *AgentService { return &AgentService{ nodeRepo: nodeRepo, @@ -40,6 +46,7 @@ func NewAgentService( recordRepo: recordRepo, storageRepo: storageRepo, cmdRepo: cmdRepo, + registry: registry, cipher: cipher, } } @@ -145,10 +152,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 +195,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 +233,101 @@ 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 := path.Clean(strings.TrimSpace(objectKey)) + if cleanKey == "." || path.IsAbs(cleanKey) || strings.HasPrefix(cleanKey, "../") || cleanKey != objectKey || strings.Contains(objectKey, "\\") { + return apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "中转上传对象路径不安全", nil) + } + checksumBytes, checksumErr := hex.DecodeString(strings.TrimSpace(checksum)) + if size < 0 || checksumErr != nil || len(checksumBytes) != 32 { + 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+size > target.QuotaBytes { + 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 +350,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 +375,60 @@ 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) error { + if _, ok := allowedTargets[targetID]; !ok { + return apperror.Unauthorized("BACKUP_STORAGE_TARGET_FORBIDDEN", "存储目标不属于该任务", nil) + } + if transferMode == "" { + return 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 != expectedMode { + return apperror.BadRequest("AGENT_STORAGE_TRANSFER_MODE_INVALID", "Agent 上报的存储传输模式与目标配置不一致", nil) + } + return nil + } + if update.StorageTargetID > 0 { + if _, ok := allowedTargets[update.StorageTargetID]; !ok { + return apperror.Unauthorized("BACKUP_STORAGE_TARGET_FORBIDDEN", "存储目标不属于该任务", nil) + } + if err := validateTransferMode(update.StorageTargetID, update.StorageTransferMode); err != nil { + return err + } + } else if update.StorageTransferMode != "" { + return apperror.BadRequest("AGENT_STORAGE_TRANSFER_MODE_INVALID", "传输模式缺少对应的存储目标", nil) + } + for _, result := range update.StorageUploadResults { + if err := validateTransferMode(result.StorageTargetID, result.TransferMode); err != nil { + return err + } + } if update.Status != "" { record.Status = update.Status } @@ -278,6 +447,9 @@ func (s *AgentService) UpdateRecord(ctx context.Context, node *model.Node, recor if update.StorageTargetID > 0 { record.StorageTargetID = update.StorageTargetID } + if update.StorageTransferMode != "" { + record.StorageTransferMode = update.StorageTransferMode + } if len(update.StorageUploadResults) > 0 { if resultsJSON, marshalErr := json.Marshal(update.StorageUploadResults); marshalErr == nil { record.StorageUploadResults = string(resultsJSON) diff --git a/server/internal/service/agent_service_test.go b/server/internal/service/agent_service_test.go index ccda6f4..96b9102 100644 --- a/server/internal/service/agent_service_test.go +++ b/server/internal/service/agent_service_test.go @@ -1,8 +1,12 @@ package service import ( + "bytes" "context" + "crypto/sha256" "errors" + "fmt" + "os" "path/filepath" "strings" "testing" @@ -13,7 +17,9 @@ import ( "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 +48,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 +82,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 +97,22 @@ 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") } 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: "tasks/1/backup.tar.gz", + 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: "tasks/1/backup.tar.gz", FileSize: 123, TransferMode: storage.TransferModeMasterRelay}, }, }); err != nil { t.Fatalf("owner UpdateRecord returned error: %v", err) @@ -114,10 +124,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 +138,69 @@ func TestAgentServicePooledTaskUsesRecordNodeForSpecAndRecordUpdates(t *testing. } } +func TestAgentServiceRelaysRemoteArtifactToMasterLocalDisk(t *testing.T) { + svc, _, _, _, owner, other := newAgentServicePoolTestHarness(t) + ctx := context.Background() + payload := []byte("artifact from remote source server") + digest := sha256.Sum256(payload) + checksum := fmt.Sprintf("%x", digest[:]) + objectKey := "file/2026/08/06/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, "file/forbidden.tar", int64(len(payload)), checksum, bytes.NewReader(payload)); err == nil { + t.Fatal("expected a different node to be forbidden from relaying the artifact") + } +} + +func TestAgentServiceKeepsExistingLocalDiskTargetsAgentLocal(t *testing.T) { + svc, _, _, _, 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) + err = svc.UploadArtifact(ctx, owner, 1, 1, "file/legacy.tar", 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) { diff --git a/server/internal/service/backup_execution_service.go b/server/internal/service/backup_execution_service.go index ea74e5f..3c90055 100644 --- a/server/internal/service/backup_execution_service.go +++ b/server/internal/service/backup_execution_service.go @@ -51,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"` } @@ -410,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 diff --git a/server/internal/service/backup_execution_service_test.go b/server/internal/service/backup_execution_service_test.go index abbcbd0..f9eb3b5 100644 --- a/server/internal/service/backup_execution_service_test.go +++ b/server/internal/service/backup_execution_service_test.go @@ -429,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() diff --git a/server/internal/service/backup_record_service.go b/server/internal/service/backup_record_service.go index e4d7246..3b2461a 100644 --- a/server/internal/service/backup_record_service.go +++ b/server/internal/service/backup_record_service.go @@ -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, } } diff --git a/server/internal/service/execution_helpers.go b/server/internal/service/execution_helpers.go index 4c04de5..2042a2c 100644 --- a/server/internal/service/execution_helpers.go +++ b/server/internal/service/execution_helpers.go @@ -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 diff --git a/server/internal/service/restore_service.go b/server/internal/service/restore_service.go index edf22e0..e5802fe 100644 --- a/server/internal/service/restore_service.go +++ b/server/internal/service/restore_service.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "io" "os" "path/filepath" "strings" @@ -601,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, @@ -628,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, @@ -639,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) diff --git a/server/internal/service/restore_service_test.go b/server/internal/service/restore_service_test.go index 5991c9c..a411728 100644 --- a/server/internal/service/restore_service_test.go +++ b/server/internal/service/restore_service_test.go @@ -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") } diff --git a/server/internal/storage/types.go b/server/internal/storage/types.go index 3abc664..aae226a 100644 --- a/server/internal/storage/types.go +++ b/server/internal/storage/types.go @@ -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"` @@ -99,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 { diff --git a/web/src/components/backup-records/BackupRecordLogDrawer.tsx b/web/src/components/backup-records/BackupRecordLogDrawer.tsx index b0e35c6..9380ef6 100644 --- a/web/src/components/backup-records/BackupRecordLogDrawer.tsx +++ b/web/src/components/backup-records/BackupRecordLogDrawer.tsx @@ -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 } )} - {record.storageUploadResults && record.storageUploadResults.length > 1 && ( + {record.storageUploadResults && (record.storageUploadResults.length > 1 || record.storageUploadResults.some((result) => result.transferMode)) && (
存储目标上传结果 ({ label: r.storageTargetName, - value: r.status === 'success' ? '上传成功' : `上传失败: ${r.error || '未知错误'}`, + value: r.status === 'success' + ? `上传成功${r.transferMode === 'master_relay' ? ' · Master 流式中转' : r.transferMode === 'direct' ? ' · Agent 直传' : ''}` + : `上传失败: ${r.error || '未知错误'}`, }))} />
diff --git a/web/src/components/backup-tasks/BackupTaskFormDrawer.tsx b/web/src/components/backup-tasks/BackupTaskFormDrawer.tsx index f4722cd..31fd1ad 100644 --- a/web/src/components/backup-tasks/BackupTaskFormDrawer.tsx +++ b/web/src/components/backup-tasks/BackupTaskFormDrawer.tsx @@ -9,6 +9,7 @@ import type { NodeSummary } from '../../types/nodes' import { DatabasePicker } from '../common/DatabasePicker' import { DirectoryPicker } from '../common/DirectoryPicker' import { StorageTargetFormDrawer } from '../storage-targets/StorageTargetFormDrawer' +import { SourceServerSelector } from './SourceServerSelector' import { backupCompressionOptions, backupTaskTypeOptions, @@ -176,21 +177,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) { setDraft((current) => ({ ...current, ...patch })) } @@ -312,42 +298,26 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa 备份类型 { - const nodeId = Number(value ?? 0) - // 固定节点与节点池互斥：切到固定节点时清空 NodePoolTag - updateDraft(nodeId > 0 - ? { - nodeId, - nodePoolTag: '', - backupMode: nodeId !== localNodeId && draft.backupMode === 'repository' ? 'full' : draft.backupMode, - } - : { nodeId }) - }} - /> - - 任务在所选节点上执行备份与恢复；源路径/数据库以该节点视角解析。远程节点需先在"节点管理"中安装 Agent。 - - -
- 节点池标签（可选） - 0} - onChange={(value) => updateDraft({ - nodePoolTag: value, - backupMode: value.trim() && draft.backupMode === 'repository' ? 'full' : draft.backupMode, - })} - /> - - 执行节点选"本机 / 未指定"时可启用；从节点 Labels 命中此 tag 的在线节点中按当前运行任务数最少的挑选一台执行。 - -
+ { + // 固定源服务器与服务器池互斥；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, + })} + />
Cron 表达式 updateDraft({ cronExpr: value })} /> @@ -602,6 +572,11 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa )} + {((draft.nodeId ?? 0) > 0 && draft.nodeId !== localNodeId) || draft.nodePoolTag?.trim() ? ( + + 远程源服务器会直传 S3、WebDAV 等网络存储；本地磁盘目标启用 Master 中转后，文件经 Agent 认证 API 流式写入中央目录。跨公网部署请为 Master 配置 HTTPS。 + + ) : null}
压缩策略 diff --git a/web/src/components/backup-tasks/SourceServerSelector.test.ts b/web/src/components/backup-tasks/SourceServerSelector.test.ts new file mode 100644 index 0000000..23baa10 --- /dev/null +++ b/web/src/components/backup-tasks/SourceServerSelector.test.ts @@ -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 }, + ]) + }) +}) diff --git a/web/src/components/backup-tasks/SourceServerSelector.tsx b/web/src/components/backup-tasks/SourceServerSelector.tsx new file mode 100644 index 0000000..86b5396 --- /dev/null +++ b/web/src/components/backup-tasks/SourceServerSelector.tsx @@ -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 ( + <> +
+ 源服务器 + 0} + onChange={onNodePoolTagChange} + /> + + 仅在选择 Master 本机时可填写；系统从 Labels 命中该标签的在线 Agent 中选择当前运行任务最少的一台。 + +
+ + ) +} diff --git a/web/src/components/storage-targets/StorageTargetFormDrawer.tsx b/web/src/components/storage-targets/StorageTargetFormDrawer.tsx index fb93d66..4aafd37 100644 --- a/web/src/components/storage-targets/StorageTargetFormDrawer.tsx +++ b/web/src/components/storage-targets/StorageTargetFormDrawer.tsx @@ -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) }} > diff --git a/web/src/components/storage-targets/field-config.test.ts b/web/src/components/storage-targets/field-config.test.ts index 515c90c..95b80c5 100644 --- a/web/src/components/storage-targets/field-config.test.ts +++ b/web/src/components/storage-targets/field-config.test.ts @@ -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', () => { diff --git a/web/src/components/storage-targets/field-config.ts b/web/src/components/storage-targets/field-config.ts index 7fd6a8a..3d15c3e 100644 --- a/web/src/components/storage-targets/field-config.ts +++ b/web/src/components/storage-targets/field-config.ts @@ -7,6 +7,7 @@ import type { StorageTargetFieldConfig, StorageTargetType } from '../../types/st const BUILTIN_FIELD_CONFIG: Record = { 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' }, diff --git a/web/src/types/backup-records.ts b/web/src/types/backup-records.ts index 0da3d32..a6debfc 100644 --- a/web/src/types/backup-records.ts +++ b/web/src/types/backup-records.ts @@ -21,6 +21,7 @@ export interface BackupRecordSummary { fileSize: number checksum: string storagePath: string + storageTransferMode?: 'direct' | 'master_relay' durationSeconds: number errorMessage: string startedAt: string @@ -49,6 +50,7 @@ export interface StorageUploadResultItem { status: 'success' | 'failed' storagePath?: string fileSize?: number + transferMode?: 'direct' | 'master_relay' error?: string }