Repository: dragonflyoss/Dragonfly2
Pull request #1366 (2022-06-08 10:30:19)

feat: add objectstorage pkg
<!--- Provide a general summary of your changes in the Title above -->

## Description
- Add objectstorage pkg.
- Add object storage config to manager.
- Refactor manager config validation.
- Refactor scheduler config validation.
<!--- Describe your changes in detail -->

## Related Issue

#1249 
<!--- This project only accepts pull requests related to open issues -->
<!--- If suggesting a new feature or change, please discuss it in an issue first -->
<!--- If fixing a bug, there should be an issue describing it with steps to reproduce -->
<!--- Please link to the issue here: -->

## Motivation and Context

<!--- Why is this change required? What problem does it solve? -->

## Screenshots (if appropriate)

## Types of changes

<!--- What types of changes does your code introduce? Put an `x` in all the boxes that apply: -->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)
- [ ] Documentation Update (if none of the other choices apply)

## Checklist

<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->

- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [x] I have read the **CONTRIBUTING** document.
- [ ] I have added tests to cover my changes.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1366?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1366](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1366?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (01cf6dc) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/ebc0ea537b965227dd3da2a346d074e563491fce?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (ebc0ea5) will **decrease** coverage by `0.08%`.
> The diff coverage is `0.00%`.

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1366/graphs/tree.svg?width=650&height=150&src=pr&token=PUICJYSAT7&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1366?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff             @@
##             main    #1366      +/-   ##
==========================================
- Coverage   45.79%   45.70%   -0.09%     
==========================================
  Files         140      140              
  Lines       11984    12000      +16     
==========================================
- Hits         5488     5485       -3     
- Misses       5873     5892      +19     
  Partials      623      623              
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| e2etests | `?` | |
| unittests | `45.70% <0.00%> (-0.09%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1366?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [manager/config/config.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1366/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-bWFuYWdlci9jb25maWcvY29uZmlnLmdv) | `0.00% <0.00%> (ø)` | |
| [scheduler/config/config.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1366/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL2NvbmZpZy9jb25maWcuZ28=) | `47.41% <0.00%> (ø)` | |
| [client/daemon/peer/peertask\_conductor.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1366/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2NvbmR1Y3Rvci5nbw==) | `58.14% <0.00%> (-0.31%)` | :arrow_down: |


Patch:
diff --git a/go.mod b/go.mod
index 64f2b44f..0a3e24ac 100644
--- a/go.mod
+++ b/go.mod
@@ -9,6 +9,7 @@ require (
 	github.com/agiledragon/gomonkey/v2 v2.3.1
 	github.com/aliyun/aliyun-oss-go-sdk v2.2.4+incompatible
 	github.com/appleboy/gin-jwt/v2 v2.8.0
+	github.com/aws/aws-sdk-go v1.44.24
 	github.com/bits-and-blooms/bitset v1.2.2
 	github.com/casbin/casbin/v2 v2.47.1
 	github.com/casbin/gorm-adapter/v3 v3.3.2
@@ -94,7 +95,6 @@ require (
 	github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible // indirect
 	github.com/KyleBanks/depth v1.2.1 // indirect
 	github.com/RichardKnop/logging v0.0.0-20190827224416-1a693bdd4fae // indirect
-	github.com/aws/aws-sdk-go v1.44.24 // indirect
 	github.com/baiyubin/aliyun-sts-go-sdk v0.0.0-20180326062324-cfa1a18b161f // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bradfitz/gomemcache v0.0.0-20220106215444-fb4bf637b56d // indirect
diff --git a/manager/config/config.go b/manager/config/config.go
index 664595b8..62dc939b 100644
--- a/manager/config/config.go
+++ b/manager/config/config.go
@@ -24,6 +24,7 @@ import (
 	"github.com/docker/go-connections/tlsconfig"
 
 	"d7y.io/dragonfly/v2/cmd/dependency/base"
+	"d7y.io/dragonfly/v2/pkg/objectstorage"
 )
 
 type Config struct {
@@ -39,8 +40,11 @@ type Config struct {
 	// Cache configuration
 	Cache *CacheConfig `yaml:"cache" mapstructure:"cache"`
 
+	// ObjectStorage configuration
+	ObjectStorage *ObjectStorageConfig `yaml:"objectStorage" mapstructure:"objectStorage"`
+
 	// Metrics configuration
-	Metrics *RestConfig `yaml:"metrics" mapstructure:"metrics"`
+	Metrics *MetricsConfig `yaml:"metrics" mapstructure:"metrics"`
 }
 
 type ServerConfig struct {
@@ -162,6 +166,14 @@ type RestConfig struct {
 	Addr string `yaml:"addr" mapstructure:"addr"`
 }
 
+type MetricsConfig struct {
+	// Enable metrics service.
+	Enable bool `yaml:"enable" mapstructure:"enable"`
+
+	// Metrics service address.
+	Addr string `yaml:"addr" mapstructure:"addr"`
+}
+
 type TCPListenConfig struct {
 	// Listen stands listen interface, like: 0.0.0.0, 192.168.0.1
 	Listen string `mapstructure:"listen" yaml:"listen"`
@@ -175,6 +187,26 @@ type TCPListenPortRange struct {
 	End   int
 }
 
+type ObjectStorageConfig struct {
+	// Enable object storage.
+	Enable bool `yaml:"enable" mapstructure:"enable"`
+
+	// Object storage name of type, it can be s3 or oss.
+	Name string `mapstructure:"name" yaml:"name"`
+
+	// Storage region.
+	Region string `mapstructure:"region" yaml:"region"`
+
+	// Datacenter endpoint.
+	Endpoint string `mapstructure:"endpoint" yaml:"endpoint"`
+
+	// Access key ID.
+	AccessKey string `mapstructure:"accessKey" yaml:"accessKey"`
+
+	// Access key secret
+	SecretKey string `mapstructure:"secretKey" yaml:"secretKey"`
+}
+
 // New config instance
 func New() *Config {
 	return &Config{
@@ -210,119 +242,145 @@ func New() *Config {
 				TTL:  30 * time.Second,
 			},
 		},
+		ObjectStorage: &ObjectStorageConfig{
+			Enable: false,
+		},
+		Metrics: &MetricsConfig{
+			Enable: false,
+		},
 	}
 }
 
 // Validate config values
 func (cfg *Config) Validate() error {
 	if cfg.Server == nil {
-		return errors.New("empty server config is not specified")
+		return errors.New("config requires parameter server")
 	}
 
 	if cfg.Server.Name == "" {
-		return errors.New("empty server name config is not specified")
+		return errors.New("server requires parameter name")
 	}
 
 	if cfg.Server.GRPC == nil {
-		return errors.New("empty grpc server config is not specified")
+		return errors.New("server requires parameter grpc")
 	}
 
 	if cfg.Server.REST == nil {
-		return errors.New("empty rest server config is not specified")
+		return errors.New("server requires parameter rest")
 	}
 
 	if cfg.Database == nil {
-		return errors.New("empty database config is not specified")
+		return errors.New("config requires parameter database")
 	}
 
 	if cfg.Database.Redis == nil {
-		return errors.New("empty database redis config is not specified")
+		return errors.New("database requires parameter redis")
 	}
 
 	if cfg.Database.Redis.Host == "" {
-		return errors.New("empty database redis host is not specified")
+		return errors.New("redis requires parameter host")
 	}
 
 	if cfg.Database.Redis.Port <= 0 {
-		return errors.New("empty database redis port is not specified")
+		return errors.New("redis requires parameter port")
 	}
 
 	if cfg.Database.Redis.CacheDB < 0 {
-		return errors.New("empty database redis cacheDB is not specified")
+		return errors.New("redis requires parameter cacheDB")
 	}
 
 	if cfg.Database.Redis.BrokerDB < 0 {
-		return errors.New("empty database redis brokerDB is not specified")
+		return errors.New("redis requires parameter brokerDB")
 	}
 
 	if cfg.Database.Redis.BackendDB < 0 {
-		return errors.New("empty database redis backendDB is not specified")
+		return errors.New("redis requires parameter backendDB")
 	}
 
 	if cfg.Database.Mysql == nil {
-		return errors.New("empty database mysql config is not specified")
+		return errors.New("database requires parameter mysql")
 	}
 
 	if cfg.Database.Mysql.User == "" {
-		return errors.New("empty database mysql user is not specified")
+		return errors.New("mysql requires parameter user")
 	}
 
 	if cfg.Database.Mysql.Password == "" {
-		return errors.New("empty database mysql password is not specified")
+		return errors.New("mysql requires parameter password")
 	}
 
 	if cfg.Database.Mysql.Host == "" {
-		return errors.New("empty database mysql host is not specified")
+		return errors.New("mysql requires parameter host")
 	}
 
 	if cfg.Database.Mysql.Port <= 0 {
-		return errors.New("empty database mysql port is not specified")
+		return errors.New("mysql requires parameter port")
 	}
 
 	if cfg.Database.Mysql.DBName == "" {
-		return errors.New("empty database mysql dbName is not specified")
+		return errors.New("mysql requires parameter dbname")
 	}
 
 	if cfg.Database.Mysql.TLS != nil {
 		if cfg.Database.Mysql.TLS.Cert == "" {
-			return errors.New("empty database mysql tls cert is not specified")
+			return errors.New("tls requires parameter cert")
 		}
 
 		if cfg.Database.Mysql.TLS.Key == "" {
-			return errors.New("empty database mysql tls key is not specified")
+			return errors.New("tls requires parameter key")
 		}
 
 		if cfg.Database.Mysql.TLS.CA == "" {
-			return errors.New("empty database mysql tls ca is not specified")
+			return errors.New("tls requires parameter ca")
 		}
 	}
 
 	if cfg.Cache == nil {
-		return errors.New("empty cache config is not specified")
+		return errors.New("config requires parameter cache")
 	}
 
 	if cfg.Cache.Redis == nil {
-		return errors.New("empty cache redis config is not specified")
+		return errors.New("cache requires parameter redis")
 	}
 
 	if cfg.Cache.Redis.TTL == 0 {
-		return errors.New("empty cache redis TTL is not specified")
+		return errors.New("redis requires parameter ttl")
 	}
 
 	if cfg.Cache.Local == nil {
-		return errors.New("empty cache local config is not specified")
+		return errors.New("cache requires parameter local")
 	}
 
 	if cfg.Cache.Local.Size == 0 {
-		return errors.New("empty cache local size is not specified")
+		return errors.New("local requires parameter size")
 	}
 
 	if cfg.Cache.Local.TTL == 0 {
-		return errors.New("empty cache local TTL is not specified")
+		return errors.New("local requires parameter ttl")
+	}
+
+	if cfg.ObjectStorage != nil && cfg.ObjectStorage.Enable {
+		if cfg.ObjectStorage.Name == "" {
+			return errors.New("objectStorage requires parameter name")
+		}
+
+		if cfg.ObjectStorage.Name != objectstorage.ServiceNameS3 && cfg.ObjectStorage.Name != objectstorage.ServiceNameOSS {
+			return errors.New("objectStorage requires parameter name")
+		}
+
+		if cfg.ObjectStorage.AccessKey == "" {
+			return errors.New("objectStorage requires parameter accessKey")
+		}
+
+		if cfg.ObjectStorage.SecretKey == "" {
+			return errors.New("objectStorage requires parameter secretKey")
+		}
 	}
 
-	if cfg.Metrics != nil && cfg.Metrics.Addr == "" {
-		return errors.New("empty metrics addr is not specified")
+	if cfg.Metrics != nil && cfg.Metrics.Enable {
+		if cfg.Metrics.Addr == "" {
+			return errors.New("metrics requires parameter addr")
+		}
 	}
 
 	return nil
diff --git a/manager/manager.go b/manager/manager.go
index 9bd1fe5f..32599d1c 100644
--- a/manager/manager.go
+++ b/manager/manager.go
@@ -118,7 +118,7 @@ func New(cfg *config.Config, d dfpath.Dfpath) (*Server, error) {
 	s.grpcServer = grpcServer
 
 	// Initialize prometheus
-	if cfg.Metrics != nil {
+	if cfg.Metrics.Enable {
 		s.metricsServer = metrics.New(cfg.Metrics, grpcServer)
 	}
 
diff --git a/manager/metrics/metrics.go b/manager/metrics/metrics.go
index 4512539e..cd79d172 100644
--- a/manager/metrics/metrics.go
+++ b/manager/metrics/metrics.go
@@ -29,7 +29,7 @@ import (
 	"d7y.io/dragonfly/v2/manager/config"
 )
 
-func New(cfg *config.RestConfig, grpcServer *grpc.Server) *http.Server {
+func New(cfg *config.MetricsConfig, grpcServer *grpc.Server) *http.Server {
 	grpc_prometheus.Register(grpcServer)
 
 	mux := http.NewServeMux()
diff --git a/pkg/objectstorage/constants.go b/pkg/objectstorage/constants.go
new file mode 100644
index 00000000..2e11f2bd
--- /dev/null
+++ b/pkg/objectstorage/constants.go
@@ -0,0 +1,25 @@
+/*
+ *     Copyright 2022 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package objectstorage
+
+const (
+	// ServiceNameS3 is name of s3 storage
+	ServiceNameS3 = "s3"
+
+	// ServiceNameOSS is name of oss storage
+	ServiceNameOSS = "oss"
+)
diff --git a/pkg/objectstorage/mocks/objectstorage_mock.go b/pkg/objectstorage/mocks/objectstorage_mock.go
new file mode 100644
index 00000000..a8805b7d
--- /dev/null
+++ b/pkg/objectstorage/mocks/objectstorage_mock.go
@@ -0,0 +1,167 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: objectstorage.go
+
+// Package mocks is a generated GoMock package.
+package mocks
+
+import (
+	io "io"
+	reflect "reflect"
+
+	objectstorage "d7y.io/dragonfly/v2/pkg/objectstorage"
+	gomock "github.com/golang/mock/gomock"
+)
+
+// MockObjectStorage is a mock of ObjectStorage interface.
+type MockObjectStorage struct {
+	ctrl     *gomock.Controller
+	recorder *MockObjectStorageMockRecorder
+}
+
+// MockObjectStorageMockRecorder is the mock recorder for MockObjectStorage.
+type MockObjectStorageMockRecorder struct {
+	mock *MockObjectStorage
+}
+
+// NewMockObjectStorage creates a new mock instance.
+func NewMockObjectStorage(ctrl *gomock.Controller) *MockObjectStorage {
+	mock := &MockObjectStorage{ctrl: ctrl}
+	mock.recorder = &MockObjectStorageMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockObjectStorage) EXPECT() *MockObjectStorageMockRecorder {
+	return m.recorder
+}
+
+// CreateBucket mocks base method.
+func (m *MockObjectStorage) CreateBucket(bucketName string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CreateBucket", bucketName)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// CreateBucket indicates an expected call of CreateBucket.
+func (mr *MockObjectStorageMockRecorder) CreateBucket(bucketName interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateBucket", reflect.TypeOf((*MockObjectStorage)(nil).CreateBucket), bucketName)
+}
+
+// CreateObject mocks base method.
+func (m *MockObjectStorage) CreateObject(bucketName, objectKey string, reader io.Reader) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CreateObject", bucketName, objectKey, reader)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// CreateObject indicates an expected call of CreateObject.
+func (mr *MockObjectStorageMockRecorder) CreateObject(bucketName, objectKey, reader interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateObject", reflect.TypeOf((*MockObjectStorage)(nil).CreateObject), bucketName, objectKey, reader)
+}
+
+// DeleteBucket mocks base method.
+func (m *MockObjectStorage) DeleteBucket(bucketName string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "DeleteBucket", bucketName)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// DeleteBucket indicates an expected call of DeleteBucket.
+func (mr *MockObjectStorageMockRecorder) DeleteBucket(bucketName interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteBucket", reflect.TypeOf((*MockObjectStorage)(nil).DeleteBucket), bucketName)
+}
+
+// DeleteObject mocks base method.
+func (m *MockObjectStorage) DeleteObject(bucketName, objectKey string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "DeleteObject", bucketName, objectKey)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// DeleteObject indicates an expected call of DeleteObject.
+func (mr *MockObjectStorageMockRecorder) DeleteObject(bucketName, objectKey interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteObject", reflect.TypeOf((*MockObjectStorage)(nil).DeleteObject), bucketName, objectKey)
+}
+
+// GetBucketMetadata mocks base method.
+func (m *MockObjectStorage) GetBucketMetadata(bucketName string) (*objectstorage.BucketMetadata, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetBucketMetadata", bucketName)
+	ret0, _ := ret[0].(*objectstorage.BucketMetadata)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetBucketMetadata indicates an expected call of GetBucketMetadata.
+func (mr *MockObjectStorageMockRecorder) GetBucketMetadata(bucketName interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetBucketMetadata", reflect.TypeOf((*MockObjectStorage)(nil).GetBucketMetadata), bucketName)
+}
+
+// GetObjectMetadata mocks base method.
+func (m *MockObjectStorage) GetObjectMetadata(bucketName, objectKey string) (*objectstorage.ObjectMetadata, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetObjectMetadata", bucketName, objectKey)
+	ret0, _ := ret[0].(*objectstorage.ObjectMetadata)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetObjectMetadata indicates an expected call of GetObjectMetadata.
+func (mr *MockObjectStorageMockRecorder) GetObjectMetadata(bucketName, objectKey interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetObjectMetadata", reflect.TypeOf((*MockObjectStorage)(nil).GetObjectMetadata), bucketName, objectKey)
+}
+
+// GetOject mocks base method.
+func (m *MockObjectStorage) GetOject(bucketName, objectKey string) (io.ReadCloser, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetOject", bucketName, objectKey)
+	ret0, _ := ret[0].(io.ReadCloser)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetOject indicates an expected call of GetOject.
+func (mr *MockObjectStorageMockRecorder) GetOject(bucketName, objectKey interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetOject", reflect.TypeOf((*MockObjectStorage)(nil).GetOject), bucketName, objectKey)
+}
+
+// ListBucketMetadatas mocks base method.
+func (m *MockObjectStorage) ListBucketMetadatas() ([]*objectstorage.BucketMetadata, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ListBucketMetadatas")
+	ret0, _ := ret[0].([]*objectstorage.BucketMetadata)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ListBucketMetadatas indicates an expected call of ListBucketMetadatas.
+func (mr *MockObjectStorageMockRecorder) ListBucketMetadatas() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListBucketMetadatas", reflect.TypeOf((*MockObjectStorage)(nil).ListBucketMetadatas))
+}
+
+// ListObjectMetadatas mocks base method.
+func (m *MockObjectStorage) ListObjectMetadatas(bucketName, prefix, marker string, limit int64) ([]*objectstorage.ObjectMetadata, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ListObjectMetadatas", bucketName, prefix, marker, limit)
+	ret0, _ := ret[0].([]*objectstorage.ObjectMetadata)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ListObjectMetadatas indicates an expected call of ListObjectMetadatas.
+func (mr *MockObjectStorageMockRecorder) ListObjectMetadatas(bucketName, prefix, marker, limit interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListObjectMetadatas", reflect.TypeOf((*MockObjectStorage)(nil).ListObjectMetadatas), bucketName, prefix, marker, limit)
+}
diff --git a/pkg/objectstorage/objectstorage.go b/pkg/objectstorage/objectstorage.go
new file mode 100644
index 00000000..ceab6c5c
--- /dev/null
+++ b/pkg/objectstorage/objectstorage.go
@@ -0,0 +1,97 @@
+/*
+ *     Copyright 2022 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+//go:generate mockgen -package mocks -source objectstorage.go -destination ./mocks/objectstorage_mock.go
+
+package objectstorage
+
+import (
+	"fmt"
+	"io"
+	"time"
+)
+
+type ObjectMetadata struct {
+	// Key is object key
+	Key string
+
+	// ContentDisposition is Content-Disposition header
+	ContentDisposition string
+
+	// ContentEncoding is Content-Encoding header
+	ContentEncoding string
+
+	// ContentLanguage is Content-Language header
+	ContentLanguage string
+
+	// ContentLanguage is Content-Length header
+	ContentLength int64
+
+	// ContentType is Content-Type header
+	ContentType string
+
+	// Etag is Etag header
+	Etag string
+}
+
+type BucketMetadata struct {
+	// Name is bucket name
+	Name string
+
+	// CreateAt is bucket create time.
+	CreateAt time.Time
+}
+
+type ObjectStorage interface {
+	// GetBucketMetadata returns metadata of bucket.
+	GetBucketMetadata(bucketName string) (*BucketMetadata, error)
+
+	// CreateBucket creates bucket of object storage.
+	CreateBucket(bucketName string) error
+
+	// DeleteBucket deletes bucket of object storage.
+	DeleteBucket(bucketName string) error
+
+	// ListBucketMetadatas returns metadata of buckets.
+	ListBucketMetadatas() ([]*BucketMetadata, error)
+
+	// GetObjectMetadata returns metadata of object.
+	GetObjectMetadata(bucketName, objectKey string) (*ObjectMetadata, error)
+
+	// GetOject returns data of object.
+	GetOject(bucketName, objectKey string) (io.ReadCloser, error)
+
+	// CreateObject creates data of object.
+	CreateObject(bucketName, objectKey string, reader io.Reader) error
+
+	// DeleteObject deletes data of object.
+	DeleteObject(bucketName, objectKey string) error
+
+	// ListObjectMetadatas returns metadata of objects.
+	ListObjectMetadatas(bucketName, prefix, marker string, limit int64) ([]*ObjectMetadata, error)
+}
+
+// New object storage interface.
+func New(name, region, endpoint, accessKey, secretKey string) (ObjectStorage, error) {
+	switch name {
+	case ServiceNameS3:
+		return newS3(region, endpoint, accessKey, secretKey)
+	case ServiceNameOSS:
+		return newOSS(region, endpoint, accessKey, secretKey)
+	}
+
+	return nil, fmt.Errorf("unknow service name %s", name)
+}
diff --git a/pkg/objectstorage/oss.go b/pkg/objectstorage/oss.go
new file mode 100644
index 00000000..eb18b669
--- /dev/null
+++ b/pkg/objectstorage/oss.go
@@ -0,0 +1,165 @@
+/*
+ *     Copyright 2022 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package objectstorage
+
+import (
+	"fmt"
+	"io"
+	"strconv"
+
+	aliyunoss "github.com/aliyun/aliyun-oss-go-sdk/oss"
+	"github.com/go-http-utils/headers"
+)
+
+type oss struct {
+	// OSS client.
+	client *aliyunoss.Client
+}
+
+// New oss instance.
+func newOSS(region, endpoint, accessKey, secretKey string) (ObjectStorage, error) {
+	client, err := aliyunoss.New(endpoint, accessKey, secretKey, aliyunoss.Region(region))
+	if err != nil {
+		return nil, fmt.Errorf("new oss client failed: %s", err)
+	}
+
+	return &oss{
+		client: client,
+	}, nil
+}
+
+// GetBucketMetadata returns metadata of bucket.
+func (o *oss) GetBucketMetadata(bucketName string) (*BucketMetadata, error) {
+	resp, err := o.client.GetBucketInfo(bucketName)
+	if err != nil {
+		return nil, err
+	}
+
+	return &BucketMetadata{
+		Name:     resp.BucketInfo.Name,
+		CreateAt: resp.BucketInfo.CreationDate,
+	}, nil
+}
+
+// GetBucketMetadata returns metadata of bucket.
+func (o *oss) CreateBucket(bucketName string) error {
+	return o.client.CreateBucket(bucketName)
+}
+
+// DeleteBucket deletes bucket of object storage.
+func (o *oss) DeleteBucket(bucketName string) error {
+	return o.client.DeleteBucket(bucketName)
+}
+
+// DeleteBucket deletes bucket of object storage.
+func (o *oss) ListBucketMetadatas() ([]*BucketMetadata, error) {
+	resp, err := o.client.ListBuckets()
+	if err != nil {
+		return nil, err
+	}
+
+	var metadatas []*BucketMetadata
+	for _, bucket := range resp.Buckets {
+		metadatas = append(metadatas, &BucketMetadata{
+			Name:     bucket.Name,
+			CreateAt: bucket.CreationDate,
+		})
+	}
+
+	return metadatas, nil
+}
+
+// GetObjectMetadata returns metadata of object.
+func (o *oss) GetObjectMetadata(bucketName, objectKey string) (*ObjectMetadata, error) {
+	bucket, err := o.client.Bucket(bucketName)
+	if err != nil {
+		return nil, err
+	}
+
+	header, err := bucket.GetObjectMeta(objectKey)
+	if err != nil {
+		return nil, err
+	}
+
+	contentLength, err := strconv.ParseInt(header.Get(headers.ContentLength), 10, 64)
+	if err != nil {
+		return nil, err
+	}
+
+	return &ObjectMetadata{
+		Key:                objectKey,
+		ContentDisposition: header.Get(headers.ContentDisposition),
+		ContentEncoding:    header.Get(headers.ContentEncoding),
+		ContentLanguage:    header.Get(headers.ContentLanguage),
+		ContentLength:      contentLength,
+		ContentType:        header.Get(headers.ContentType),
+		Etag:               header.Get(headers.ETag),
+	}, nil
+}
+
+// GetOject returns data of object.
+func (o *oss) GetOject(bucketName, objectKey string) (io.ReadCloser, error) {
+	bucket, err := o.client.Bucket(bucketName)
+	if err != nil {
+		return nil, err
+	}
+
+	return bucket.GetObject(bucketName)
+}
+
+// CreateObject creates data of object.
+func (o *oss) CreateObject(bucketName, objectKey string, reader io.Reader) error {
+	bucket, err := o.client.Bucket(bucketName)
+	if err != nil {
+		return err
+	}
+
+	return bucket.PutObject(objectKey, reader)
+}
+
+// DeleteObject deletes data of object.
+func (o *oss) DeleteObject(bucketName, objectKey string) error {
+	bucket, err := o.client.Bucket(bucketName)
+	if err != nil {
+		return err
+	}
+
+	return bucket.DeleteObject(objectKey)
+}
+
+// ListObjectMetadatas returns metadata of objects.
+func (o *oss) ListObjectMetadatas(bucketName, prefix, marker string, limit int64) ([]*ObjectMetadata, error) {
+	bucket, err := o.client.Bucket(bucketName)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := bucket.ListObjects(aliyunoss.Prefix(prefix), aliyunoss.Marker(marker), aliyunoss.MaxKeys(int(limit)))
+	if err != nil {
+		return nil, err
+	}
+
+	var metadatas []*ObjectMetadata
+	for _, object := range resp.Objects {
+		metadatas = append(metadatas, &ObjectMetadata{
+			Key:  object.Key,
+			Etag: object.ETag,
+		})
+	}
+
+	return metadatas, nil
+}
diff --git a/pkg/objectstorage/s3.go b/pkg/objectstorage/s3.go
new file mode 100644
index 00000000..2d57b175
--- /dev/null
+++ b/pkg/objectstorage/s3.go
@@ -0,0 +1,165 @@
+/*
+ *     Copyright 2022 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package objectstorage
+
+import (
+	"fmt"
+	"io"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/session"
+	awss3 "github.com/aws/aws-sdk-go/service/s3"
+)
+
+type s3 struct {
+	// S3 client.
+	client *awss3.S3
+}
+
+// New s3 instance.
+func newS3(region, endpoint, accessKey, secretKey string) (ObjectStorage, error) {
+	cfg := aws.NewConfig().WithCredentials(credentials.NewStaticCredentials(accessKey, secretKey, ""))
+	s, err := session.NewSession(cfg)
+	if err != nil {
+		return nil, fmt.Errorf("new aws session failed: %s", err)
+	}
+
+	return &s3{
+		client: awss3.New(s, cfg.WithRegion(region), cfg.WithEndpoint(endpoint)),
+	}, nil
+}
+
+// GetBucketMetadata returns metadata of bucket.
+func (s *s3) GetBucketMetadata(bucketName string) (*BucketMetadata, error) {
+	_, err := s.client.HeadBucket(&awss3.HeadBucketInput{Bucket: aws.String(bucketName)})
+	if err != nil {
+		return nil, err
+	}
+
+	return &BucketMetadata{
+		Name: bucketName,
+	}, nil
+}
+
+// CreateBucket creates bucket of object storage.
+func (s *s3) CreateBucket(bucketName string) error {
+	_, err := s.client.CreateBucket(&awss3.CreateBucketInput{Bucket: aws.String(bucketName)})
+	return err
+}
+
+// DeleteBucket deletes bucket of object storage.
+func (s *s3) DeleteBucket(bucketName string) error {
+	_, err := s.client.DeleteBucket(&awss3.DeleteBucketInput{Bucket: aws.String(bucketName)})
+	return err
+}
+
+// DeleteBucket deletes bucket of object storage.
+func (s *s3) ListBucketMetadatas() ([]*BucketMetadata, error) {
+	resp, err := s.client.ListBuckets(&awss3.ListBucketsInput{})
+	if err != nil {
+		return nil, err
+	}
+
+	var metadatas []*BucketMetadata
+	for _, bucket := range resp.Buckets {
+		metadatas = append(metadatas, &BucketMetadata{
+			Name:     aws.StringValue(bucket.Name),
+			CreateAt: aws.TimeValue(bucket.CreationDate),
+		})
+	}
+
+	return metadatas, nil
+}
+
+// GetObjectMetadata returns metadata of object.
+func (s *s3) GetObjectMetadata(bucketName, objectKey string) (*ObjectMetadata, error) {
+	resp, err := s.client.HeadObject(&awss3.HeadObjectInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &ObjectMetadata{
+		Key:                objectKey,
+		ContentDisposition: aws.StringValue(resp.ContentDisposition),
+		ContentEncoding:    aws.StringValue(resp.ContentEncoding),
+		ContentLanguage:    aws.StringValue(resp.ContentLanguage),
+		ContentLength:      aws.Int64Value(resp.ContentLength),
+		ContentType:        aws.StringValue(resp.ContentType),
+		Etag:               aws.StringValue(resp.ETag),
+	}, nil
+}
+
+// GetOject returns data of object.
+func (s *s3) GetOject(bucketName, objectKey string) (io.ReadCloser, error) {
+	resp, err := s.client.GetObject(&awss3.GetObjectInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return resp.Body, nil
+}
+
+// CreateObject creates data of object.
+func (s *s3) CreateObject(bucketName, objectKey string, reader io.Reader) error {
+	_, err := s.client.PutObject(&awss3.PutObjectInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+		Body:   aws.ReadSeekCloser(reader),
+	})
+
+	return err
+}
+
+// DeleteObject deletes data of object.
+func (s *s3) DeleteObject(bucketName, objectKey string) error {
+	_, err := s.client.DeleteObject(&awss3.DeleteObjectInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(objectKey),
+	})
+
+	return err
+}
+
+// DeleteObject deletes data of object.
+func (s *s3) ListObjectMetadatas(bucketName, prefix, marker string, limit int64) ([]*ObjectMetadata, error) {
+	resp, err := s.client.ListObjects(&awss3.ListObjectsInput{
+		Bucket:  aws.String(bucketName),
+		Prefix:  aws.String(prefix),
+		Marker:  aws.String(marker),
+		MaxKeys: aws.Int64(limit),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	var metadatas []*ObjectMetadata
+	for _, object := range resp.Contents {
+		metadatas = append(metadatas, &ObjectMetadata{
+			Key:  aws.StringValue(object.Key),
+			Etag: aws.StringValue(object.ETag),
+		})
+	}
+
+	return metadatas, nil
+}
diff --git a/scheduler/config/config.go b/scheduler/config/config.go
index 58071559..2bfd8e22 100644
--- a/scheduler/config/config.go
+++ b/scheduler/config/config.go
@@ -120,123 +120,123 @@ func New() *Config {
 }
 
 // Validate config parameters.
-func (c *Config) Validate() error {
-	if c.Server == nil {
+func (cfg *Config) Validate() error {
+	if cfg.Server == nil {
 		return errors.New("server requires parameter server")
 	}
 
-	if c.Server.IP == "" {
+	if cfg.Server.IP == "" {
 		return errors.New("server requires parameter ip")
 	}
 
-	if c.Server.Host == "" {
+	if cfg.Server.Host == "" {
 		return errors.New("server requires parameter host")
 	}
 
-	if c.Server.Port <= 0 {
+	if cfg.Server.Port <= 0 {
 		return errors.New("server requires parameter port")
 	}
 
-	if c.Server.Listen == "" {
+	if cfg.Server.Listen == "" {
 		return errors.New("server requires parameter listen")
 	}
 
-	if c.Scheduler.Algorithm == "" {
+	if cfg.Scheduler.Algorithm == "" {
 		return errors.New("scheduler requires parameter algorithm")
 	}
 
-	if c.Scheduler.RetryLimit <= 0 {
+	if cfg.Scheduler.RetryLimit <= 0 {
 		return errors.New("scheduler requires parameter retryLimit")
 	}
 
-	if c.Scheduler.RetryInterval <= 0 {
+	if cfg.Scheduler.RetryInterval <= 0 {
 		return errors.New("scheduler requires parameter retryInterval")
 	}
 
-	if c.Scheduler.GC == nil {
+	if cfg.Scheduler.GC == nil {
 		return errors.New("scheduler requires parameter gc")
 	}
 
-	if c.Scheduler.GC.PeerGCInterval <= 0 {
+	if cfg.Scheduler.GC.PeerGCInterval <= 0 {
 		return errors.New("scheduler requires parameter peerGCInterval")
 	}
 
-	if c.Scheduler.GC.PeerTTL <= 0 {
+	if cfg.Scheduler.GC.PeerTTL <= 0 {
 		return errors.New("scheduler requires parameter peerTTL")
 	}
 
-	if c.Scheduler.GC.TaskGCInterval <= 0 {
+	if cfg.Scheduler.GC.TaskGCInterval <= 0 {
 		return errors.New("scheduler requires parameter taskGCInterval")
 	}
 
-	if c.Scheduler.GC.TaskTTL <= 0 {
+	if cfg.Scheduler.GC.TaskTTL <= 0 {
 		return errors.New("scheduler requires parameter taskTTL")
 	}
 
-	if c.DynConfig.RefreshInterval <= 0 {
+	if cfg.DynConfig.RefreshInterval <= 0 {
 		return errors.New("dynconfig requires parameter refreshInterval")
 	}
 
-	if c.Manager.Addr == "" {
+	if cfg.Manager.Addr == "" {
 		return errors.New("manager requires parameter addr")
 	}
 
-	if c.Manager.SchedulerClusterID == 0 {
+	if cfg.Manager.SchedulerClusterID == 0 {
 		return errors.New("manager requires parameter schedulerClusterID")
 	}
 
-	if c.Manager.KeepAlive.Interval <= 0 {
+	if cfg.Manager.KeepAlive.Interval <= 0 {
 		return errors.New("manager requires parameter keepAlive interval")
 	}
 
-	if c.Job != nil && c.Job.Enable {
-		if c.Job.GlobalWorkerNum == 0 {
+	if cfg.Job != nil && cfg.Job.Enable {
+		if cfg.Job.GlobalWorkerNum == 0 {
 			return errors.New("job requires parameter globalWorkerNum")
 		}
 
-		if c.Job.SchedulerWorkerNum == 0 {
+		if cfg.Job.SchedulerWorkerNum == 0 {
 			return errors.New("job requires parameter schedulerWorkerNum")
 		}
 
-		if c.Job.LocalWorkerNum == 0 {
+		if cfg.Job.LocalWorkerNum == 0 {
 			return errors.New("job requires parameter localWorkerNum")
 		}
 
-		if c.Job.Redis.Host == "" {
+		if cfg.Job.Redis.Host == "" {
 			return errors.New("job requires parameter redis host")
 		}
 
-		if c.Job.Redis.Port <= 0 {
+		if cfg.Job.Redis.Port <= 0 {
 			return errors.New("job requires parameter redis port")
 		}
 
-		if c.Job.Redis.BrokerDB <= 0 {
+		if cfg.Job.Redis.BrokerDB <= 0 {
 			return errors.New("job requires parameter redis brokerDB")
 		}
 
-		if c.Job.Redis.BackendDB <= 0 {
+		if cfg.Job.Redis.BackendDB <= 0 {
 			return errors.New("job requires parameter redis backendDB")
 		}
 	}
 
-	if c.Storage == nil {
+	if cfg.Storage == nil {
 		return errors.New("server requires parameter storage")
 	}
 
-	if c.Storage.MaxSize <= 0 {
+	if cfg.Storage.MaxSize <= 0 {
 		return errors.New("storage requires parameter maxSize")
 	}
 
-	if c.Storage.MaxBackups <= 0 {
+	if cfg.Storage.MaxBackups <= 0 {
 		return errors.New("storage requires parameter maxBackups")
 	}
 
-	if c.Storage.BufferSize <= 0 {
+	if cfg.Storage.BufferSize <= 0 {
 		return errors.New("storage requires parameter bufferSize")
 	}
 
-	if c.Metrics != nil && c.Metrics.Enable {
-		if c.Metrics.Addr == "" {
+	if cfg.Metrics != nil && cfg.Metrics.Enable {
+		if cfg.Metrics.Addr == "" {
 			return errors.New("metrics requires parameter addr")
 		}
 	}


Test patch:
diff --git a/manager/config/config_test.go b/manager/config/config_test.go
index 7e1a7362..b9ca8a92 100644
--- a/manager/config/config_test.go
+++ b/manager/config/config_test.go
@@ -23,6 +23,8 @@ import (
 	"github.com/mitchellh/mapstructure"
 	testifyassert "github.com/stretchr/testify/assert"
 	"gopkg.in/yaml.v3"
+
+	"d7y.io/dragonfly/v2/pkg/objectstorage"
 )
 
 func TestManagerConfig_Load(t *testing.T) {
@@ -76,8 +78,16 @@ func TestManagerConfig_Load(t *testing.T) {
 				TTL:  1000,
 			},
 		},
-		Metrics: &RestConfig{
-			Addr: ":8000",
+		ObjectStorage: &ObjectStorageConfig{
+			Enable:    true,
+			Name:      objectstorage.ServiceNameS3,
+			Endpoint:  "127.0.0.1",
+			AccessKey: "foo",
+			SecretKey: "bar",
+		},
+		Metrics: &MetricsConfig{
+			Enable: true,
+			Addr:   ":8000",
 		},
 	}
 
diff --git a/manager/config/testdata/manager.yaml b/manager/config/testdata/manager.yaml
index c78e158a..45f5d1de 100644
--- a/manager/config/testdata/manager.yaml
+++ b/manager/config/testdata/manager.yaml
@@ -37,5 +37,13 @@ cache:
     size: 10000
     ttl: 1000
 
+objectStorage:
+  enable: true
+  name: s3
+  endpoint: 127.0.0.1
+  accessKey: foo
+  secretKey: bar
+
 metrics:
+  enable: true
   addr: :8000