mirror of
https://github.com/Awuqing/BackupX.git
synced 2026-08-28 19:47:27 +08:00
Awuqing
c06bbec383
Remove obsolete implementations, centralize background task ownership and terminal-state recovery, consolidate frontend routing and log streaming, and enforce project-wide verification in CI.
397 lines
14 KiB
Go
397 lines
14 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"backupx/server/internal/apperror"
|
|
"backupx/server/internal/metrics"
|
|
"backupx/server/internal/model"
|
|
"backupx/server/internal/repository"
|
|
"backupx/server/internal/storage"
|
|
"backupx/server/internal/storage/codec"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
// ReplicationService 实现备份复制（3-2-1 规则核心）。
|
|
// 语义：把源备份对象从 source storage target 镜像到 dest target，保持 StoragePath。
|
|
//
|
|
// 触发路径：
|
|
// 1. 自动：BackupExecutionService 备份成功后调用 TriggerAutoReplication
|
|
// 2. 手动：前端通过 BackupRecord 详情页触发 Start
|
|
//
|
|
// 执行模型：异步 + 节点无关（复制在 Master 本地 download → upload）。
|
|
// 跨节点 local_disk 场景不支持（与 Download/Delete 保护一致）。
|
|
type ReplicationService struct {
|
|
replications repository.ReplicationRecordRepository
|
|
records repository.BackupRecordRepository
|
|
targets repository.StorageTargetRepository
|
|
nodeRepo repository.NodeRepository
|
|
storageRegistry *storage.Registry
|
|
cipher *codec.ConfigCipher
|
|
eventDispatcher EventDispatcher
|
|
tempDir string
|
|
semaphore chan struct{}
|
|
async func(func(context.Context)) bool
|
|
now func() time.Time
|
|
metrics *metrics.Metrics
|
|
logger *zap.Logger
|
|
}
|
|
|
|
// SetMetrics 注入 Prometheus 采集器。
|
|
func (s *ReplicationService) SetMetrics(m *metrics.Metrics) {
|
|
s.metrics = m
|
|
}
|
|
|
|
func (s *ReplicationService) SetLogger(logger *zap.Logger) {
|
|
if logger != nil {
|
|
s.logger = logger
|
|
}
|
|
}
|
|
|
|
// SetBackgroundRunner binds replication work to the application lifecycle.
|
|
func (s *ReplicationService) SetBackgroundRunner(runner BackgroundRunner) {
|
|
if runner != nil {
|
|
s.async = runner.Go
|
|
}
|
|
}
|
|
|
|
func NewReplicationService(
|
|
replications repository.ReplicationRecordRepository,
|
|
records repository.BackupRecordRepository,
|
|
targets repository.StorageTargetRepository,
|
|
nodeRepo repository.NodeRepository,
|
|
storageRegistry *storage.Registry,
|
|
cipher *codec.ConfigCipher,
|
|
tempDir string,
|
|
maxConcurrent int,
|
|
) *ReplicationService {
|
|
if tempDir == "" {
|
|
tempDir = "/tmp/backupx-replicate"
|
|
}
|
|
if maxConcurrent <= 0 {
|
|
maxConcurrent = 2
|
|
}
|
|
return &ReplicationService{
|
|
replications: replications,
|
|
records: records,
|
|
targets: targets,
|
|
nodeRepo: nodeRepo,
|
|
storageRegistry: storageRegistry,
|
|
cipher: cipher,
|
|
tempDir: tempDir,
|
|
semaphore: make(chan struct{}, maxConcurrent),
|
|
async: runDetached,
|
|
now: func() time.Time { return time.Now().UTC() },
|
|
logger: zap.NewNop(),
|
|
}
|
|
}
|
|
|
|
func (s *ReplicationService) SetEventDispatcher(dispatcher EventDispatcher) {
|
|
s.eventDispatcher = dispatcher
|
|
}
|
|
|
|
// ReplicationRecordSummary 列表项。
|
|
type ReplicationRecordSummary struct {
|
|
ID uint `json:"id"`
|
|
BackupRecordID uint `json:"backupRecordId"`
|
|
TaskID uint `json:"taskId"`
|
|
SourceTargetID uint `json:"sourceTargetId"`
|
|
SourceTargetName string `json:"sourceTargetName"`
|
|
DestTargetID uint `json:"destTargetId"`
|
|
DestTargetName string `json:"destTargetName"`
|
|
Status string `json:"status"`
|
|
StoragePath string `json:"storagePath"`
|
|
FileSize int64 `json:"fileSize"`
|
|
Checksum string `json:"checksum"`
|
|
ErrorMessage string `json:"errorMessage"`
|
|
DurationSeconds int `json:"durationSeconds"`
|
|
TriggeredBy string `json:"triggeredBy"`
|
|
StartedAt time.Time `json:"startedAt"`
|
|
CompletedAt *time.Time `json:"completedAt,omitempty"`
|
|
}
|
|
|
|
type ReplicationRecordListInput struct {
|
|
TaskID *uint
|
|
BackupRecordID *uint
|
|
DestTargetID *uint
|
|
Status string
|
|
DateFrom *time.Time
|
|
DateTo *time.Time
|
|
Limit int
|
|
Offset int
|
|
}
|
|
|
|
// TriggerAutoReplication 备份成功钩子：根据 task.ReplicationTargetIDs 自动派发复制。
|
|
// best-effort：单个目标失败不影响其他。
|
|
func (s *ReplicationService) TriggerAutoReplication(ctx context.Context, task *model.BackupTask, record *model.BackupRecord) {
|
|
if task == nil || record == nil {
|
|
return
|
|
}
|
|
destIDs := parseUintCSV(task.ReplicationTargetIDs)
|
|
if len(destIDs) == 0 {
|
|
return
|
|
}
|
|
// 跨节点 local_disk 场景保护：Master 无法访问远程节点本地文件
|
|
if err := s.validateClusterAccessible(ctx, record); err != nil {
|
|
s.logger.Warn("automatic replication skipped: source is not accessible", zap.Uint("backup_record_id", record.ID), zap.Error(err))
|
|
return
|
|
}
|
|
for _, destID := range destIDs {
|
|
if destID == record.StorageTargetID {
|
|
continue // 源与目标相同，跳过
|
|
}
|
|
if _, err := s.Start(ctx, record.ID, destID, "system"); err != nil {
|
|
s.logger.Warn("automatic replication start failed", zap.Uint("backup_record_id", record.ID), zap.Uint("dest_target_id", destID), zap.Error(err))
|
|
}
|
|
}
|
|
}
|
|
|
|
// Start 开始一次复制。同步创建 ReplicationRecord → 异步执行。
|
|
func (s *ReplicationService) Start(ctx context.Context, backupRecordID, destTargetID uint, triggeredBy string) (*ReplicationRecordSummary, error) {
|
|
record, err := s.records.FindByID(ctx, backupRecordID)
|
|
if err != nil {
|
|
return nil, apperror.Internal("BACKUP_RECORD_GET_FAILED", "无法获取备份记录", err)
|
|
}
|
|
if record == nil {
|
|
return nil, apperror.New(404, "BACKUP_RECORD_NOT_FOUND", "备份记录不存在", nil)
|
|
}
|
|
if record.Status != model.BackupRecordStatusSuccess {
|
|
return nil, apperror.BadRequest("REPLICATION_SOURCE_INVALID", "只能复制成功的备份记录", nil)
|
|
}
|
|
if destTargetID == 0 || destTargetID == record.StorageTargetID {
|
|
return nil, apperror.BadRequest("REPLICATION_DEST_INVALID", "目标存储无效或与源相同", nil)
|
|
}
|
|
if err := s.validateClusterAccessible(ctx, record); err != nil {
|
|
return nil, err
|
|
}
|
|
dest, err := s.targets.FindByID(ctx, destTargetID)
|
|
if err != nil || dest == nil {
|
|
return nil, apperror.BadRequest("REPLICATION_DEST_INVALID", "目标存储不存在", err)
|
|
}
|
|
if !dest.Enabled {
|
|
return nil, apperror.BadRequest("REPLICATION_DEST_DISABLED", "目标存储已禁用", nil)
|
|
}
|
|
startedAt := s.now()
|
|
rep := &model.ReplicationRecord{
|
|
BackupRecordID: backupRecordID,
|
|
TaskID: record.TaskID,
|
|
SourceTargetID: record.StorageTargetID,
|
|
DestTargetID: destTargetID,
|
|
Status: model.ReplicationStatusRunning,
|
|
StoragePath: record.StoragePath,
|
|
TriggeredBy: strings.TrimSpace(triggeredBy),
|
|
StartedAt: startedAt,
|
|
}
|
|
if err := s.replications.Create(ctx, rep); err != nil {
|
|
return nil, apperror.Internal("REPLICATION_CREATE_FAILED", "无法创建复制记录", err)
|
|
}
|
|
repForRun := *rep
|
|
if !s.async(func(runCtx context.Context) {
|
|
s.executeReplication(runCtx, &repForRun)
|
|
}) {
|
|
message := "服务正在关闭，复制任务未启动"
|
|
if finalizeErr := s.finalizeReplication(ctx, rep, model.ReplicationStatusFailed, message, 0); finalizeErr != nil {
|
|
return nil, apperror.Internal("REPLICATION_FINALIZE_FAILED", "无法写回复制失败状态", finalizeErr)
|
|
}
|
|
return nil, backgroundTaskUnavailable("REPLICATION_SERVICE_SHUTTING_DOWN")
|
|
}
|
|
summary := s.toSummary(rep, "", dest.Name)
|
|
return &summary, nil
|
|
}
|
|
|
|
// executeReplication 实际执行：下载源对象到本地临时文件 → 上传到目标存储。
|
|
func (s *ReplicationService) executeReplication(ctx context.Context, rep *model.ReplicationRecord) {
|
|
if rep == nil {
|
|
s.logger.Error("replication record is nil")
|
|
return
|
|
}
|
|
status := model.ReplicationStatusFailed
|
|
errMessage := ""
|
|
fileSize := int64(0)
|
|
|
|
defer func() {
|
|
persistCtx, cancel := finalizationContext(ctx)
|
|
defer cancel()
|
|
if finalizeErr := s.finalizeReplication(persistCtx, rep, status, errMessage, fileSize); finalizeErr != nil {
|
|
s.logger.Error("finalize replication record failed", zap.Uint("replication_id", rep.ID), zap.Error(finalizeErr))
|
|
}
|
|
s.metrics.ObserveReplication(status)
|
|
if status == model.ReplicationStatusFailed {
|
|
if dispatchErr := s.dispatchFailed(persistCtx, rep, errMessage); dispatchErr != nil {
|
|
s.logger.Warn("dispatch replication failure event failed", zap.Uint("replication_id", rep.ID), zap.Error(dispatchErr))
|
|
}
|
|
}
|
|
}()
|
|
|
|
if !acquireBackgroundSlot(ctx, s.semaphore) {
|
|
errMessage = ctx.Err().Error()
|
|
return
|
|
}
|
|
defer func() { <-s.semaphore }()
|
|
|
|
sourceProvider, err := s.resolveProvider(ctx, rep.SourceTargetID)
|
|
if err != nil {
|
|
errMessage = err.Error()
|
|
return
|
|
}
|
|
destProvider, err := s.resolveProvider(ctx, rep.DestTargetID)
|
|
if err != nil {
|
|
errMessage = err.Error()
|
|
return
|
|
}
|
|
if err := os.MkdirAll(s.tempDir, 0o755); err != nil {
|
|
errMessage = err.Error()
|
|
return
|
|
}
|
|
tempDir, err := os.MkdirTemp(s.tempDir, "replicate-*")
|
|
if err != nil {
|
|
errMessage = err.Error()
|
|
return
|
|
}
|
|
defer os.RemoveAll(tempDir)
|
|
|
|
reader, err := sourceProvider.Download(ctx, rep.StoragePath)
|
|
if err != nil {
|
|
errMessage = fmt.Sprintf("下载源对象失败: %v", err)
|
|
return
|
|
}
|
|
localPath := filepath.Join(tempDir, filepath.Base(rep.StoragePath))
|
|
if err := writeReaderToFile(localPath, reader); err != nil {
|
|
errMessage = fmt.Sprintf("写入临时文件失败: %v", err)
|
|
return
|
|
}
|
|
info, err := os.Stat(localPath)
|
|
if err != nil {
|
|
errMessage = err.Error()
|
|
return
|
|
}
|
|
fileSize = info.Size()
|
|
file, err := os.Open(localPath)
|
|
if err != nil {
|
|
errMessage = err.Error()
|
|
return
|
|
}
|
|
defer file.Close()
|
|
meta := map[string]string{
|
|
"replicationId": strconv.FormatUint(uint64(rep.ID), 10),
|
|
"sourceRecord": strconv.FormatUint(uint64(rep.BackupRecordID), 10),
|
|
}
|
|
if err := destProvider.Upload(ctx, rep.StoragePath, file, fileSize, meta); err != nil {
|
|
errMessage = fmt.Sprintf("上传到目标存储失败: %v", err)
|
|
return
|
|
}
|
|
rep.Checksum = "" // 可选：调用方可按需复算 SHA-256
|
|
status = model.ReplicationStatusSuccess
|
|
}
|
|
|
|
func (s *ReplicationService) resolveProvider(ctx context.Context, targetID uint) (storage.StorageProvider, error) {
|
|
return resolveStorageProvider(ctx, s.targets, s.storageRegistry, s.cipher, targetID)
|
|
}
|
|
|
|
// validateClusterAccessible 拒绝跨节点 local_disk 源（Master 无法拉取）。
|
|
func (s *ReplicationService) validateClusterAccessible(ctx context.Context, record *model.BackupRecord) error {
|
|
return validateCrossNodeLocalDisk(ctx, s.nodeRepo, s.targets, record,
|
|
"REPLICATION_CROSS_NODE_LOCAL_DISK", "复制。请改用云存储作为主备份")
|
|
}
|
|
|
|
func (s *ReplicationService) finalizeReplication(ctx context.Context, rep *model.ReplicationRecord, status, message string, fileSize int64) error {
|
|
completedAt := s.now()
|
|
rep.Status = status
|
|
rep.FileSize = fileSize
|
|
rep.ErrorMessage = strings.TrimSpace(message)
|
|
rep.DurationSeconds = int(completedAt.Sub(rep.StartedAt).Seconds())
|
|
rep.CompletedAt = &completedAt
|
|
return s.replications.Update(ctx, rep)
|
|
}
|
|
|
|
func (s *ReplicationService) dispatchFailed(ctx context.Context, rep *model.ReplicationRecord, message string) error {
|
|
if s.eventDispatcher == nil || rep == nil {
|
|
return nil
|
|
}
|
|
title := "BackupX 备份复制失败"
|
|
body := fmt.Sprintf("备份记录：#%d\n源 → 目标：#%d → #%d\n错误：%s", rep.BackupRecordID, rep.SourceTargetID, rep.DestTargetID, message)
|
|
fields := map[string]any{
|
|
"replicationId": rep.ID,
|
|
"backupRecordId": rep.BackupRecordID,
|
|
"taskId": rep.TaskID,
|
|
"sourceTargetId": rep.SourceTargetID,
|
|
"destTargetId": rep.DestTargetID,
|
|
"error": message,
|
|
}
|
|
return s.eventDispatcher.DispatchEvent(ctx, model.NotificationEventReplicationFailed, title, body, fields)
|
|
}
|
|
|
|
// List / Get / toSummary
|
|
func (s *ReplicationService) List(ctx context.Context, input ReplicationRecordListInput) ([]ReplicationRecordSummary, error) {
|
|
items, err := s.replications.List(ctx, repository.ReplicationRecordListOptions{
|
|
TaskID: input.TaskID, BackupRecordID: input.BackupRecordID, DestTargetID: input.DestTargetID,
|
|
Status: strings.TrimSpace(input.Status), DateFrom: input.DateFrom, DateTo: input.DateTo,
|
|
Limit: input.Limit, Offset: input.Offset,
|
|
})
|
|
if err != nil {
|
|
return nil, apperror.Internal("REPLICATION_LIST_FAILED", "无法获取复制记录", err)
|
|
}
|
|
result := make([]ReplicationRecordSummary, 0, len(items))
|
|
for i := range items {
|
|
item := items[i]
|
|
result = append(result, s.toSummary(&item, item.SourceTarget.Name, item.DestTarget.Name))
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *ReplicationService) Get(ctx context.Context, id uint) (*ReplicationRecordSummary, error) {
|
|
item, err := s.replications.FindByID(ctx, id)
|
|
if err != nil {
|
|
return nil, apperror.Internal("REPLICATION_GET_FAILED", "无法获取复制记录", err)
|
|
}
|
|
if item == nil {
|
|
return nil, apperror.New(404, "REPLICATION_NOT_FOUND", "复制记录不存在", nil)
|
|
}
|
|
summary := s.toSummary(item, item.SourceTarget.Name, item.DestTarget.Name)
|
|
return &summary, nil
|
|
}
|
|
|
|
func (s *ReplicationService) toSummary(rep *model.ReplicationRecord, sourceName, destName string) ReplicationRecordSummary {
|
|
return ReplicationRecordSummary{
|
|
ID: rep.ID, BackupRecordID: rep.BackupRecordID, TaskID: rep.TaskID,
|
|
SourceTargetID: rep.SourceTargetID, SourceTargetName: sourceName,
|
|
DestTargetID: rep.DestTargetID, DestTargetName: destName,
|
|
Status: rep.Status, StoragePath: rep.StoragePath, FileSize: rep.FileSize,
|
|
Checksum: rep.Checksum, ErrorMessage: rep.ErrorMessage, DurationSeconds: rep.DurationSeconds,
|
|
TriggeredBy: rep.TriggeredBy, StartedAt: rep.StartedAt, CompletedAt: rep.CompletedAt,
|
|
}
|
|
}
|
|
|
|
// parseUintCSV 解析逗号分隔的 uint 列表，跳过非法项。
|
|
func parseUintCSV(value string) []uint {
|
|
if strings.TrimSpace(value) == "" {
|
|
return nil
|
|
}
|
|
parts := strings.Split(value, ",")
|
|
out := make([]uint, 0, len(parts))
|
|
seen := map[uint]bool{}
|
|
for _, p := range parts {
|
|
trimmed := strings.TrimSpace(p)
|
|
if trimmed == "" {
|
|
continue
|
|
}
|
|
parsed, err := strconv.ParseUint(trimmed, 10, 32)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
id := uint(parsed)
|
|
if seen[id] {
|
|
continue
|
|
}
|
|
seen[id] = true
|
|
out = append(out, id)
|
|
}
|
|
return out
|
|
}
|