support CTYun OOS (#87)

* support CTYun OOS

* change loglevl

Co-authored-by: liyong <chnliyong@gmail.com>

Author: davies

README.md

@@ -100,6 +100,7 @@ SRC and DST must be an URI of the following object storage:
 - b2: Backblaze B2
 - space: Digital Ocean Space
 - obs: Huawei Object Storage Service
+- oos: CTYun OOS
 
 SRC and DST should be in the following format:
 

object/object_storage_test.go

@@ -362,3 +362,12 @@ func TestHDFS(t *testing.T) {
 	dfs := newHDFS(os.Getenv("HDFS_ADDR"), "", "")
 	testStorage(t, dfs)
 }
+
+func TestOOS(t *testing.T) {
+	if os.Getenv("OOS_ACCESS_KEY") == "" {
+		t.SkipNow()
+	}
+	b := newOOS(fmt.Sprintf("https://%s", os.Getenv("OOS_TEST_BUCKET")),
+		os.Getenv("OOS_ACCESS_KEY"), os.Getenv("OOS_SECRET_KEY"))
+	testStorage(t, b)
+}

object/oos.go

@@ -0,0 +1,69 @@
+// Copyright (C) 2018-present Juicedata Inc.
+
+package object
+
+import (
+	"fmt"
+	"net/url"
+	"strings"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/s3"
+)
+
+type oos struct {
+	s3client
+}
+
+func (s *oos) String() string {
+	return fmt.Sprintf("oos://%s", s.s3client.bucket)
+}
+
+func (s *oos) Create() error {
+	_, err := s.List("", "", 1)
+	if err != nil {
+		return fmt.Errorf("please create bucket %s manually", s.s3client.bucket)
+	}
+	return err
+}
+
+func (s *oos) List(prefix, marker string, limit int64) ([]*Object, error) {
+	if limit > 1000 {
+		limit = 1000
+	}
+	objs, err := s.s3client.List(prefix, marker, limit)
+	if marker != "" && len(objs) > 0 && objs[0].Key == marker {
+		objs = objs[1:]
+	}
+	return objs, err
+}
+
+func newOOS(endpoint, accessKey, secretKey string) ObjectStorage {
+	uri, err := url.ParseRequestURI(endpoint)
+	if err != nil {
+		logger.Fatalf("Invalid endpoint %s: %s", endpoint, err)
+	}
+	ssl := strings.ToLower(uri.Scheme) == "https"
+	hostParts := strings.Split(uri.Host, ".")
+	bucket := hostParts[0]
+	region := hostParts[1][4:]
+	endpoint = uri.Host[len(bucket)+1:]
+
+	awsConfig := &aws.Config{
+		Region:           &region,
+		Endpoint:         &endpoint,
+		DisableSSL:       aws.Bool(!ssl),
+		S3ForcePathStyle: aws.Bool(true),
+		// HTTPClient:       httpClient,
+		Credentials:      credentials.NewStaticCredentials(accessKey, secretKey, ""),
+	}
+
+	ses := session.New(awsConfig)
+	return &oos{s3client{bucket, s3.New(ses), ses}}
+}
+
+func init() {
+	register("oos", newOOS)
+}