9 Commits

2 changed files with 104 additions and 34 deletions

View File

@@ -5,6 +5,7 @@ import (
"fmt"
"io"
"log"
"math/rand"
"os"
"path/filepath"
"time"
@@ -13,9 +14,16 @@ import (
"github.com/minio/minio-go/v7/pkg/credentials"
)
type cleanupFunc func() error
type Storage interface {
ReadFile(ctx context.Context, fileHash string) (io.ReadSeekCloser, error)
SaveFile(ctx context.Context, fileHash string, srcLen int64, src io.Reader) error
// TempLocalFilepath returns a local file path and a cleanup function that
// can be used to access the file through the filesystem.
// It's an alternative to ReadFile.
TempLocalFilepath(ctx context.Context, fileHash string) (string, cleanupFunc, error)
}
//
@@ -63,6 +71,19 @@ func (ls *localStorage) SaveFile(ctx context.Context, fileHash string, srcLen in
return nil
}
func (ls *localStorage) TempLocalFilepath(ctx context.Context, fileHash string) (string, cleanupFunc, error) {
path := filepath.Join(ls.dataDir, fileHash)
cleanup := func() error { return nil }
if _, err := os.Stat(path); err != nil {
return "", nil, err // e.g. not exists
}
// Exists (but may be TOCTTOU)
return path, cleanup, nil
}
var _ Storage = &localStorage{} // interface assertion
//
@@ -100,7 +121,36 @@ func (ss *s3Storage) ReadFile(ctx context.Context, fileHash string) (io.ReadSeek
func (ss *s3Storage) SaveFile(ctx context.Context, fileHash string, srcLen int64, src io.Reader) error {
_, err := ss.s3client.PutObject(ctx, ss.Bucket, ss.Prefix+fileHash, src, srcLen, minio.PutObjectOptions{})
return err
}
func (ss *s3Storage) TempLocalFilepath(ctx context.Context, fileHash string) (string, cleanupFunc, error) {
// Download to temporary file
fh, err := os.CreateTemp("", "contented-temp-*")
if err != nil {
return "", nil, err
}
name := fh.Name() // n.b. This is the absolute path to the file
r, err := ss.ReadFile(ctx, fileHash)
if err != nil {
return "", nil, err
}
_, err = io.Copy(fh, r)
if err != nil {
return "", nil, err
}
_ = r.Close()
_ = fh.Close()
cleanup := func() error {
return os.Remove(name)
}
return name, cleanup, nil
}
var _ Storage = &s3Storage{} // interface assertion
@@ -155,20 +205,39 @@ func (ts *tieredStorage) migrateNow() error {
return fmt.Errorf("Reading hot storage files: %w", err)
}
if len(dirents) == 0 {
return nil // Directory empty, nothing to do
}
cutOff := time.Now().Add(-TierMigrationAfter)
// Shuffle files to avoid getting stuck
rand.Shuffle(len(dirents), func(i, j int) {
dirents[i], dirents[j] = dirents[j], dirents[i]
})
log.Printf("tier-migration: Scanning %d items...", len(dirents))
var countMigrated int64 = 0
for _, dirent := range dirents {
fi, err := dirent.Info()
if err != nil {
return fmt.Errorf("Reading hot storage files: %w", err) // local files can't be stat'd = important error
}
if !fi.ModTime().After(cutOff) {
if fi.IsDir() {
continue // probably . or ..
}
if fi.ModTime().After(cutOff) {
continue // not eligible
}
fileHash := dirent.Name()
log.Printf("tier-migration: Migrating %q...", fileHash)
// Copy to cold storage
// Any concurrent reads will be serviced from the hot storage, so this
// is a safe operation
@@ -186,11 +255,15 @@ func (ts *tieredStorage) migrateNow() error {
// Copy was successful. Delete local file
err = os.Remove(filepath.Join(ts.hot.dataDir, fileHash))
if err != nil {
return fmt.Errorf("Remove %q from hot storage: %w", err) // can't rm local file
return fmt.Errorf("Remove %q from hot storage: %w", fileHash, err) // can't rm local file
}
countMigrated++
}
// Migrated everything we can for now
log.Printf("tier-migration: Sleeping (migrated %d/%d items)", countMigrated, len(dirents))
return nil
}
@@ -206,4 +279,19 @@ func (ts *tieredStorage) SaveFile(ctx context.Context, fileHash string, srcLen i
return ts.hot.SaveFile(ctx, fileHash, srcLen, src)
}
func (ts *tieredStorage) TempLocalFilepath(ctx context.Context, fileHash string) (string, cleanupFunc, error) {
path, cleanup, err_hot := ts.hot.TempLocalFilepath(ctx, fileHash)
if err_hot == nil {
return path, cleanup, nil
}
path, cleanup, err_cold := ts.cold.TempLocalFilepath(ctx, fileHash)
if err_cold == nil {
return path, cleanup, nil
}
// Neither err_hot nor err_cold worked
return "", nil, fmt.Errorf("TempLocalFilepath: failed to allocate local path for %q (hot error: %q, cold error: %q)", fileHash, err_hot.Error(), err_cold.Error())
}
var _ Storage = &tieredStorage{} // interface assertion

View File

@@ -4,11 +4,8 @@ import (
"context"
"errors"
"fmt"
"io"
"log"
"net/http"
"os"
"path/filepath"
"code.ivysaur.me/thumbnail"
)
@@ -93,45 +90,30 @@ func (this *Server) handleThumbInternal(ctx context.Context, w http.ResponseWrit
// Load metadata
m, err := this.Metadata(fileId)
if err != nil {
return err
return fmt.Errorf("Metadata: %w", err)
}
if m.FileSize > this.opts.MaxThumbSizeBytes {
return errors.New("Don't want to thumbnail very large files, sorry")
}
var filePath string
if this.opts.StorageType == STORAGE_LOCAL {
filePath = filepath.Join(this.opts.DataDirectory, m.FileHash)
} else if this.opts.StorageType == STORAGE_S3 {
// Need to temporarily download it for thumbnailing (slow and costs money)
destFh, err := os.CreateTemp("", "contented-thumbcache-*")
defer os.Remove(destFh.Name())
srcFh, err := this.store.ReadFile(ctx, m.FileHash)
if err != nil {
return err
}
_, err = io.CopyN(destFh, srcFh, m.FileSize)
srcFh.Close()
if err != nil {
return err
}
destFh.Seek(0, io.SeekStart)
filePath = destFh.Name()
} else {
panic("bad StorageType")
// Some storage backends can supply a local filepath immediately for thumbnailing
// Others (s3, tiered) need to download the file temporarily first
filePath, cleanup, err := this.store.TempLocalFilepath(ctx, m.FileHash)
if err != nil {
return fmt.Errorf("TempLocalFilepath: %w", err)
}
defer func() {
// delete temporary file, if necessary
if cleanupErr := cleanup(); cleanupErr != nil {
log.Printf("cleaning up temporary thumbnail file: %v", cleanupErr)
}
}()
thumb, err := t.RenderFileAs(filePath, m.MimeType)
if err != nil {
return err
return fmt.Errorf("RenderFileAs: %w", err)
}
w.Header().Set(`Cache-Control`, `max-age=31536000, immutable`)