feat: optional S3-compatible storage for uploaded images, instead of the local upload directory
This commit is contained in:
fsylvain committed
1 parent
d752a61e52
commit
eb9bfabed5
23 files changed
+1962
-7
No files matched your search
@@ -0,0 +1,42 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
domain "RenewCMS/internal/domain/image"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
type ReadCountingStorage struct {
|
||||
domain.Storage
|
||||
opened atomic.Int64
|
||||
}
|
||||
|
||||
var _ domain.Storage = &ReadCountingStorage{}
|
||||
|
||||
func NewReadCountingStorage(inner domain.Storage) *ReadCountingStorage {
|
||||
return &ReadCountingStorage{Storage: inner}
|
||||
}
|
||||
|
||||
func (s *ReadCountingStorage) Open(storedName string) (domain.Blob, error) {
|
||||
blob, err := s.Storage.Open(storedName)
|
||||
if err == nil {
|
||||
s.opened.Add(1)
|
||||
}
|
||||
return blob, err
|
||||
}
|
||||
|
||||
func (s *ReadCountingStorage) Opened() int {
|
||||
return int(s.opened.Load())
|
||||
}
|
||||
|
||||
type Logger interface {
|
||||
Printf(format string, args ...any)
|
||||
}
|
||||
|
||||
func RunBackfillReportingReads(inner domain.Storage, logger Logger, run func(domain.Storage) error) error {
|
||||
counting := NewReadCountingStorage(inner)
|
||||
err := run(counting)
|
||||
if opened := counting.Opened(); opened > 0 {
|
||||
logger.Printf("image metadata backfill read %d images from the bucket", opened)
|
||||
}
|
||||
return err
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
)
|
||||
|
||||
type rangeOpener func(offset int64) (io.ReadCloser, context.CancelFunc, error)
|
||||
|
||||
type s3Reader struct {
|
||||
size int64
|
||||
position int64
|
||||
bodyOffset int64
|
||||
body io.ReadCloser
|
||||
cancel context.CancelFunc
|
||||
reopen rangeOpener
|
||||
}
|
||||
|
||||
func newS3Reader(size int64, body io.ReadCloser, cancel context.CancelFunc, reopen rangeOpener) *s3Reader {
|
||||
return &s3Reader{size: size, body: body, cancel: cancel, reopen: reopen}
|
||||
}
|
||||
|
||||
func (r *s3Reader) Read(p []byte) (int, error) {
|
||||
if r.position >= r.size {
|
||||
return 0, io.EOF
|
||||
}
|
||||
if r.body == nil || r.bodyOffset != r.position {
|
||||
if err := r.openAtPosition(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
}
|
||||
n, err := r.body.Read(p)
|
||||
r.position += int64(n)
|
||||
r.bodyOffset = r.position
|
||||
return n, err
|
||||
}
|
||||
|
||||
func (r *s3Reader) openAtPosition() error {
|
||||
r.release()
|
||||
body, cancel, err := r.reopen(r.position)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
r.body, r.cancel, r.bodyOffset = body, cancel, r.position
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *s3Reader) Seek(offset int64, whence int) (int64, error) {
|
||||
var target int64
|
||||
switch whence {
|
||||
case io.SeekStart:
|
||||
target = offset
|
||||
case io.SeekCurrent:
|
||||
target = r.position + offset
|
||||
case io.SeekEnd:
|
||||
target = r.size + offset
|
||||
default:
|
||||
return 0, errors.New("invalid seek whence")
|
||||
}
|
||||
if target < 0 {
|
||||
return 0, fmt.Errorf("negative seek position %d", target)
|
||||
}
|
||||
r.position = target
|
||||
return target, nil
|
||||
}
|
||||
|
||||
func (r *s3Reader) Close() error {
|
||||
r.release()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *s3Reader) release() {
|
||||
if r.body != nil {
|
||||
_ = r.body.Close()
|
||||
r.body = nil
|
||||
}
|
||||
if r.cancel != nil {
|
||||
r.cancel()
|
||||
r.cancel = nil
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,211 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
domain "RenewCMS/internal/domain/image"
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
|
||||
awsconfig "github.com/aws/aws-sdk-go-v2/config"
|
||||
"github.com/aws/aws-sdk-go-v2/credentials"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
const s3CallTimeout = 30 * time.Second
|
||||
|
||||
type S3Config struct {
|
||||
Bucket string
|
||||
Region string
|
||||
Endpoint string
|
||||
AccessKeyID string
|
||||
SecretAccessKey string
|
||||
KeyPrefix string
|
||||
PathStyle bool
|
||||
TLS bool
|
||||
MaxAttempts int
|
||||
}
|
||||
|
||||
type S3Storage struct {
|
||||
client *s3.Client
|
||||
bucket string
|
||||
keyPrefix string
|
||||
}
|
||||
|
||||
var _ domain.Storage = &S3Storage{}
|
||||
|
||||
func OpenS3Storage(config S3Config) (*S3Storage, error) {
|
||||
client, err := newS3Client(config)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("unable to configure the S3 storage for bucket %q: %s", config.Bucket, redact(err, config))
|
||||
}
|
||||
store := &S3Storage{client: client, bucket: config.Bucket, keyPrefix: config.KeyPrefix}
|
||||
if err := store.checkBucket(config); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return store, nil
|
||||
}
|
||||
|
||||
func newS3Client(config S3Config) (*s3.Client, error) {
|
||||
options := []func(*awsconfig.LoadOptions) error{awsconfig.WithRegion(config.Region)}
|
||||
if config.AccessKeyID != "" {
|
||||
options = append(options, awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(config.AccessKeyID, config.SecretAccessKey, "")))
|
||||
}
|
||||
loaded, err := awsconfig.LoadDefaultConfig(context.Background(), options...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
loaded.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
|
||||
loaded.ResponseChecksumValidation = aws.ResponseChecksumValidationWhenRequired
|
||||
if config.MaxAttempts > 0 {
|
||||
loaded.RetryMaxAttempts = config.MaxAttempts
|
||||
}
|
||||
endpoint := s3Endpoint(config)
|
||||
return s3.NewFromConfig(loaded, func(o *s3.Options) {
|
||||
o.UsePathStyle = config.PathStyle
|
||||
if endpoint != "" {
|
||||
o.BaseEndpoint = aws.String(endpoint)
|
||||
}
|
||||
}), nil
|
||||
}
|
||||
|
||||
func s3Endpoint(config S3Config) string {
|
||||
if config.Endpoint == "" || strings.Contains(config.Endpoint, "://") {
|
||||
return config.Endpoint
|
||||
}
|
||||
if config.TLS {
|
||||
return "https://" + config.Endpoint
|
||||
}
|
||||
return "http://" + config.Endpoint
|
||||
}
|
||||
|
||||
func (s *S3Storage) checkBucket(config S3Config) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), s3CallTimeout)
|
||||
defer cancel()
|
||||
_, err := s.client.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: aws.String(s.bucket)})
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("S3 bucket %q is not usable: %s", s.bucket, bucketFailureReason(err, config))
|
||||
}
|
||||
|
||||
func bucketFailureReason(err error, config S3Config) string {
|
||||
switch statusOf(err) {
|
||||
case http.StatusNotFound:
|
||||
return "the bucket does not exist at the configured endpoint and region"
|
||||
case http.StatusForbidden:
|
||||
return "access denied, check the credentials and the bucket policy"
|
||||
case 0:
|
||||
return redact(err, config)
|
||||
default:
|
||||
return fmt.Sprintf("unexpected response from the storage service (HTTP %d)", statusOf(err))
|
||||
}
|
||||
}
|
||||
|
||||
func redact(err error, config S3Config) string {
|
||||
text := err.Error()
|
||||
for _, secret := range []string{config.SecretAccessKey, config.AccessKeyID} {
|
||||
if secret != "" {
|
||||
text = strings.ReplaceAll(text, secret, "***")
|
||||
}
|
||||
}
|
||||
return text
|
||||
}
|
||||
|
||||
func statusOf(err error) int {
|
||||
var responseErr *awshttp.ResponseError
|
||||
if errors.As(err, &responseErr) {
|
||||
return responseErr.HTTPStatusCode()
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func isNotFound(err error) bool {
|
||||
return statusOf(err) == http.StatusNotFound
|
||||
}
|
||||
|
||||
func (s *S3Storage) key(storedName string) string {
|
||||
return s.keyPrefix + storedName
|
||||
}
|
||||
|
||||
func (s *S3Storage) Put(contentType string, content io.Reader) (string, error) {
|
||||
data, err := io.ReadAll(content)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
storedName := uuid.NewString() + contentTypeExtensions[contentType]
|
||||
ctx, cancel := context.WithTimeout(context.Background(), s3CallTimeout)
|
||||
defer cancel()
|
||||
_, err = s.client.PutObject(ctx, &s3.PutObjectInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
Key: aws.String(s.key(storedName)),
|
||||
Body: bytes.NewReader(data),
|
||||
ContentType: aws.String(contentType),
|
||||
})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("unable to store the image in the S3 bucket: %w", err)
|
||||
}
|
||||
return storedName, nil
|
||||
}
|
||||
|
||||
func (s *S3Storage) Open(storedName string) (domain.Blob, error) {
|
||||
if !isPlainName(storedName) {
|
||||
return domain.Blob{}, domain.ErrBlobNotFound
|
||||
}
|
||||
key := s.key(storedName)
|
||||
output, cancel, err := s.getObject(key, "")
|
||||
if err != nil {
|
||||
if isNotFound(err) {
|
||||
return domain.Blob{}, domain.ErrBlobNotFound
|
||||
}
|
||||
return domain.Blob{}, fmt.Errorf("unable to read the image from the S3 bucket: %w", err)
|
||||
}
|
||||
size := aws.ToInt64(output.ContentLength)
|
||||
reader := newS3Reader(size, output.Body, cancel, func(offset int64) (io.ReadCloser, context.CancelFunc, error) {
|
||||
ranged, rangedCancel, err := s.getObject(key, fmt.Sprintf("bytes=%d-", offset))
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("unable to read the image from the S3 bucket: %w", err)
|
||||
}
|
||||
return ranged.Body, rangedCancel, nil
|
||||
})
|
||||
return domain.Blob{Content: reader, Size: size, ModTime: aws.ToTime(output.LastModified)}, nil
|
||||
}
|
||||
|
||||
func (s *S3Storage) getObject(key string, byteRange string) (*s3.GetObjectOutput, context.CancelFunc, error) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
deadline := time.AfterFunc(s3CallTimeout, cancel)
|
||||
input := &s3.GetObjectInput{Bucket: aws.String(s.bucket), Key: aws.String(key)}
|
||||
if byteRange != "" {
|
||||
input.Range = aws.String(byteRange)
|
||||
}
|
||||
output, err := s.client.GetObject(ctx, input)
|
||||
deadline.Stop()
|
||||
if err != nil {
|
||||
cancel()
|
||||
return nil, nil, err
|
||||
}
|
||||
return output, cancel, nil
|
||||
}
|
||||
|
||||
func (s *S3Storage) Delete(storedName string) error {
|
||||
if !isPlainName(storedName) {
|
||||
return nil
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), s3CallTimeout)
|
||||
defer cancel()
|
||||
_, err := s.client.DeleteObject(ctx, &s3.DeleteObjectInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
Key: aws.String(s.key(storedName)),
|
||||
})
|
||||
if err != nil && !isNotFound(err) {
|
||||
return fmt.Errorf("unable to delete the image from the S3 bucket: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -5,17 +5,23 @@ import (
|
||||
"fmt"
|
||||
)
|
||||
|
||||
const DriverLocal = "local"
|
||||
const (
|
||||
DriverLocal = "local"
|
||||
DriverS3 = "s3"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
Driver string
|
||||
Location string
|
||||
S3 S3Config
|
||||
}
|
||||
|
||||
func Open(config Config) (domain.Storage, error) {
|
||||
switch config.Driver {
|
||||
case DriverLocal:
|
||||
return NewLocalStorage(config.Location), nil
|
||||
case DriverS3:
|
||||
return OpenS3Storage(config.S3)
|
||||
default:
|
||||
return nil, fmt.Errorf("unsupported storage driver %q", config.Driver)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user