Merge pull request #1372 from aibaars/gcs-delete
StorageDriver GCS: improve test suite clean-up and add retrying to all GCS api calls
This commit is contained in:
commit
fdd2e4d77e
@ -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
|
||||
|
@ -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) {
|
||||
|
@ -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)
|
||||
|
Loading…
Reference in New Issue
Block a user