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 listPrefix := s.appPrefix(app)
154 jobs := make([]domain.VolumeBackupJob, 0)
155 var token *string
156 for {
157 out, err := s.client.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{
158 Bucket: aws.String(s.bucket),
159 Prefix: aws.String(listPrefix),
160 ContinuationToken: token,
161 })
162 if err != nil {
163 return nil, fmt.Errorf("failed to list volume archives: %w", err)
164 }
165 for _, obj := range out.Contents {
166 if obj.Key == nil {
167 continue
168 }
169 job, ok := s.jobFromKey(app, *obj.Key)
170 if !ok {
171 continue
172 }
173 if obj.Size != nil {
174 job.SizeBytes = *obj.Size
175 }
176 s.hydrateVolumeBackupJobMetadata(ctx, &job, *obj.Key)
177 jobs = append(jobs, job)
178 }
179 if out.IsTruncated == nil || !*out.IsTruncated {
180 break
181 }
182 token = out.NextContinuationToken
183 }
184
185 sortVolumeBackupJobs(jobs)
186 return jobs, nil
187}
188
189// DeleteVolumeArchive deletes a volume archive by artifact reference.
190func (s *VolumeBackupStorage) hydrateVolumeBackupJobMetadata(ctx context.Context, job *domain.VolumeBackupJob, key string) {
191 out, err := s.client.HeadObject(ctx, &awss3.HeadObjectInput{Bucket: aws.String(s.bucket), Key: aws.String(key)})
192 if err != nil || out == nil || out.Metadata == nil {
193 return
194 }
195 job.Service = out.Metadata["gordon-service"]
196 job.ContainerName = out.Metadata["gordon-container"]
197 job.MountPath = out.Metadata["gordon-mount-path"]
198 if compression := out.Metadata["gordon-compression"]; compression != "" {
199 if job.Metadata == nil {
200 job.Metadata = make(map[string]string)
201 }
202 job.Metadata["compression"] = compression
203 }
204}
205
206func (s *VolumeBackupStorage) DeleteVolumeArchive(ctx context.Context, artifactRef string) error {
207 key, err := s.keyFromArtifactRef(artifactRef)
208 if err != nil {
209 return err
210 }
211 _, err = s.client.DeleteObject(ctx, &awss3.DeleteObjectInput{
212 Bucket: aws.String(s.bucket),
213 Key: aws.String(key),
214 })
215 if err != nil {
216 return fmt.Errorf("failed to delete volume archive: %w", err)
217 }
218 return nil
219}
220
221// ApplyVolumeRetention deletes old completed volume archives according to the keep count.
222func (s *VolumeBackupStorage) ApplyVolumeRetention(ctx context.Context, app string, policy domain.VolumeBackupRetentionPolicy) (int, error) {
223 if policy.Keep < 0 {
224 return 0, fmt.Errorf("volume backup retention keep cannot be negative")
225 }
226 if policy.Keep == 0 {
227 return 0, nil
228 }
229
230 jobs, err := s.ListVolumeArchives(ctx, app)
231 if err != nil {
232 return 0, err
233 }
234
235 byVolume := make(map[string][]domain.VolumeBackupJob)
236 for _, job := range jobs {
237 byVolume[job.VolumeName] = append(byVolume[job.VolumeName], job)
238 }
239
240 deleted := 0
241 for _, group := range byVolume {
242 sortVolumeBackupJobs(group)
243 for i := policy.Keep; i < len(group); i++ {
244 if err := s.DeleteVolumeArchive(ctx, group[i].ArtifactRef); err != nil {
245 return deleted, err
246 }
247 deleted++
248 }
249 }
250 return deleted, nil
251}
252
253func (s *VolumeBackupStorage) objectKey(job domain.VolumeBackupJob) string {
254 started := job.StartedAt.UTC()
255 if started.IsZero() {
256 started = time.Now().UTC()
257 }
258 id := job.ID
259 if id == "" {
260 id = randomObjectSuffix()
261 }
262
263 ext := "tar.zst"
264 if job.Metadata != nil && job.Metadata["compression"] == string(domain.VolumeBackupCompressionGzip) {
265 ext = "tar.gz"
266 }
267 fileName := fmt.Sprintf("%s-%s.%s", started.Format(volumeBackupTimestampLayout), sanitizeS3KeyComponent(id), ext)
268 parts := []string{s.prefix, "apps", sanitizeS3KeyComponent(job.App), "volumes", sanitizeS3KeyComponent(job.VolumeName), fileName}
269 return joinS3Key(parts...)
270}
271
272// appPrefix is the object prefix of one app's volume archives, or of every
273// app when app is empty. Object keys are app-based; artifacts written by
274// older versions under a domain prefix are never read or migrated here.
275func (s *VolumeBackupStorage) appPrefix(app string) string {
276 if strings.TrimSpace(app) == "" {
277 return joinS3Key(s.prefix, "apps") + "/"
278 }
279 return joinS3Key(s.prefix, "apps", sanitizeS3KeyComponent(app), "volumes") + "/"
280}
281
282func (s *VolumeBackupStorage) artifactRef(key string) string {
283 return "s3://" + s.bucket + "/" + key
284}
285
286func (s *VolumeBackupStorage) keyFromArtifactRef(artifactRef string) (string, error) {
287 if strings.HasPrefix(artifactRef, "s3://") {
288 u, err := url.Parse(artifactRef)
289 if err != nil {
290 return "", fmt.Errorf("invalid s3 artifact ref: %w", err)
291 }
292 if u.Host != s.bucket {
293 return "", fmt.Errorf("artifact bucket %q does not match configured bucket", u.Host)
294 }
295 return strings.TrimPrefix(u.Path, "/"), nil
296 }
297 if strings.TrimSpace(artifactRef) == "" {
298 return "", fmt.Errorf("artifact ref is required")
299 }
300 return strings.TrimPrefix(artifactRef, "/"), nil
301}
302
303func (s *VolumeBackupStorage) jobFromKey(app, key string) (domain.VolumeBackupJob, bool) {
304 var rel string
305 if strings.TrimSpace(app) == "" {
306 rel = strings.TrimPrefix(key, joinS3Key(s.prefix, "apps")+"/")
307 parts := strings.Split(rel, "/")
308 if len(parts) != 4 || parts[1] != "volumes" {
309 return domain.VolumeBackupJob{}, false
310 }
311 app = parts[0]
312 rel = strings.Join(parts[2:], "/")
313 } else {
314 rel = strings.TrimPrefix(key, joinS3Key(s.prefix, "apps", sanitizeS3KeyComponent(app), "volumes")+"/")
315 }
316 parts := strings.Split(rel, "/")
317 if len(parts) != 2 {
318 return domain.VolumeBackupJob{}, false
319 }
320 volumeName := parts[0]
321 fileName := parts[1]
322 if volumeName == "" || fileName == "" {
323 return domain.VolumeBackupJob{}, false
324 }
325 started, id, ok := parseVolumeBackupFileName(fileName)
326 if !ok {
327 return domain.VolumeBackupJob{}, false
328 }
329 return domain.VolumeBackupJob{
330 ID: id,
331 App: app,
332 VolumeName: volumeName,
333 Type: domain.BackupTypeVolumeArchive,
334 Status: domain.BackupStatusCompleted,
335 StartedAt: started,
336 CompletedAt: started,
337 ArtifactRef: s.artifactRef(key),
338 }, true
339}
340
341func parseVolumeBackupFileName(fileName string) (time.Time, string, bool) {
342 trimmed := strings.TrimSuffix(strings.TrimSuffix(fileName, ".tar.zst"), ".tar.gz")
343 if trimmed == fileName {
344 return time.Time{}, "", false
345 }
346 timestampPart, id, ok := strings.Cut(trimmed, "-")
347 if !ok {
348 return time.Time{}, "", false
349 }
350 started, err := time.Parse(volumeBackupTimestampLayout, timestampPart)
351 if err != nil {
352 return time.Time{}, "", false
353 }
354 return started.UTC(), id, true
355}
356
357func sortVolumeBackupJobs(jobs []domain.VolumeBackupJob) {
358 sort.Slice(jobs, func(i, j int) bool {
359 if !jobs[i].StartedAt.Equal(jobs[j].StartedAt) {
360 return jobs[i].StartedAt.After(jobs[j].StartedAt)
361 }
362 return jobs[i].ArtifactRef < jobs[j].ArtifactRef
363 })
364}
365
366func normalizeS3Prefix(prefix string) string {
367 return strings.Trim(strings.TrimSpace(prefix), "/")
368}
369
370func joinS3Key(parts ...string) string {
371 clean := make([]string, 0, len(parts))
372 for _, part := range parts {
373 part = strings.Trim(part, "/")
374 if part != "" {
375 clean = append(clean, part)
376 }
377 }
378 return path.Join(clean...)
379}
380
381func sanitizeS3KeyComponent(input string) string {
382 clean := strings.TrimSpace(input)
383 if clean == "" || strings.Trim(clean, ".") == "" {
384 return "unknown"
385 }
386 clean = strings.Trim(clean, ".")
387 replacer := strings.NewReplacer("/", "_", "\\", "_", ":", "_", " ", "_")
388 return replacer.Replace(clean)
389}
390
391func contentTypeForCompression(compression string) string {
392 switch compression {
393 case string(domain.VolumeBackupCompressionGzip):
394 return "application/gzip"
395 case string(domain.VolumeBackupCompressionZstd):
396 return "application/zstd"
397 default:
398 return "application/octet-stream"
399 }
400}
401
402func randomObjectSuffix() string {
403 var b [8]byte
404 if _, err := rand.Read(b[:]); err != nil {
405 return fmt.Sprintf("%d", time.Now().UTC().UnixNano())
406 }
407 return hex.EncodeToString(b[:])
408}