feat(cluster): 支持远程源服务器集中备份

为 Master 本地磁盘增加可选流式中转,远程 Agent 可直接把产物写入中央存储并通过反向通道恢复。

持久化并校验传输模式,兼容既有 Agent 本机磁盘目标,补齐鉴权、完整性、配额、访问保护、前端配置及双向链路测试。

Closes #101
This commit is contained in:
Awuqing
2026-08-07 05:54:34 +08:00
parent 05dd1baa61
commit cc50637b4b
28 changed files with 1012 additions and 180 deletions
+88 -12
View File
@@ -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) {