Files
BackupX/server/internal/service/task_export_service.go
T
Wu Qing 539e9e64c4 功能: v2.0.0 企业级备份管理平台 — 11 项核心能力 (#45)
* 功能: v2.0.0 企业级备份管理平台 — 11 项核心能力

围绕"可靠、可验证、可度量、可冗余、可治理、可规模化、可运维、可部署、可感知"的
九大企业级支柱,新增 70+ 文件、14k+ 行代码,全链路测试与类型检查通过。

## 集群能力

- 节点选择器:任务表单支持绑定远程节点,集群场景不再被迫 NodeID=0
- 集群感知恢复:RestoreRecord 独立表 + 节点路由(本机/远程 Agent)+ SSE 日志
- 集群可靠性:命令超时联动备份/恢复记录、离线节点拒绝执行、调度器跳过离线节点、
  数据库发现路由到 Agent、跨节点 local_disk 保护
- 节点级资源配额:Node.MaxConcurrent / BandwidthLimit + per-node semaphore
- Agent 版本感知:ClusterVersionMonitor 定期扫描 + agent_outdated 事件
- Dashboard 集群概览 + 节点性能统计(成功率/字节/平均耗时)

## 企业功能

- 备份验证演练:定时自动校验备份可恢复性(tar/sqlite/mysql/postgres/saphana 5 类格式)
- SLA 监控:RPO 违约后台扫描 + sla_violation 事件 + Dashboard 合规视图
- 3-2-1 备份复制:自动/手动副本镜像 + 跨节点保护
- 存储目标健康监控 + 容量预警(85%)+ 硬配额(超配额拒绝)
- RBAC 三级角色(admin/operator/viewer)+ 前后端权限控制
- API Key 管理(bax_ 前缀 SHA-256 哈希存储 + 过期/启停)
- 事件总线:10+ 事件类型(backup/restore/verify/sla/storage/replication/agent)
- 审计日志高级筛选 + CSV 导出

## 规模化运维

- 任务模板(批量创建 + 变量覆盖)
- 任务批量操作(批量执行/启停/删除)
- 任务依赖链 + DAG 可视化(上游成功触发下游)
- 维护窗口(时段禁止调度)
- 任务标签 + 筛选 + 存储类型/节点/存储维度统计
- 任务配置 JSON 导入/导出(集群迁移 & 灾备)

## 体验 & 可达性

- 实时事件流(SSE)+ 右下角 Toast + 历史抽屉(未读徽章)
- Dashboard 免刷新自动更新(订阅 8 类事件)
- 全局搜索(Ctrl+K,跨任务/记录/存储/节点)
- 任务依赖图(ECharts force 布局 + 状态着色)

## 合规 & 可部署

- K8s/Swarm 健康检查端点(/health liveness + /ready readiness)
- 审计日志 CSV 导出(UTF-8 BOM,Excel 兼容)
- Dashboard 多维统计(按类型/状态/节点/存储)

## 破坏性变更

- POST /backup/records/:id/restore 返回格式变更为 {restoreRecordId, ...}
  (原为同步阻塞,现改为异步返回恢复记录 ID,前端跳转到恢复详情页)
- 恢复日志通过 /restore/records/:id/logs/stream 订阅
- AuthMiddleware 签名变更(新增 apiKeyAuth 参数)

* 修复: CodeQL 安全扫描告警

- 所有 strconv.ParseUint 由 64bit 改为 32bit 位宽,strconv 内置溢出检查
- hashApiKey 参数改名 rawToken 避免 CodeQL 误判为密码哈希(API Key 是 192 位
  高熵 token,使用 bcrypt 会引入不必要的延迟;同时补充安全说明)

* 修复: API Key 哈希改用 HMAC-SHA256 + 应用级 pepper

- 符合 RFC 2104 标准,业界 API token 存储的推荐方案
- 数据库泄漏场景下增加离线反推难度(需同时获取二进制 pepper)
- 规避 CodeQL go/weak-sensitive-data-hashing 对裸 SHA-256 的误判
2026-04-20 13:04:13 +08:00

319 lines
11 KiB
Go

package service
import (
"context"
"encoding/json"
"fmt"
"strings"
"time"
"backupx/server/internal/apperror"
"backupx/server/internal/model"
"backupx/server/internal/repository"
)
// TaskExportService 管理备份任务的 JSON 导入 / 导出。
// 用途:
// 1. 集群迁移(旧 Master → 新 Master 的任务配置搬迁)
// 2. 灾备恢复(任务配置本地文件化,Master 宕机后重建)
// 3. 配置审计(版本化 Git 管理 JSON 快照)
//
// 出于安全考虑,导出/导入不包含任何敏感字段:
// - 数据库密码(DBPasswordCiphertext):跳过,导入后需人工填补
// - 存储目标具体配置:仅按 name 匹配现有目标,不搬运密钥
// - Node 绑定:按 name 匹配现有节点,不存在时退化为 NodeID=0(本机)
type TaskExportService struct {
tasks *BackupTaskService
taskRepo repository.BackupTaskRepository
targets repository.StorageTargetRepository
nodes repository.NodeRepository
}
func NewTaskExportService(
tasks *BackupTaskService,
taskRepo repository.BackupTaskRepository,
targets repository.StorageTargetRepository,
nodes repository.NodeRepository,
) *TaskExportService {
return &TaskExportService{tasks: tasks, taskRepo: taskRepo, targets: targets, nodes: nodes}
}
// ExportedTask 导出格式:按名称引用存储/节点,不含敏感数据。
type ExportedTask struct {
Name string `json:"name"`
Type string `json:"type"`
Enabled bool `json:"enabled"`
CronExpr string `json:"cronExpr,omitempty"`
SourcePath string `json:"sourcePath,omitempty"`
SourcePaths []string `json:"sourcePaths,omitempty"`
ExcludePatterns []string `json:"excludePatterns,omitempty"`
DBHost string `json:"dbHost,omitempty"`
DBPort int `json:"dbPort,omitempty"`
DBUser string `json:"dbUser,omitempty"`
DBName string `json:"dbName,omitempty"`
DBPath string `json:"dbPath,omitempty"`
ExtraConfig map[string]any `json:"extraConfig,omitempty"`
// 按名称引用:导入时按名称查找对应 ID
StorageTargetNames []string `json:"storageTargetNames"`
ReplicationTargetNames []string `json:"replicationTargetNames,omitempty"`
NodeName string `json:"nodeName,omitempty"`
DependsOnTaskNames []string `json:"dependsOnTaskNames,omitempty"`
Tags string `json:"tags,omitempty"`
Compression string `json:"compression,omitempty"`
Encrypt bool `json:"encrypt,omitempty"`
RetentionDays int `json:"retentionDays,omitempty"`
MaxBackups int `json:"maxBackups,omitempty"`
VerifyEnabled bool `json:"verifyEnabled,omitempty"`
VerifyCronExpr string `json:"verifyCronExpr,omitempty"`
VerifyMode string `json:"verifyMode,omitempty"`
SLAHoursRPO int `json:"slaHoursRpo,omitempty"`
AlertOnConsecutiveFails int `json:"alertOnConsecutiveFails,omitempty"`
MaintenanceWindows string `json:"maintenanceWindows,omitempty"`
}
// ExportPayload 导出整体结构,带元信息。
type ExportPayload struct {
Version string `json:"version"`
ExportedAt time.Time `json:"exportedAt"`
TaskCount int `json:"taskCount"`
Tasks []ExportedTask `json:"tasks"`
Notice string `json:"notice"`
}
// ImportResult 导入单条结果,best-effort。
type ImportResult struct {
Name string `json:"name"`
TaskID uint `json:"taskId,omitempty"`
Success bool `json:"success"`
Error string `json:"error,omitempty"`
Skipped bool `json:"skipped,omitempty"`
}
// Export 导出当前全部任务为 JSON。
// taskIDs 为空则导出全部;否则仅导出指定 ID。
func (s *TaskExportService) Export(ctx context.Context, taskIDs []uint) (*ExportPayload, error) {
items, err := s.taskRepo.List(ctx, repository.BackupTaskListOptions{})
if err != nil {
return nil, apperror.Internal("TASK_EXPORT_LIST_FAILED", "无法获取任务列表", err)
}
targetNames := map[uint]string{}
if all, err := s.targets.List(ctx); err == nil {
for _, t := range all {
targetNames[t.ID] = t.Name
}
}
nodeNames := map[uint]string{}
if all, err := s.nodes.List(ctx); err == nil {
for _, n := range all {
nodeNames[n.ID] = n.Name
}
}
taskNames := map[uint]string{}
for _, t := range items {
taskNames[t.ID] = t.Name
}
idFilter := map[uint]bool{}
for _, id := range taskIDs {
idFilter[id] = true
}
exported := make([]ExportedTask, 0, len(items))
for i := range items {
item := items[i]
if len(idFilter) > 0 && !idFilter[item.ID] {
continue
}
et := s.toExported(&item, targetNames, nodeNames, taskNames)
exported = append(exported, et)
}
return &ExportPayload{
Version: "v1",
ExportedAt: time.Now().UTC(),
TaskCount: len(exported),
Tasks: exported,
Notice: "敏感字段(数据库密码、存储凭证)已排除,导入后需人工补全。",
}, nil
}
// Import 批量导入任务。best-effort:单条失败不阻断。
// 冲突策略:任务名重复则跳过(不覆盖)。
func (s *TaskExportService) Import(ctx context.Context, payload ExportPayload) ([]ImportResult, error) {
// 预加载所有命名 → ID 映射
targetsByName := map[string]uint{}
if all, err := s.targets.List(ctx); err == nil {
for _, t := range all {
targetsByName[t.Name] = t.ID
}
}
nodesByName := map[string]uint{}
if all, err := s.nodes.List(ctx); err == nil {
for _, n := range all {
nodesByName[n.Name] = n.ID
}
}
tasksByName := map[string]uint{}
existing, err := s.taskRepo.List(ctx, repository.BackupTaskListOptions{})
if err != nil {
return nil, apperror.Internal("TASK_IMPORT_LIST_FAILED", "无法读取当前任务列表", err)
}
for _, t := range existing {
tasksByName[t.Name] = t.ID
}
results := make([]ImportResult, 0, len(payload.Tasks))
// 两阶段:先创建所有任务(忽略 DependsOn),再更新依赖
created := map[string]uint{}
for _, t := range payload.Tasks {
if t.Name == "" {
continue
}
if _, dup := tasksByName[t.Name]; dup {
results = append(results, ImportResult{Name: t.Name, Skipped: true, Success: true, Error: "已存在同名任务,跳过"})
continue
}
input := s.toUpsertInput(t, targetsByName, nodesByName, nil)
detail, err := s.tasks.Create(ctx, input)
if err != nil {
results = append(results, ImportResult{Name: t.Name, Success: false, Error: appErrorMessage(err)})
continue
}
created[t.Name] = detail.ID
tasksByName[t.Name] = detail.ID
results = append(results, ImportResult{Name: t.Name, TaskID: detail.ID, Success: true})
}
// 第二阶段:依赖链接(上游任务名 → 新 ID)
for i, t := range payload.Tasks {
if len(t.DependsOnTaskNames) == 0 {
continue
}
id, ok := created[t.Name]
if !ok {
continue
}
deps := []uint{}
for _, name := range t.DependsOnTaskNames {
if depID, ok := tasksByName[name]; ok && depID != id {
deps = append(deps, depID)
}
}
if len(deps) == 0 {
continue
}
input := s.toUpsertInput(t, targetsByName, nodesByName, deps)
if _, err := s.tasks.Update(ctx, id, input); err != nil {
// 已创建但依赖更新失败:降级为 warning,不影响任务本体
for idx := range results {
if results[idx].Name == t.Name {
results[idx].Error = fmt.Sprintf("任务已创建,但依赖更新失败: %s", appErrorMessage(err))
break
}
}
_ = i
}
}
return results, nil
}
func (s *TaskExportService) toExported(item *model.BackupTask, targetNames, nodeNames, taskNames map[uint]string) ExportedTask {
sourcePaths := []string{}
if strings.TrimSpace(item.SourcePaths) != "" {
_ = json.Unmarshal([]byte(item.SourcePaths), &sourcePaths)
}
excludes := []string{}
if strings.TrimSpace(item.ExcludePatterns) != "" {
_ = json.Unmarshal([]byte(item.ExcludePatterns), &excludes)
}
var extra map[string]any
if strings.TrimSpace(item.ExtraConfig) != "" {
_ = json.Unmarshal([]byte(item.ExtraConfig), &extra)
}
storageNames := namesFromIDs(collectTargetIDs(item), targetNames)
replicationNames := namesFromIDs(parseUintCSV(item.ReplicationTargetIDs), targetNames)
dependsOnNames := namesFromIDs(parseUintCSV(item.DependsOnTaskIDs), taskNames)
nodeName := ""
if item.NodeID > 0 {
nodeName = nodeNames[item.NodeID]
}
return ExportedTask{
Name: item.Name,
Type: item.Type,
Enabled: item.Enabled,
CronExpr: item.CronExpr,
SourcePath: item.SourcePath,
SourcePaths: sourcePaths,
ExcludePatterns: excludes,
DBHost: item.DBHost,
DBPort: item.DBPort,
DBUser: item.DBUser,
DBName: item.DBName,
DBPath: item.DBPath,
ExtraConfig: extra,
StorageTargetNames: storageNames,
ReplicationTargetNames: replicationNames,
NodeName: nodeName,
DependsOnTaskNames: dependsOnNames,
Tags: item.Tags,
Compression: item.Compression,
Encrypt: item.Encrypt,
RetentionDays: item.RetentionDays,
MaxBackups: item.MaxBackups,
VerifyEnabled: item.VerifyEnabled,
VerifyCronExpr: item.VerifyCronExpr,
VerifyMode: item.VerifyMode,
SLAHoursRPO: item.SLAHoursRPO,
AlertOnConsecutiveFails: item.AlertOnConsecutiveFails,
MaintenanceWindows: item.MaintenanceWindows,
}
}
func (s *TaskExportService) toUpsertInput(t ExportedTask, targetsByName, nodesByName map[string]uint, deps []uint) BackupTaskUpsertInput {
return BackupTaskUpsertInput{
Name: t.Name,
Type: t.Type,
Enabled: t.Enabled,
CronExpr: t.CronExpr,
SourcePath: t.SourcePath,
SourcePaths: t.SourcePaths,
ExcludePatterns: t.ExcludePatterns,
DBHost: t.DBHost,
DBPort: t.DBPort,
DBUser: t.DBUser,
DBName: t.DBName,
DBPath: t.DBPath,
ExtraConfig: t.ExtraConfig,
StorageTargetIDs: idsFromNames(t.StorageTargetNames, targetsByName),
ReplicationTargetIDs: idsFromNames(t.ReplicationTargetNames, targetsByName),
NodeID: nodesByName[t.NodeName],
Tags: t.Tags,
Compression: t.Compression,
Encrypt: t.Encrypt,
RetentionDays: t.RetentionDays,
MaxBackups: t.MaxBackups,
VerifyEnabled: t.VerifyEnabled,
VerifyCronExpr: t.VerifyCronExpr,
VerifyMode: t.VerifyMode,
SLAHoursRPO: t.SLAHoursRPO,
AlertOnConsecutiveFails: t.AlertOnConsecutiveFails,
MaintenanceWindows: t.MaintenanceWindows,
DependsOnTaskIDs: deps,
}
}
func namesFromIDs(ids []uint, lookup map[uint]string) []string {
out := make([]string, 0, len(ids))
for _, id := range ids {
if name, ok := lookup[id]; ok {
out = append(out, name)
}
}
return out
}
func idsFromNames(names []string, lookup map[string]uint) []uint {
out := make([]uint, 0, len(names))
for _, name := range names {
if id, ok := lookup[name]; ok {
out = append(out, id)
}
}
return out
}