Repository: dragonflyoss/Dragonfly2
Pull request #1546 (2022-08-09 04:46:11)

feat: add MultiReadCloser and storage add open func
Signed-off-by: Gaius <gaius.qi@gmail.com>

<!--- Provide a general summary of your changes in the Title above -->

## Description
- Add `MultiReadCloser` and storage add open func.
<!--- Describe your changes in detail -->

## Related Issue

<!--- 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/1546?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1546](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1546?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (aa69951) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/9ad2c1ee4dda1ae82a987d1317f2d4f629c68712?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (9ad2c1e) will **increase** coverage by `0.12%`.
> The diff coverage is `88.46%`.

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1546/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/1546?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff             @@
##             main    #1546      +/-   ##
==========================================
+ Coverage   43.07%   43.20%   +0.12%     
==========================================
  Files         116      117       +1     
  Lines       10875    10898      +23     
==========================================
+ Hits         4684     4708      +24     
+ Misses       5704     5702       -2     
- Partials      487      488       +1     
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| e2etests | `?` | |
| unittests | `43.20% <88.46%> (+0.12%)` | :arrow_up: |

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/1546?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [scheduler/storage/storage.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1546/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3N0b3JhZ2Uvc3RvcmFnZS5nbw==) | `83.17% <78.57%> (-0.16%)` | :arrow_down: |
| [pkg/io/multi\_readcloser.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1546/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL2lvL211bHRpX3JlYWRjbG9zZXIuZ28=) | `100.00% <100.00%> (ø)` | |
| [client/daemon/peer/peertask\_conductor.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1546/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2NvbmR1Y3Rvci5nbw==) | `56.13% <0.00%> (+0.28%)` | :arrow_up: |


Patch:
diff --git a/go.mod b/go.mod
index 79d0b06f..7811eed4 100644
--- a/go.mod
+++ b/go.mod
@@ -58,7 +58,7 @@ require (
 	github.com/swaggo/files v0.0.0-20220610200504-28940afbdbfe
 	github.com/swaggo/gin-swagger v1.5.1
 	github.com/swaggo/swag v1.8.3
-	go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin v0.32.0
+	go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin v0.34.0
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.33.0
 	go.opentelemetry.io/otel v1.9.0
 	go.opentelemetry.io/otel/exporters/jaeger v1.9.0
diff --git a/go.sum b/go.sum
index 4793eff1..fe20b953 100644
--- a/go.sum
+++ b/go.sum
@@ -1073,17 +1073,15 @@ go.opencensus.io v0.23.0 h1:gqCw0LfLxScz8irSi8exQc7fyQ0fKQU/qnC/X8+V/1M=
 go.opencensus.io v0.23.0/go.mod h1:XItmlyltB5F7CS4xOC1DcqMoFqwtC6OG2xF7mCv7P7E=
 go.opentelemetry.io/contrib v0.20.0 h1:ubFQUn0VCZ0gPwIoJfBJVpeBlyRMxu8Mm/huKWYd9p0=
 go.opentelemetry.io/contrib v0.20.0/go.mod h1:G/EtFaa6qaN7+LxqfIAT3GiZa7Wv5DTBUzl5H4LY0Kc=
-go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin v0.32.0 h1:ht6IqV6njVN4cMHYpN7pX5oDXZqGtl4fqvbGax1QFNU=
-go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin v0.32.0/go.mod h1:1126nNcUXEt2PRo3E5pJ4x98Gyu6K+bQIl5KECEJ6Qk=
+go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin v0.34.0 h1:wy5fZjhOPoh9b8GMJQ4M91f8rXqRZisTn0vdi5nwJD4=
+go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin v0.34.0/go.mod h1:Z4izXLrEw517ux2YzM2qnbTlirPilcwbF/uh/9xWDio=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.33.0 h1:z6rnla1Asjzn0FrhohzIbDi4bxbtc6EMmQ7f5ZPn+pA=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.33.0/go.mod h1:y/SlJpJQPd2UzfBCj0E9Flk9FDCtTyqUmaCB41qFrWI=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.20.0/go.mod h1:2AboqHi0CiIZU0qwhtUfCYD1GeUzvvIXWNkhDt7ZMG4=
-go.opentelemetry.io/contrib/propagators/b3 v1.7.0 h1:oRAenUhj+GFttfIp3gj7HYVzBhPOHgq/dWPDSmLCXSY=
-go.opentelemetry.io/contrib/propagators/b3 v1.7.0/go.mod h1:gXx7AhL4xXCF42gpm9dQvdohoDa2qeyEx4eIIxqK+h4=
+go.opentelemetry.io/contrib/propagators/b3 v1.9.0 h1:Lzb9zU98jCE2kyfCjWfSSsiQoGtvBL+COxvUBf7FNhU=
 go.opentelemetry.io/otel v0.11.0/go.mod h1:G8UCk+KooF2HLkgo8RHX9epABH/aRGYET7gQOqBVdB0=
 go.opentelemetry.io/otel v0.17.0/go.mod h1:Oqtdxmf7UtEvL037ohlgnaYa1h7GtMh0NcSd9eqkC9s=
 go.opentelemetry.io/otel v0.20.0/go.mod h1:Y3ugLH2oa81t5QO+Lty+zXf8zC9L26ax4Nzoxm/dooo=
-go.opentelemetry.io/otel v1.7.0/go.mod h1:5BdUoMIz5WEs0vt0CUEMtSSaTSHBBVwrhnz7+nrD5xk=
 go.opentelemetry.io/otel v1.9.0 h1:8WZNQFIB2a71LnANS9JeyidJKKGOOremcUtb/OtHISw=
 go.opentelemetry.io/otel v1.9.0/go.mod h1:np4EoPGzoPs3O67xUVNoPPcmSvsfOxNlNA4F4AC+0Eo=
 go.opentelemetry.io/otel/exporters/jaeger v1.9.0 h1:gAEgEVGDWwFjcis9jJTOJqZNxDzoZfR12WNIxr7g9Ww=
@@ -1100,7 +1098,6 @@ go.opentelemetry.io/otel/sdk/export/metric v0.20.0/go.mod h1:h7RBNMsDJ5pmI1zExLi
 go.opentelemetry.io/otel/sdk/metric v0.20.0/go.mod h1:knxiS8Xd4E/N+ZqKmUPf3gTTZ4/0TjTXukfxjzSTpHE=
 go.opentelemetry.io/otel/trace v0.17.0/go.mod h1:bIujpqg6ZL6xUTubIUgziI1jSaUPthmabA/ygf/6Cfg=
 go.opentelemetry.io/otel/trace v0.20.0/go.mod h1:6GjCW8zgDjwGHGa6GkyeB8+/5vjT16gUEi0Nf1iBdgw=
-go.opentelemetry.io/otel/trace v1.7.0/go.mod h1:fzLSB9nqR2eXzxPXb2JW9IKE+ScyXA48yyE4TNvoHqU=
 go.opentelemetry.io/otel/trace v1.9.0 h1:oZaCNJUjWcg60VXWee8lJKlqhPbXAPB51URuR47pQYc=
 go.opentelemetry.io/otel/trace v1.9.0/go.mod h1:2737Q0MuG8q1uILYm2YYVkAyLtOofiTNGg6VODnOiPo=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
@@ -1400,7 +1397,6 @@ golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220209214540-3681064d5158/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220227234510-4e6760a101f9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220328115105-d36c6a25d886/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220502124256-b6088ccd6cba/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
diff --git a/pkg/io/multi_readcloser.go b/pkg/io/multi_readcloser.go
new file mode 100644
index 00000000..f6d4ddfc
--- /dev/null
+++ b/pkg/io/multi_readcloser.go
@@ -0,0 +1,62 @@
+/*
+ *     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 io
+
+import (
+	"io"
+
+	"github.com/hashicorp/go-multierror"
+)
+
+// multiReadCloser is MultiReadCloser instance.
+type multiReadCloser struct {
+	reader  io.Reader
+	closers []io.Closer
+}
+
+// MultiReadCloser returns a ReadCloser that's the logical concatenation of
+// the provided input readClosers.
+func MultiReadCloser(readClosers ...io.ReadCloser) io.ReadCloser {
+	readers := make([]io.Reader, len(readClosers))
+	closers := make([]io.Closer, len(readClosers))
+	for readCloser := range readClosers {
+		closers[readCloser] = readClosers[readCloser]
+		readers[readCloser] = readClosers[readCloser]
+	}
+
+	return &multiReadCloser{
+		reader:  io.MultiReader(readers...),
+		closers: closers,
+	}
+}
+
+// Reader is the interface that wraps the basic Read method for multiReadCloser.
+func (m *multiReadCloser) Read(p []byte) (int, error) {
+	return m.reader.Read(p)
+}
+
+// Closer is the interface that wraps the basic Close method for multiReadCloser.
+func (m *multiReadCloser) Close() error {
+	var merr error
+	for i := range m.closers {
+		if err := m.closers[i].Close(); err != nil {
+			merr = multierror.Append(merr, err)
+		}
+	}
+
+	return merr
+}
diff --git a/scheduler/config/dynconfig.go b/scheduler/config/dynconfig.go
index 999c2cb0..940e8965 100644
--- a/scheduler/config/dynconfig.go
+++ b/scheduler/config/dynconfig.go
@@ -41,6 +41,14 @@ var (
 )
 
 type DynconfigData struct {
+	ID               uint64            `yaml:"id" mapstructure:"id" json:"id"`
+	Hostname         string            `yaml:"hostname" mapstructure:"hostname" json:"host_name"`
+	Idc              string            `yaml:"idc" mapstructure:"idc" json:"idc"`
+	Location         string            `yaml:"location" mapstructure:"location" json:"location"`
+	NetTopology      string            `yaml:"netTopology" mapstructure:"netTopology" json:"net_topology"`
+	IP               string            `yaml:"ip" mapstructure:"ip" json:"ip"`
+	Port             int32             `yaml:"port" mapstructure:"port" json:"port"`
+	State            string            `yaml:"state" mapstructure:"state" json:"state"`
 	SeedPeers        []*SeedPeer       `yaml:"seedPeers" mapstructure:"seedPeers" json:"seed_peers"`
 	SchedulerCluster *SchedulerCluster `yaml:"schedulerCluster" mapstructure:"schedulerCluster" json:"scheduler_cluster"`
 }
@@ -72,10 +80,14 @@ func (c *SeedPeer) GetSeedPeerClusterConfig() (types.SeedPeerClusterConfig, bool
 }
 
 type SeedPeerCluster struct {
+	ID     uint64 `yaml:"id" mapstructure:"id" json:"id"`
+	Name   string `yaml:"name" mapstructure:"name" json:"name"`
 	Config []byte `yaml:"config" mapstructure:"config" json:"config"`
 }
 
 type SchedulerCluster struct {
+	ID           uint64 `yaml:"id" mapstructure:"id" json:"id"`
+	Name         string `yaml:"name" mapstructure:"name" json:"name"`
 	Config       []byte `yaml:"config" mapstructure:"config" json:"config"`
 	ClientConfig []byte `yaml:"clientConfig" mapstructure:"clientConfig" json:"client_config"`
 }
diff --git a/scheduler/storage/mocks/storage_mock.go b/scheduler/storage/mocks/storage_mock.go
index c5761949..412bf0d0 100644
--- a/scheduler/storage/mocks/storage_mock.go
+++ b/scheduler/storage/mocks/storage_mock.go
@@ -5,6 +5,7 @@
 package mocks
 
 import (
+	io "io"
 	reflect "reflect"
 
 	storage "d7y.io/dragonfly/v2/scheduler/storage"
@@ -76,3 +77,18 @@ func (mr *MockStorageMockRecorder) List() *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockStorage)(nil).List))
 }
+
+// Open mocks base method.
+func (m *MockStorage) Open() (io.ReadCloser, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Open")
+	ret0, _ := ret[0].(io.ReadCloser)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Open indicates an expected call of Open.
+func (mr *MockStorageMockRecorder) Open() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Open", reflect.TypeOf((*MockStorage)(nil).Open))
+}
diff --git a/scheduler/storage/storage.go b/scheduler/storage/storage.go
index 21162b75..1f64466a 100644
--- a/scheduler/storage/storage.go
+++ b/scheduler/storage/storage.go
@@ -34,6 +34,7 @@ import (
 	"github.com/gocarina/gocsv"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
+	pkgio "d7y.io/dragonfly/v2/pkg/io"
 )
 
 const (
@@ -182,6 +183,9 @@ type Storage interface {
 	// List returns all of records in csv file.
 	List() ([]Record, error)
 
+	// Open opens storage for read, it returns io.ReadCloser of storage files.
+	Open() (io.ReadCloser, error)
+
 	// Clear removes all record files.
 	Clear() error
 }
@@ -279,10 +283,10 @@ func (s *storage) List() ([]Record, error) {
 	}
 
 	var readers []io.Reader
-	var closers []io.ReadCloser
+	var readClosers []io.ReadCloser
 	defer func() {
-		for _, closer := range closers {
-			if err := closer.Close(); err != nil {
+		for _, readCloser := range readClosers {
+			if err := readCloser.Close(); err != nil {
 				logger.Error(err)
 			}
 		}
@@ -295,7 +299,7 @@ func (s *storage) List() ([]Record, error) {
 		}
 
 		readers = append(readers, file)
-		closers = append(closers, file)
+		readClosers = append(readClosers, file)
 	}
 
 	var records []Record
@@ -306,6 +310,29 @@ func (s *storage) List() ([]Record, error) {
 	return records, nil
 }
 
+// Open opens storage for read, it returns io.ReadCloser of storage files.
+func (s *storage) Open() (io.ReadCloser, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	fileInfos, err := s.backups()
+	if err != nil {
+		return nil, err
+	}
+
+	var readClosers []io.ReadCloser
+	for _, fileInfo := range fileInfos {
+		file, err := os.Open(filepath.Join(s.baseDir, fileInfo.Name()))
+		if err != nil {
+			return nil, err
+		}
+
+		readClosers = append(readClosers, file)
+	}
+
+	return pkgio.MultiReadCloser(readClosers...), nil
+}
+
 // Clear removes all records.
 func (s *storage) Clear() error {
 	s.mu.Lock()


Test patch:
diff --git a/pkg/io/multi_readcloser_test.go b/pkg/io/multi_readcloser_test.go
new file mode 100644
index 00000000..c25d9fdc
--- /dev/null
+++ b/pkg/io/multi_readcloser_test.go
@@ -0,0 +1,81 @@
+/*
+ *     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 io
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type mockReadCloser struct{}
+
+func (m *mockReadCloser) Read(p []byte) (int, error) {
+	return 1, nil
+}
+
+func (m *mockReadCloser) Close() error {
+	return nil
+}
+
+type mockReadCloserWithReadError struct{}
+
+func (m *mockReadCloserWithReadError) Read(p []byte) (int, error) {
+	return 0, errors.New("foo")
+}
+
+func (m *mockReadCloserWithReadError) Close() error {
+	return nil
+}
+
+type mockReadCloserWithCloseError struct{}
+
+func (m *mockReadCloserWithCloseError) Read(p []byte) (int, error) {
+	return 1, nil
+}
+
+func (m *mockReadCloserWithCloseError) Close() error {
+	return errors.New("foo")
+}
+
+func TestMultiReadCloser(t *testing.T) {
+	assert := assert.New(t)
+	readCloser := MultiReadCloser(&mockReadCloser{})
+	n, err := readCloser.Read([]byte{})
+	assert.NoError(err)
+	assert.Equal(n, 1)
+
+	err = readCloser.Close()
+	assert.NoError(err)
+
+	readCloser = MultiReadCloser(&mockReadCloserWithReadError{})
+	n, err = readCloser.Read([]byte{})
+	assert.Error(err)
+	assert.Equal(n, 0)
+
+	err = readCloser.Close()
+	assert.NoError(err)
+
+	readCloser = MultiReadCloser(&mockReadCloserWithCloseError{})
+	n, err = readCloser.Read([]byte{})
+	assert.NoError(err)
+	assert.Equal(n, 1)
+
+	err = readCloser.Close()
+	assert.Error(err)
+}
diff --git a/scheduler/config/dynconfig_test.go b/scheduler/config/dynconfig_test.go
index 4e5c604b..51812087 100644
--- a/scheduler/config/dynconfig_test.go
+++ b/scheduler/config/dynconfig_test.go
@@ -28,6 +28,7 @@ import (
 
 	managerv1 "d7y.io/api/pkg/apis/manager/v1"
 
+	"d7y.io/dragonfly/v2/manager/model"
 	"d7y.io/dragonfly/v2/pkg/rpc/manager/client/mocks"
 )
 
@@ -63,22 +64,76 @@ func TestDynconfig_GetManagerSourceType(t *testing.T) {
 			sleep: func() {},
 			mock: func(m *mocks.MockClientMockRecorder) {
 				m.GetScheduler(gomock.Any()).Return(&managerv1.Scheduler{
+					Id:          1,
+					HostName:    "foo",
+					Idc:         "idc",
+					Location:    "location",
+					NetTopology: "net_topology",
+					Ip:          "127.0.0.1",
+					Port:        8002,
+					State:       "active",
 					SeedPeers: []*managerv1.SeedPeer{
 						{
+							Id:           1,
 							HostName:     "bar",
+							Type:         model.SeedPeerTypeSuperSeed,
+							Idc:          "idc",
+							NetTopology:  "net_topology",
+							Location:     "location",
 							Ip:           "127.0.0.1",
 							Port:         8001,
 							DownloadPort: 8003,
+							SeedPeerCluster: &managerv1.SeedPeerCluster{
+								Id:     1,
+								Name:   "baz",
+								Config: []byte{1},
+							},
 						},
 					},
+					SchedulerCluster: &managerv1.SchedulerCluster{
+						Id:           1,
+						Name:         "bas",
+						Config:       []byte{1},
+						ClientConfig: []byte{1},
+					},
 				}, nil).Times(1)
 			},
 			expect: func(t *testing.T, data *DynconfigData, err error) {
 				assert := assert.New(t)
-				assert.Equal(data.SeedPeers[0].Hostname, "bar")
-				assert.Equal(data.SeedPeers[0].IP, "127.0.0.1")
-				assert.Equal(data.SeedPeers[0].Port, int32(8001))
-				assert.Equal(data.SeedPeers[0].DownloadPort, int32(8003))
+				assert.EqualValues(data, &DynconfigData{
+					ID:          1,
+					Hostname:    "foo",
+					Idc:         "idc",
+					Location:    "location",
+					NetTopology: "net_topology",
+					IP:          "127.0.0.1",
+					Port:        8002,
+					State:       "active",
+					SeedPeers: []*SeedPeer{
+						{
+							ID:           1,
+							Hostname:     "bar",
+							Type:         model.SeedPeerTypeSuperSeed,
+							IDC:          "idc",
+							NetTopology:  "net_topology",
+							Location:     "location",
+							IP:           "127.0.0.1",
+							Port:         8001,
+							DownloadPort: 8003,
+							SeedPeerCluster: &SeedPeerCluster{
+								ID:     1,
+								Name:   "baz",
+								Config: []byte{1},
+							},
+						},
+					},
+					SchedulerCluster: &SchedulerCluster{
+						ID:           1,
+						Name:         "bas",
+						Config:       []byte{1},
+						ClientConfig: []byte{1},
+					},
+				})
 			},
 		},
 		{
@@ -95,24 +150,78 @@ func TestDynconfig_GetManagerSourceType(t *testing.T) {
 			mock: func(m *mocks.MockClientMockRecorder) {
 				gomock.InOrder(
 					m.GetScheduler(gomock.Any()).Return(&managerv1.Scheduler{
+						Id:          1,
+						HostName:    "foo",
+						Idc:         "idc",
+						Location:    "location",
+						NetTopology: "net_topology",
+						Ip:          "127.0.0.1",
+						Port:        8002,
+						State:       "active",
 						SeedPeers: []*managerv1.SeedPeer{
 							{
+								Id:           1,
 								HostName:     "bar",
+								Type:         model.SeedPeerTypeSuperSeed,
+								Idc:          "idc",
+								NetTopology:  "net_topology",
+								Location:     "location",
 								Ip:           "127.0.0.1",
 								Port:         8001,
 								DownloadPort: 8003,
+								SeedPeerCluster: &managerv1.SeedPeerCluster{
+									Id:     1,
+									Name:   "baz",
+									Config: []byte{1},
+								},
 							},
 						},
+						SchedulerCluster: &managerv1.SchedulerCluster{
+							Id:           1,
+							Name:         "bas",
+							Config:       []byte{1},
+							ClientConfig: []byte{1},
+						},
 					}, nil).Times(1),
 					m.GetScheduler(gomock.Any()).Return(nil, errors.New("foo")).Times(1),
 				)
 			},
 			expect: func(t *testing.T, data *DynconfigData, err error) {
 				assert := assert.New(t)
-				assert.Equal(data.SeedPeers[0].Hostname, "bar")
-				assert.Equal(data.SeedPeers[0].IP, "127.0.0.1")
-				assert.Equal(data.SeedPeers[0].Port, int32(8001))
-				assert.Equal(data.SeedPeers[0].DownloadPort, int32(8003))
+				assert.EqualValues(data, &DynconfigData{
+					ID:          1,
+					Hostname:    "foo",
+					Idc:         "idc",
+					Location:    "location",
+					NetTopology: "net_topology",
+					IP:          "127.0.0.1",
+					Port:        8002,
+					State:       "active",
+					SeedPeers: []*SeedPeer{
+						{
+							ID:           1,
+							Hostname:     "bar",
+							Type:         model.SeedPeerTypeSuperSeed,
+							IDC:          "idc",
+							NetTopology:  "net_topology",
+							Location:     "location",
+							IP:           "127.0.0.1",
+							Port:         8001,
+							DownloadPort: 8003,
+							SeedPeerCluster: &SeedPeerCluster{
+								ID:     1,
+								Name:   "baz",
+								Config: []byte{1},
+							},
+						},
+					},
+					SchedulerCluster: &SchedulerCluster{
+						ID:           1,
+						Name:         "bas",
+						Config:       []byte{1},
+						ClientConfig: []byte{1},
+					},
+				})
 			},
 		},
 	}
diff --git a/scheduler/resource/peer_test.go b/scheduler/resource/peer_test.go
index e3687a55..070abba3 100644
--- a/scheduler/resource/peer_test.go
+++ b/scheduler/resource/peer_test.go
@@ -331,6 +331,10 @@ func TestPeer_MainParent(t *testing.T) {
 				peer.Pieces.Add(&schedulerv1.PieceResult{
 					DstPid: seedPeer.ID,
 				})
+				peer.Pieces.Add(&schedulerv1.PieceResult{
+					DstPid: seedPeer.ID,
+				})
+
 				peer.Task.StorePeer(peer)
 				peer.Task.StorePeer(seedPeer)
 				parent, err := peer.MainParent()
diff --git a/scheduler/storage/storage_test.go b/scheduler/storage/storage_test.go
index b9aeb96a..56e1f1a0 100644
--- a/scheduler/storage/storage_test.go
+++ b/scheduler/storage/storage_test.go
@@ -209,7 +209,7 @@ func TestStorage_List(t *testing.T) {
 			baseDir: os.TempDir(),
 			options: []Option{},
 			mock: func(t *testing.T, s Storage, baseDir string, record Record) {
-				s.(*storage).baseDir = "bar"
+				s.(*storage).baseDir = "bae"
 			},
 			expect: func(t *testing.T, s Storage, baseDir string, record Record) {
 				assert := assert.New(t)
@@ -339,6 +339,153 @@ func TestStorage_List(t *testing.T) {
 	}
 }
 
+func TestStorage_Open(t *testing.T) {
+	tests := []struct {
+		name    string
+		baseDir string
+		options []Option
+		record  Record
+		mock    func(t *testing.T, s Storage, baseDir string, record Record)
+		expect  func(t *testing.T, s Storage, baseDir string, record Record)
+	}{
+		{
+			name:    "open storage withempty csv file given",
+			baseDir: os.TempDir(),
+			options: []Option{},
+			mock:    func(t *testing.T, s Storage, baseDir string, record Record) {},
+			expect: func(t *testing.T, s Storage, baseDir string, record Record) {
+				assert := assert.New(t)
+				_, err := s.Open()
+				assert.NoError(err)
+			},
+		},
+		{
+			name:    "open file infos failed",
+			baseDir: os.TempDir(),
+			options: []Option{},
+			mock: func(t *testing.T, s Storage, baseDir string, record Record) {
+				s.(*storage).baseDir = "bas"
+			},
+			expect: func(t *testing.T, s Storage, baseDir string, record Record) {
+				assert := assert.New(t)
+				_, err := s.Open()
+				assert.Error(err)
+				s.(*storage).baseDir = baseDir
+			},
+		},
+		{
+			name:    "open storage with records of a file",
+			baseDir: os.TempDir(),
+			options: []Option{WithBufferSize(1)},
+			record: Record{
+				ID:                   "1",
+				IP:                   "127.0.0.1",
+				Hostname:             "hostname",
+				Tag:                  "tag",
+				Cost:                 1,
+				PieceCount:           1,
+				TotalPieceCount:      1,
+				ContentLength:        1,
+				SecurityDomain:       "security_domain",
+				IDC:                  "idc",
+				NetTopology:          "net_topology",
+				Location:             "location",
+				FreeUploadLoad:       1,
+				State:                PeerStateSucceeded,
+				CreateAt:             time.Now().UnixNano(),
+				UpdateAt:             time.Now().UnixNano(),
+				ParentID:             "2",
+				ParentIP:             "127.0.0.1",
+				ParentHostname:       "parent_hostname",
+				ParentTag:            "parent_tag",
+				ParentPieceCount:     1,
+				ParentSecurityDomain: "parent_security_domain",
+				ParentIDC:            "parent_idc",
+				ParentNetTopology:    "parent_net_topology",
+				ParentLocation:       "parent_location",
+				ParentFreeUploadLoad: 1,
+				ParentCreateAt:       time.Now().UnixNano(),
+				ParentUpdateAt:       time.Now().UnixNano(),
+			},
+			mock: func(t *testing.T, s Storage, baseDir string, record Record) {
+				if err := s.Create(record); err != nil {
+					t.Fatal(err)
+				}
+			},
+			expect: func(t *testing.T, s Storage, baseDir string, record Record) {
+				assert := assert.New(t)
+				_, err := s.Open()
+				assert.NoError(err)
+
+				if err := s.Create(record); err != nil {
+					t.Fatal(err)
+				}
+
+				readCloser, err := s.Open()
+				assert.NoError(err)
+
+				var records []Record
+				err = gocsv.UnmarshalWithoutHeaders(readCloser, &records)
+				assert.NoError(err)
+				assert.Equal(len(records), 1)
+				assert.EqualValues(records[0], record)
+			},
+		},
+		{
+			name:    "open storage with records of multi files",
+			baseDir: os.TempDir(),
+			options: []Option{WithBufferSize(1)},
+			record:  Record{},
+			mock: func(t *testing.T, s Storage, baseDir string, record Record) {
+				file, err := os.OpenFile(filepath.Join(baseDir, "record-test.csv"), os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0600)
+				if err != nil {
+					t.Fatal(err)
+				}
+				defer file.Close()
+
+				if err := gocsv.MarshalWithoutHeaders([]Record{{ID: "2"}}, file); err != nil {
+					t.Fatal(err)
+				}
+
+				if err := s.Create(Record{ID: "1"}); err != nil {
+					t.Fatal(err)
+				}
+
+				if err := s.Create(Record{ID: "3"}); err != nil {
+					t.Fatal(err)
+				}
+			},
+			expect: func(t *testing.T, s Storage, baseDir string, record Record) {
+				assert := assert.New(t)
+				readCloser, err := s.Open()
+				assert.NoError(err)
+
+				var records []Record
+				err = gocsv.UnmarshalWithoutHeaders(readCloser, &records)
+				assert.NoError(err)
+				assert.Equal(len(records), 2)
+				assert.Equal(records[0].ID, "2")
+				assert.Equal(records[1].ID, "1")
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			s, err := New(tc.baseDir, tc.options...)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			tc.mock(t, s, tc.baseDir, tc.record)
+			tc.expect(t, s, tc.baseDir, tc.record)
+			if err := s.Clear(); err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}
+
 func TestStorage_Clear(t *testing.T) {
 	tests := []struct {
 		name    string
@@ -462,7 +609,7 @@ func TestStorage_openFile(t *testing.T) {
 			baseDir: os.TempDir(),
 			options: []Option{},
 			mock: func(t *testing.T, s Storage) {
-				s.(*storage).baseDir = "baz"
+				s.(*storage).baseDir = "bat"
 			},
 			expect: func(t *testing.T, s Storage, baseDir string) {
 				assert := assert.New(t)