vbatts/registry
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Merge pull request #1372 from aibaars/gcs-delete

Browse source 
StorageDriver GCS: improve test suite clean-up and add retrying to all GCS api calls
This commit is contained in:
Richard Scothern 2016-01-20 12:21:22 -08:00
commit 05b8317291
3 changed files with 105 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

71
docs/storage/driver/gcs/gcs.go
View file

@ -206,7 +206,7 @@ func (d *driver) ReadStream(context ctx.Context, path string, offset int64) (io.
}
if res.StatusCode == http.StatusRequestedRangeNotSatisfiable {
res.Body.Close()
obj, err := storage.StatObject(d.context(context), d.bucket, name)
obj, err := storageStatObject(d.context(context), d.bucket, name)
if err != nil {
return nil, err
}
@ -287,7 +287,7 @@ func (d *driver) WriteStream(context ctx.Context, path string, offset int64, rea
}
// wc was closed succesfully, so the temporary part exists, schedule it for deletion at the end
// of the function
defer storage.DeleteObject(gcsContext, d.bucket, partName)
defer storageDeleteObject(gcsContext, d.bucket, partName)
req := &storageapi.ComposeRequest{
Destination: &storageapi.Object{Bucket: obj.Bucket, Name: obj.Name, ContentType: obj.ContentType},
@ -386,7 +386,7 @@ func (d *driver) Stat(context ctx.Context, path string) (storagedriver.FileInfo,
var fi storagedriver.FileInfoFields
//try to get as file
gcsContext := d.context(context)
obj, err := storage.StatObject(gcsContext, d.bucket, d.pathToKey(path))
obj, err := storageStatObject(gcsContext, d.bucket, d.pathToKey(path))
if err == nil {
fi = storagedriver.FileInfoFields{
Path: path,
@ -404,7 +404,7 @@ func (d *driver) Stat(context ctx.Context, path string) (storagedriver.FileInfo,
query.Prefix = dirpath
query.MaxResults = 1
objects, err := storage.ListObjects(gcsContext, d.bucket, query)
objects, err := storageListObjects(gcsContext, d.bucket, query)
if err != nil {
return nil, err
}
@ -432,7 +432,7 @@ func (d *driver) List(context ctx.Context, path string) ([]string, error) {
query.Prefix = d.pathToDirKey(path)
list := make([]string, 0, 64)
for {
objects, err := storage.ListObjects(d.context(context), d.bucket, query)
objects, err := storageListObjects(d.context(context), d.bucket, query)
if err != nil {
return nil, err
}
@ -482,7 +482,7 @@ func (d *driver) Move(context ctx.Context, sourcePath string, destPath string) e
var err error
for _, key := range keys {
dest := destPrefix + key[len(prefix):]
_, err = storage.CopyObject(gcsContext, d.bucket, key, d.bucket, dest, nil)
_, err = storageCopyObject(gcsContext, d.bucket, key, d.bucket, dest, nil)
if err == nil {
copies = append(copies, dest)
} else {
@ -492,20 +492,20 @@ func (d *driver) Move(context ctx.Context, sourcePath string, destPath string) e
// if an error occurred, attempt to cleanup the copies made
if err != nil {
for i := len(copies) - 1; i >= 0; i-- {
_ = storage.DeleteObject(gcsContext, d.bucket, copies[i])
_ = storageDeleteObject(gcsContext, d.bucket, copies[i])
}
return err
}
// delete originals
for i := len(keys) - 1; i >= 0; i-- {
err2 := storage.DeleteObject(gcsContext, d.bucket, keys[i])
err2 := storageDeleteObject(gcsContext, d.bucket, keys[i])
if err2 != nil {
err = err2
}
}
return err
}
_, err = storage.CopyObject(gcsContext, d.bucket, d.pathToKey(sourcePath), d.bucket, d.pathToKey(destPath), nil)
_, err = storageCopyObject(gcsContext, d.bucket, d.pathToKey(sourcePath), d.bucket, d.pathToKey(destPath), nil)
if err != nil {
if status := err.(*googleapi.Error); status != nil {
if status.Code == http.StatusNotFound {
@ -514,7 +514,7 @@ func (d *driver) Move(context ctx.Context, sourcePath string, destPath string) e
}
return err
}
return storage.DeleteObject(gcsContext, d.bucket, d.pathToKey(sourcePath))
return storageDeleteObject(gcsContext, d.bucket, d.pathToKey(sourcePath))
}
// listAll recursively lists all names of objects stored at "prefix" and its subpaths.
@ -524,7 +524,7 @@ func (d *driver) listAll(context context.Context, prefix string) ([]string, erro
query.Prefix = prefix
query.Versions = false
for {
objects, err := storage.ListObjects(d.context(context), d.bucket, query)
objects, err := storageListObjects(d.context(context), d.bucket, query)
if err != nil {
return nil, err
}
@ -555,13 +555,22 @@ func (d *driver) Delete(context ctx.Context, path string) error {
if len(keys) > 0 {
sort.Sort(sort.Reverse(sort.StringSlice(keys)))
for _, key := range keys {
if err := storage.DeleteObject(gcsContext, d.bucket, key); err != nil {
err := storageDeleteObject(gcsContext, d.bucket, key)
// GCS only guarantees eventual consistency, so listAll might return
// paths that no longer exist. If this happens, just ignore any not
// found error
if status, ok := err.(*googleapi.Error); ok {
if status.Code == http.StatusNotFound {
err = nil
}
}
if err != nil {
return err
}
}
return nil
}
err = storage.DeleteObject(gcsContext, d.bucket, d.pathToKey(path))
err = storageDeleteObject(gcsContext, d.bucket, d.pathToKey(path))
if err != nil {
if status := err.(*googleapi.Error); status != nil {
if status.Code == http.StatusNotFound {
@ -572,6 +581,42 @@ func (d *driver) Delete(context ctx.Context, path string) error {
return err
}
func storageDeleteObject(context context.Context, bucket string, name string) error {
return retry(5, func() error {
return storage.DeleteObject(context, bucket, name)
})
}
func storageStatObject(context context.Context, bucket string, name string) (*storage.Object, error) {
var obj *storage.Object
err := retry(5, func() error {
var err error
obj, err = storage.StatObject(context, bucket, name)
return err
})
return obj, err
}
func storageListObjects(context context.Context, bucket string, q *storage.Query) (*storage.Objects, error) {
var objs *storage.Objects
err := retry(5, func() error {
var err error
objs, err = storage.ListObjects(context, bucket, q)
return err
})
return objs, err
}
func storageCopyObject(context context.Context, srcBucket, srcName string, destBucket, destName string, attrs *storage.ObjectAttrs) (*storage.Object, error) {
var obj *storage.Object
err := retry(5, func() error {
var err error
obj, err = storage.CopyObject(context, srcBucket, srcName, destBucket, destName, attrs)
return err
})
return obj, err
}
// URLFor returns a URL which may be used to retrieve the content stored at
// the given path, possibly using the given options.
// Returns ErrUnsupportedMethod if this driver has no privateKey

8
docs/storage/driver/gcs/gcs_test.go
View file

@ -155,8 +155,12 @@ func TestEmptyRootList(t *testing.T) {
if err != nil {
t.Fatalf("unexpected error creating content: %v", err)
}
defer rootedDriver.Delete(ctx, filename)
defer func() {
err := rootedDriver.Delete(ctx, filename)
if err != nil {
t.Fatalf("failed to remove %v due to %v\n", filename, err)
}
}()
keys, err := emptyRootDriver.List(ctx, "/")
for _, path := range keys {
if !storagedriver.PathRegexp.MatchString(path) {

64
docs/storage/driver/testsuites/testsuites.go
View file

@ -120,7 +120,7 @@ func (suite *DriverSuite) TestValidPaths(c *check.C) {
for _, filename := range validFiles {
err := suite.StorageDriver.PutContent(suite.ctx, filename, contents)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(filename))
defer suite.deletePath(c, firstPart(filename))
c.Assert(err, check.IsNil)
received, err := suite.StorageDriver.GetContent(suite.ctx, filename)
@ -129,6 +129,21 @@ func (suite *DriverSuite) TestValidPaths(c *check.C) {
}
}
func (suite *DriverSuite) deletePath(c *check.C, path string) {
for tries := 2; tries > 0; tries-- {
err := suite.StorageDriver.Delete(suite.ctx, path)
if _, ok := err.(storagedriver.PathNotFoundError); ok {
err = nil
}
c.Assert(err, check.IsNil)
paths, err := suite.StorageDriver.List(suite.ctx, path)
if len(paths) == 0 {
break
}
time.Sleep(time.Second * 2)
}
}
// TestInvalidPaths checks that various invalid file paths are rejected by the
// storage driver.
func (suite *DriverSuite) TestInvalidPaths(c *check.C) {
@ -143,7 +158,10 @@ func (suite *DriverSuite) TestInvalidPaths(c *check.C) {
for _, filename := range invalidFiles {
err := suite.StorageDriver.PutContent(suite.ctx, filename, contents)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(filename))
// only delete if file was succesfully written
if err == nil {
defer suite.deletePath(c, firstPart(filename))
}
c.Assert(err, check.NotNil)
c.Assert(err, check.FitsTypeOf, storagedriver.InvalidPathError{})
c.Assert(strings.Contains(err.Error(), suite.Name()), check.Equals, true)
@ -258,7 +276,7 @@ func (suite *DriverSuite) TestWriteReadLargeStreams(c *check.C) {
}
filename := randomPath(32)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(filename))
defer suite.deletePath(c, firstPart(filename))
checksum := sha1.New()
var fileSize int64 = 5 * 1024 * 1024 * 1024
@ -282,7 +300,7 @@ func (suite *DriverSuite) TestWriteReadLargeStreams(c *check.C) {
// reading with a given offset.
func (suite *DriverSuite) TestReadStreamWithOffset(c *check.C) {
filename := randomPath(32)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(filename))
defer suite.deletePath(c, firstPart(filename))
chunkSize := int64(32)
@ -372,7 +390,7 @@ func (suite *DriverSuite) TestContinueStreamAppendSmall(c *check.C) {
func (suite *DriverSuite) testContinueStreamAppend(c *check.C, chunkSize int64) {
filename := randomPath(32)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(filename))
defer suite.deletePath(c, firstPart(filename))
contentsChunk1 := randomContents(chunkSize)
contentsChunk2 := randomContents(chunkSize)
@ -470,7 +488,7 @@ func (suite *DriverSuite) TestReadNonexistentStream(c *check.C) {
// TestList checks the returned list of keys after populating a directory tree.
func (suite *DriverSuite) TestList(c *check.C) {
rootDirectory := "/" + randomFilename(int64(8+rand.Intn(8)))
defer suite.StorageDriver.Delete(suite.ctx, rootDirectory)
defer suite.deletePath(c, rootDirectory)
doesnotexist := path.Join(rootDirectory, "nonexistent")
_, err := suite.StorageDriver.List(suite.ctx, doesnotexist)
@ -516,8 +534,8 @@ func (suite *DriverSuite) TestMove(c *check.C) {
sourcePath := randomPath(32)
destPath := randomPath(32)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(sourcePath))
defer suite.StorageDriver.Delete(suite.ctx, firstPart(destPath))
defer suite.deletePath(c, firstPart(sourcePath))
defer suite.deletePath(c, firstPart(destPath))
err := suite.StorageDriver.PutContent(suite.ctx, sourcePath, contents)
c.Assert(err, check.IsNil)
@ -543,8 +561,8 @@ func (suite *DriverSuite) TestMoveOverwrite(c *check.C) {
sourceContents := randomContents(32)
destContents := randomContents(64)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(sourcePath))
defer suite.StorageDriver.Delete(suite.ctx, firstPart(destPath))
defer suite.deletePath(c, firstPart(sourcePath))
defer suite.deletePath(c, firstPart(destPath))
err := suite.StorageDriver.PutContent(suite.ctx, sourcePath, sourceContents)
c.Assert(err, check.IsNil)
@ -572,7 +590,7 @@ func (suite *DriverSuite) TestMoveNonexistent(c *check.C) {
sourcePath := randomPath(32)
destPath := randomPath(32)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(destPath))
defer suite.deletePath(c, firstPart(destPath))
err := suite.StorageDriver.PutContent(suite.ctx, destPath, contents)
c.Assert(err, check.IsNil)
@ -594,7 +612,7 @@ func (suite *DriverSuite) TestMoveInvalid(c *check.C) {
// Create a regular file.
err := suite.StorageDriver.PutContent(suite.ctx, "/notadir", contents)
c.Assert(err, check.IsNil)
defer suite.StorageDriver.Delete(suite.ctx, "/notadir")
defer suite.deletePath(c, "/notadir")
// Now try to move a non-existent file under it.
err = suite.StorageDriver.Move(suite.ctx, "/notadir/foo", "/notadir/bar")
@ -607,7 +625,7 @@ func (suite *DriverSuite) TestDelete(c *check.C) {
filename := randomPath(32)
contents := randomContents(32)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(filename))
defer suite.deletePath(c, firstPart(filename))
err := suite.StorageDriver.PutContent(suite.ctx, filename, contents)
c.Assert(err, check.IsNil)
@ -627,7 +645,7 @@ func (suite *DriverSuite) TestURLFor(c *check.C) {
filename := randomPath(32)
contents := randomContents(32)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(filename))
defer suite.deletePath(c, firstPart(filename))
err := suite.StorageDriver.PutContent(suite.ctx, filename, contents)
c.Assert(err, check.IsNil)
@ -674,7 +692,7 @@ func (suite *DriverSuite) TestDeleteFolder(c *check.C) {
filename3 := randomPath(32)
contents := randomContents(32)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(dirname))
defer suite.deletePath(c, firstPart(dirname))
err := suite.StorageDriver.PutContent(suite.ctx, path.Join(dirname, filename1), contents)
c.Assert(err, check.IsNil)
@ -725,7 +743,7 @@ func (suite *DriverSuite) TestStatCall(c *check.C) {
fileName := randomFilename(32)
filePath := path.Join(dirPath, fileName)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(dirPath))
defer suite.deletePath(c, firstPart(dirPath))
// Call on non-existent file/dir, check error.
fi, err := suite.StorageDriver.Stat(suite.ctx, dirPath)
@ -788,7 +806,7 @@ func (suite *DriverSuite) TestPutContentMultipleTimes(c *check.C) {
filename := randomPath(32)
contents := randomContents(4096)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(filename))
defer suite.deletePath(c, firstPart(filename))
err := suite.StorageDriver.PutContent(suite.ctx, filename, contents)
c.Assert(err, check.IsNil)
@ -814,7 +832,7 @@ func (suite *DriverSuite) TestConcurrentStreamReads(c *check.C) {
filename := randomPath(32)
contents := randomContents(filesize)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(filename))
defer suite.deletePath(c, firstPart(filename))
err := suite.StorageDriver.PutContent(suite.ctx, filename, contents)
c.Assert(err, check.IsNil)
@ -872,7 +890,7 @@ func (suite *DriverSuite) TestEventualConsistency(c *check.C) {
}
filename := randomPath(32)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(filename))
defer suite.deletePath(c, firstPart(filename))
var offset int64
var misswrites int
@ -1033,7 +1051,7 @@ func (suite *DriverSuite) BenchmarkDelete50Files(c *check.C) {
func (suite *DriverSuite) benchmarkDeleteFiles(c *check.C, numFiles int64) {
for i := 0; i < c.N; i++ {
parentDir := randomPath(8)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(parentDir))
defer suite.deletePath(c, firstPart(parentDir))
c.StopTimer()
for j := int64(0); j < numFiles; j++ {
@ -1055,7 +1073,7 @@ func (suite *DriverSuite) testFileStreams(c *check.C, size int64) {
defer tf.Close()
filename := randomPath(32)
defer suite.StorageDriver.Delete(suite.ctx, firstPart(filename))
defer suite.deletePath(c, firstPart(filename))
contents := randomContents(size)
@ -1080,7 +1098,7 @@ func (suite *DriverSuite) testFileStreams(c *check.C, size int64) {
}
func (suite *DriverSuite) writeReadCompare(c *check.C, filename string, contents []byte) {
defer suite.StorageDriver.Delete(suite.ctx, firstPart(filename))
defer suite.deletePath(c, firstPart(filename))
err := suite.StorageDriver.PutContent(suite.ctx, filename, contents)
c.Assert(err, check.IsNil)
@ -1092,7 +1110,7 @@ func (suite *DriverSuite) writeReadCompare(c *check.C, filename string, contents
}
func (suite *DriverSuite) writeReadCompareStreams(c *check.C, filename string, contents []byte) {
defer suite.StorageDriver.Delete(suite.ctx, firstPart(filename))
defer suite.deletePath(c, firstPart(filename))
nn, err := suite.StorageDriver.WriteStream(suite.ctx, filename, 0, bytes.NewReader(contents))
c.Assert(err, check.IsNil)