mirror of
https://github.com/Awuqing/BackupX.git
synced 2026-08-31 13:07:36 +08:00
Awuqing
cc50637b4b
为 Master 本地磁盘增加可选流式中转，远程 Agent 可直接把产物写入中央存储并通过反向通道恢复。 持久化并校验传输模式，兼容既有 Agent 本机磁盘目标，补齐鉴权、完整性、配额、访问保护、前端配置及双向链路测试。 Closes #101
1453 lines
57 KiB
Go
1453 lines
57 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"hash"
|
||
"io"
|
||
"os"
|
||
"path/filepath"
|
||
"sort"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"backupx/server/internal/apperror"
|
||
"backupx/server/internal/backup"
|
||
backupretention "backupx/server/internal/backup/retention"
|
||
"backupx/server/internal/metrics"
|
||
"backupx/server/internal/model"
|
||
"backupx/server/internal/repository"
|
||
"backupx/server/internal/storage"
|
||
"backupx/server/internal/storage/codec"
|
||
"backupx/server/internal/storage/rclone"
|
||
"backupx/server/pkg/compress"
|
||
backupcrypto "backupx/server/pkg/crypto"
|
||
)
|
||
|
||
type BackupExecutionNotification struct {
|
||
Task *model.BackupTask
|
||
Record *model.BackupRecord
|
||
Error error
|
||
}
|
||
|
||
type BackupResultNotifier interface {
|
||
NotifyBackupResult(context.Context, BackupExecutionNotification) error
|
||
}
|
||
|
||
type noopBackupNotifier struct{}
|
||
|
||
func (noopBackupNotifier) NotifyBackupResult(context.Context, BackupExecutionNotification) error {
|
||
return nil
|
||
}
|
||
|
||
type StorageUploadResultItem struct {
|
||
StorageTargetID uint `json:"storageTargetId"`
|
||
StorageTargetName string `json:"storageTargetName"`
|
||
Status string `json:"status"`
|
||
StoragePath string `json:"storagePath,omitempty"`
|
||
FileSize int64 `json:"fileSize,omitempty"`
|
||
TransferMode string `json:"transferMode,omitempty"`
|
||
Error string `json:"error,omitempty"`
|
||
}
|
||
|
||
const (
|
||
uploadMaxAttempts = 3
|
||
uploadRetryBackoff = 10 * time.Second
|
||
)
|
||
|
||
type DownloadedArtifact struct {
|
||
FileName string
|
||
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 {
|
||
ids := make([]uint, len(task.StorageTargets))
|
||
for i, t := range task.StorageTargets {
|
||
ids[i] = t.ID
|
||
}
|
||
return ids
|
||
}
|
||
if task.StorageTargetID > 0 {
|
||
return []uint{task.StorageTargetID}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
type BackupExecutionService struct {
|
||
tasks repository.BackupTaskRepository
|
||
records repository.BackupRecordRepository
|
||
targets repository.StorageTargetRepository
|
||
nodeRepo repository.NodeRepository
|
||
storageRegistry *storage.Registry
|
||
runnerRegistry *backup.Registry
|
||
logHub *backup.LogHub
|
||
retention *backupretention.Service
|
||
cipher *codec.ConfigCipher
|
||
notifier BackupResultNotifier
|
||
agentDispatcher AgentDispatcher
|
||
replicationHook ReplicationTrigger
|
||
dependentsResolver DependentsResolver
|
||
async func(func())
|
||
now func() time.Time
|
||
tempDir string
|
||
semaphore chan struct{}
|
||
// nodeSemaphores 节点级并发限制（按 NodeID 映射）。
|
||
// 没命中的 NodeID 走全局 semaphore，节点配置 MaxConcurrent>0 时按该节点独立排队。
|
||
nodeSemaphores sync.Map
|
||
retries int // rclone 底层重试次数
|
||
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。
|
||
func (s *BackupExecutionService) SetMetrics(m *metrics.Metrics) {
|
||
s.metrics = m
|
||
}
|
||
|
||
// ReplicationTrigger 抽象备份成功后的副本派发（实现者：ReplicationService）。
|
||
type ReplicationTrigger interface {
|
||
TriggerAutoReplication(ctx context.Context, task *model.BackupTask, record *model.BackupRecord)
|
||
}
|
||
|
||
// SetReplicationTrigger 注入备份复制触发器。可选注入：未注入时不自动复制。
|
||
func (s *BackupExecutionService) SetReplicationTrigger(trigger ReplicationTrigger) {
|
||
s.replicationHook = trigger
|
||
}
|
||
|
||
// DependentsResolver 根据 upstream 任务 ID 返回应触发的下游任务 ID。
|
||
// 由 BackupTaskService 实现。抽象接口避免执行服务直接查仓储。
|
||
type DependentsResolver interface {
|
||
TriggerDependents(ctx context.Context, upstreamID uint) ([]uint, error)
|
||
}
|
||
|
||
// SetDependentsResolver 注入下游依赖解析器。
|
||
func (s *BackupExecutionService) SetDependentsResolver(r DependentsResolver) {
|
||
s.dependentsResolver = r
|
||
}
|
||
|
||
// AgentDispatcher 抽象把任务下发给 Agent 的能力，由 AgentService 实现。
|
||
// 用接口避免 execution service ↔ agent service 的循环依赖风险。
|
||
type AgentDispatcher interface {
|
||
EnqueueCommand(ctx context.Context, nodeID uint, cmdType string, payload any) (uint, error)
|
||
}
|
||
|
||
// SetClusterDependencies 注入集群相关的依赖，使备份执行时可把任务路由到远程节点。
|
||
func (s *BackupExecutionService) SetClusterDependencies(nodeRepo repository.NodeRepository, dispatcher AgentDispatcher) {
|
||
s.nodeRepo = nodeRepo
|
||
s.agentDispatcher = dispatcher
|
||
}
|
||
|
||
func NewBackupExecutionService(
|
||
tasks repository.BackupTaskRepository,
|
||
records repository.BackupRecordRepository,
|
||
targets repository.StorageTargetRepository,
|
||
storageRegistry *storage.Registry,
|
||
runnerRegistry *backup.Registry,
|
||
logHub *backup.LogHub,
|
||
retention *backupretention.Service,
|
||
cipher *codec.ConfigCipher,
|
||
notifier BackupResultNotifier,
|
||
tempDir string,
|
||
maxConcurrent int,
|
||
retries int,
|
||
bandwidthLimit string,
|
||
) *BackupExecutionService {
|
||
if notifier == nil {
|
||
notifier = noopBackupNotifier{}
|
||
}
|
||
if tempDir == "" {
|
||
tempDir = "/tmp/backupx"
|
||
}
|
||
if maxConcurrent <= 0 {
|
||
maxConcurrent = 2
|
||
}
|
||
return &BackupExecutionService{
|
||
tasks: tasks,
|
||
records: records,
|
||
targets: targets,
|
||
storageRegistry: storageRegistry,
|
||
runnerRegistry: runnerRegistry,
|
||
logHub: logHub,
|
||
retention: retention,
|
||
cipher: cipher,
|
||
notifier: notifier,
|
||
async: func(job func()) {
|
||
go job()
|
||
},
|
||
now: func() time.Time { return time.Now().UTC() },
|
||
tempDir: tempDir,
|
||
semaphore: make(chan struct{}, maxConcurrent),
|
||
retries: retries,
|
||
bandwidthLimit: bandwidthLimit,
|
||
}
|
||
}
|
||
|
||
func (s *BackupExecutionService) RunTaskByID(ctx context.Context, id uint) (*BackupRecordDetail, error) {
|
||
return s.startTask(ctx, id, true)
|
||
}
|
||
|
||
func (s *BackupExecutionService) RunTaskByIDSync(ctx context.Context, id uint) (*BackupRecordDetail, error) {
|
||
return s.startTask(ctx, id, false)
|
||
}
|
||
|
||
func (s *BackupExecutionService) DownloadRecord(ctx context.Context, recordID uint) (*DownloadedArtifact, error) {
|
||
record, err := s.records.FindByID(ctx, recordID)
|
||
if err != nil {
|
||
return nil, apperror.Internal("BACKUP_RECORD_GET_FAILED", "无法获取备份记录详情", err)
|
||
}
|
||
if record == nil {
|
||
return nil, apperror.New(404, "BACKUP_RECORD_NOT_FOUND", "备份记录不存在", fmt.Errorf("backup record %d not found", recordID))
|
||
}
|
||
// 集群场景保护：local_disk 类型的存储文件只在执行节点本地可见，Master 不能跨节点访问
|
||
if err := s.validateClusterAccessible(ctx, record); err != nil {
|
||
return nil, err
|
||
}
|
||
provider, err := s.resolveProvider(ctx, record.StorageTargetID)
|
||
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, 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)
|
||
}
|
||
fileName := record.FileName
|
||
if strings.TrimSpace(fileName) == "" {
|
||
fileName = filepath.Base(record.StoragePath)
|
||
}
|
||
return &DownloadedArtifact{FileName: fileName, Reader: reader}, nil
|
||
}
|
||
|
||
func (s *BackupExecutionService) RestoreRecord(ctx context.Context, recordID uint) error {
|
||
record, provider, err := s.loadRecordProvider(ctx, recordID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
task, err := s.tasks.FindByID(ctx, record.TaskID)
|
||
if err != nil {
|
||
return apperror.Internal("BACKUP_TASK_GET_FAILED", "无法获取关联备份任务", err)
|
||
}
|
||
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, 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)
|
||
}
|
||
defer os.RemoveAll(tempDir)
|
||
artifactPath := filepath.Join(tempDir, filepath.Base(record.FileName))
|
||
if strings.TrimSpace(filepath.Base(record.FileName)) == "" {
|
||
artifactPath = filepath.Join(tempDir, filepath.Base(record.StoragePath))
|
||
}
|
||
reader, err := provider.Download(ctx, record.StoragePath)
|
||
if err != nil {
|
||
return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "无法下载备份文件", err)
|
||
}
|
||
if err := writeReaderToFile(artifactPath, reader); err != nil {
|
||
return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "无法写入恢复文件", err)
|
||
}
|
||
preparedPath, err := s.prepareArtifactForRestore(artifactPath)
|
||
if err != nil {
|
||
return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "无法准备恢复文件", err)
|
||
}
|
||
spec, err := s.buildTaskSpec(task, record.StartedAt)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
runner, err := s.runnerRegistry.Runner(spec.Type)
|
||
if err != nil {
|
||
return apperror.BadRequest("BACKUP_TASK_INVALID", "不支持的备份任务类型", err)
|
||
}
|
||
if err := runner.Restore(ctx, spec, preparedPath, backup.NopLogWriter{}); err != nil {
|
||
return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "恢复备份失败", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *BackupExecutionService) DeleteRecord(ctx context.Context, recordID uint) error {
|
||
record, err := s.records.FindByID(ctx, recordID)
|
||
if err != nil {
|
||
return apperror.Internal("BACKUP_RECORD_GET_FAILED", "无法获取备份记录详情", err)
|
||
}
|
||
if record == nil {
|
||
return apperror.New(404, "BACKUP_RECORD_NOT_FOUND", "备份记录不存在", fmt.Errorf("backup record %d not found", recordID))
|
||
}
|
||
if record.Locked {
|
||
return apperror.BadRequest("BACKUP_RECORD_LOCKED",
|
||
"该备份已保留锁定（法律保留），请先解锁再删除", nil)
|
||
}
|
||
// 差异链保护：禁止删除仍被差异备份依赖的全量，否则这些差异将无法恢复（与保留清理的保护一致）。
|
||
if record.BackupKind == model.BackupKindFull {
|
||
deps, depErr := s.records.CountDependentDifferentials(ctx, record.ID)
|
||
if depErr != nil {
|
||
return apperror.Internal("BACKUP_RECORD_DELETE_FAILED", "无法检查差异备份依赖", depErr)
|
||
}
|
||
if deps > 0 {
|
||
return apperror.BadRequest("BACKUP_RECORD_HAS_DEPENDENTS",
|
||
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) != "" {
|
||
provider, err := s.resolveProvider(ctx, record.StorageTargetID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := provider.Delete(ctx, record.StoragePath); err != nil {
|
||
return apperror.Internal("BACKUP_RECORD_DELETE_FAILED", "无法删除备份文件", err)
|
||
}
|
||
}
|
||
if err := s.records.Delete(ctx, recordID); err != nil {
|
||
return apperror.Internal("BACKUP_RECORD_DELETE_FAILED", "无法删除备份记录", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *BackupExecutionService) deleteRemoteLocalDiskObject(ctx context.Context, record *model.BackupRecord) (bool, error) {
|
||
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
|
||
}
|
||
target, err := s.targets.FindByID(ctx, record.StorageTargetID)
|
||
if err != nil {
|
||
return false, apperror.Internal("BACKUP_STORAGE_TARGET_GET_FAILED", "无法获取存储目标详情", err)
|
||
}
|
||
if target == nil || !strings.EqualFold(target.Type, "local_disk") {
|
||
return false, nil
|
||
}
|
||
if s.agentDispatcher == nil {
|
||
return true, apperror.BadRequest("BACKUP_RECORD_CROSS_NODE_LOCAL_DISK",
|
||
fmt.Sprintf("该备份位于节点 %s 的本地磁盘（local_disk），Master 无法跨节点删除。请确保 Agent 在线后再操作。", node.Name),
|
||
nil)
|
||
}
|
||
configMap := map[string]any{}
|
||
if err := s.cipher.DecryptJSON(target.ConfigCiphertext, &configMap); err != nil {
|
||
return true, apperror.Internal("BACKUP_STORAGE_TARGET_DECRYPT_FAILED", "无法解密存储目标配置", err)
|
||
}
|
||
if _, err := s.agentDispatcher.EnqueueCommand(ctx, record.NodeID, model.AgentCommandTypeDeleteStorageObject, map[string]any{
|
||
"targetType": target.Type,
|
||
"targetConfig": configMap,
|
||
"storagePath": record.StoragePath,
|
||
}); err != nil {
|
||
return true, apperror.Internal("AGENT_COMMAND_ENQUEUE_FAILED", "无法下发远程备份文件删除命令", err)
|
||
}
|
||
return true, nil
|
||
}
|
||
|
||
// validateClusterAccessible 在跨节点 + local_disk 场景下拒绝 Master 端直接访问。
|
||
// 场景说明：远程 Agent 把备份写到其本机磁盘（local_disk basePath）时，Master 的
|
||
// provider 指向的是 Master 本机的同名路径，访问会静默取错文件或 404。明确拒绝
|
||
// 让用户知情，避免假成功。
|
||
func (s *BackupExecutionService) validateClusterAccessible(ctx context.Context, record *model.BackupRecord) error {
|
||
return validateCrossNodeLocalDisk(ctx, s.nodeRepo, s.targets, record,
|
||
"BACKUP_RECORD_CROSS_NODE_LOCAL_DISK", "访问。请登录该节点或改用云存储后再操作")
|
||
}
|
||
|
||
func (s *BackupExecutionService) startTask(ctx context.Context, id uint, async bool) (*BackupRecordDetail, error) {
|
||
task, err := s.tasks.FindByID(ctx, id)
|
||
if err != nil {
|
||
return nil, apperror.Internal("BACKUP_TASK_GET_FAILED", "无法获取备份任务详情", err)
|
||
}
|
||
if task == nil {
|
||
return nil, apperror.New(404, "BACKUP_TASK_NOT_FOUND", "备份任务不存在", fmt.Errorf("backup task %d not found", id))
|
||
}
|
||
unlock := s.acquireTaskStartLock(task.ID)
|
||
defer unlock()
|
||
if err := s.ensureTaskNotRunning(ctx, task); err != nil {
|
||
return nil, err
|
||
}
|
||
// 维护窗口校验：手动执行同样尊重窗口，避免业务高峰期误触发。
|
||
if strings.TrimSpace(task.MaintenanceWindows) != "" {
|
||
windows := backup.ParseMaintenanceWindows(task.MaintenanceWindows)
|
||
if len(windows) > 0 && !backup.IsWithinWindow(s.now(), windows) {
|
||
return nil, apperror.BadRequest("BACKUP_TASK_OUTSIDE_WINDOW",
|
||
fmt.Sprintf("当前时间不在任务「%s」的维护窗口内（%s），已拒绝执行。", task.Name, task.MaintenanceWindows),
|
||
nil)
|
||
}
|
||
}
|
||
// 节点池动态选择：task.NodeID=0 且 NodePoolTag 非空时，从匹配的在线节点中挑一台。
|
||
// 选择策略：正在运行任务数最少者优先；并列时按 ID 升序稳定。
|
||
// 选中节点仅影响本次运行（task.NodeID 不持久化改动），保证任务在池内轮转。
|
||
resolvedNodeID := task.NodeID
|
||
if task.NodeID == 0 && strings.TrimSpace(task.NodePoolTag) != "" {
|
||
if pooled, perr := s.selectPoolNode(ctx, task.NodePoolTag); perr == nil && pooled != nil {
|
||
resolvedNodeID = pooled.ID
|
||
} else if perr != nil {
|
||
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
|
||
if tids := collectTargetIDs(task); len(tids) > 0 {
|
||
primaryTargetID = tids[0]
|
||
}
|
||
record := &model.BackupRecord{TaskID: task.ID, StorageTargetID: primaryTargetID, NodeID: resolvedNodeID, Status: "running", StartedAt: startedAt}
|
||
if err := s.records.Create(ctx, record); err != nil {
|
||
return nil, apperror.Internal("BACKUP_RECORD_CREATE_FAILED", "无法创建备份记录", err)
|
||
}
|
||
runTask := *task
|
||
runTask.NodeID = resolvedNodeID
|
||
task.LastRunAt = &startedAt
|
||
task.LastStatus = "running"
|
||
if err := s.tasks.Update(ctx, task); err != nil {
|
||
return nil, apperror.Internal("BACKUP_TASK_UPDATE_FAILED", "无法更新任务状态", err)
|
||
}
|
||
// 多节点路由：task.NodeID 指向远程节点时，把执行任务入队给 Agent；
|
||
// NodeID=0 或本机节点时由 Master 直接执行。
|
||
if remoteNode := s.resolveRemoteNode(ctx, resolvedNodeID); remoteNode != nil {
|
||
// 节点离线 → 立即把刚创建的 running 记录标记 failed，返回明确错误
|
||
if remoteNode.Status != model.NodeStatusOnline {
|
||
offlineMsg := fmt.Sprintf("节点 %s 当前离线，无法执行备份任务", remoteNode.Name)
|
||
_ = s.finalizeRecord(ctx, &runTask, record.ID, startedAt, model.BackupRecordStatusFailed,
|
||
offlineMsg, "", "", 0, "", "", primaryTargetID)
|
||
return nil, apperror.BadRequest("NODE_OFFLINE", offlineMsg, nil)
|
||
}
|
||
if _, enqueueErr := s.agentDispatcher.EnqueueCommand(ctx, resolvedNodeID, model.AgentCommandTypeRunTask, map[string]any{
|
||
"taskId": task.ID,
|
||
"recordId": record.ID,
|
||
}); enqueueErr != nil {
|
||
// 入队失败 → 在记录中标记失败，继续返回详情
|
||
_ = s.finalizeRecord(ctx, &runTask, record.ID, startedAt, model.BackupRecordStatusFailed,
|
||
"无法下发任务到远程节点: "+enqueueErr.Error(), "", "", 0, "", "", primaryTargetID)
|
||
return nil, apperror.Internal("AGENT_COMMAND_ENQUEUE_FAILED", "无法下发任务到远程节点", enqueueErr)
|
||
}
|
||
return s.getRecordDetail(ctx, record.ID)
|
||
}
|
||
run := func() {
|
||
s.executeTask(context.Background(), &runTask, record.ID, startedAt)
|
||
}
|
||
if async {
|
||
s.async(run)
|
||
} else {
|
||
run()
|
||
}
|
||
return s.getRecordDetail(ctx, record.ID)
|
||
}
|
||
|
||
func (s *BackupExecutionService) acquireTaskStartLock(taskID uint) func() {
|
||
value, _ := s.taskLocks.LoadOrStore(taskID, &sync.Mutex{})
|
||
mu := value.(*sync.Mutex)
|
||
mu.Lock()
|
||
return mu.Unlock
|
||
}
|
||
|
||
func (s *BackupExecutionService) ensureTaskNotRunning(ctx context.Context, task *model.BackupTask) error {
|
||
taskID := task.ID
|
||
items, err := s.records.List(ctx, repository.BackupRecordListOptions{TaskID: &taskID, Status: model.BackupRecordStatusRunning})
|
||
if err != nil {
|
||
return apperror.Internal("BACKUP_RECORD_LIST_FAILED", "无法检查任务运行状态", err)
|
||
}
|
||
if len(items) == 0 {
|
||
return nil
|
||
}
|
||
return apperror.BadRequest("BACKUP_TASK_ALREADY_RUNNING",
|
||
fmt.Sprintf("任务「%s」正在运行（记录 #%d），请等待完成后再触发。", task.Name, items[0].ID),
|
||
nil)
|
||
}
|
||
|
||
// shouldNotify 按任务的告警策略决定是否发送本次通知。
|
||
// 成功结果：始终发送（方便用户确认备份状态）。
|
||
// 失败结果：仅当"最近 N 条记录（含本次）均为 failed"时发送，N = AlertOnConsecutiveFails。
|
||
// 该策略降低单次偶发失败的告警噪音，企业运维场景下更友好。
|
||
func (s *BackupExecutionService) shouldNotify(ctx context.Context, task *model.BackupTask, status string) bool {
|
||
if task == nil {
|
||
return true
|
||
}
|
||
threshold := task.AlertOnConsecutiveFails
|
||
if threshold <= 1 {
|
||
return true
|
||
}
|
||
if status != model.BackupRecordStatusFailed {
|
||
return true
|
||
}
|
||
items, err := s.records.ListByTask(ctx, task.ID)
|
||
if err != nil || len(items) < threshold {
|
||
return true
|
||
}
|
||
// ListByTask 默认按 id desc 返回：取前 threshold 条
|
||
count := threshold
|
||
if len(items) < count {
|
||
count = len(items)
|
||
}
|
||
for i := 0; i < count; i++ {
|
||
if items[i].Status != model.BackupRecordStatusFailed {
|
||
return false
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
|
||
// selectPoolNode 从所有 Labels 包含 poolTag 的在线节点中选择"当前运行中任务最少"的一台。
|
||
// 返回 (nil, error) 表示硬错误（仓储访问失败）；(nil, nil) 表示没有匹配节点（退化走本机 Master）。
|
||
// 本方法不修改任何持久化状态，仅做选择。
|
||
func (s *BackupExecutionService) selectPoolNode(ctx context.Context, poolTag string) (*model.Node, error) {
|
||
if s.nodeRepo == nil {
|
||
// 没接入集群依赖时，降级为让调用方走本机 Master
|
||
return nil, nil
|
||
}
|
||
nodes, err := s.nodeRepo.List(ctx)
|
||
if err != nil {
|
||
return nil, apperror.Internal("NODE_LIST_FAILED", "无法枚举节点池", err)
|
||
}
|
||
candidates := make([]*model.Node, 0)
|
||
for i := range nodes {
|
||
n := &nodes[i]
|
||
if n.Status != model.NodeStatusOnline {
|
||
continue
|
||
}
|
||
if !n.HasLabel(poolTag) {
|
||
continue
|
||
}
|
||
candidates = append(candidates, n)
|
||
}
|
||
if len(candidates) == 0 {
|
||
return nil, apperror.BadRequest("NODE_POOL_EMPTY",
|
||
fmt.Sprintf("节点池 %q 下无在线节点，任务无法调度", poolTag), nil)
|
||
}
|
||
// 运行中记录数越少越优先。并列按 ID 升序（稳定、可预期）。
|
||
best := candidates[0]
|
||
bestLoad := s.countRunningOnNode(ctx, best.ID)
|
||
for _, n := range candidates[1:] {
|
||
load := s.countRunningOnNode(ctx, n.ID)
|
||
if load < bestLoad || (load == bestLoad && n.ID < best.ID) {
|
||
best = n
|
||
bestLoad = load
|
||
}
|
||
}
|
||
return best, nil
|
||
}
|
||
|
||
// countRunningOnNode 近似返回节点当前 running 记录数。失败按 0 处理（不影响功能，仅退化调度精度）。
|
||
func (s *BackupExecutionService) countRunningOnNode(ctx context.Context, nodeID uint) int {
|
||
if s.records == nil {
|
||
return 0
|
||
}
|
||
items, err := s.records.List(ctx, repository.BackupRecordListOptions{Status: model.BackupRecordStatusRunning})
|
||
if err != nil {
|
||
return 0
|
||
}
|
||
count := 0
|
||
for i := range items {
|
||
if items[i].NodeID == nodeID {
|
||
count++
|
||
}
|
||
}
|
||
return count
|
||
}
|
||
|
||
// effectiveBandwidth 返回当前上下文应用的带宽限速字符串。
|
||
// 优先级：Node.BandwidthLimit（非空） > 全局 s.bandwidthLimit。
|
||
func (s *BackupExecutionService) effectiveBandwidth(ctx context.Context, nodeID uint) string {
|
||
if nodeID == 0 || s.nodeRepo == nil {
|
||
return s.bandwidthLimit
|
||
}
|
||
node, err := s.nodeRepo.FindByID(ctx, nodeID)
|
||
if err != nil || node == nil {
|
||
return s.bandwidthLimit
|
||
}
|
||
if strings.TrimSpace(node.BandwidthLimit) != "" {
|
||
return node.BandwidthLimit
|
||
}
|
||
return s.bandwidthLimit
|
||
}
|
||
|
||
// acquireNodeSemaphore 返回节点级并发通道。懒初始化：第一次为某节点排队时创建。
|
||
// 如果节点未配置 MaxConcurrent 或 nodeRepo 未注入，返回 nil（调用方走全局 semaphore）。
|
||
// 节点容量仅在首次创建时采用，后续变更需重启服务才生效（避免运行时 resize 通道的复杂度）。
|
||
func (s *BackupExecutionService) acquireNodeSemaphore(ctx context.Context, nodeID uint) chan struct{} {
|
||
if nodeID == 0 || s.nodeRepo == nil {
|
||
return nil
|
||
}
|
||
if v, ok := s.nodeSemaphores.Load(nodeID); ok {
|
||
return v.(chan struct{})
|
||
}
|
||
node, err := s.nodeRepo.FindByID(ctx, nodeID)
|
||
if err != nil || node == nil || node.MaxConcurrent <= 0 {
|
||
return nil
|
||
}
|
||
created := make(chan struct{}, node.MaxConcurrent)
|
||
actual, _ := s.nodeSemaphores.LoadOrStore(nodeID, created)
|
||
return actual.(chan struct{})
|
||
}
|
||
|
||
// isRemoteNode 判断 NodeID 是否指向一个有效的远程（非本机）节点。
|
||
// 当未注入集群依赖、nodeID 为 0、或节点为本机时，均返回 false（走本地执行）。
|
||
func (s *BackupExecutionService) isRemoteNode(ctx context.Context, nodeID uint) bool {
|
||
return s.resolveRemoteNode(ctx, nodeID) != nil
|
||
}
|
||
|
||
// resolveRemoteNode 返回 NodeID 对应的远程节点指针，或 nil 表示本机执行。
|
||
// 相比 isRemoteNode，它让调用方能读取节点状态（在线/离线）做进一步判断。
|
||
func (s *BackupExecutionService) resolveRemoteNode(ctx context.Context, nodeID uint) *model.Node {
|
||
return resolveRemoteExecutionNode(ctx, s.nodeRepo, s.agentDispatcher != nil, nodeID)
|
||
}
|
||
|
||
// resolveDifferentialBase 为差异备份解析基线全量：仅本机（NodeID=0）文件任务且 BackupMode=differential 时生效。
|
||
// 返回最近一次「成功、含清单、未超过 DiffFullIntervalDays」的全量记录 ID 及其清单；
|
||
// 无合适基线（首次备份 / 最近全量已过期 / 清单缺失）时 ok=false，调用方回退为全量。
|
||
func (s *BackupExecutionService) resolveDifferentialBase(ctx context.Context, task *model.BackupTask) (uint, backup.Manifest, bool) {
|
||
if task.Type != model.BackupTaskTypeFile || task.NodeID != 0 || !strings.EqualFold(task.BackupMode, model.BackupModeDifferential) {
|
||
return 0, backup.Manifest{}, false
|
||
}
|
||
records, err := s.records.ListSuccessfulByTask(ctx, task.ID)
|
||
if err != nil {
|
||
return 0, backup.Manifest{}, false
|
||
}
|
||
intervalDays := task.DiffFullIntervalDays
|
||
if intervalDays <= 0 {
|
||
intervalDays = 7
|
||
}
|
||
cutoff := time.Now().Add(-time.Duration(intervalDays) * 24 * time.Hour)
|
||
for i := range records {
|
||
rec := records[i]
|
||
if rec.BackupKind != model.BackupKindFull {
|
||
continue
|
||
}
|
||
// 最近的全量已超过强制全量间隔 → 触发新全量，限制差异链跨度与单个差异体积。
|
||
if rec.StartedAt.Before(cutoff) {
|
||
return 0, backup.Manifest{}, false
|
||
}
|
||
// 列表查询已省略 Manifest 列，这里按需单独加载最近全量的清单（FindByID 含 Manifest）。
|
||
full, ferr := s.records.FindByID(ctx, rec.ID)
|
||
if ferr != nil || full == nil || strings.TrimSpace(full.Manifest) == "" {
|
||
return 0, backup.Manifest{}, false
|
||
}
|
||
manifest, decErr := backup.DecodeManifest([]byte(full.Manifest))
|
||
if decErr != nil || len(manifest.Entries) == 0 {
|
||
return 0, backup.Manifest{}, false
|
||
}
|
||
return rec.ID, manifest, true
|
||
}
|
||
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，互相排队
|
||
nodeSem := s.acquireNodeSemaphore(ctx, task.NodeID)
|
||
if nodeSem != nil {
|
||
nodeSem <- struct{}{}
|
||
defer func() { <-nodeSem }()
|
||
}
|
||
s.semaphore <- struct{}{}
|
||
defer func() { <-s.semaphore }()
|
||
|
||
// Prometheus: running gauge + 完成时 observe 耗时/字节/状态
|
||
s.metrics.IncTaskRunning()
|
||
defer s.metrics.DecTaskRunning()
|
||
|
||
logger := backup.NewExecutionLogger(recordID, s.logHub)
|
||
status := "failed"
|
||
errMessage := ""
|
||
var fileName string
|
||
var fileSize int64
|
||
var checksum string
|
||
var storagePath string
|
||
selectedStorageTargetID := task.StorageTargetID
|
||
var uploadResults []StorageUploadResultItem
|
||
// 差异备份链信息：实际类型（全量/差异）、基线全量 ID、全量清单 JSON。
|
||
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 {
|
||
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
|
||
}
|
||
}
|
||
}
|
||
// 持久化差异链信息：全量记录其清单（供后续差异比对），差异记录其基线全量 ID。
|
||
if status == model.BackupRecordStatusSuccess && (backupKind != model.BackupKindFull || baseRecordID != 0 || manifestJSON != "") {
|
||
if record, findErr := s.records.FindByID(ctx, recordID); findErr == nil && record != nil {
|
||
record.BackupKind = backupKind
|
||
record.BaseRecordID = baseRecordID
|
||
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)
|
||
}
|
||
}
|
||
}
|
||
if s.shouldNotify(ctx, task, status) {
|
||
if err := s.notifier.NotifyBackupResult(ctx, BackupExecutionNotification{Task: task, Record: &model.BackupRecord{ID: recordID, TaskID: task.ID, Status: status, FileName: fileName, FileSize: fileSize, StoragePath: storagePath, ErrorMessage: errMessage, StartedAt: startedAt}, Error: buildOptionalError(errMessage)}); err != nil {
|
||
logger.Warnf("发送备份通知失败：%v", err)
|
||
}
|
||
} else {
|
||
logger.Infof("连续失败次数未达通知阈值（%d），跳过本次告警", task.AlertOnConsecutiveFails)
|
||
}
|
||
s.logHub.Complete(recordID, status)
|
||
}
|
||
defer completeRecord()
|
||
|
||
spec, err := s.buildTaskSpec(task, startedAt)
|
||
if err != nil {
|
||
errMessage = err.Error()
|
||
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
|
||
spec.BaseManifest = baseManifest
|
||
baseRecordID = baseID
|
||
logger.Infof("差异备份模式：基于全量备份 #%d 仅打包变更", baseID)
|
||
}
|
||
runner, err := s.runnerRegistry.Runner(spec.Type)
|
||
if err != nil {
|
||
errMessage = err.Error()
|
||
logger.Errorf("获取备份执行器失败：%v", err)
|
||
return
|
||
}
|
||
result, err := runner.Run(ctx, spec, logger)
|
||
if err != nil {
|
||
errMessage = err.Error()
|
||
logger.Errorf("执行备份失败：%v", err)
|
||
return
|
||
}
|
||
defer os.RemoveAll(result.TempDir)
|
||
// 依据运行器产出判定实际类型：产出清单 → 全量（记录清单供后续差异比对）；否则为差异。
|
||
if result.Manifest != nil {
|
||
backupKind = model.BackupKindFull
|
||
if data, encErr := backup.EncodeManifest(*result.Manifest); encErr == nil {
|
||
manifestJSON = string(data)
|
||
} else {
|
||
logger.Warnf("备份清单序列化失败（不影响本次备份，但将禁用后续差异）：%v", encErr)
|
||
}
|
||
} else {
|
||
backupKind = model.BackupKindDifferential
|
||
}
|
||
finalPath := result.ArtifactPath
|
||
if strings.EqualFold(task.Compression, "gzip") && !strings.HasSuffix(strings.ToLower(finalPath), ".gz") {
|
||
logger.Infof("开始压缩备份文件")
|
||
compressedPath, compressErr := compress.GzipFile(finalPath)
|
||
if compressErr != nil {
|
||
errMessage = compressErr.Error()
|
||
logger.Errorf("压缩备份文件失败：%v", compressErr)
|
||
return
|
||
}
|
||
finalPath = compressedPath
|
||
} else if strings.EqualFold(task.Compression, "zstd") && !strings.HasSuffix(strings.ToLower(finalPath), ".zst") {
|
||
logger.Infof("开始压缩备份文件（zstd）")
|
||
compressedPath, compressErr := compress.ZstdFile(finalPath)
|
||
if compressErr != nil {
|
||
errMessage = compressErr.Error()
|
||
logger.Errorf("压缩备份文件失败：%v", compressErr)
|
||
return
|
||
}
|
||
finalPath = compressedPath
|
||
}
|
||
if task.Encrypt {
|
||
logger.Infof("开始加密备份文件")
|
||
encryptedPath, encryptErr := backupcrypto.EncryptFile(s.cipher.Key(), finalPath)
|
||
if encryptErr != nil {
|
||
errMessage = encryptErr.Error()
|
||
logger.Errorf("加密备份文件失败：%v", encryptErr)
|
||
return
|
||
}
|
||
finalPath = encryptedPath
|
||
}
|
||
info, err := os.Stat(finalPath)
|
||
if err != nil {
|
||
errMessage = err.Error()
|
||
logger.Errorf("获取备份文件信息失败：%v", err)
|
||
return
|
||
}
|
||
fileSize = info.Size()
|
||
fileName = filepath.Base(finalPath)
|
||
storagePath = backup.BuildStorageKey(task.Type, startedAt, fileName)
|
||
|
||
// 收集所有存储目标
|
||
targetIDs := collectTargetIDs(task)
|
||
if len(targetIDs) == 0 {
|
||
errMessage = "没有关联的存储目标"
|
||
logger.Errorf("没有关联的存储目标")
|
||
return
|
||
}
|
||
storageUsage, err := s.storageUsageSnapshot(ctx)
|
||
if err != nil {
|
||
logger.Warnf("读取存储目标用量失败，跳过本次软配额校验：%v", err)
|
||
storageUsage = map[uint]int64{}
|
||
}
|
||
|
||
// 并行上传到所有目标
|
||
uploadResults = make([]StorageUploadResultItem, len(targetIDs))
|
||
var checksumOnce sync.Once
|
||
var wg sync.WaitGroup
|
||
for i, tid := range targetIDs {
|
||
wg.Add(1)
|
||
go func(index int, targetID uint) {
|
||
defer wg.Done()
|
||
target, findErr := s.targets.FindByID(ctx, targetID)
|
||
targetName := fmt.Sprintf("target-%d", targetID)
|
||
if findErr == nil && target != nil {
|
||
targetName = target.Name
|
||
}
|
||
// 节点级带宽覆盖：若 task 绑定节点并配置了 BandwidthLimit，覆盖全局限速
|
||
provider, resolveErr := s.resolveProviderForNode(ctx, targetID, task.NodeID)
|
||
if resolveErr != nil {
|
||
uploadResults[index] = StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: resolveErr.Error()}
|
||
logger.Warnf("存储目标 %s 创建客户端失败：%v", targetName, resolveErr)
|
||
return
|
||
}
|
||
// 软限额校验：QuotaBytes > 0 时，已累计 + 本次 > 配额 → 拒绝上传
|
||
if target != nil && target.QuotaBytes > 0 {
|
||
currentUsed := storageUsage[targetID]
|
||
if currentUsed+fileSize > target.QuotaBytes {
|
||
quotaMsg := fmt.Sprintf("超出存储目标 %s 的配额（%d + %d > %d）", targetName, currentUsed, fileSize, target.QuotaBytes)
|
||
uploadResults[index] = StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: quotaMsg}
|
||
logger.Warnf("%s", quotaMsg)
|
||
return
|
||
}
|
||
}
|
||
logger.Infof("开始上传备份到存储目标：%s", targetName)
|
||
// 上传级重试：最多 3 次，等待时间随 context 取消及时退出。
|
||
var lastUploadErr error
|
||
var hr *hashingReader
|
||
for attempt := 1; attempt <= uploadMaxAttempts; attempt++ {
|
||
if attempt > 1 {
|
||
backoff := time.Duration(attempt-1) * uploadRetryBackoff
|
||
logger.Warnf("存储目标 %s 第 %d 次重试（等待 %v）：%v", targetName, attempt, backoff, lastUploadErr)
|
||
if waitErr := waitForUploadRetry(ctx, backoff); waitErr != nil {
|
||
uploadResults[index] = StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: waitErr.Error()}
|
||
logger.Warnf("存储目标 %s 上传重试已取消：%v", targetName, waitErr)
|
||
return
|
||
}
|
||
}
|
||
artifact, openErr := os.Open(finalPath)
|
||
if openErr != nil {
|
||
uploadResults[index] = StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: openErr.Error()}
|
||
logger.Warnf("存储目标 %s 打开备份文件失败：%v", targetName, openErr)
|
||
return
|
||
}
|
||
hr = newHashingReader(artifact)
|
||
pr := newProgressReader(hr, fileSize, func(bytesRead int64, speedBps float64) {
|
||
percent := float64(0)
|
||
if fileSize > 0 {
|
||
percent = float64(bytesRead) / float64(fileSize) * 100
|
||
}
|
||
s.logHub.AppendProgress(recordID, backup.ProgressInfo{
|
||
BytesSent: bytesRead,
|
||
TotalBytes: fileSize,
|
||
Percent: percent,
|
||
SpeedBps: speedBps,
|
||
TargetName: targetName,
|
||
})
|
||
})
|
||
lastUploadErr = provider.Upload(ctx, storagePath, pr, fileSize, map[string]string{"taskId": fmt.Sprintf("%d", task.ID), "recordId": fmt.Sprintf("%d", recordID)})
|
||
artifact.Close()
|
||
if lastUploadErr == nil {
|
||
break
|
||
}
|
||
}
|
||
if lastUploadErr != nil {
|
||
uploadResults[index] = StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: lastUploadErr.Error()}
|
||
logger.Warnf("存储目标 %s 上传失败（已重试 %d 次）：%v", targetName, uploadMaxAttempts, lastUploadErr)
|
||
return
|
||
}
|
||
// 完整性校验：对比实际传输字节数
|
||
if hr.n != fileSize {
|
||
uploadResults[index] = StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: fmt.Sprintf("完整性校验失败: 预期 %d bytes, 实际传输 %d bytes", fileSize, hr.n)}
|
||
logger.Errorf("存储目标 %s 完整性校验失败：预期 %d bytes, 实际传输 %d bytes", targetName, fileSize, hr.n)
|
||
_ = provider.Delete(ctx, storagePath)
|
||
return
|
||
}
|
||
// 取第一个成功目标的哈希写入 record（所有目标读同一文件，哈希一定相同）
|
||
targetChecksum := hr.Sum()
|
||
checksumOnce.Do(func() { checksum = targetChecksum })
|
||
uploadResults[index] = StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "success", StoragePath: storagePath, FileSize: fileSize}
|
||
logger.Infof("存储目标 %s 上传成功 (%d bytes, SHA-256=%s)", targetName, fileSize, targetChecksum)
|
||
// 每个成功目标独立执行保留策略
|
||
if s.retention != nil {
|
||
cleanupResult, cleanupErr := s.retention.Cleanup(ctx, task, provider)
|
||
if cleanupErr != nil {
|
||
logger.Warnf("存储目标 %s 执行保留策略失败：%v", targetName, cleanupErr)
|
||
} else {
|
||
for _, warning := range cleanupResult.Warnings {
|
||
logger.Warnf("存储目标 %s 保留策略警告：%s", targetName, warning)
|
||
}
|
||
}
|
||
}
|
||
}(i, tid)
|
||
}
|
||
wg.Wait()
|
||
|
||
// 汇总结果：任意一个 success → 整体 success
|
||
anySuccess := false
|
||
var failedMessages []string
|
||
for _, r := range uploadResults {
|
||
if r.Status == "success" {
|
||
anySuccess = true
|
||
if selectedStorageTargetID == task.StorageTargetID {
|
||
selectedStorageTargetID = r.StorageTargetID
|
||
}
|
||
} else if r.Error != "" {
|
||
failedMessages = append(failedMessages, fmt.Sprintf("%s: %s", r.StorageTargetName, r.Error))
|
||
}
|
||
}
|
||
if anySuccess {
|
||
status = "success"
|
||
if len(failedMessages) > 0 {
|
||
logger.Warnf("部分存储目标上传失败：%s", strings.Join(failedMessages, "; "))
|
||
}
|
||
logger.Infof("备份执行完成")
|
||
// 自动派发复制（3-2-1）：任务配置 ReplicationTargetIDs 且本次有任意目标成功时生效
|
||
// 触发下游依赖任务（best-effort，失败仅 warn）
|
||
if s.dependentsResolver != nil {
|
||
go func(upstreamID uint, upstreamName string) {
|
||
dependents, err := s.dependentsResolver.TriggerDependents(context.Background(), upstreamID)
|
||
if err != nil {
|
||
return
|
||
}
|
||
for _, depID := range dependents {
|
||
_, runErr := s.RunTaskByID(context.Background(), depID)
|
||
if runErr != nil {
|
||
logger.Warnf("触发下游任务 #%d 失败（上游: %s）: %v", depID, upstreamName, runErr)
|
||
} else {
|
||
logger.Infof("已触发下游任务 #%d（上游: %s）", depID, upstreamName)
|
||
}
|
||
}
|
||
}(task.ID, task.Name)
|
||
}
|
||
if s.replicationHook != nil && strings.TrimSpace(task.ReplicationTargetIDs) != "" {
|
||
record := &model.BackupRecord{
|
||
ID: recordID,
|
||
TaskID: task.ID,
|
||
StorageTargetID: selectedStorageTargetID,
|
||
NodeID: task.NodeID,
|
||
Status: "success",
|
||
FileName: fileName,
|
||
FileSize: fileSize,
|
||
Checksum: checksum,
|
||
StoragePath: storagePath,
|
||
StartedAt: startedAt,
|
||
}
|
||
// 取第一个成功的上传作为源 target，避免从失败目标拉取
|
||
for _, r := range uploadResults {
|
||
if r.Status == "success" {
|
||
record.StorageTargetID = r.StorageTargetID
|
||
break
|
||
}
|
||
}
|
||
logger.Infof("触发自动复制（3-2-1 规则）：%s", task.ReplicationTargetIDs)
|
||
s.replicationHook.TriggerAutoReplication(context.Background(), task, record)
|
||
}
|
||
} else {
|
||
errMessage = strings.Join(failedMessages, "; ")
|
||
logger.Errorf("所有存储目标上传均失败")
|
||
}
|
||
}
|
||
|
||
func (s *BackupExecutionService) finalizeRecord(ctx context.Context, task *model.BackupTask, recordID uint, startedAt time.Time, status string, errorMessage string, logContent string, fileName string, fileSize int64, checksum string, storagePath string, storageTargetID uint) error {
|
||
record, err := s.records.FindByID(ctx, recordID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if record == nil {
|
||
return fmt.Errorf("backup record %d not found", recordID)
|
||
}
|
||
completedAt := s.now()
|
||
record.Status = status
|
||
if storageTargetID > 0 {
|
||
record.StorageTargetID = storageTargetID
|
||
}
|
||
record.FileName = fileName
|
||
record.FileSize = fileSize
|
||
record.Checksum = checksum
|
||
record.StoragePath = storagePath
|
||
record.DurationSeconds = int(completedAt.Sub(startedAt).Seconds())
|
||
record.ErrorMessage = strings.TrimSpace(errorMessage)
|
||
record.LogContent = strings.TrimSpace(logContent)
|
||
record.CompletedAt = &completedAt
|
||
if err := s.records.Update(ctx, record); err != nil {
|
||
return err
|
||
}
|
||
task.LastRunAt = &startedAt
|
||
task.LastStatus = status
|
||
return s.tasks.Update(ctx, task)
|
||
}
|
||
|
||
func (s *BackupExecutionService) storageUsageSnapshot(ctx context.Context) (map[uint]int64, error) {
|
||
items, err := s.records.StorageUsage(ctx)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("storage usage snapshot: %w", err)
|
||
}
|
||
usage := make(map[uint]int64, len(items))
|
||
for _, item := range items {
|
||
usage[item.StorageTargetID] = item.TotalSize
|
||
}
|
||
return usage, nil
|
||
}
|
||
|
||
func waitForUploadRetry(ctx context.Context, delay time.Duration) error {
|
||
if delay <= 0 {
|
||
return nil
|
||
}
|
||
timer := time.NewTimer(delay)
|
||
defer timer.Stop()
|
||
select {
|
||
case <-ctx.Done():
|
||
return ctx.Err()
|
||
case <-timer.C:
|
||
return nil
|
||
}
|
||
}
|
||
|
||
func (s *BackupExecutionService) resolveProvider(ctx context.Context, targetID uint) (storage.StorageProvider, error) {
|
||
return s.resolveProviderForNode(ctx, targetID, 0)
|
||
}
|
||
|
||
// resolveProviderForNode 根据节点的 BandwidthLimit 覆盖全局默认。
|
||
// nodeID=0 或节点未配置时退化为全局默认。
|
||
// 仅在 Master 本地执行生效；Agent 会收到自身 Node 配置，并在独立 runtime 中应用。
|
||
func (s *BackupExecutionService) resolveProviderForNode(ctx context.Context, targetID uint, nodeID uint) (storage.StorageProvider, error) {
|
||
// 注入 rclone 传输配置（重试、节点级带宽覆盖全局）
|
||
ctx = rclone.ConfiguredContext(ctx, rclone.TransferConfig{
|
||
LowLevelRetries: s.retries,
|
||
BandwidthLimit: s.effectiveBandwidth(ctx, nodeID),
|
||
})
|
||
return resolveStorageProvider(ctx, s.targets, s.storageRegistry, s.cipher, targetID)
|
||
}
|
||
|
||
func (s *BackupExecutionService) buildTaskSpec(task *model.BackupTask, startedAt time.Time) (backup.TaskSpec, error) {
|
||
return buildBackupTaskSpec(s.cipher, task, startedAt, s.tempDir)
|
||
}
|
||
|
||
// applyHANAExtraConfig 从 ExtraConfig map 中提取 SAP HANA 字段填入 DatabaseSpec。
|
||
// 不识别的键被忽略，保持向后兼容。
|
||
func applyHANAExtraConfig(spec *backup.DatabaseSpec, extra map[string]any) {
|
||
if v, ok := extra["instanceNumber"].(string); ok {
|
||
spec.InstanceNumber = strings.TrimSpace(v)
|
||
}
|
||
if v, ok := extra["backupLevel"].(string); ok {
|
||
spec.BackupLevel = strings.ToLower(strings.TrimSpace(v))
|
||
}
|
||
if v, ok := extra["backupType"].(string); ok {
|
||
spec.BackupType = strings.ToLower(strings.TrimSpace(v))
|
||
}
|
||
if v, ok := extra["backupChannels"].(float64); ok {
|
||
spec.BackupChannels = int(v)
|
||
}
|
||
if v, ok := extra["maxRetries"].(float64); ok {
|
||
spec.MaxRetries = int(v)
|
||
}
|
||
}
|
||
|
||
func (s *BackupExecutionService) loadRecordProvider(ctx context.Context, recordID uint) (*model.BackupRecord, storage.StorageProvider, error) {
|
||
record, err := s.records.FindByID(ctx, recordID)
|
||
if err != nil {
|
||
return nil, nil, apperror.Internal("BACKUP_RECORD_GET_FAILED", "无法获取备份记录详情", err)
|
||
}
|
||
if record == nil {
|
||
return nil, nil, apperror.New(404, "BACKUP_RECORD_NOT_FOUND", "备份记录不存在", fmt.Errorf("backup record %d not found", recordID))
|
||
}
|
||
if err := s.validateClusterAccessible(ctx, record); err != nil {
|
||
return nil, nil, err
|
||
}
|
||
provider, err := s.resolveProvider(ctx, record.StorageTargetID)
|
||
if err != nil {
|
||
return nil, nil, err
|
||
}
|
||
return record, provider, nil
|
||
}
|
||
|
||
func (s *BackupExecutionService) prepareArtifactForRestore(artifactPath string) (string, error) {
|
||
return prepareBackupArtifact(s.cipher, artifactPath, nil)
|
||
}
|
||
|
||
func (s *BackupExecutionService) getRecordDetail(ctx context.Context, recordID uint) (*BackupRecordDetail, error) {
|
||
record, err := s.records.FindByID(ctx, recordID)
|
||
if err != nil {
|
||
return nil, apperror.Internal("BACKUP_RECORD_GET_FAILED", "无法获取备份记录详情", err)
|
||
}
|
||
if record == nil {
|
||
return nil, apperror.New(404, "BACKUP_RECORD_NOT_FOUND", "备份记录不存在", fmt.Errorf("backup record %d not found", recordID))
|
||
}
|
||
return toBackupRecordDetail(record, s.logHub), nil
|
||
}
|
||
|
||
func writeReaderToFile(targetPath string, reader io.ReadCloser) error {
|
||
defer reader.Close()
|
||
if err := os.MkdirAll(filepath.Dir(targetPath), 0o755); err != nil {
|
||
return err
|
||
}
|
||
file, err := os.Create(targetPath)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer file.Close()
|
||
_, err = io.Copy(file, reader)
|
||
return err
|
||
}
|
||
|
||
func buildOptionalError(message string) error {
|
||
if strings.TrimSpace(message) == "" {
|
||
return nil
|
||
}
|
||
return fmt.Errorf("%s", message)
|
||
}
|
||
|
||
func buildStorageProviderFromRepos(ctx context.Context, storageTargetID uint, storageTargets repository.StorageTargetRepository, storageRegistry *storage.Registry, cipher *codec.ConfigCipher) (storage.StorageProvider, *model.StorageTarget, error) {
|
||
target, err := storageTargets.FindByID(ctx, storageTargetID)
|
||
if err != nil {
|
||
return nil, nil, apperror.Internal("BACKUP_STORAGE_TARGET_LOOKUP_FAILED", "无法读取存储目标", err)
|
||
}
|
||
if target == nil {
|
||
return nil, nil, apperror.BadRequest("BACKUP_STORAGE_TARGET_INVALID", "存储目标不存在", nil)
|
||
}
|
||
var configMap map[string]any
|
||
if err := cipher.DecryptJSON(target.ConfigCiphertext, &configMap); err != nil {
|
||
return nil, nil, apperror.Internal("BACKUP_STORAGE_TARGET_DECRYPT_FAILED", "无法解密存储目标配置", err)
|
||
}
|
||
provider, err := storageRegistry.Create(ctx, storage.ParseProviderType(target.Type), configMap)
|
||
if err != nil {
|
||
return nil, nil, err
|
||
}
|
||
return provider, target, nil
|
||
}
|
||
|
||
// hashingReader 在上传过程中同步计算字节数和 SHA-256，零额外 I/O
|
||
type hashingReader struct {
|
||
reader io.Reader
|
||
hash hash.Hash
|
||
n int64
|
||
}
|
||
|
||
func newHashingReader(reader io.Reader) *hashingReader {
|
||
h := sha256.New()
|
||
return &hashingReader{
|
||
reader: io.TeeReader(reader, h),
|
||
hash: h,
|
||
}
|
||
}
|
||
|
||
func (r *hashingReader) Read(p []byte) (int, error) {
|
||
n, err := r.reader.Read(p)
|
||
r.n += int64(n)
|
||
return n, err
|
||
}
|
||
|
||
func (r *hashingReader) Sum() string {
|
||
return hex.EncodeToString(r.hash.Sum(nil))
|
||
}
|