Fix s3 backend bug (#4)

Fix bug: pass the region when create s3 session.
Fix bug: when the backup root dir is the s3 root, s3 list will not get the right file name.
Improve: add more readable storage type and add more log.

pkg/proto/util.go

@@ -21,6 +21,21 @@ const (
 	S3Type
 )
 
+func (t BackendType) String() string {
+	switch t {
+	case LocalType:
+		return "Local"
+	case HdfsType:
+		return "HDFS"
+	case S3Type:
+		return "S3-compatible"
+	case InvalidType:
+		return "Invalid"
+	default:
+		return "Unknown"
+	}
+}
+
 func ParseType(uri string) BackendType {
 	if strings.HasPrefix(uri, LocalPrefix) {
 		return LocalType

pkg/storage/s3.go

@@ -30,13 +30,23 @@ func NewS3(b *pb.Backend) (*S3, error) {
 
 	creds := credentials.NewStaticCredentials(b.GetS3().AccessKey, b.GetS3().SecretKey, "")
 	forcePath := strings.ContainsAny(b.GetS3().GetEndpoint(), ":")
+	region := "default"
+	if b.GetS3().GetRegion() != "" {
+		region = b.GetS3().GetRegion()
+	}
+
 	sess := session.Must(session.NewSession(&aws.Config{
-		Region:           aws.String("default"),
+		Region:           aws.String(region),
 		Endpoint:         aws.String(b.GetS3().GetEndpoint()),
 		S3ForcePathStyle: aws.Bool(forcePath), // ip:port
 		Credentials:      creds,
 	}))
 
+	log.WithField("region", region).
+		WithField("endpoint", b.GetS3().GetEndpoint()).
+		WithField("forcePath", forcePath).
+		Debugf("Try to create s3 backend")
+
 	return &S3{
 		backend: b,
 		sess:    sess,
@@ -266,10 +276,15 @@ func (s *S3) ListDir(ctx context.Context, uri string) ([]string, error) {
 	names := make([]string, 0)
 	s.client.ListObjectsV2Pages(req, func(p *s3.ListObjectsV2Output, lastPage bool) bool {
 		for _, obj := range p.CommonPrefixes {
-			name, err := filepath.Rel(prefix, *obj.Prefix)
-			if err != nil {
-				log.WithError(err).WithField("key", *obj.Prefix).WithField("prefix", prefix).Error("Get relative path failed")
-				return false
+			var name string
+			if prefix == "/" { // special case: backup root is the bucket root
+				name = *obj.Prefix
+			} else {
+				name, err = filepath.Rel(prefix, *obj.Prefix)
+				if err != nil {
+					log.WithError(err).WithField("key", *obj.Prefix).WithField("prefix", prefix).Error("Get relative path failed")
+					return false
+				}
 			}
 			names = append(names, name)
 		}

pkg/storage/storage.go

@@ -47,7 +47,7 @@ type ExternalStorage interface {
 }
 
 func New(b *pb.Backend) (ExternalStorage, error) {
-	log.WithField("uri", b.Uri()).Debugf("Create %v stoarge", b.Type())
+	log.WithField("uri", b.Uri()).Debugf("Create type: %s stoarge", b.Type())
 
 	switch b.Type() {
 	case pb.LocalType: