1package backup
2
3import (
4 "context"
5 "crypto/rand"
6 "encoding/hex"
7 "fmt"
8 "io"
9 "sort"
10 "sync"
11 "time"
12
13 "github.com/bnema/zerowrap"
14
15 "github.com/bnema/gordon/internal/boundaries/out"
16 "github.com/bnema/gordon/internal/domain"
17)
18
19// VolumeService orchestrates declarative app volume archive backups.
20// Targets come from the app's ACTIVE record: a service declares its
21// volumes and which of them are backed up, and the app name is the
22// backup identity. No container label or attachment is ever consulted.
23type VolumeService struct {
24 exporter out.VolumeArchiveExporter
25 storage out.VolumeBackupStorage
26 config domain.VolumeBackupConfig
27 log zerowrap.Logger
28 state out.AppStateReader
29 metrics out.BackupMetrics
30
31 mu sync.Mutex
32 recent map[string]domain.VolumeBackupJob
33}
34
35// NewVolumeService creates a volume backup service.
36func NewVolumeService(exporter out.VolumeArchiveExporter, storage out.VolumeBackupStorage, config domain.VolumeBackupConfig, log zerowrap.Logger) *VolumeService {
37 return &VolumeService{
38 exporter: exporter,
39 storage: storage,
40 config: config,
41 log: log,
42 recent: make(map[string]domain.VolumeBackupJob),
43 }
44}
45
46// WithAppState wires the ACTIVE app state target resolution reads.
47func (s *VolumeService) WithAppState(state out.AppStateReader) *VolumeService {
48 s.state = state
49 return s
50}
51
52// WithMetrics wires the backup metrics recorder. Without it no metric is
53// recorded.
54func (s *VolumeService) WithMetrics(metrics out.BackupMetrics) *VolumeService {
55 s.metrics = metrics
56 return s
57}
58
59// SeedMetrics reconciles backup metrics with the declared volume targets:
60// it forgets targets that are no longer declared and reports the newest
61// stored archive of each declared one as its last success, so the metric
62// survives a restart. It counts no run.
63func (s *VolumeService) SeedMetrics(ctx context.Context) error {
64 if s.metrics == nil {
65 return nil
66 }
67 asOf := time.Now()
68 targets, err := s.allVolumeTargets(ctx)
69 if err != nil {
70 return fmt.Errorf("seed volume backup metrics: %w", err)
71 }
72 refs := make([]domain.BackupTargetRef, 0, len(targets))
73 byVolume := make(map[string]domain.BackupTargetRef, len(targets))
74 for _, target := range targets {
75 ref := volumeTargetRef(target)
76 refs = append(refs, ref)
77 byVolume[target.App+"/"+target.VolumeName] = ref
78 }
79 s.metrics.RetainBackupTargets(domain.BackupKindVolume, refs, asOf)
80
81 archives, err := s.storage.LatestVolumeArchives(ctx)
82 if err != nil {
83 return fmt.Errorf("seed volume backup metrics: %w", err)
84 }
85 for _, archive := range archives {
86 ref, declared := byVolume[archive.App+"/"+archive.VolumeName]
87 if !declared {
88 continue
89 }
90 s.metrics.ObserveStoredBackup(domain.BackupRunRecord{
91 BackupTargetRef: ref, Succeeded: true, SizeBytes: archive.SizeBytes, StartedAt: archive.StartedAt,
92 })
93 }
94 return nil
95}
96
97func volumeTargetRef(target domain.VolumeBackupTarget) domain.BackupTargetRef {
98 return domain.BackupTargetRef{Kind: domain.BackupKindVolume, App: target.App, Service: target.Service, Target: target.VolumeName}
99}
100
101// VolumeTargets returns every declared volume target of one app.
102func (s *VolumeService) VolumeTargets(ctx context.Context, app string) ([]domain.VolumeBackupTarget, error) {
103 if s.state == nil {
104 return nil, fmt.Errorf("backup: app state is not wired")
105 }
106 _, targets, err := declaredTargets(ctx, s.state, app)
107 if err != nil {
108 return nil, err
109 }
110 return targets, nil
111}
112
113// allVolumeTargets returns the declared volume targets of every app.
114func (s *VolumeService) allVolumeTargets(ctx context.Context) ([]domain.VolumeBackupTarget, error) {
115 if s.state == nil {
116 return nil, fmt.Errorf("backup: app state is not wired")
117 }
118 apps, err := appNames(ctx, s.state)
119 if err != nil {
120 return nil, err
121 }
122 var targets []domain.VolumeBackupTarget
123 for _, app := range apps {
124 appTargets, err := s.VolumeTargets(ctx, app)
125 if err != nil {
126 return nil, err
127 }
128 targets = append(targets, appTargets...)
129 }
130 return targets, nil
131}
132
133// ListVolumeBackups lists completed volume backups of one app, or every
134// app when app is empty.
135func (s *VolumeService) ListVolumeBackups(ctx context.Context, app string) ([]domain.VolumeBackupJob, error) {
136 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
137 zerowrap.FieldLayer: "usecase",
138 zerowrap.FieldUseCase: "ListVolumeBackups",
139 "app": app,
140 })
141 jobs, err := s.storage.ListVolumeArchives(ctx, app)
142 if err != nil {
143 return nil, fmt.Errorf("list volume archives: %w", err)
144 }
145 return jobs, nil
146}
147
148// VolumeBackupStatus returns completed backup artifacts plus current or
149// recent in-memory jobs.
150func (s *VolumeService) VolumeBackupStatus(ctx context.Context) ([]domain.VolumeBackupJob, error) {
151 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
152 zerowrap.FieldLayer: "usecase",
153 zerowrap.FieldUseCase: "VolumeBackupStatus",
154 })
155 jobs, err := s.storage.ListVolumeArchives(ctx, "")
156 if err != nil {
157 return nil, fmt.Errorf("list volume backup status: %w", err)
158 }
159
160 s.mu.Lock()
161 for _, job := range s.recent {
162 if job.Status == domain.BackupStatusRunning || job.Status == domain.BackupStatusFailed {
163 jobs = append(jobs, job)
164 }
165 }
166 s.mu.Unlock()
167
168 sortVolumeJobs(jobs)
169 return jobs, nil
170}
171
172// RunVolumeBackups runs the declared volume backups of one app. service
173// and volume are explicit selectors: an omitted selector succeeds only
174// when exactly one compatible target exists.
175func (s *VolumeService) RunVolumeBackups(ctx context.Context, app, service, volume string) ([]domain.VolumeBackupJob, error) {
176 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
177 zerowrap.FieldLayer: "usecase",
178 zerowrap.FieldUseCase: "RunVolumeBackups",
179 "app": app,
180 "volume": volume,
181 })
182 if !s.config.Enabled {
183 return []domain.VolumeBackupJob{}, nil
184 }
185 targets, err := s.VolumeTargets(ctx, app)
186 if err != nil {
187 return nil, err
188 }
189 target, err := selectTarget(targets, "volume", service, volume, volumeTargetKey)
190 if err != nil {
191 return nil, err
192 }
193 job := s.runVolumeBackup(ctx, target)
194 if job.Status == domain.BackupStatusCompleted {
195 if _, err := s.storage.ApplyVolumeRetention(ctx, job.App, s.config.Retention); err != nil {
196 return []domain.VolumeBackupJob{job}, fmt.Errorf("apply volume backup retention for %s: %w", job.App, err)
197 }
198 }
199 if job.Status == domain.BackupStatusFailed {
200 return []domain.VolumeBackupJob{job}, fmt.Errorf("volume backup failed for %s/%s: %s", job.App, job.VolumeName, job.Error)
201 }
202 return []domain.VolumeBackupJob{job}, nil
203}
204
205// RunVolumeBackupsForSchedule runs every declared volume backup of every
206// app under the installation's schedule, then applies retention. Volume
207// declarations carry no schedule of their own: the installation preset is
208// the schedule, and the label is only used for the caller's own filter.
209func (s *VolumeService) RunVolumeBackupsForSchedule(ctx context.Context, schedule domain.BackupSchedule) error {
210 if !s.config.Enabled {
211 return nil
212 }
213 if schedule != "" && !isValidBackupSchedule(schedule) {
214 return fmt.Errorf("invalid backup schedule: %q", schedule)
215 }
216 targets, err := s.allVolumeTargets(ctx)
217 if err != nil {
218 return err
219 }
220 jobs := s.runVolumeBackupsConcurrently(ctx, targets)
221 var firstErr error
222 apps := map[string]struct{}{}
223 for _, job := range jobs {
224 if job.Status != domain.BackupStatusCompleted {
225 if firstErr == nil {
226 firstErr = fmt.Errorf("volume backup %s/%s: %s", job.App, job.VolumeName, job.Error)
227 }
228 continue
229 }
230 apps[job.App] = struct{}{}
231 }
232 // Retention runs only after every archive finished, so a pass never
233 // deletes an app's older archive while its replacement is uploading.
234 for _, app := range sortedSet(apps) {
235 if _, err := s.storage.ApplyVolumeRetention(ctx, app, s.config.Retention); err != nil {
236 if firstErr == nil {
237 firstErr = fmt.Errorf("apply volume backup retention for %s: %w", app, err)
238 }
239 }
240 }
241 return firstErr
242}
243
244// runVolumeBackupsConcurrently runs every target with at most
245// MaxConcurrency exports in flight and returns the jobs in target order.
246func (s *VolumeService) runVolumeBackupsConcurrently(ctx context.Context, targets []domain.VolumeBackupTarget) []domain.VolumeBackupJob {
247 limit := s.config.MaxConcurrency
248 if limit < 1 {
249 limit = 1
250 }
251 jobs := make([]domain.VolumeBackupJob, len(targets))
252 sem := make(chan struct{}, limit)
253 var wg sync.WaitGroup
254 for i, target := range targets {
255 wg.Add(1)
256 sem <- struct{}{}
257 go func(i int, target domain.VolumeBackupTarget) {
258 defer wg.Done()
259 defer func() { <-sem }()
260 jobs[i] = s.runVolumeBackup(ctx, target)
261 }(i, target)
262 }
263 wg.Wait()
264 return jobs
265}
266
267// runVolumeBackup archives one target and records the outcome in metrics.
268func (s *VolumeService) runVolumeBackup(ctx context.Context, target domain.VolumeBackupTarget) domain.VolumeBackupJob {
269 job := s.archiveVolume(ctx, target)
270 if s.metrics != nil {
271 s.metrics.RecordBackupRun(ctx, domain.BackupRunRecord{
272 BackupTargetRef: volumeTargetRef(target),
273 Succeeded: job.Status == domain.BackupStatusCompleted,
274 Duration: job.CompletedAt.Sub(job.StartedAt),
275 SizeBytes: job.SizeBytes,
276 StartedAt: job.StartedAt,
277 })
278 }
279 return job
280}
281
282func (s *VolumeService) archiveVolume(ctx context.Context, target domain.VolumeBackupTarget) domain.VolumeBackupJob {
283 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
284 "app": target.App,
285 "service": target.Service,
286 "volume": target.VolumeName,
287 "runtime_volume": target.RuntimeVolumeName,
288 "mount_path": target.MountPath,
289 })
290 log := zerowrap.FromCtx(ctx)
291 started := time.Now().UTC()
292 job := domain.VolumeBackupJob{
293 ID: newVolumeBackupJobID(),
294 App: target.App,
295 Service: target.Service,
296 VolumeName: target.VolumeName,
297 RuntimeVolumeName: target.RuntimeVolumeName,
298 MountPath: target.MountPath,
299 Type: domain.BackupTypeVolumeArchive,
300 Status: domain.BackupStatusRunning,
301 StartedAt: started,
302 Metadata: map[string]string{
303 "compression": string(s.config.Compression),
304 },
305 }
306 s.remember(job)
307
308 timeout := s.config.Timeout
309 if timeout <= 0 {
310 timeout = 2 * time.Hour
311 }
312 exportCtx, cancel := context.WithTimeout(ctx, timeout)
313 defer cancel()
314
315 archive, err := s.exporter.ExportVolumeArchive(exportCtx, domain.VolumeArchiveRequest{
316 VolumeName: target.RuntimeVolumeName,
317 MountPath: target.MountPath,
318 Compression: s.config.Compression,
319 HelperImage: s.config.HelperImage,
320 })
321 if err != nil {
322 log.Error().Err(err).Msg("volume archive export failed")
323 return s.failJob(ctx, job, err)
324 }
325 defer archive.Stream.Close()
326
327 counter := &byteCounter{}
328 artifactRef, err := s.storage.StoreVolumeArchive(exportCtx, job, io.TeeReader(archive.Stream, counter))
329 if err != nil {
330 log.Error().Err(err).Msg("volume archive upload failed")
331 return s.failJob(ctx, job, err)
332 }
333
334 job.Status = domain.BackupStatusCompleted
335 job.CompletedAt = time.Now().UTC()
336 job.SizeBytes = counter.n
337 job.ArtifactRef = artifactRef
338 log.Info().
339 Int64("size_bytes", job.SizeBytes).
340 Dur("duration", job.CompletedAt.Sub(job.StartedAt)).
341 Msg("volume backup completed")
342 s.forget(job)
343 return job
344}
345
346// newVolumeBackupJobID returns a short random job ID. Storage keys already
347// carry the start timestamp, so the ID only has to make each run unique.
348func newVolumeBackupJobID() string {
349 random := make([]byte, 4)
350 if _, err := rand.Read(random); err != nil {
351 return fmt.Sprintf("%x", time.Now().UnixNano())
352 }
353 return hex.EncodeToString(random)
354}
355
356// volumeTargetKey reports the (service, volume) identity of one target.
357func volumeTargetKey(target domain.VolumeBackupTarget) (string, string) {
358 return target.Service, target.VolumeName
359}
360
361func (s *VolumeService) failJob(ctx context.Context, job domain.VolumeBackupJob, err error) domain.VolumeBackupJob {
362 job.Status = domain.BackupStatusFailed
363 job.CompletedAt = time.Now().UTC()
364 job.Error = err.Error()
365 log := zerowrap.FromCtx(ctx)
366 log.Error().Err(err).Msg("volume backup failed")
367 s.remember(job)
368 return job
369}
370
371func (s *VolumeService) remember(job domain.VolumeBackupJob) {
372 s.mu.Lock()
373 defer s.mu.Unlock()
374 s.recent[job.App+"/"+job.VolumeName] = job
375 if len(s.recent) > 100 {
376 for k := range s.recent {
377 delete(s.recent, k)
378 break
379 }
380 }
381}
382
383func (s *VolumeService) forget(job domain.VolumeBackupJob) {
384 s.mu.Lock()
385 defer s.mu.Unlock()
386 delete(s.recent, job.App+"/"+job.VolumeName)
387}
388
389func sortVolumeJobs(jobs []domain.VolumeBackupJob) {
390 sort.Slice(jobs, func(i, j int) bool {
391 if !jobs[i].StartedAt.Equal(jobs[j].StartedAt) {
392 return jobs[i].StartedAt.After(jobs[j].StartedAt)
393 }
394 return jobs[i].VolumeName < jobs[j].VolumeName
395 })
396}