mirrors/remark42
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

implement two-stage image commit with background cleanup

Browse Source
This commit is contained in:
Umputun
2019-03-23 02:56:58 -05:00
parent debd914e39
commit eb79c3d9f9
11 changed files with 189 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/store/image/image.go
+68 -8
View File
@@ -3,6 +3,7 @@
package image
import (
"context"
"fmt"
"hash/crc64"
"io"
@@ -13,6 +14,7 @@ import (
"strconv"
"strings"
"sync"
"time"
log "github.com/go-pkgz/lgr"
"github.com/google/uuid"
@@ -28,8 +30,10 @@ type Interface interface {
// FileSystem provides image Interface for local files. Saves and loads files from Location, restricts max size
type FileSystem struct {
Location string
Staging string
MaxSize int
Partitions int
TTL time.Duration // for how long file allowed on staging
crc struct {
*crc64.Table
@@ -39,7 +43,7 @@ type FileSystem struct {
}
}
// Save data from reader for given file name to local FS. Returns id as user/uuid.ext
// Save data from reader for given file name to local FS, staging directory. Returns id as user/uuid.ext
// Files partitioned across multiple subdirectories and the final path includes part, i.e. /location/user1/03/123-4567.png
func (f *FileSystem) Save(fileName string, userID string, r io.Reader) (id string, err error) {
@@ -49,7 +53,7 @@ func (f *FileSystem) Save(fileName string, userID string, r io.Reader) (id strin
}
id = path.Join(userID, uid.String()) + filepath.Ext(fileName) // make id as user/uuid.ext
dst := f.location(id)
dst := f.location(f.Staging, id)
if err = os.MkdirAll(path.Dir(dst), 0700); err != nil {
return "", errors.Wrap(err, "can't make image directory")
@@ -77,14 +81,36 @@ func (f *FileSystem) Save(fileName string, userID string, r io.Reader) (id strin
return id, nil
}
// Commit file stored in staging location by moving it to permanent location
func (f *FileSystem) Commit(id string) error {
stagingImage, permImage := f.location(f.Staging, id), f.location(f.Location, id)
if err := os.MkdirAll(path.Dir(permImage), 0700); err != nil {
return errors.Wrap(err, "can't make image directory")
}
err := os.Rename(stagingImage, permImage)
return errors.Wrapf(err, "failed to commit image %s", id)
}
// Load image from FS. Uses id to get partition subdirectory.
// returns ReadCloser and caller should call close after processing completed.
func (f *FileSystem) Load(id string) (io.ReadCloser, int64, error) {
imgFile := f.location(id)
st, err := os.Stat(imgFile)
// get image file by id. first try permanent location and if not found - staging
img := func(id string) (file string, st os.FileInfo, err error) {
file = f.location(f.Location, id)
st, err = os.Stat(file)
if err != nil {
file = f.location(f.Staging, id)
st, err = os.Stat(file)
}
return file, st, errors.Wrapf(err, "can't get image stats for %s", id)
}
imgFile, st, err := img(id)
if err != nil {
return nil, 0, errors.Wrapf(err, "can't get image size for %s", id)
return nil, 0, errors.Wrapf(err, "can't get image file for %s", id)
}
fh, err := os.Open(imgFile)
@@ -94,11 +120,45 @@ func (f *FileSystem) Load(id string) (io.ReadCloser, int64, error) {
return fh, st.Size(), nil
}
// Cleanup runs periodic scan of staging and removes old files based on TTL
func (f *FileSystem) Cleanup(ctx context.Context) {
cleanup := func() {
err := filepath.Walk(f.Staging, func(path string, info os.FileInfo, err error) error {
if err != nil {
return err
}
if info.IsDir() {
return nil
}
age := time.Since(info.ModTime())
if age > f.TTL {
log.Printf("[INFO] remove staging image %s, age %v", path, age)
return os.Remove(path)
}
return nil
})
if err != nil {
log.Printf("[WARN] failed to cleanup images, %v", err)
}
}
for {
select {
case <-ctx.Done():
log.Printf("[INFO] cleanup terminated, %v", ctx.Err())
return
case <-time.After(f.TTL / 2):
cleanup()
}
}
}
// location gets full path for id by adding partition to the final path in order to keep files in different subdirectories
// and avoid too many files in a single place.
// the end result is a full path like this - /tmp/images/user1/92/xxx-yyy.png.
// Number of partitions defined by FileSystem.Partitions
func (f *FileSystem) location(id string) string {
func (f *FileSystem) location(base string, id string) string {
partition := func(id string) string {
f.crc.Do(func() {
@@ -118,8 +178,8 @@ func (f *FileSystem) location(id string) string {
}
if f.Partitions == 0 {
return path.Join(f.Location, user, file) // avoid partition directory if 0 Partitions
return path.Join(base, user, file) // avoid partition directory if 0 Partitions
}
return path.Join(f.Location, user, partition(id), file)
return path.Join(base, user, partition(id), file)
}
backend/app/store/image/image_test.go
+93 -4
View File
@@ -1,12 +1,14 @@
package image
import (
"context"
"io/ioutil"
"math/rand"
"os"
"strconv"
"strings"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -22,7 +24,29 @@ func TestImage_Save(t *testing.T) {
assert.Contains(t, id, ".png")
t.Log(id)
data, err := ioutil.ReadFile(svc.location(id))
img := svc.location(svc.Staging, id)
t.Log(img)
data, err := ioutil.ReadFile(img)
assert.NoError(t, err)
assert.Equal(t, "blah blah", string(data))
}
func TestImage_SaveAndCommit(t *testing.T) {
svc, teardown := prepareImageTest(t)
defer teardown()
id, err := svc.Save("file1.png", "user1", strings.NewReader("blah blah"))
require.NoError(t, err)
err = svc.Commit(id)
require.NoError(t, err)
imgStaging := svc.location(svc.Staging, id)
_, err = os.Stat(imgStaging)
assert.NotNil(t, err, "no file on staging anymore")
img := svc.location(svc.Location, id)
t.Log(img)
data, err := ioutil.ReadFile(img)
assert.NoError(t, err)
assert.Equal(t, "blah blah", string(data))
}
@@ -36,7 +60,7 @@ func TestImage_SaveTooLarge(t *testing.T) {
assert.Contains(t, err.Error(), "is too large")
}
func TestImage_Load(t *testing.T) {
func TestImage_LoadAfterSave(t *testing.T) {
svc, teardown := prepareImageTest(t)
defer teardown()
@@ -56,6 +80,28 @@ func TestImage_Load(t *testing.T) {
assert.NotNil(t, err)
}
func TestImage_LoadAfterCommit(t *testing.T) {
svc, teardown := prepareImageTest(t)
defer teardown()
id, err := svc.Save("blah_ff1.png", "user1", strings.NewReader("blah blah"))
assert.NoError(t, err)
t.Log(id)
err = svc.Commit(id)
require.NoError(t, err)
r, sz, err := svc.Load(id)
assert.NoError(t, err)
defer func() { assert.NoError(t, r.Close()) }()
data, err := ioutil.ReadAll(r)
assert.NoError(t, err)
assert.Equal(t, "blah blah", string(data))
assert.Equal(t, int64(9), sz)
_, _, err = svc.Load("abcd")
assert.NotNil(t, err)
}
func TestImage_location(t *testing.T) {
tbl := []struct {
partitions int
@@ -74,7 +120,7 @@ func TestImage_location(t *testing.T) {
for n, tt := range tbl {
t.Run(strconv.Itoa(n), func(t *testing.T) {
svc := FileSystem{Location: "/tmp", Partitions: tt.partitions}
assert.Equal(t, tt.res, svc.location(tt.id))
assert.Equal(t, tt.res, svc.location("/tmp", tt.id))
})
}
@@ -91,7 +137,7 @@ func TestImage_location(t *testing.T) {
svc := FileSystem{Location: "/tmp", Partitions: 10}
for i := 0; i < 1000; i++ {
v := randomID(rand.Intn(64))
location := svc.location(v)
location := svc.location("/tmp", v)
elems := strings.Split(location, "/")
p, err := strconv.Atoi(elems[3])
require.NoError(t, err, location)
@@ -99,12 +145,54 @@ func TestImage_location(t *testing.T) {
}
}
func TestImage_Cleanup(t *testing.T) {
svc, teardown := prepareImageTest(t)
defer teardown()
save := func(file string, user string, content string) (path string) {
id, err := svc.Save(file, user, strings.NewReader(content))
require.NoError(t, err)
img := svc.location(svc.Staging, id)
data, err := ioutil.ReadFile(img)
require.NoError(t, err)
require.Equal(t, content, string(data))
return img
}
// save 3 images to staging
img1 := save("blah_ff1.png", "user1", "blah blah1")
time.Sleep(100 * time.Millisecond)
img2 := save("blah_ff2.png", "user1", "blah blah2")
time.Sleep(100 * time.Millisecond)
img3 := save("blah_ff3.png", "user2", "blah blah3")
svc.TTL = time.Millisecond * 300
ctx, cancel := context.WithCancel(context.Background())
go func() {
time.Sleep(1000 * time.Millisecond)
cancel()
}()
svc.Cleanup(ctx)
_, err := os.Stat(img1)
assert.NotNil(t, err, "no file on staging anymore")
_, err = os.Stat(img2)
assert.NotNil(t, err, "no file on staging anymore")
_, err = os.Stat(img3)
assert.NotNil(t, err, "no file on staging anymore")
}
func prepareImageTest(t *testing.T) (svc FileSystem, teardown func()) {
loc, err := ioutil.TempDir("", "test_image_r42")
require.NoError(t, err, "failed to make temp dir")
staging, err := ioutil.TempDir("", "test_image_r42.staging")
require.NoError(t, err, "failed to make temp staging dir")
svc = FileSystem{
Location: loc,
Staging: staging,
Partitions: 100,
MaxSize: 50,
}
@@ -112,6 +200,7 @@ func prepareImageTest(t *testing.T) (svc FileSystem, teardown func()) {
teardown = func() {
defer func() {
assert.NoError(t, os.RemoveAll(loc))
assert.NoError(t, os.RemoveAll(staging))
}()
}
backend/go.sum
-1
View File
@@ -81,7 +81,6 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/umputun/remark v1.2.0 h1:RoKBgzjow7+t4Z1XbhCIOiLcZNuE6LGuvj+Gxh4mopI=
golang.org/x/crypto v0.0.0-20181030102418-4d3f4d9ffa16 h1:y6ce7gCWtnH+m3dCjzQ1PCuwl28DDIc3VNnvY29DlIA=
golang.org/x/crypto v0.0.0-20181030102418-4d3f4d9ffa16/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
golang.org/x/image v0.0.0-20181116024801-cd38e8056d9b h1:VHyIDlv3XkfCa5/a81uzaoDkHH4rr81Z62g+xlnO8uM=
backend/vendor/github.com/hashicorp/errwrap/README.md
Generated Vendored
+1 -1
View File
@@ -48,7 +48,7 @@ func main() {
// We can use the Contains helpers to check if an error contains
// another error. It is safe to do this with a nil error, or with
// an error that doesn't even use the errwrap package.
if errwrap.Contains(err, ErrNotExist) {
if errwrap.Contains(err, "does not exist") {
// Do something
}
if errwrap.ContainsType(err, new(os.PathError)) {
backend/vendor/github.com/hashicorp/errwrap/go.mod
Generated Vendored
+1
View File
@@ -0,0 +1 @@
module github.com/hashicorp/errwrap
backend/vendor/github.com/hashicorp/golang-lru/lru.go
Generated Vendored
+12 -18
View File
@@ -40,35 +40,31 @@ func (c *Cache) Purge() {
// Add adds a value to the cache. Returns true if an eviction occurred.
func (c *Cache) Add(key, value interface{}) (evicted bool) {
c.lock.Lock()
evicted = c.lru.Add(key, value)
c.lock.Unlock()
return evicted
defer c.lock.Unlock()
return c.lru.Add(key, value)
}
// Get looks up a key's value from the cache.
func (c *Cache) Get(key interface{}) (value interface{}, ok bool) {
c.lock.Lock()
value, ok = c.lru.Get(key)
c.lock.Unlock()
return value, ok
defer c.lock.Unlock()
return c.lru.Get(key)
}
// Contains checks if a key is in the cache, without updating the
// recent-ness or deleting it for being stale.
func (c *Cache) Contains(key interface{}) bool {
c.lock.RLock()
containKey := c.lru.Contains(key)
c.lock.RUnlock()
return containKey
defer c.lock.RUnlock()
return c.lru.Contains(key)
}
// Peek returns the key value (or undefined if not found) without updating
// the "recently used"-ness of the key.
func (c *Cache) Peek(key interface{}) (value interface{}, ok bool) {
c.lock.RLock()
value, ok = c.lru.Peek(key)
c.lock.RUnlock()
return value, ok
defer c.lock.RUnlock()
return c.lru.Peek(key)
}
// ContainsOrAdd checks if a key is in the cache without updating the
@@ -102,15 +98,13 @@ func (c *Cache) RemoveOldest() {
// Keys returns a slice of the keys in the cache, from oldest to newest.
func (c *Cache) Keys() []interface{} {
c.lock.RLock()
keys := c.lru.Keys()
c.lock.RUnlock()
return keys
defer c.lock.RUnlock()
return c.lru.Keys()
}
// Len returns the number of items in the cache.
func (c *Cache) Len() int {
c.lock.RLock()
length := c.lru.Len()
c.lock.RUnlock()
return length
defer c.lock.RUnlock()
return c.lru.Len()
}
backend/vendor/github.com/shurcooL/sanitized_anchor_name/go.mod
Generated Vendored
+1
View File
@@ -0,0 +1 @@
module github.com/shurcooL/sanitized_anchor_name
backend/vendor/golang.org/x/time/rate/rate.go
Generated Vendored
+9 -15
View File
@@ -6,6 +6,7 @@
package rate
import (
"context"
"fmt"
"math"
"sync"
@@ -212,19 +213,8 @@ func (lim *Limiter) ReserveN(now time.Time, n int) *Reservation {
return &r
}
// contextContext is a temporary(?) copy of the context.Context type
// to support both Go 1.6 using golang.org/x/net/context and Go 1.7+
// with the built-in context package. If people ever stop using Go 1.6
// we can remove this.
type contextContext interface {
Deadline() (deadline time.Time, ok bool)
Done() <-chan struct{}
Err() error
Value(key interface{}) interface{}
}
// Wait is shorthand for WaitN(ctx, 1).
func (lim *Limiter) wait(ctx contextContext) (err error) {
func (lim *Limiter) Wait(ctx context.Context) (err error) {
return lim.WaitN(ctx, 1)
}
@@ -232,7 +222,7 @@ func (lim *Limiter) wait(ctx contextContext) (err error) {
// It returns an error if n exceeds the Limiter's burst size, the Context is
// canceled, or the expected wait time exceeds the Context's Deadline.
// The burst limit is ignored if the rate limit is Inf.
func (lim *Limiter) waitN(ctx contextContext, n int) (err error) {
func (lim *Limiter) WaitN(ctx context.Context, n int) (err error) {
if n > lim.burst && lim.limit != Inf {
return fmt.Errorf("rate: Wait(n=%d) exceeds limiter's burst %d", n, lim.burst)
}
@@ -253,8 +243,12 @@ func (lim *Limiter) waitN(ctx contextContext, n int) (err error) {
if !r.ok {
return fmt.Errorf("rate: Wait(n=%d) would exceed context deadline", n)
}
// Wait
t := time.NewTimer(r.DelayFrom(now))
// Wait if necessary
delay := r.DelayFrom(now)
if delay == 0 {
return nil
}
t := time.NewTimer(delay)
defer t.Stop()
select {
case <-t.C:
backend/vendor/golang.org/x/time/rate/rate_go16.go
Generated Vendored
-21
View File
@@ -1,21 +0,0 @@
// Copyright 2017 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// +build !go1.7
package rate
import "golang.org/x/net/context"
// Wait is shorthand for WaitN(ctx, 1).
func (lim *Limiter) Wait(ctx context.Context) (err error) {
return lim.waitN(ctx, 1)
}
// WaitN blocks until lim permits n events to happen.
// It returns an error if n exceeds the Limiter's burst size, the Context is
// canceled, or the expected wait time exceeds the Context's Deadline.
func (lim *Limiter) WaitN(ctx context.Context, n int) (err error) {
return lim.waitN(ctx, n)
}
backend/vendor/golang.org/x/time/rate/rate_go17.go
Generated Vendored
-21
View File
@@ -1,21 +0,0 @@
// Copyright 2017 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// +build go1.7
package rate
import "context"
// Wait is shorthand for WaitN(ctx, 1).
func (lim *Limiter) Wait(ctx context.Context) (err error) {
return lim.waitN(ctx, 1)
}
// WaitN blocks until lim permits n events to happen.
// It returns an error if n exceeds the Limiter's burst size, the Context is
// canceled, or the expected wait time exceeds the Context's Deadline.
func (lim *Limiter) WaitN(ctx context.Context, n int) (err error) {
return lim.waitN(ctx, n)
}
backend/vendor/modules.txt
Vendored
+4 -4
View File
@@ -58,11 +58,11 @@ github.com/golang/protobuf/proto
github.com/google/uuid
# github.com/gorilla/feeds v1.1.0
github.com/gorilla/feeds
# github.com/hashicorp/errwrap v0.0.0-20141028054710-7554cd9344ce
# github.com/hashicorp/errwrap v1.0.0
github.com/hashicorp/errwrap
# github.com/hashicorp/go-multierror v0.0.0-20171204182908-b7773ae21874
github.com/hashicorp/go-multierror
# github.com/hashicorp/golang-lru v0.5.1
# github.com/hashicorp/golang-lru v0.5.0
github.com/hashicorp/golang-lru
github.com/hashicorp/golang-lru/simplelru
# github.com/jessevdk/go-flags v0.0.0-20180331124232-1c38ed7ad0cc
@@ -79,7 +79,7 @@ github.com/pkg/errors
github.com/pmezard/go-difflib/difflib
# github.com/rakyll/statik v0.1.3
github.com/rakyll/statik/fs
# github.com/shurcooL/sanitized_anchor_name v0.0.0-20170918181015-86672fcb3f95
# github.com/shurcooL/sanitized_anchor_name v1.0.0
github.com/shurcooL/sanitized_anchor_name
# github.com/stretchr/testify v1.3.0
github.com/stretchr/testify/assert
@@ -106,7 +106,7 @@ golang.org/x/oauth2/jws
golang.org/x/oauth2/jwt
# golang.org/x/sys v0.0.0-20190109145017-48ac38b7c8cb
golang.org/x/sys/unix
# golang.org/x/time v0.0.0-20170927054726-6dc17368e09b
# golang.org/x/time v0.0.0-20190308202827-9d24e82272b4
golang.org/x/time/rate
# google.golang.org/appengine v1.4.0
google.golang.org/appengine