package agent import ( "context" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "io" "log" "os" "path/filepath" "strings" "time" "gitea.hsrv.site/mrleo1nid/backup-master/internal/api" "gitea.hsrv.site/mrleo1nid/backup-master/internal/archive" "gitea.hsrv.site/mrleo1nid/backup-master/internal/catalog" "gitea.hsrv.site/mrleo1nid/backup-master/internal/manifest" ) type Options struct { ConfigPath string AgentVersion string // Output — записать части локально в каталог вместо отправки (для отладки). Output string // ForceHeavy — включить heavy-части независимо от расписания. ForceHeavy bool } // prepare загружает конфиг, получает переопределения с сервера и собирает план. func prepare(ctx context.Context, opts Options, spool string) (*Plan, *api.Client, error) { cfg, err := LoadConfig(opts.ConfigPath) if err != nil { return nil, nil, err } client := api.NewClient(cfg.Server, cfg.Token) remote, err := client.AgentConfig(ctx) if err != nil { return nil, nil, fmt.Errorf("конфиг с сервера: %w", err) } cat, err := catalog.Default() if err != nil { return nil, nil, err } if strings.TrimSpace(remote.ConfigYAML) != "" { over, err := catalog.Parse([]byte(remote.ConfigYAML)) if err != nil { return nil, nil, fmt.Errorf("конфиг клиента с сервера: %w", err) } cat = catalog.Merge(cat, over) } if err := cat.Validate(); err != nil { return nil, nil, err } if spool == "" { spool = cfg.SpoolDir } plan, err := Collect(ctx, cat, spool, remote.HeavyDue || opts.ForceHeavy) if err != nil { return nil, nil, err } plan.Manifest.Domain = remote.Domain plan.Manifest.AgentVersion = opts.AgentVersion return plan, client, nil } func newSpool(base string) (string, error) { dir := filepath.Join(base, time.Now().Format("20060102-150405")) return dir, os.MkdirAll(dir, 0o700) } // PrintPlan выводит, что попадёт в бэкап. Дампы БД при этом реально выполняются (во временный каталог). func PrintPlan(ctx context.Context, opts Options, w io.Writer) error { cfg, err := LoadConfig(opts.ConfigPath) if err != nil { return err } spool, err := newSpool(cfg.SpoolDir) if err != nil { return err } defer os.RemoveAll(spool) plan, _, err := prepare(ctx, opts, spool) if err != nil { return err } m := &plan.Manifest fmt.Fprintf(w, "Домен: %s, hostname: %s, %s (%s)\n", m.Domain, m.Hostname, m.OS.PrettyName, m.OS.Arch) if len(m.Packages) > 0 { fmt.Fprintf(w, "Пакеты:") for k, v := range m.Packages { fmt.Fprintf(w, " %s=%s", k, v) } fmt.Fprintln(w) } for _, c := range m.Categories { line := fmt.Sprintf("\n[%s] %s", c.Name, c.Type) if c.Auto { line += " (найден автоматически)" } if c.Version != "" { line += ", версия " + c.Version } if c.Service != nil { line += fmt.Sprintf(", сервис %s enabled=%v active=%v", c.Service.Unit, c.Service.Enabled, c.Service.Active) } fmt.Fprintln(w, line) if c.Compose != nil { fmt.Fprintf(w, " проект %s, запущен=%v\n", c.Compose.Project, c.Compose.Running) for svc, img := range c.Compose.Images { ref := img.Digest if img.Built { ref = "собирается из исходников" } else if ref == "" { ref = img.Ref + " (digest не определён)" } fmt.Fprintf(w, " образ %s: %s\n", svc, ref) } } for _, d := range c.Dumps { fmt.Fprintf(w, " дамп %s %s\n", d.Kind, d.Name) } for _, v := range c.Volumes { fmt.Fprintf(w, " том %s\n", v.Name) } for _, h := range c.Heavy { note := "в этот раз" if c.HeavySkipped { note = "по расписанию, не в этот раз" } fmt.Fprintf(w, " heavy %s (%s)\n", h, note) } } fmt.Fprintln(w, "\nСодержимое частей:") for _, part := range m.Parts { var total int64 var files int for _, it := range plan.Parts[part] { size, n := treeSize(it.Src, it.Skip) total += size files += n } fmt.Fprintf(w, " %s: %d файлов, %s до сжатия\n", part, files, human(total)) } if len(m.Warnings) > 0 { fmt.Fprintln(w, "\nПредупреждения:") for _, wr := range m.Warnings { fmt.Fprintln(w, " ! "+wr) } } return nil } // Backup выполняет полный цикл: сбор, отправка частей, подтверждение. func Backup(ctx context.Context, opts Options) (err error) { unlock, err := lock("/run/backup-agent.lock") if err != nil { return err } defer unlock() cfg, err := LoadConfig(opts.ConfigPath) if err != nil { return err } spool, err := newSpool(cfg.SpoolDir) if err != nil { return err } defer os.RemoveAll(spool) start := time.Now() plan, client, err := prepare(ctx, opts, spool) if err != nil { return err } m := &plan.Manifest log.Printf("собрано: %d категорий, части %v, предупреждений %d", len(m.Categories), m.Parts, len(m.Warnings)) for _, w := range m.Warnings { log.Printf("! %s", w) } if len(m.Categories) == 0 { // пустой «успешный» бэкап хуже ошибки: сервер решит, что всё в порядке return errors.New("ни одна категория не найдена на сервере — проверьте каталог (backupctl plan)") } if opts.Output != "" { return writeLocal(plan, opts.Output) } id, err := client.CreateBackup(ctx, m) if err != nil { return fmt.Errorf("создание бэкапа на сервере: %w", err) } defer func() { if err != nil { fctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() if ferr := client.Fail(fctx, id, err.Error()); ferr != nil { log.Printf("не удалось сообщить серверу об ошибке: %v", ferr) } } }() var done api.CompleteRequest for _, part := range m.Parts { info, warns, err := uploadWithRetry(ctx, client, id, plan, part) if err != nil { return fmt.Errorf("часть %s: %w", part, err) } done.Parts = append(done.Parts, *info) done.Warnings = append(done.Warnings, warns...) log.Printf("часть %s отправлена: %s", part, human(info.Size)) } for _, w := range done.Warnings { log.Printf("! %s", w) } if err := client.Complete(ctx, id, done); err != nil { return fmt.Errorf("подтверждение: %w", err) } log.Printf("бэкап #%d готов за %s", id, time.Since(start).Round(time.Second)) return nil } func uploadWithRetry(ctx context.Context, client *api.Client, id int64, plan *Plan, part string) (*api.PartInfo, []string, error) { var lastErr error for attempt := 1; attempt <= 3; attempt++ { if attempt > 1 { wait := time.Duration(attempt*attempt) * 20 * time.Second log.Printf("часть %s: попытка %d через %s (%v)", part, attempt, wait, lastErr) select { case <-ctx.Done(): return nil, nil, ctx.Err() case <-time.After(wait): } } info, warns, err := uploadPart(ctx, client, id, plan, part) if err == nil { return info, warns, nil } lastErr = err var apiErr *api.Error if errors.As(err, &apiErr) && apiErr.Status >= 400 && apiErr.Status < 500 && apiErr.Status != 408 && apiErr.Status != 429 { break // повтор не поможет } } return nil, nil, lastErr } // uploadPart архивирует часть прямо в тело PUT-запроса, параллельно считая sha256. func uploadPart(ctx context.Context, client *api.Client, id int64, plan *Plan, part string) (*api.PartInfo, []string, error) { pr, pw := io.Pipe() h := sha256.New() cw := &countWriter{w: io.MultiWriter(pw, h)} type result struct { warns []string err error } done := make(chan result, 1) go func() { warns, err := writePart(cw, plan, part) pw.CloseWithError(err) done <- result{warns, err} }() info, err := client.UploadPart(ctx, id, part, pr) pr.CloseWithError(errors.New("загрузка прервана")) // освобождаем писателя, если сервер ответил раньше res := <-done if err != nil { if res.err != nil && !errors.Is(res.err, io.ErrClosedPipe) && !strings.Contains(res.err.Error(), "загрузка прервана") { return nil, nil, fmt.Errorf("архивация: %w", res.err) } return nil, nil, err } if res.err != nil { return nil, nil, fmt.Errorf("архивация: %w", res.err) } warns := res.warns sum := hex.EncodeToString(h.Sum(nil)) if info.SHA256 != sum || info.Size != cw.n { return nil, nil, fmt.Errorf("сервер получил %d байт (%s), отправлено %d (%s)", info.Size, info.SHA256, cw.n, sum) } return info, warns, nil } func writePart(w io.Writer, plan *Plan, part string) ([]string, error) { aw, err := archive.NewWriter(w) if err != nil { return nil, err } if part == manifest.MainPart { data, err := json.MarshalIndent(plan.Manifest, "", " ") if err != nil { return nil, err } if err := aw.AddBytes(manifest.ManifestFile, data, 0o600); err != nil { return nil, err } } for _, it := range plan.Parts[part] { if err := aw.AddTree(it.Src, it.Name, it.Skip); err != nil { return aw.Warnings, err } } return aw.Warnings, aw.Close() } func writeLocal(plan *Plan, dir string) error { if err := os.MkdirAll(dir, 0o700); err != nil { return err } for _, part := range plan.Manifest.Parts { p := filepath.Join(dir, part+".tar.zst") f, err := os.OpenFile(p, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600) if err != nil { return err } warns, err := writePart(f, plan, part) if cerr := f.Close(); err == nil { err = cerr } if err != nil { return err } for _, w := range warns { log.Printf("! %s", w) } log.Printf("записано %s", p) } return nil } type countWriter struct { w io.Writer n int64 } func (c *countWriter) Write(p []byte) (int, error) { n, err := c.w.Write(p) c.n += int64(n) return n, err } func human(n int64) string { const unit = 1024 if n < unit { return fmt.Sprintf("%d Б", n) } div, exp := int64(unit), 0 for m := n / unit; m >= unit; m /= unit { div *= unit exp++ } return fmt.Sprintf("%.1f %cБ", float64(n)/float64(div), []rune("КМГТП")[exp]) }