br.storage: add delete api (pingcap#27237) (pingcap#27766)

br/pkg/storage/gcs.go

@@ -88,6 +88,13 @@ type gcsStorage struct {
 	bucket *storage.BucketHandle
 }
 
+// DeleteFile delete the file in storage
+func (s *gcsStorage) DeleteFile(ctx context.Context, name string) error {
+	object := s.objectName(name)
+	err := s.bucket.Object(object).Delete(ctx)
+	return errors.Trace(err)
+}
+
 func (s *gcsStorage) objectName(name string) string {
 	return path.Join(s.gcs.Prefix, name)
 }

br/pkg/storage/gcs_test.go

@@ -65,6 +65,25 @@ func (r *testStorageSuite) TestGCS(c *C) {
 	c.Assert(err, IsNil)
 	c.Assert(exist, IsFalse)
 
+	keyDelete := "key_delete"
+	exist, err = stg.FileExists(ctx, keyDelete)
+	c.Assert(err, IsNil)
+	c.Assert(exist, IsFalse)
+
+	err = stg.WriteFile(ctx, keyDelete, []byte("data"))
+	c.Assert(err, IsNil)
+
+	exist, err = stg.FileExists(ctx, keyDelete)
+	c.Assert(err, IsNil)
+	c.Assert(exist, IsTrue)
+
+	err = stg.DeleteFile(ctx, keyDelete)
+	c.Assert(err, IsNil)
+
+	exist, err = stg.FileExists(ctx, keyDelete)
+	c.Assert(err, IsNil)
+	c.Assert(exist, IsFalse)
+
 	list := ""
 	var totalSize int64 = 0
 	err = stg.WalkDir(ctx, nil, func(name string, size int64) error {

br/pkg/storage/local.go

@@ -25,6 +25,11 @@ type LocalStorage struct {
 	base string
 }
 
+func (l *LocalStorage) DeleteFile(ctx context.Context, name string) error {
+	path := filepath.Join(l.base, name)
+	return os.Remove(path)
+}
+
 // WriteFile writes data to a file to storage.
 func (l *LocalStorage) WriteFile(ctx context.Context, name string, data []byte) error {
 	path := filepath.Join(l.base, name)

br/pkg/storage/local_test.go

@@ -15,6 +15,33 @@ type testLocalSuite struct{}
 
 var _ = Suite(&testLocalSuite{})
 
+func (r *testStorageSuite) TestDeleteFile(c *C) {
+	dir := c.MkDir()
+	sb, err := ParseBackend("file://"+filepath.ToSlash(dir), &BackendOptions{})
+	c.Assert(err, IsNil)
+	store, err := Create(context.TODO(), sb, true)
+	c.Assert(err, IsNil)
+
+	name := "test_delete"
+	ret, err := store.FileExists(context.Background(), name)
+	c.Assert(err, IsNil)
+	c.Assert(ret, Equals, false)
+
+	_, err = store.Create(context.Background(), name)
+	c.Assert(err, IsNil)
+
+	ret, err = store.FileExists(context.Background(), name)
+	c.Assert(err, IsNil)
+	c.Assert(ret, Equals, true)
+
+	err = store.DeleteFile(context.Background(), name)
+	c.Assert(err, IsNil)
+
+	ret, err = store.FileExists(context.Background(), name)
+	c.Assert(err, IsNil)
+	c.Assert(ret, Equals, false)
+}
+
 func (r *testStorageSuite) TestWalkDirWithSoftLinkFile(c *C) {
 	if runtime.GOOS == "windows" {
 		// skip the test on windows. typically windows users don't have symlink permission.
@@ -26,7 +53,6 @@ func (r *testStorageSuite) TestWalkDirWithSoftLinkFile(c *C) {
 	path1 := filepath.Join(dir1, name1)
 	f1, err := os.Create(path1)
 	c.Assert(err, IsNil)
-
 	data := "/* whatever pragmas */;" +
 		"INSERT INTO `namespaced`.`table` (columns, more, columns) VALUES (1,-2, 3),\n(4,5., 6);" +
 		"INSERT `namespaced`.`table` (x,y,z) VALUES (7,8,9);" +

br/pkg/storage/noop.go

@@ -8,6 +8,11 @@ import (
 
 type noopStorage struct{}
 
+// DeleteFile delete the file in storage
+func (s *noopStorage) DeleteFile(ctx context.Context, name string) error {
+	return nil
+}
+
 // WriteFile file to storage.
 func (*noopStorage) WriteFile(ctx context.Context, name string, data []byte) error {
 	return nil

br/pkg/storage/s3.go

@@ -403,6 +403,17 @@ func (rs *S3Storage) ReadFile(ctx context.Context, file string) ([]byte, error)
 	return data, nil
 }
 
+// DeleteFile delete the file in s3 storage
+func (rs *S3Storage) DeleteFile(ctx context.Context, file string) error {
+	input := &s3.DeleteObjectInput{
+		Bucket: aws.String(rs.options.Bucket),
+		Key:    aws.String(rs.options.Prefix + file),
+	}
+
+	_, err := rs.svc.DeleteObjectWithContext(ctx, input)
+	return errors.Trace(err)
+}
+
 // FileExists check if file exists on s3 storage.
 func (rs *S3Storage) FileExists(ctx context.Context, file string) (bool, error) {
 	input := &s3.HeadObjectInput{

br/pkg/storage/s3_test.go

@@ -508,6 +508,52 @@ func (s *s3Suite) TestFileExistsNoError(c *C) {
 	c.Assert(exists, IsTrue)
 }
 
+func (s *s3Suite) TestDeleteFileNoError(c *C) {
+	s.setUpTest(c)
+	defer s.tearDownTest()
+	ctx := aws.BackgroundContext()
+
+	s.s3.EXPECT().
+		DeleteObjectWithContext(ctx, gomock.Any()).
+		DoAndReturn(func(_ context.Context, input *s3.DeleteObjectInput, opt ...request.Option) (*s3.DeleteObjectInput, error) {
+			c.Assert(aws.StringValue(input.Bucket), Equals, "bucket")
+			c.Assert(aws.StringValue(input.Key), Equals, "prefix/file")
+			return &s3.DeleteObjectInput{}, nil
+		})
+
+	err := s.storage.DeleteFile(ctx, "file")
+	c.Assert(err, IsNil)
+}
+
+func (s *s3Suite) TestDeleteFileMissing(c *C) {
+	s.setUpTest(c)
+	defer s.tearDownTest()
+	ctx := aws.BackgroundContext()
+
+	awserr := awserr.New(s3.ErrCodeNoSuchKey, "no such key", nil)
+	s.s3.EXPECT().
+		DeleteObjectWithContext(ctx, gomock.Any()).
+		Return(nil, awserr)
+
+	err := s.storage.DeleteFile(ctx, "file-missing")
+	c.Assert(err, ErrorMatches, awserr.Error())
+}
+
+func (s *s3Suite) TestDeleteFileError(c *C) {
+	s.setUpTest(c)
+	defer s.tearDownTest()
+	ctx := aws.BackgroundContext()
+
+	expectedErr := errors.New("just some unrelated error")
+
+	s.s3.EXPECT().
+		DeleteObjectWithContext(ctx, gomock.Any()).
+		Return(nil, expectedErr)
+
+	err := s.storage.DeleteFile(ctx, "file3")
+	c.Assert(err, ErrorMatches, `\Q`+expectedErr.Error()+`\E`)
+}
+
 // TestFileExistsNoSuckKey ensures FileExists API reports file missing if S3's
 // HeadObject request replied NoSuchKey.
 func (s *s3Suite) TestFileExistsMissing(c *C) {

br/pkg/storage/storage.go

@@ -77,6 +77,8 @@ type ExternalStorage interface {
 	ReadFile(ctx context.Context, name string) ([]byte, error)
 	// FileExists return true if file exists
 	FileExists(ctx context.Context, name string) (bool, error)
+	// DeleteFile delete the file in storage
+	DeleteFile(ctx context.Context, name string) error
 	// Open a Reader by file path. path is relative path to storage base path
 	Open(ctx context.Context, path string) (ExternalFileReader, error)
 	// WalkDir traverse all the files in a dir.