1package s3
2
3import (
4 "context"
5 "crypto/rand"
6 "encoding/hex"
7 "fmt"
8 "io"
9 "net/url"
10 "path"
11 "sort"
12 "strings"
13 "time"
14
15 "github.com/aws/aws-sdk-go-v2/aws"
16 "github.com/aws/aws-sdk-go-v2/config"
17 "github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager"
18 tmtypes "github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager/types"
19 awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
20
21 "github.com/bnema/gordon/internal/domain"
22)
23
24const volumeBackupTimestampLayout = "20060102T150405Z"
25
26type volumeS3Client interface {
27 GetObject(ctx context.Context, params *awss3.GetObjectInput, optFns ...func(*awss3.Options)) (*awss3.GetObjectOutput, error)
28 HeadObject(ctx context.Context, params *awss3.HeadObjectInput, optFns ...func(*awss3.Options)) (*awss3.HeadObjectOutput, error)
29 ListObjectsV2(ctx context.Context, params *awss3.ListObjectsV2Input, optFns ...func(*awss3.Options)) (*awss3.ListObjectsV2Output, error)
30 DeleteObject(ctx context.Context, params *awss3.DeleteObjectInput, optFns ...func(*awss3.Options)) (*awss3.DeleteObjectOutput, error)
31}
32
33type volumeUploader interface {
34 UploadObject(ctx context.Context, input *transfermanager.UploadObjectInput, opts ...func(*transfermanager.Options)) (*transfermanager.UploadObjectOutput, error)
35}
36
37// VolumeBackupStorage stores volume archive backups in S3.
38type VolumeBackupStorage struct {
39 bucket string
40 prefix string
41 sseAlgorithm string
42 sseKMSKeyID string
43 storageClass string
44 client volumeS3Client
45 uploader volumeUploader
46}
47
48// NewVolumeBackupStorage creates an S3-backed volume backup storage adapter.
49func NewVolumeBackupStorage(ctx context.Context, cfg domain.VolumeBackupConfig) (*VolumeBackupStorage, error) {
50 awsCfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(cfg.S3Region))
51 if err != nil {
52 return nil, fmt.Errorf("failed to load AWS config: %w", err)
53 }
54
55 client := awss3.NewFromConfig(awsCfg, func(o *awss3.Options) {
56 if cfg.S3Endpoint != "" {
57 o.BaseEndpoint = aws.String(cfg.S3Endpoint)
58 }
59 o.UsePathStyle = cfg.S3PathStyle
60 })
61 uploader := transfermanager.New(client, func(o *transfermanager.Options) {
62 o.Concurrency = 2
63 o.PartSizeBytes = 16 * 1024 * 1024
64 o.MultipartUploadThreshold = 16 * 1024 * 1024
65 o.FailTimeout = time.Minute
66 })
67
68 return NewVolumeBackupStorageWithClients(cfg, client, uploader), nil
69}
70
71// NewVolumeBackupStorageWithClients creates storage with injected clients for tests.
72func NewVolumeBackupStorageWithClients(cfg domain.VolumeBackupConfig, client volumeS3Client, uploader volumeUploader) *VolumeBackupStorage {
73 return &VolumeBackupStorage{
74 bucket: strings.TrimSpace(cfg.S3Bucket),
75 prefix: normalizeS3Prefix(cfg.S3Prefix),
76 sseAlgorithm: strings.TrimSpace(cfg.S3SSEAlgorithm),
77 sseKMSKeyID: strings.TrimSpace(cfg.S3SSEKMSKeyID),
78 storageClass: strings.TrimSpace(cfg.S3StorageClass),
79 client: client,
80 uploader: uploader,
81 }
82}
83
84// StoreVolumeArchive uploads a volume archive stream to S3 and returns its artifact reference.
85func (s *VolumeBackupStorage) StoreVolumeArchive(ctx context.Context, job domain.VolumeBackupJob, data io.Reader) (string, error) {
86 if s.bucket == "" {
87 return "", fmt.Errorf("s3 bucket is required")
88 }
89 if job.App == "" {
90 return "", fmt.Errorf("backup app is required")
91 }
92 if job.VolumeName == "" {
93 return "", fmt.Errorf("backup volume name is required")
94 }
95 if data == nil {
96 return "", fmt.Errorf("backup data is required")
97 }
98
99 key := s.objectKey(job)
100 compression := ""
101 if job.Metadata != nil {
102 compression = job.Metadata["compression"]
103 }
104 input := &transfermanager.UploadObjectInput{
105 Bucket: aws.String(s.bucket),
106 Key: aws.String(key),
107 Body: data,
108 ContentType: aws.String(contentTypeForCompression(compression)),
109 Metadata: map[string]string{
110 "gordon-app": job.App,
111 "gordon-service": job.Service,
112 "gordon-volume": job.VolumeName,
113 "gordon-container": job.ContainerName,
114 "gordon-mount-path": job.MountPath,
115 "gordon-compression": compression,
116 },
117 }
118 if s.sseAlgorithm != "" {
119 input.ServerSideEncryption = tmtypes.ServerSideEncryption(s.sseAlgorithm)
120 }
121 if s.sseKMSKeyID != "" {
122 input.SSEKMSKeyID = aws.String(s.sseKMSKeyID)
123 }
124 if s.storageClass != "" {
125 input.StorageClass = tmtypes.StorageClass(s.storageClass)
126 }
127
128 if _, err := s.uploader.UploadObject(ctx, input); err != nil {
129 return "", fmt.Errorf("failed to upload volume archive: %w", err)
130 }
131
132 return s.artifactRef(key), nil
133}
134
135// GetVolumeArchive retrieves a volume archive by artifact reference.
136func (s *VolumeBackupStorage) GetVolumeArchive(ctx context.Context, artifactRef string) (io.ReadCloser, error) {
137 key, err := s.keyFromArtifactRef(artifactRef)
138 if err != nil {
139 return nil, err
140 }
141 out, err := s.client.GetObject(ctx, &awss3.GetObjectInput{
142 Bucket: aws.String(s.bucket),
143 Key: aws.String(key),
144 })
145 if err != nil {
146 return nil, fmt.Errorf("failed to get volume archive: %w", err)
147 }
148 return out.Body, nil
149}
150
151// ListVolumeArchives lists completed volume archive backups for a domain.
152func (s *VolumeBackupStorage) ListVolumeArchives(ctx context.Context, app string) ([]domain.VolumeBackupJob, error) {
153 jobs, err := s.listArchiveObjects(ctx, app)
154 if err != nil {
155 return nil, err
156 }
157 for i := range jobs {
158 s.hydrateVolumeBackupJobMetadata(ctx, &jobs[i], s.keyOf(jobs[i]))
159 }
160 sortVolumeBackupJobs(jobs)
161 return jobs, nil
162}
163
164// LatestVolumeArchives returns the newest archive of every app volume from
165// the object listing alone, without one metadata request per object.
166func (s *VolumeBackupStorage) LatestVolumeArchives(ctx context.Context) ([]domain.VolumeBackupJob, error) {
167 jobs, err := s.listArchiveObjects(ctx, "")
168 if err != nil {
169 return nil, err
170 }
171 latest := make(map[string]domain.VolumeBackupJob)
172 for _, job := range jobs {
173 key := job.App + "/" + job.VolumeName
174 if prev, ok := latest[key]; !ok || job.StartedAt.After(prev.StartedAt) {
175 latest[key] = job
176 }
177 }
178 out := make([]domain.VolumeBackupJob, 0, len(latest))
179 for _, job := range latest {
180 out = append(out, job)
181 }
182 sortVolumeBackupJobs(out)
183 return out, nil
184}
185
186// listArchiveObjects lists the archives of one app, or of every app when
187// app is empty, from object keys and sizes only.
188func (s *VolumeBackupStorage) listArchiveObjects(ctx context.Context, app string) ([]domain.VolumeBackupJob, error) {
189 listPrefix := s.appPrefix(app)
190 jobs := make([]domain.VolumeBackupJob, 0)
191 var token *string
192 for {
193 out, err := s.client.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{
194 Bucket: aws.String(s.bucket),
195 Prefix: aws.String(listPrefix),
196 ContinuationToken: token,
197 })
198 if err != nil {
199 return nil, fmt.Errorf("failed to list volume archives: %w", err)
200 }
201 for _, obj := range out.Contents {
202 if obj.Key == nil {
203 continue
204 }
205 job, ok := s.jobFromKey(app, *obj.Key)
206 if !ok {
207 continue
208 }
209 if obj.Size != nil {
210 job.SizeBytes = *obj.Size
211 }
212 jobs = append(jobs, job)
213 }
214 if out.IsTruncated == nil || !*out.IsTruncated {
215 break
216 }
217 token = out.NextContinuationToken
218 }
219 return jobs, nil
220}
221
222// keyOf returns the object key of a listed job.
223func (s *VolumeBackupStorage) keyOf(job domain.VolumeBackupJob) string {
224 return strings.TrimPrefix(job.ArtifactRef, "s3://"+s.bucket+"/")
225}
226
227// DeleteVolumeArchive deletes a volume archive by artifact reference.
228func (s *VolumeBackupStorage) hydrateVolumeBackupJobMetadata(ctx context.Context, job *domain.VolumeBackupJob, key string) {
229 out, err := s.client.HeadObject(ctx, &awss3.HeadObjectInput{Bucket: aws.String(s.bucket), Key: aws.String(key)})
230 if err != nil || out == nil || out.Metadata == nil {
231 return
232 }
233 job.Service = out.Metadata["gordon-service"]
234 job.ContainerName = out.Metadata["gordon-container"]
235 job.MountPath = out.Metadata["gordon-mount-path"]
236 if compression := out.Metadata["gordon-compression"]; compression != "" {
237 if job.Metadata == nil {
238 job.Metadata = make(map[string]string)
239 }
240 job.Metadata["compression"] = compression
241 }
242}
243
244func (s *VolumeBackupStorage) DeleteVolumeArchive(ctx context.Context, artifactRef string) error {
245 key, err := s.keyFromArtifactRef(artifactRef)
246 if err != nil {
247 return err
248 }
249 _, err = s.client.DeleteObject(ctx, &awss3.DeleteObjectInput{
250 Bucket: aws.String(s.bucket),
251 Key: aws.String(key),
252 })
253 if err != nil {
254 return fmt.Errorf("failed to delete volume archive: %w", err)
255 }
256 return nil
257}
258
259// ApplyVolumeRetention deletes old completed volume archives according to the keep count.
260func (s *VolumeBackupStorage) ApplyVolumeRetention(ctx context.Context, app string, policy domain.VolumeBackupRetentionPolicy) (int, error) {
261 if policy.Keep < 0 {
262 return 0, fmt.Errorf("volume backup retention keep cannot be negative")
263 }
264 if policy.Keep == 0 {
265 return 0, nil
266 }
267
268 jobs, err := s.ListVolumeArchives(ctx, app)
269 if err != nil {
270 return 0, err
271 }
272
273 byVolume := make(map[string][]domain.VolumeBackupJob)
274 for _, job := range jobs {
275 byVolume[job.VolumeName] = append(byVolume[job.VolumeName], job)
276 }
277
278 deleted := 0
279 for _, group := range byVolume {
280 sortVolumeBackupJobs(group)
281 for i := policy.Keep; i < len(group); i++ {
282 if err := s.DeleteVolumeArchive(ctx, group[i].ArtifactRef); err != nil {
283 return deleted, err
284 }
285 deleted++
286 }
287 }
288 return deleted, nil
289}
290
291func (s *VolumeBackupStorage) objectKey(job domain.VolumeBackupJob) string {
292 started := job.StartedAt.UTC()
293 if started.IsZero() {
294 started = time.Now().UTC()
295 }
296 id := job.ID
297 if id == "" {
298 id = randomObjectSuffix()
299 }
300
301 ext := "tar.zst"
302 if job.Metadata != nil && job.Metadata["compression"] == string(domain.VolumeBackupCompressionGzip) {
303 ext = "tar.gz"
304 }
305 fileName := fmt.Sprintf("%s-%s.%s", started.Format(volumeBackupTimestampLayout), sanitizeS3KeyComponent(id), ext)
306 parts := []string{s.prefix, "apps", sanitizeS3KeyComponent(job.App), "volumes", sanitizeS3KeyComponent(job.VolumeName), fileName}
307 return joinS3Key(parts...)
308}
309
310// appPrefix is the object prefix of one app's volume archives, or of every
311// app when app is empty. Object keys are app-based; artifacts written by
312// older versions under a domain prefix are never read or migrated here.
313func (s *VolumeBackupStorage) appPrefix(app string) string {
314 if strings.TrimSpace(app) == "" {
315 return joinS3Key(s.prefix, "apps") + "/"
316 }
317 return joinS3Key(s.prefix, "apps", sanitizeS3KeyComponent(app), "volumes") + "/"
318}
319
320func (s *VolumeBackupStorage) artifactRef(key string) string {
321 return "s3://" + s.bucket + "/" + key
322}
323
324func (s *VolumeBackupStorage) keyFromArtifactRef(artifactRef string) (string, error) {
325 if strings.HasPrefix(artifactRef, "s3://") {
326 u, err := url.Parse(artifactRef)
327 if err != nil {
328 return "", fmt.Errorf("invalid s3 artifact ref: %w", err)
329 }
330 if u.Host != s.bucket {
331 return "", fmt.Errorf("artifact bucket %q does not match configured bucket", u.Host)
332 }
333 return strings.TrimPrefix(u.Path, "/"), nil
334 }
335 if strings.TrimSpace(artifactRef) == "" {
336 return "", fmt.Errorf("artifact ref is required")
337 }
338 return strings.TrimPrefix(artifactRef, "/"), nil
339}
340
341func (s *VolumeBackupStorage) jobFromKey(app, key string) (domain.VolumeBackupJob, bool) {
342 var rel string
343 if strings.TrimSpace(app) == "" {
344 rel = strings.TrimPrefix(key, joinS3Key(s.prefix, "apps")+"/")
345 parts := strings.Split(rel, "/")
346 if len(parts) != 4 || parts[1] != "volumes" {
347 return domain.VolumeBackupJob{}, false
348 }
349 app = parts[0]
350 rel = strings.Join(parts[2:], "/")
351 } else {
352 rel = strings.TrimPrefix(key, joinS3Key(s.prefix, "apps", sanitizeS3KeyComponent(app), "volumes")+"/")
353 }
354 parts := strings.Split(rel, "/")
355 if len(parts) != 2 {
356 return domain.VolumeBackupJob{}, false
357 }
358 volumeName := parts[0]
359 fileName := parts[1]
360 if volumeName == "" || fileName == "" {
361 return domain.VolumeBackupJob{}, false
362 }
363 started, id, ok := parseVolumeBackupFileName(fileName)
364 if !ok {
365 return domain.VolumeBackupJob{}, false
366 }
367 return domain.VolumeBackupJob{
368 ID: id,
369 App: app,
370 VolumeName: volumeName,
371 Type: domain.BackupTypeVolumeArchive,
372 Status: domain.BackupStatusCompleted,
373 StartedAt: started,
374 CompletedAt: started,
375 ArtifactRef: s.artifactRef(key),
376 }, true
377}
378
379func parseVolumeBackupFileName(fileName string) (time.Time, string, bool) {
380 trimmed := strings.TrimSuffix(strings.TrimSuffix(fileName, ".tar.zst"), ".tar.gz")
381 if trimmed == fileName {
382 return time.Time{}, "", false
383 }
384 timestampPart, id, ok := strings.Cut(trimmed, "-")
385 if !ok {
386 return time.Time{}, "", false
387 }
388 started, err := time.Parse(volumeBackupTimestampLayout, timestampPart)
389 if err != nil {
390 return time.Time{}, "", false
391 }
392 return started.UTC(), id, true
393}
394
395func sortVolumeBackupJobs(jobs []domain.VolumeBackupJob) {
396 sort.Slice(jobs, func(i, j int) bool {
397 if !jobs[i].StartedAt.Equal(jobs[j].StartedAt) {
398 return jobs[i].StartedAt.After(jobs[j].StartedAt)
399 }
400 return jobs[i].ArtifactRef < jobs[j].ArtifactRef
401 })
402}
403
404func normalizeS3Prefix(prefix string) string {
405 return strings.Trim(strings.TrimSpace(prefix), "/")
406}
407
408func joinS3Key(parts ...string) string {
409 clean := make([]string, 0, len(parts))
410 for _, part := range parts {
411 part = strings.Trim(part, "/")
412 if part != "" {
413 clean = append(clean, part)
414 }
415 }
416 return path.Join(clean...)
417}
418
419func sanitizeS3KeyComponent(input string) string {
420 clean := strings.TrimSpace(input)
421 if clean == "" || strings.Trim(clean, ".") == "" {
422 return "unknown"
423 }
424 clean = strings.Trim(clean, ".")
425 replacer := strings.NewReplacer("/", "_", "\\", "_", ":", "_", " ", "_")
426 return replacer.Replace(clean)
427}
428
429func contentTypeForCompression(compression string) string {
430 switch compression {
431 case string(domain.VolumeBackupCompressionGzip):
432 return "application/gzip"
433 case string(domain.VolumeBackupCompressionZstd):
434 return "application/zstd"
435 default:
436 return "application/octet-stream"
437 }
438}
439
440func randomObjectSuffix() string {
441 var b [8]byte
442 if _, err := rand.Read(b[:]); err != nil {
443 return fmt.Sprintf("%d", time.Now().UTC().UnixNano())
444 }
445 return hex.EncodeToString(b[:])
446}