diff --git a/docs/storage/driver/gcs/gcs.go b/docs/storage/driver/gcs/gcs.go index 765d5492..0e3480f2 100644 --- a/docs/storage/driver/gcs/gcs.go +++ b/docs/storage/driver/gcs/gcs.go @@ -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 diff --git a/docs/storage/driver/gcs/gcs_test.go b/docs/storage/driver/gcs/gcs_test.go index 31494bde..554d95e4 100644 --- a/docs/storage/driver/gcs/gcs_test.go +++ b/docs/storage/driver/gcs/gcs_test.go @@ -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) { diff --git a/docs/storage/driver/testsuites/testsuites.go b/docs/storage/driver/testsuites/testsuites.go index 6fea2def..5c34cca6 100644 --- a/docs/storage/driver/testsuites/testsuites.go +++ b/docs/storage/driver/testsuites/testsuites.go @@ -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)