octoleo/restic
2
2
Fork 0
mirror of https://github.com/octoleo/restic.git synced 2024-11-02 11:46:36 +00:00
Code Issues Releases Wiki Activity

s3: Switch back to high-level API, limit connections

Browse Source
This commit is contained in:
Alexander Neumann 2017-06-11 11:15:15 +02:00
parent a4e8dc3371
commit 08e1d9ffad
4 changed files with 21 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/restic/backend/s3/config.go
View File

@ -25,7 +25,7 @@ type Config struct {
// NewConfig returns a new Config with the default values filled in.
func NewConfig() Config {
return Config{
Connections: 20,
Connections: 5,
}
}

28
src/restic/backend/s3/config_test.go
View File

@ -10,89 +10,89 @@ var configTests = []struct {
Endpoint: "eu-central-1",
Bucket: "bucketname",
Prefix: "restic",
Connections: 20,
Connections: 5,
}},
{"s3://eu-central-1/bucketname/", Config{
Endpoint: "eu-central-1",
Bucket: "bucketname",
Prefix: "restic",
Connections: 20,
Connections: 5,
}},
{"s3://eu-central-1/bucketname/prefix/directory", Config{
Endpoint: "eu-central-1",
Bucket: "bucketname",
Prefix: "prefix/directory",
Connections: 20,
Connections: 5,
}},
{"s3://eu-central-1/bucketname/prefix/directory/", Config{
Endpoint: "eu-central-1",
Bucket: "bucketname",
Prefix: "prefix/directory",
Connections: 20,
Connections: 5,
}},
{"s3:eu-central-1/foobar", Config{
Endpoint: "eu-central-1",
Bucket: "foobar",
Prefix: "restic",
Connections: 20,
Connections: 5,
}},
{"s3:eu-central-1/foobar/", Config{
Endpoint: "eu-central-1",
Bucket: "foobar",
Prefix: "restic",
Connections: 20,
Connections: 5,
}},
{"s3:eu-central-1/foobar/prefix/directory", Config{
Endpoint: "eu-central-1",
Bucket: "foobar",
Prefix: "prefix/directory",
Connections: 20,
Connections: 5,
}},
{"s3:eu-central-1/foobar/prefix/directory/", Config{
Endpoint: "eu-central-1",
Bucket: "foobar",
Prefix: "prefix/directory",
Connections: 20,
Connections: 5,
}},
{"s3:https://hostname:9999/foobar", Config{
Endpoint: "hostname:9999",
Bucket: "foobar",
Prefix: "restic",
Connections: 20,
Connections: 5,
}},
{"s3:https://hostname:9999/foobar/", Config{
Endpoint: "hostname:9999",
Bucket: "foobar",
Prefix: "restic",
Connections: 20,
Connections: 5,
}},
{"s3:http://hostname:9999/foobar", Config{
Endpoint: "hostname:9999",
Bucket: "foobar",
Prefix: "restic",
UseHTTP: true,
Connections: 20,
Connections: 5,
}},
{"s3:http://hostname:9999/foobar/", Config{
Endpoint: "hostname:9999",
Bucket: "foobar",
Prefix: "restic",
UseHTTP: true,
Connections: 20,
Connections: 5,
}},
{"s3:http://hostname:9999/bucket/prefix/directory", Config{
Endpoint: "hostname:9999",
Bucket: "bucket",
Prefix: "prefix/directory",
UseHTTP: true,
Connections: 20,
Connections: 5,
}},
{"s3:http://hostname:9999/bucket/prefix/directory/", Config{
Endpoint: "hostname:9999",
Bucket: "bucket",
Prefix: "prefix/directory",
UseHTTP: true,
Connections: 20,
Connections: 5,
}},
}

71
src/restic/backend/s3/s3.go
View File

@ -157,63 +157,15 @@ func (be *Backend) Path() string {
return be.prefix
}
// getRemainingSize returns number of bytes remaining. If it is not possible to
// determine the size, panic() is called.
func getRemainingSize(rd io.Reader) (size int64, err error) {
type Sizer interface {
Size() int64
}
type Lenner interface {
Len() int
}
if r, ok := rd.(Lenner); ok {
size = int64(r.Len())
} else if r, ok := rd.(Sizer); ok {
size = r.Size()
} else if f, ok := rd.(*os.File); ok {
fi, err := f.Stat()
if err != nil {
return 0, err
}
pos, err := f.Seek(0, io.SeekCurrent)
if err != nil {
return 0, err
}
size = fi.Size() - pos
} else {
panic(fmt.Sprintf("Save() got passed a reader without a method to determine the data size, type is %T", rd))
}
return size, nil
}
// preventCloser wraps an io.Reader to run a function instead of the original Close() function.
type preventCloser struct {
io.Reader
f func()
}
func (wr preventCloser) Close() error {
wr.f()
return nil
}
// Save stores data in the backend at the handle.
func (be *Backend) Save(ctx context.Context, h restic.Handle, rd io.Reader) (err error) {
debug.Log("Save %v", h)
if err := h.Valid(); err != nil {
return err
}
objName := be.Filename(h)
size, err := getRemainingSize(rd)
if err != nil {
return err
}
debug.Log("Save %v at %v (%d bytes)", h, objName, size)
// Check key does not already exist
_, err = be.client.StatObject(be.bucketname, objName)
@ -222,23 +174,12 @@ func (be *Backend) Save(ctx context.Context, h restic.Handle, rd io.Reader) (err
return errors.New("key already exists")
}
be.sem.GetToken()
// wrap the reader so that net/http client cannot close the reader, return
// the token instead.
rd = preventCloser{
Reader: rd,
f: func() {
debug.Log("Close()")
},
}
debug.Log("PutObject(%v, %v)", be.bucketname, objName)
coreClient := minio.Core{Client: be.client}
info, err := coreClient.PutObject(be.bucketname, objName, size, rd, nil, nil, nil)
be.sem.GetToken()
n, err := be.client.PutObject(be.bucketname, objName, rd, "application/octet-stream")
be.sem.ReleaseToken()
debug.Log("%v -> %v bytes, err %#v: %v", objName, info.Size, err, err)
debug.Log("%v -> %v bytes, err %#v: %v", objName, n, err, err)
return errors.Wrap(err, "client.PutObject")
}

71
src/restic/backend/s3/s3_internal_test.go
View File

@ -1,71 +0,0 @@
package s3
import (
"bytes"
"io"
"io/ioutil"
"os"
"restic/test"
"testing"
)
func writeFile(t testing.TB, data []byte, offset int64) *os.File {
tempfile, err := ioutil.TempFile("", "restic-test-")
if err != nil {
t.Fatal(err)
}
if _, err = tempfile.Write(data); err != nil {
t.Fatal(err)
}
if _, err = tempfile.Seek(offset, io.SeekStart); err != nil {
t.Fatal(err)
}
return tempfile
}
func TestGetRemainingSize(t *testing.T) {
length := 18 * 1123
partialRead := 1005
data := test.Random(23, length)
partReader := bytes.NewReader(data)
buf := make([]byte, partialRead)
_, _ = io.ReadFull(partReader, buf)
partFileReader := writeFile(t, data, int64(partialRead))
defer func() {
if err := partFileReader.Close(); err != nil {
t.Fatal(err)
}
if err := os.Remove(partFileReader.Name()); err != nil {
t.Fatal(err)
}
}()
var tests = []struct {
io.Reader
size int64
}{
{bytes.NewReader([]byte("foobar test")), 11},
{partReader, int64(length - partialRead)},
{partFileReader, int64(length - partialRead)},
}
for _, test := range tests {
t.Run("", func(t *testing.T) {
size, err := getRemainingSize(test.Reader)
if err != nil {
t.Fatal(err)
}
if size != test.size {
t.Fatalf("invalid size returned, want %v, got %v", test.size, size)
}
})
}
}