Repository: dragonflyoss/Dragonfly2
Pull request #1234 (2022-04-08 09:36:17)

Add csv storage to scheduler
<!--- Provide a general summary of your changes in the Title above -->

## Description
- Add csv storage to scheduler
<!--- 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! -->

- [x] Code compiles correctly.
- [x] My code follows the code style of this project.
- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [x] I have read the **CONTRIBUTING** document.
- [x] I have added tests to cover my changes.
- [x] All new and existing tests passed.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1234?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1234](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1234?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (903c609) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/1bfa0643a3d2feb679e185e5eeb5fa1158a8b2de?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (1bfa064) will **decrease** coverage by `0.51%`.
> The diff coverage is `44.15%`.

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

```diff
@@            Coverage Diff             @@
##             main    #1234      +/-   ##
==========================================
- Coverage   46.65%   46.14%   -0.52%     
==========================================
  Files         133      134       +1     
  Lines       10584    10661      +77     
==========================================
- Hits         4938     4919      -19     
- Misses       5090     5184      +94     
- Partials      556      558       +2     
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| e2etests | `?` | |
| unittests | `46.14% <44.15%> (-0.52%)` | :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/1234?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/1234/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3N0b3JhZ2Uvc3RvcmFnZS5nbw==) | `44.15% <44.15%> (ø)` | |
| [client/daemon/peer/peertask\_piecetask\_poller.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1234/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3BpZWNldGFza19wb2xsZXIuZ28=) | `30.37% <0.00%> (-22.97%)` | :arrow_down: |
| [...ent/daemon/peer/peertask\_piecetask\_synchronizer.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1234/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3BpZWNldGFza19zeW5jaHJvbml6ZXIuZ28=) | `47.27% <0.00%> (-11.52%)` | :arrow_down: |
| [client/daemon/peer/peertask\_conductor.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1234/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2NvbmR1Y3Rvci5nbw==) | `56.86% <0.00%> (-0.33%)` | :arrow_down: |


Patch:
diff --git a/go.mod b/go.mod
index 0583cb78..5cbe8771 100644
--- a/go.mod
+++ b/go.mod
@@ -27,6 +27,7 @@ require (
 	github.com/go-redis/cache/v8 v8.4.1
 	github.com/go-redis/redis/v8 v8.10.0
 	github.com/go-sql-driver/mysql v1.5.0
+	github.com/gocarina/gocsv v0.0.0-20220310154401-d4df709ca055
 	github.com/gofrs/flock v0.8.0
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da
 	github.com/golang/mock v1.6.0
diff --git a/go.sum b/go.sum
index f5ad301a..d957610a 100644
--- a/go.sum
+++ b/go.sum
@@ -317,6 +317,8 @@ github.com/gobuffalo/packd v0.1.0/go.mod h1:M2Juc+hhDXf/PnmBANFCqx4DM3wRbgDvnVWe
 github.com/gobuffalo/packr/v2 v2.0.9/go.mod h1:emmyGweYTm6Kdper+iywB6YK5YzuKchGtJQZ0Odn4pQ=
 github.com/gobuffalo/packr/v2 v2.2.0/go.mod h1:CaAwI0GPIAv+5wKLtv8Afwl+Cm78K/I/VCm/3ptBN+0=
 github.com/gobuffalo/syncx v0.0.0-20190224160051-33c29581e754/go.mod h1:HhnNqWY95UYwwW3uSASeV7vtgYkT2t16hJgV3AEPUpw=
+github.com/gocarina/gocsv v0.0.0-20220310154401-d4df709ca055 h1:UfcDMw41lSx3XM7UvD1i7Fsu3rMgD55OU5LYwLoR/Yk=
+github.com/gocarina/gocsv v0.0.0-20220310154401-d4df709ca055/go.mod h1:5YoVOkjYAQumqlV356Hj3xeYh4BdZuLE0/nRkf2NKkI=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/gofrs/flock v0.8.0 h1:MSdYClljsF3PbENUUEx85nkWfJSGfzYI9yEBZOJz6CY=
 github.com/gofrs/flock v0.8.0/go.mod h1:F1TvTiK9OcQqauNUHlbJvyl9Qa1QvF/gOUDKA14jxHU=
diff --git a/scheduler/config/config.go b/scheduler/config/config.go
index 30c02c2f..2768b198 100644
--- a/scheduler/config/config.go
+++ b/scheduler/config/config.go
@@ -24,6 +24,7 @@ import (
 	"d7y.io/dragonfly/v2/cmd/dependency/base"
 	"d7y.io/dragonfly/v2/pkg/util/hostutils"
 	"d7y.io/dragonfly/v2/pkg/util/net/iputils"
+	"d7y.io/dragonfly/v2/scheduler/storage"
 )
 
 type Config struct {
@@ -51,6 +52,9 @@ type Config struct {
 	// Job configuration
 	Job *JobConfig `yaml:"job" mapstructure:"job"`
 
+	// Storage configuration
+	Storage *StorageConfig `yaml:"storage" mapstructure:"storage"`
+
 	// Metrics configuration
 	Metrics *MetricsConfig `yaml:"metrics" mapstructure:"metrics"`
 }
@@ -103,6 +107,11 @@ func New() *Config {
 				BackendDB: 2,
 			},
 		},
+		Storage: &StorageConfig{
+			MaxSize:    storage.DefaultMaxSize,
+			MaxBackups: storage.DefaultMaxBackups,
+			BufferSize: storage.DefaultBufferSize,
+		},
 		Metrics: &MetricsConfig{
 			Enable:         false,
 			EnablePeerHost: false,
@@ -202,6 +211,18 @@ func (c *Config) Validate() error {
 		}
 	}
 
+	if c.Storage.MaxSize <= 0 {
+		return errors.New("storage requires parameter maxSize")
+	}
+
+	if c.Storage.MaxBackups <= 0 {
+		return errors.New("storage requires parameter maxBackups")
+	}
+
+	if c.Storage.BufferSize <= 0 {
+		return errors.New("storage requires parameter bufferSize")
+	}
+
 	if c.Metrics.Enable {
 		if c.Metrics.Addr == "" {
 			return errors.New("metrics requires parameter addr")
@@ -328,6 +349,18 @@ type JobConfig struct {
 	Redis *RedisConfig `yaml:"redis" mapstructure:"redis"`
 }
 
+type StorageConfig struct {
+	// MaxSize sets the maximum size in megabytes of storage file
+	MaxSize int `yaml:"maxSize" mapstructure:"maxSize"`
+
+	// MaxBackups sets the maximum number of storage files to retain
+	MaxBackups int `yaml:"maxBackups" mapstructure:"maxBackups"`
+
+	// BufferSize sets the size of buffer container,
+	// if the buffer is full, write all the records in the buffer to the file
+	BufferSize int `yaml:"bufferSize" mapstructure:"bufferSize"`
+}
+
 type RedisConfig struct {
 	// Server hostname
 	Host string `yaml:"host" mapstructure:"host"`
diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go
index 641ab6db..0bf1b4b1 100644
--- a/scheduler/scheduler.go
+++ b/scheduler/scheduler.go
@@ -38,6 +38,7 @@ import (
 	"d7y.io/dragonfly/v2/scheduler/rpcserver"
 	"d7y.io/dragonfly/v2/scheduler/scheduler"
 	"d7y.io/dragonfly/v2/scheduler/service"
+	"d7y.io/dragonfly/v2/scheduler/storage"
 )
 
 const (
@@ -131,8 +132,14 @@ func New(ctx context.Context, cfg *config.Config, d dfpath.Dfpath) (*Server, err
 	// Initialize scheduler
 	scheduler := scheduler.New(cfg.Scheduler, dynconfig, d.PluginDir())
 
+	// Initialize Storage
+	storage, err := storage.New(d.DataDir())
+	if err != nil {
+		return nil, err
+	}
+
 	// Initialize scheduler service
-	service := service.New(cfg, resource, scheduler, dynconfig)
+	service := service.New(cfg, resource, scheduler, dynconfig, storage)
 
 	// Initialize grpc service
 	svr := rpcserver.New(service, serverOptions...)
diff --git a/scheduler/service/service.go b/scheduler/service/service.go
index bc73187b..0ea0748c 100644
--- a/scheduler/service/service.go
+++ b/scheduler/service/service.go
@@ -35,6 +35,7 @@ import (
 	"d7y.io/dragonfly/v2/scheduler/metrics"
 	"d7y.io/dragonfly/v2/scheduler/resource"
 	"d7y.io/dragonfly/v2/scheduler/scheduler"
+	"d7y.io/dragonfly/v2/scheduler/storage"
 )
 
 type Service struct {
@@ -49,6 +50,9 @@ type Service struct {
 
 	// Dynamic config
 	dynconfig config.DynconfigInterface
+
+	// Storage interface
+	storage storage.Storage
 }
 
 // New service instance
@@ -57,6 +61,7 @@ func New(
 	resource resource.Resource,
 	scheduler scheduler.Scheduler,
 	dynconfig config.DynconfigInterface,
+	storage storage.Storage,
 ) *Service {
 
 	return &Service{
@@ -64,6 +69,7 @@ func New(
 		scheduler: scheduler,
 		config:    cfg,
 		dynconfig: dynconfig,
+		storage:   storage,
 	}
 }
 
@@ -321,27 +327,36 @@ func (s *Service) ReportPeerResult(ctx context.Context, req *rpcscheduler.PeerRe
 	}
 
 	metrics.DownloadCount.WithLabelValues(peer.BizTag).Inc()
-
 	if !req.Success {
 		peer.Log.Errorf("report peer failed result: %s %#v", req.Code, req)
 		if peer.FSM.Is(resource.PeerStateBackToSource) {
-			s.handleTaskFail(ctx, peer.Task)
+			s.createRecord(peer, storage.PeerStateBackToSourceFailed, req)
 			metrics.DownloadFailureCount.WithLabelValues(peer.BizTag, metrics.DownloadFailureBackToSourceType).Inc()
-		} else {
-			metrics.DownloadFailureCount.WithLabelValues(peer.BizTag, metrics.DownloadFailureP2PType).Inc()
+
+			s.handleTaskFail(ctx, peer.Task)
+			s.handlePeerFail(ctx, peer)
+			return nil
 		}
 
+		s.createRecord(peer, storage.PeerStateFailed, req)
+		metrics.DownloadFailureCount.WithLabelValues(peer.BizTag, metrics.DownloadFailureP2PType).Inc()
+
 		s.handlePeerFail(ctx, peer)
 		return nil
 	}
 
 	peer.Log.Infof("report peer result: %#v", req)
+	metrics.PeerTaskDownloadDuration.WithLabelValues(peer.BizTag).Observe(float64(req.Cost))
+
 	if peer.FSM.Is(resource.PeerStateBackToSource) {
+		s.createRecord(peer, storage.PeerStateBackToSourceSucceeded, req)
 		s.handleTaskSuccess(ctx, peer.Task, req)
+		s.handlePeerSuccess(ctx, peer)
+		return nil
 	}
-	s.handlePeerSuccess(ctx, peer)
 
-	metrics.PeerTaskDownloadDuration.WithLabelValues(peer.BizTag).Observe(float64(req.Cost))
+	s.createRecord(peer, storage.PeerStateSucceeded, req)
+	s.handlePeerSuccess(ctx, peer)
 	return nil
 }
 
@@ -642,3 +657,45 @@ func (s *Service) handleTaskFail(ctx context.Context, task *resource.Task) {
 		return
 	}
 }
+
+// createRecord stores peer download records
+func (s *Service) createRecord(peer *resource.Peer, peerState int, req *rpcscheduler.PeerResult) {
+	record := storage.Record{
+		ID:              peer.ID,
+		IP:              peer.Host.IP,
+		Hostname:        peer.Host.Hostname,
+		BizTag:          peer.BizTag,
+		Cost:            req.Cost,
+		PieceCount:      int32(peer.Pieces.Count()),
+		TotalPieceCount: peer.Task.TotalPieceCount.Load(),
+		ContentLength:   peer.Task.ContentLength.Load(),
+		SecurityDomain:  peer.Host.SecurityDomain,
+		IDC:             peer.Host.IDC,
+		NetTopology:     peer.Host.NetTopology,
+		Location:        peer.Host.Location,
+		FreeUploadLoad:  peer.Host.FreeUploadLoad(),
+		State:           peerState,
+		CreateAt:        peer.CreateAt.Load().UnixNano(),
+		UpdateAt:        peer.UpdateAt.Load().UnixNano(),
+	}
+
+	if parent, ok := peer.LoadParent(); ok {
+		record.ParentID = parent.ID
+		record.ParentIP = parent.Host.IP
+		record.ParentHostname = parent.Host.Hostname
+		record.ParentBizTag = parent.BizTag
+		record.ParentPieceCount = int32(parent.Pieces.Count())
+		record.ParentSecurityDomain = parent.Host.SecurityDomain
+		record.ParentIDC = parent.Host.IDC
+		record.ParentNetTopology = parent.Host.NetTopology
+		record.ParentLocation = parent.Host.Location
+		record.ParentFreeUploadLoad = parent.Host.FreeUploadLoad()
+		record.ParentIsCDN = parent.Host.IsCDN
+		record.ParentCreateAt = parent.CreateAt.Load().UnixNano()
+		record.ParentUpdateAt = parent.UpdateAt.Load().UnixNano()
+	}
+
+	if err := s.storage.Create(record); err != nil {
+		peer.Log.Error(err)
+	}
+}
diff --git a/scheduler/storage/mocks/storage_mock.go b/scheduler/storage/mocks/storage_mock.go
new file mode 100644
index 00000000..c5761949
--- /dev/null
+++ b/scheduler/storage/mocks/storage_mock.go
@@ -0,0 +1,78 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: storage.go
+
+// Package mocks is a generated GoMock package.
+package mocks
+
+import (
+	reflect "reflect"
+
+	storage "d7y.io/dragonfly/v2/scheduler/storage"
+	gomock "github.com/golang/mock/gomock"
+)
+
+// MockStorage is a mock of Storage interface.
+type MockStorage struct {
+	ctrl     *gomock.Controller
+	recorder *MockStorageMockRecorder
+}
+
+// MockStorageMockRecorder is the mock recorder for MockStorage.
+type MockStorageMockRecorder struct {
+	mock *MockStorage
+}
+
+// NewMockStorage creates a new mock instance.
+func NewMockStorage(ctrl *gomock.Controller) *MockStorage {
+	mock := &MockStorage{ctrl: ctrl}
+	mock.recorder = &MockStorageMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockStorage) EXPECT() *MockStorageMockRecorder {
+	return m.recorder
+}
+
+// Clear mocks base method.
+func (m *MockStorage) Clear() error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Clear")
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Clear indicates an expected call of Clear.
+func (mr *MockStorageMockRecorder) Clear() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Clear", reflect.TypeOf((*MockStorage)(nil).Clear))
+}
+
+// Create mocks base method.
+func (m *MockStorage) Create(arg0 storage.Record) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Create", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Create indicates an expected call of Create.
+func (mr *MockStorageMockRecorder) Create(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockStorage)(nil).Create), arg0)
+}
+
+// List mocks base method.
+func (m *MockStorage) List() ([]storage.Record, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "List")
+	ret0, _ := ret[0].([]storage.Record)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// List indicates an expected call of List.
+func (mr *MockStorageMockRecorder) List() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockStorage)(nil).List))
+}
diff --git a/scheduler/storage/storage.go b/scheduler/storage/storage.go
new file mode 100644
index 00000000..18d29113
--- /dev/null
+++ b/scheduler/storage/storage.go
@@ -0,0 +1,404 @@
+/*
+ *     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 -destination mocks/storage_mock.go -source storage.go -package mocks
+
+package storage
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"regexp"
+	"sort"
+	"sync"
+	"time"
+
+	"github.com/gocarina/gocsv"
+
+	logger "d7y.io/dragonfly/v2/internal/dflog"
+)
+
+const (
+	// DefaultMaxSize is the default maximum size of record file
+	DefaultMaxSize = 100
+
+	// DefaultMaxBackups is the default maximum count of backup
+	DefaultMaxBackups = 10
+
+	// DefaultBufferSize is the default size of buffer container
+	DefaultBufferSize = 100
+)
+
+const (
+	// RecordFilePrefix is prefix of record file name
+	RecordFilePrefix = "record"
+
+	// RecordFileExt is extension of record file name
+	RecordFileExt = "csv"
+)
+
+const (
+	// Peer has been downloaded successfully
+	PeerStateSucceeded = iota
+
+	// Peer has been downloaded failed
+	PeerStateFailed
+
+	// Peer has been back-to-source downloaded successfully
+	PeerStateBackToSourceSucceeded
+
+	// Peer has been back-to-source downloaded failed
+	PeerStateBackToSourceFailed
+)
+
+const (
+	// megabyte is the converted factor of MaxSize and bytes
+	megabyte = 1024 * 1024
+
+	// backupTimeFormat is the timestamp format of backup filename
+	backupTimeFormat = "2006-01-02T15-04-05.000"
+)
+
+// Record contains content for record
+type Record struct {
+	// ID is peer id
+	ID string `csv:"id"`
+
+	// IP is host ip
+	IP string `csv:"ip"`
+
+	// Hostname is host name
+	Hostname string `csv:"hostname"`
+
+	// BizTag is peer biz tag
+	BizTag string `csv:"bizTag"`
+
+	// Cost is the task download time(millisecond)
+	Cost uint32 `csv:"cost"`
+
+	// PieceCount is total piece count
+	PieceCount int32 `csv:"pieceCount"`
+
+	// TotalPieceCount is total piece count
+	TotalPieceCount int32 `csv:"totalPieceCount"`
+
+	// ContentLength is task total content length
+	ContentLength int64 `csv:"contentLength"`
+
+	// SecurityDomain is security domain of host
+	SecurityDomain string `csv:"securityDomain"`
+
+	// IDC is internet data center of host
+	IDC string `csv:"idc"`
+
+	// NetTopology is network topology of host
+	// Example: switch|router|...
+	NetTopology string `csv:"netTopology"`
+
+	// Location is location of host
+	// Example: country|province|...
+	Location string `csv:"location"`
+
+	// FreeUploadLoad is free upload load of host
+	FreeUploadLoad int32 `csv:"freeUpoladLoad"`
+
+	// State is the download state of the peer
+	State int `csv:"state"`
+
+	// CreateAt is peer create nanosecond time
+	CreateAt int64 `csv:"createAt"`
+
+	// UpdateAt is peer update nanosecond time
+	UpdateAt int64 `csv:"updateAt"`
+
+	// ParentID is parent peer id
+	ParentID string `csv:"parentID"`
+
+	// ParentIP is parent host ip
+	ParentIP string `csv:"parentIP"`
+
+	// ParentHostname is parent host name
+	ParentHostname string `csv:"parentHostname"`
+
+	// ParentBizTag is parent peer biz tag
+	ParentBizTag string `csv:"parentBizTag"`
+
+	// ParentPieceCount is parent total piece count
+	ParentPieceCount int32 `csv:"parentPieceCount"`
+
+	// ParentSecurityDomain is parent security domain of host
+	ParentSecurityDomain string `csv:"parentSecurityDomain"`
+
+	// ParentIDC is parent internet data center of host
+	ParentIDC string `csv:"parentIDC"`
+
+	// ParentNetTopology is parent network topology of host
+	// Example: switch|router|...
+	ParentNetTopology string `csv:"parentNetTopology"`
+
+	// ParentLocation is parent location of host
+	// Example: country|province|...
+	ParentLocation string `csv:"parentLocation"`
+
+	// ParentFreeUploadLoad is parent free upload load of host
+	ParentFreeUploadLoad int32 `csv:"parentFreeUploadLoad"`
+
+	// ParentIsCDN is used as tag cdn
+	ParentIsCDN bool `csv:"parentIsCDN"`
+
+	// ParentCreateAt is parent peer create nanosecond time
+	ParentCreateAt int64 `csv:"parentCreateAt"`
+
+	// ParentUpdateAt is parent peer update nanosecond time
+	ParentUpdateAt int64 `csv:"parentUpdateAt"`
+}
+
+// Storage is the interface used for storage
+type Storage interface {
+	// Create inserts the record into csv file
+	Create(Record) error
+
+	// List returns all of records in csv file
+	List() ([]Record, error)
+
+	// Clear removes all record files
+	Clear() error
+}
+
+// storage provides storage function
+type storage struct {
+	baseDir    string
+	filename   string
+	maxSize    int64
+	maxBackups int
+	buffer     []Record
+	bufferSize int
+	mu         *sync.RWMutex
+}
+
+// Option is a functional option for configuring the Storage
+type Option func(s *storage)
+
+// WithMaxSize sets the maximum size in megabytes of storage file
+func WithMaxSize(maxSize int) Option {
+	return func(s *storage) {
+		s.maxSize = int64(maxSize * megabyte)
+	}
+}
+
+// WithMaxBackups sets the maximum number of storage files to retain
+func WithMaxBackups(maxBackups int) Option {
+	return func(s *storage) {
+		s.maxBackups = maxBackups
+	}
+}
+
+// WithCacheSize sets the size of buffer container,
+// if the buffer is full, write all the records in the buffer to the file
+func WithBufferSize(bufferSize int) Option {
+	return func(s *storage) {
+		s.bufferSize = bufferSize
+		s.buffer = make([]Record, 0, bufferSize)
+	}
+}
+
+// New returns a new Storage instence
+func New(baseDir string, options ...Option) (Storage, error) {
+	s := &storage{
+		baseDir:    baseDir,
+		filename:   filepath.Join(baseDir, fmt.Sprintf("%s.%s", RecordFilePrefix, RecordFileExt)),
+		maxSize:    DefaultMaxSize * megabyte,
+		maxBackups: DefaultMaxBackups,
+		buffer:     make([]Record, 0, DefaultBufferSize),
+		bufferSize: DefaultBufferSize,
+		mu:         &sync.RWMutex{},
+	}
+
+	for _, opt := range options {
+		opt(s)
+	}
+
+	file, err := os.OpenFile(s.filename, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0600)
+	if err != nil {
+		return nil, err
+	}
+	file.Close()
+
+	return s, nil
+}
+
+// Create inserts the record into csv file
+func (s *storage) Create(record Record) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	// Write records to file
+	if len(s.buffer) >= s.bufferSize {
+		if err := s.create(s.buffer...); err != nil {
+			return err
+		}
+
+		// Keep allocated memory
+		s.buffer = s.buffer[:0]
+	}
+
+	// Write records to buffer
+	s.buffer = append(s.buffer, record)
+	return nil
+}
+
+// List returns all of records in csv file
+func (s *storage) List() ([]Record, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	fileInfos, err := s.backups()
+	if err != nil {
+		return nil, err
+	}
+
+	var readers []io.Reader
+	var closers []io.ReadCloser
+	defer func() {
+		for _, closer := range closers {
+			if err := closer.Close(); err != nil {
+				logger.Error(err)
+			}
+		}
+	}()
+
+	for _, fileInfo := range fileInfos {
+		file, err := os.Open(filepath.Join(s.baseDir, fileInfo.Name()))
+		if err != nil {
+			return nil, err
+		}
+
+		readers = append(readers, file)
+		closers = append(closers, file)
+	}
+
+	var records []Record
+	if err := gocsv.UnmarshalWithoutHeaders(io.MultiReader(readers...), &records); err != nil {
+		return nil, err
+	}
+
+	return records, nil
+}
+
+// Clear removes all records
+func (s *storage) Clear() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	fileInfos, err := s.backups()
+	if err != nil {
+		return err
+	}
+
+	for _, fileInfo := range fileInfos {
+		filename := filepath.Join(s.baseDir, fileInfo.Name())
+		if err := os.Remove(filename); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// create inserts the records into csv file
+func (s *storage) create(records ...Record) error {
+	file, err := s.openFile()
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	if err := gocsv.MarshalWithoutHeaders(records, file); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// openFile opens the record file and removes record files that exceed the total size
+func (s *storage) openFile() (*os.File, error) {
+	fileInfo, err := os.Stat(s.filename)
+	if err != nil {
+		return nil, err
+	}
+
+	if s.maxSize <= fileInfo.Size() {
+		if err := os.Rename(s.filename, s.backupFilename()); err != nil {
+			return nil, err
+		}
+	}
+
+	fileInfos, err := s.backups()
+	if err != nil {
+		return nil, err
+	}
+
+	if s.maxBackups < len(fileInfos)+1 {
+		filename := filepath.Join(s.baseDir, fileInfos[0].Name())
+		if err := os.Remove(filename); err != nil {
+			return nil, err
+		}
+	}
+
+	file, err := os.OpenFile(s.filename, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
+	if err != nil {
+		return nil, err
+	}
+
+	return file, nil
+}
+
+// backupFilename generates file name of backup files
+func (s *storage) backupFilename() string {
+	timestamp := time.Now().Format(backupTimeFormat)
+	return filepath.Join(s.baseDir, fmt.Sprintf("%s-%s.%s", RecordFilePrefix, timestamp, RecordFileExt))
+}
+
+// backupFilename returns backup file information
+func (s *storage) backups() ([]fs.FileInfo, error) {
+	fileInfos, err := ioutil.ReadDir(s.baseDir)
+	if err != nil {
+		return nil, err
+	}
+
+	var backups []fs.FileInfo
+	regexp := regexp.MustCompile(RecordFilePrefix)
+	for _, fileInfo := range fileInfos {
+		if !fileInfo.IsDir() && regexp.MatchString(fileInfo.Name()) {
+			backups = append(backups, fileInfo)
+		}
+	}
+
+	if len(backups) <= 0 {
+		return nil, errors.New("backup does not exist")
+	}
+
+	sort.Slice(backups, func(i, j int) bool {
+		return backups[i].ModTime().Before(backups[j].ModTime())
+	})
+
+	return backups, nil
+}


Test patch:
diff --git a/scheduler/config/config_test.go b/scheduler/config/config_test.go
index 70ad2389..99a49f4c 100644
--- a/scheduler/config/config_test.go
+++ b/scheduler/config/config_test.go
@@ -24,6 +24,10 @@ import (
 	"github.com/mitchellh/mapstructure"
 	testifyassert "github.com/stretchr/testify/assert"
 	"gopkg.in/yaml.v3"
+
+	"d7y.io/dragonfly/v2/pkg/util/hostutils"
+	"d7y.io/dragonfly/v2/pkg/util/net/iputils"
+	"d7y.io/dragonfly/v2/scheduler/storage"
 )
 
 func TestConfig_Load(t *testing.T) {
@@ -85,6 +89,11 @@ func TestConfig_Load(t *testing.T) {
 				BackendDB: 2,
 			},
 		},
+		Storage: &StorageConfig{
+			MaxSize:    1,
+			MaxBackups: 1,
+			BufferSize: 1,
+		},
 		Metrics: &MetricsConfig{
 			Enable:         false,
 			Addr:           ":8000",
@@ -105,3 +114,65 @@ func TestConfig_Load(t *testing.T) {
 
 	assert.EqualValues(config, schedulerConfigYAML)
 }
+
+func TestConfig_New(t *testing.T) {
+	assert := testifyassert.New(t)
+	config := New()
+
+	assert.EqualValues(config, &Config{
+		Server: &ServerConfig{
+			IP:     iputils.IPv4,
+			Host:   hostutils.FQDNHostname,
+			Listen: "0.0.0.0",
+			Port:   8002,
+		},
+		Scheduler: &SchedulerConfig{
+			Algorithm:            "default",
+			BackSourceCount:      3,
+			RetryBackSourceLimit: 5,
+			RetryLimit:           10,
+			RetryInterval:        50 * time.Millisecond,
+			GC: &GCConfig{
+				PeerGCInterval: 10 * time.Minute,
+				PeerTTL:        24 * time.Hour,
+				TaskGCInterval: 10 * time.Minute,
+				TaskTTL:        24 * time.Hour,
+				HostGCInterval: 30 * time.Minute,
+				HostTTL:        48 * time.Hour,
+			},
+		},
+		DynConfig: &DynConfig{
+			RefreshInterval: 1 * time.Minute,
+		},
+		Host: &HostConfig{},
+		Manager: &ManagerConfig{
+			SchedulerClusterID: 1,
+			KeepAlive: KeepAliveConfig{
+				Interval: 5 * time.Second,
+			},
+		},
+		CDN: &CDNConfig{
+			Enable: true,
+		},
+		Job: &JobConfig{
+			Enable:             true,
+			GlobalWorkerNum:    10,
+			SchedulerWorkerNum: 10,
+			LocalWorkerNum:     10,
+			Redis: &RedisConfig{
+				Port:      6379,
+				BrokerDB:  1,
+				BackendDB: 2,
+			},
+		},
+		Storage: &StorageConfig{
+			MaxSize:    storage.DefaultMaxSize,
+			MaxBackups: storage.DefaultMaxBackups,
+			BufferSize: storage.DefaultBufferSize,
+		},
+		Metrics: &MetricsConfig{
+			Enable:         false,
+			EnablePeerHost: false,
+		},
+	})
+}
diff --git a/scheduler/config/testdata/scheduler.yaml b/scheduler/config/testdata/scheduler.yaml
index ed3f7aa2..bc9fcd33 100644
--- a/scheduler/config/testdata/scheduler.yaml
+++ b/scheduler/config/testdata/scheduler.yaml
@@ -50,6 +50,11 @@ job:
     brokerDB: 1
     backendDB: 2
 
+storage:
+  maxSize: 1
+  maxBackups: 1
+  bufferSize: 1
+
 metrics:
   enable: false
   addr: ":8000"
diff --git a/scheduler/rpcserver/rpcserver_test.go b/scheduler/rpcserver/rpcserver_test.go
index 3080e62a..7c320aa7 100644
--- a/scheduler/rpcserver/rpcserver_test.go
+++ b/scheduler/rpcserver/rpcserver_test.go
@@ -29,6 +29,7 @@ import (
 	"d7y.io/dragonfly/v2/scheduler/resource"
 	"d7y.io/dragonfly/v2/scheduler/scheduler/mocks"
 	"d7y.io/dragonfly/v2/scheduler/service"
+	storagemocks "d7y.io/dragonfly/v2/scheduler/storage/mocks"
 )
 
 var (
@@ -61,7 +62,8 @@ func TestRPCServer_New(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
-			svc := service.New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig)
+			storage := storagemocks.NewMockStorage(ctl)
+			svc := service.New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig, storage)
 
 			svr := New(svc)
 			tc.expect(t, svr)
diff --git a/scheduler/service/service_test.go b/scheduler/service/service_test.go
index e9650766..6d05cff1 100644
--- a/scheduler/service/service_test.go
+++ b/scheduler/service/service_test.go
@@ -46,6 +46,7 @@ import (
 	"d7y.io/dragonfly/v2/scheduler/resource"
 	"d7y.io/dragonfly/v2/scheduler/scheduler"
 	"d7y.io/dragonfly/v2/scheduler/scheduler/mocks"
+	storagemocks "d7y.io/dragonfly/v2/scheduler/storage/mocks"
 )
 
 var (
@@ -114,7 +115,8 @@ func TestService_New(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			resource := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
-			tc.expect(t, New(&config.Config{Scheduler: mockSchedulerConfig}, resource, scheduler, dynconfig))
+			storage := storagemocks.NewMockStorage(ctl)
+			tc.expect(t, New(&config.Config{Scheduler: mockSchedulerConfig}, resource, scheduler, dynconfig, storage))
 		})
 	}
 }
@@ -592,10 +594,11 @@ func TestService_RegisterPeerTask(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
+			storage := storagemocks.NewMockStorage(ctl)
 			hostManager := resource.NewMockHostManager(ctl)
 			taskManager := resource.NewMockTaskManager(ctl)
 			peerManager := resource.NewMockPeerManager(ctl)
-			svc := New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig)
+			svc := New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig, storage)
 
 			mockHost := resource.NewHost(mockRawHost)
 			mockTask := resource.NewTask(mockTaskID, mockTaskURL, mockTaskBackToSourceLimit, mockTaskURLMeta)
@@ -851,9 +854,10 @@ func TestService_ReportPieceResult(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
+			storage := storagemocks.NewMockStorage(ctl)
 			peerManager := resource.NewMockPeerManager(ctl)
 			stream := rpcschedulermocks.NewMockScheduler_ReportPieceResultServer(ctl)
-			svc := New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig)
+			svc := New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig, storage)
 
 			mockHost := resource.NewHost(mockRawHost)
 			mockTask := resource.NewTask(mockTaskID, mockTaskURL, mockTaskBackToSourceLimit, mockTaskURLMeta)
@@ -871,7 +875,7 @@ func TestService_ReportPeerResult(t *testing.T) {
 		mock func(
 			mockPeer *resource.Peer,
 			res resource.Resource, peerManager resource.PeerManager,
-			mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder,
+			mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder, ms *storagemocks.MockStorageMockRecorder,
 		)
 		expect func(t *testing.T, peer *resource.Peer, err error)
 	}{
@@ -883,7 +887,7 @@ func TestService_ReportPeerResult(t *testing.T) {
 			mock: func(
 				mockPeer *resource.Peer,
 				res resource.Resource, peerManager resource.PeerManager,
-				mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder,
+				mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder, ms *storagemocks.MockStorageMockRecorder,
 			) {
 				gomock.InOrder(
 					mr.PeerManager().Return(peerManager).Times(1),
@@ -906,12 +910,14 @@ func TestService_ReportPeerResult(t *testing.T) {
 			mock: func(
 				mockPeer *resource.Peer,
 				res resource.Resource, peerManager resource.PeerManager,
-				mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder,
+				mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder, ms *storagemocks.MockStorageMockRecorder,
 			) {
 				mockPeer.FSM.SetState(resource.PeerStateFailed)
+				mockPeer.StoreParent(mockPeer)
 				gomock.InOrder(
 					mr.PeerManager().Return(peerManager).Times(1),
 					mp.Load(gomock.Eq(mockPeerID)).Return(mockPeer, true).Times(1),
+					ms.Create(gomock.Any()).Return(nil).Times(1),
 				)
 			},
 			expect: func(t *testing.T, peer *resource.Peer, err error) {
@@ -928,12 +934,13 @@ func TestService_ReportPeerResult(t *testing.T) {
 			mock: func(
 				mockPeer *resource.Peer,
 				res resource.Resource, peerManager resource.PeerManager,
-				mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder,
+				mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder, ms *storagemocks.MockStorageMockRecorder,
 			) {
 				mockPeer.FSM.SetState(resource.PeerStateBackToSource)
 				gomock.InOrder(
 					mr.PeerManager().Return(peerManager).Times(1),
 					mp.Load(gomock.Eq(mockPeerID)).Return(mockPeer, true).Times(1),
+					ms.Create(gomock.Any()).Return(nil).Times(1),
 				)
 			},
 			expect: func(t *testing.T, peer *resource.Peer, err error) {
@@ -950,12 +957,14 @@ func TestService_ReportPeerResult(t *testing.T) {
 			mock: func(
 				mockPeer *resource.Peer,
 				res resource.Resource, peerManager resource.PeerManager,
-				mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder,
+				mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder, ms *storagemocks.MockStorageMockRecorder,
 			) {
 				mockPeer.FSM.SetState(resource.PeerStateFailed)
+				mockPeer.StoreParent(mockPeer)
 				gomock.InOrder(
 					mr.PeerManager().Return(peerManager).Times(1),
 					mp.Load(gomock.Eq(mockPeerID)).Return(mockPeer, true).Times(1),
+					ms.Create(gomock.Any()).Return(nil).Times(1),
 				)
 			},
 			expect: func(t *testing.T, peer *resource.Peer, err error) {
@@ -972,12 +981,60 @@ func TestService_ReportPeerResult(t *testing.T) {
 			mock: func(
 				mockPeer *resource.Peer,
 				res resource.Resource, peerManager resource.PeerManager,
-				mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder,
+				mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder, ms *storagemocks.MockStorageMockRecorder,
 			) {
 				mockPeer.FSM.SetState(resource.PeerStateBackToSource)
 				gomock.InOrder(
 					mr.PeerManager().Return(peerManager).Times(1),
 					mp.Load(gomock.Eq(mockPeerID)).Return(mockPeer, true).Times(1),
+					ms.Create(gomock.Any()).Return(nil).Times(1),
+				)
+			},
+			expect: func(t *testing.T, peer *resource.Peer, err error) {
+				assert := assert.New(t)
+				assert.NoError(err)
+			},
+		},
+		{
+			name: "receive peer success, and peer state is PeerStateBackToSource",
+			req: &rpcscheduler.PeerResult{
+				Success: true,
+				PeerId:  mockPeerID,
+			},
+			mock: func(
+				mockPeer *resource.Peer,
+				res resource.Resource, peerManager resource.PeerManager,
+				mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder, ms *storagemocks.MockStorageMockRecorder,
+			) {
+				mockPeer.FSM.SetState(resource.PeerStateBackToSource)
+				gomock.InOrder(
+					mr.PeerManager().Return(peerManager).Times(1),
+					mp.Load(gomock.Eq(mockPeerID)).Return(mockPeer, true).Times(1),
+					ms.Create(gomock.Any()).Return(nil).Times(1),
+				)
+			},
+			expect: func(t *testing.T, peer *resource.Peer, err error) {
+				assert := assert.New(t)
+				assert.NoError(err)
+			},
+		},
+		{
+			name: "receive peer success and create record failed",
+			req: &rpcscheduler.PeerResult{
+				Success: true,
+				PeerId:  mockPeerID,
+			},
+			mock: func(
+				mockPeer *resource.Peer,
+				res resource.Resource, peerManager resource.PeerManager,
+				mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder, ms *storagemocks.MockStorageMockRecorder,
+			) {
+				mockPeer.FSM.SetState(resource.PeerStateFailed)
+				mockPeer.StoreParent(mockPeer)
+				gomock.InOrder(
+					mr.PeerManager().Return(peerManager).Times(1),
+					mp.Load(gomock.Eq(mockPeerID)).Return(mockPeer, true).Times(1),
+					ms.Create(gomock.Any()).Return(errors.New("foo")).Times(1),
 				)
 			},
 			expect: func(t *testing.T, peer *resource.Peer, err error) {
@@ -994,13 +1051,14 @@ func TestService_ReportPeerResult(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
+			storage := storagemocks.NewMockStorage(ctl)
 			peerManager := resource.NewMockPeerManager(ctl)
-			svc := New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig)
+			svc := New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig, storage)
 
 			mockHost := resource.NewHost(mockRawHost)
 			mockTask := resource.NewTask(mockTaskID, mockTaskURL, mockTaskBackToSourceLimit, mockTaskURLMeta)
 			mockPeer := resource.NewPeer(mockPeerID, mockTask, mockHost)
-			tc.mock(mockPeer, res, peerManager, res.EXPECT(), peerManager.EXPECT())
+			tc.mock(mockPeer, res, peerManager, res.EXPECT(), peerManager.EXPECT(), storage.EXPECT())
 			tc.expect(t, mockPeer, svc.ReportPeerResult(context.Background(), tc.req))
 		})
 	}
@@ -1257,8 +1315,9 @@ func TestService_LeaveTask(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
+			storage := storagemocks.NewMockStorage(ctl)
 			peerManager := resource.NewMockPeerManager(ctl)
-			svc := New(&config.Config{Scheduler: mockSchedulerConfig, Metrics: &config.MetricsConfig{EnablePeerHost: true}}, res, scheduler, dynconfig)
+			svc := New(&config.Config{Scheduler: mockSchedulerConfig, Metrics: &config.MetricsConfig{EnablePeerHost: true}}, res, scheduler, dynconfig, storage)
 
 			tc.mock(tc.peer, tc.child, peerManager, scheduler.EXPECT(), res.EXPECT(), peerManager.EXPECT())
 			tc.expect(t, tc.peer, svc.LeaveTask(context.Background(), &rpcscheduler.PeerTarget{}))
@@ -1511,7 +1570,8 @@ func TestService_registerTask(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
-			svc := New(tc.config, res, scheduler, dynconfig)
+			storage := storagemocks.NewMockStorage(ctl)
+			svc := New(tc.config, res, scheduler, dynconfig, storage)
 
 			taskManager := resource.NewMockTaskManager(ctl)
 			mockHost := resource.NewHost(mockRawHost)
@@ -1600,7 +1660,8 @@ func TestService_registerHost(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
-			svc := New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig)
+			storage := storagemocks.NewMockStorage(ctl)
+			svc := New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig, storage)
 			hostManager := resource.NewMockHostManager(ctl)
 			mockHost := resource.NewHost(mockRawHost)
 
@@ -1661,11 +1722,12 @@ func TestService_triggerCDNTask(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
+			storage := storagemocks.NewMockStorage(ctl)
 			cdn := resource.NewMockCDN(ctl)
 			mockHost := resource.NewHost(mockRawHost)
 			task := resource.NewTask(mockTaskID, mockTaskURL, mockTaskBackToSourceLimit, mockTaskURLMeta)
 			peer := resource.NewPeer(mockPeerID, task, mockHost)
-			svc := New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig)
+			svc := New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig, storage)
 
 			tc.mock(task, peer, cdn, res.EXPECT(), cdn.EXPECT())
 			svc.triggerCDNTask(context.Background(), task)
@@ -1740,10 +1802,11 @@ func TestService_handleBeginOfPiece(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
+			storage := storagemocks.NewMockStorage(ctl)
 			mockHost := resource.NewHost(mockRawHost)
 			mockTask := resource.NewTask(mockTaskID, mockTaskURL, mockTaskBackToSourceLimit, mockTaskURLMeta)
 			peer := resource.NewPeer(mockPeerID, mockTask, mockHost)
-			svc := New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig)
+			svc := New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig, storage)
 
 			tc.mock(peer, scheduler.EXPECT())
 			svc.handleBeginOfPiece(context.Background(), peer)
@@ -1804,7 +1867,8 @@ func TestService_registerPeer(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
-			svc := New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig)
+			storage := storagemocks.NewMockStorage(ctl)
+			svc := New(&config.Config{Scheduler: mockSchedulerConfig}, res, scheduler, dynconfig, storage)
 			peerManager := resource.NewMockPeerManager(ctl)
 			mockHost := resource.NewHost(mockRawHost)
 			mockTask := resource.NewTask(mockTaskID, mockTaskURL, mockTaskBackToSourceLimit, mockTaskURLMeta)
@@ -1884,7 +1948,8 @@ func TestService_handlePieceSuccess(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
-			svc := New(&config.Config{Scheduler: mockSchedulerConfig, Metrics: &config.MetricsConfig{EnablePeerHost: true}}, res, scheduler, dynconfig)
+			storage := storagemocks.NewMockStorage(ctl)
+			svc := New(&config.Config{Scheduler: mockSchedulerConfig, Metrics: &config.MetricsConfig{EnablePeerHost: true}}, res, scheduler, dynconfig, storage)
 
 			tc.mock(tc.peer)
 			svc.handlePieceSuccess(context.Background(), tc.peer, tc.piece)
@@ -2204,9 +2269,10 @@ func TestService_handlePieceFail(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
+			storage := storagemocks.NewMockStorage(ctl)
 			peerManager := resource.NewMockPeerManager(ctl)
 			cdn := resource.NewMockCDN(ctl)
-			svc := New(tc.config, res, scheduler, dynconfig)
+			svc := New(tc.config, res, scheduler, dynconfig, storage)
 
 			tc.run(t, svc, tc.peer, tc.parent, tc.piece, peerManager, cdn, scheduler.EXPECT(), res.EXPECT(), peerManager.EXPECT(), cdn.EXPECT())
 		})
@@ -2285,6 +2351,7 @@ func TestService_handlePeerSuccess(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
+			storage := storagemocks.NewMockStorage(ctl)
 
 			url, err := url.Parse(s.URL)
 			if err != nil {
@@ -2306,7 +2373,7 @@ func TestService_handlePeerSuccess(t *testing.T) {
 			mockHost := resource.NewHost(mockRawHost)
 			mockTask := resource.NewTask(mockTaskID, mockTaskURL, mockTaskBackToSourceLimit, mockTaskURLMeta)
 			peer := resource.NewPeer(mockPeerID, mockTask, mockHost)
-			svc := New(&config.Config{Scheduler: mockSchedulerConfig, Metrics: &config.MetricsConfig{EnablePeerHost: true}}, res, scheduler, dynconfig)
+			svc := New(&config.Config{Scheduler: mockSchedulerConfig, Metrics: &config.MetricsConfig{EnablePeerHost: true}}, res, scheduler, dynconfig, storage)
 
 			tc.mock(peer)
 			svc.handlePeerSuccess(context.Background(), peer)
@@ -2387,7 +2454,8 @@ func TestService_handlePeerFail(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
-			svc := New(&config.Config{Scheduler: mockSchedulerConfig, Metrics: &config.MetricsConfig{EnablePeerHost: true}}, res, scheduler, dynconfig)
+			storage := storagemocks.NewMockStorage(ctl)
+			svc := New(&config.Config{Scheduler: mockSchedulerConfig, Metrics: &config.MetricsConfig{EnablePeerHost: true}}, res, scheduler, dynconfig, storage)
 
 			tc.mock(tc.peer, tc.child, scheduler.EXPECT())
 			svc.handlePeerFail(context.Background(), tc.peer)
@@ -2466,7 +2534,8 @@ func TestService_handleTaskSuccess(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
-			svc := New(&config.Config{Scheduler: mockSchedulerConfig, Metrics: &config.MetricsConfig{EnablePeerHost: true}}, res, scheduler, dynconfig)
+			storage := storagemocks.NewMockStorage(ctl)
+			svc := New(&config.Config{Scheduler: mockSchedulerConfig, Metrics: &config.MetricsConfig{EnablePeerHost: true}}, res, scheduler, dynconfig, storage)
 			task := resource.NewTask(mockTaskID, mockTaskURL, mockTaskBackToSourceLimit, mockTaskURLMeta)
 
 			tc.mock(task)
@@ -2543,7 +2612,8 @@ func TestService_handleTaskFail(t *testing.T) {
 			scheduler := mocks.NewMockScheduler(ctl)
 			res := resource.NewMockResource(ctl)
 			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
-			svc := New(&config.Config{Scheduler: mockSchedulerConfig, Metrics: &config.MetricsConfig{EnablePeerHost: true}}, res, scheduler, dynconfig)
+			storage := storagemocks.NewMockStorage(ctl)
+			svc := New(&config.Config{Scheduler: mockSchedulerConfig, Metrics: &config.MetricsConfig{EnablePeerHost: true}}, res, scheduler, dynconfig, storage)
 			task := resource.NewTask(mockTaskID, mockTaskURL, mockTaskBackToSourceLimit, mockTaskURLMeta)
 
 			tc.mock(task)
diff --git a/scheduler/storage/storage_test.go b/scheduler/storage/storage_test.go
new file mode 100644
index 00000000..51f77663
--- /dev/null
+++ b/scheduler/storage/storage_test.go
@@ -0,0 +1,639 @@
+/*
+ *     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 storage
+
+import (
+	"fmt"
+	"io/fs"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"reflect"
+	"regexp"
+	"testing"
+	"time"
+
+	"github.com/gocarina/gocsv"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestStorage_New(t *testing.T) {
+	tests := []struct {
+		name    string
+		baseDir string
+		options []Option
+		expect  func(t *testing.T, s Storage, err error)
+	}{
+		{
+			name:    "new storage",
+			baseDir: os.TempDir(),
+			options: []Option{},
+			expect: func(t *testing.T, s Storage, err error) {
+				assert := assert.New(t)
+				assert.NoError(err)
+				assert.Equal(reflect.TypeOf(s).Elem().Name(), "storage")
+				assert.Equal(s.(*storage).maxSize, int64(DefaultMaxSize*megabyte))
+				assert.Equal(s.(*storage).maxBackups, DefaultMaxBackups)
+				assert.Equal(s.(*storage).bufferSize, DefaultBufferSize)
+				assert.Equal(cap(s.(*storage).buffer), DefaultBufferSize)
+				assert.Equal(len(s.(*storage).buffer), 0)
+
+				if err := s.Clear(); err != nil {
+					t.Fatal(err)
+				}
+			},
+		},
+		{
+			name:    "new storage with maxSize",
+			baseDir: os.TempDir(),
+			options: []Option{WithMaxSize(1)},
+			expect: func(t *testing.T, s Storage, err error) {
+				assert := assert.New(t)
+				assert.NoError(err)
+				assert.Equal(reflect.TypeOf(s).Elem().Name(), "storage")
+				assert.Equal(s.(*storage).maxSize, int64(1*megabyte))
+				assert.Equal(s.(*storage).maxBackups, DefaultMaxBackups)
+				assert.Equal(s.(*storage).bufferSize, DefaultBufferSize)
+				assert.Equal(cap(s.(*storage).buffer), DefaultBufferSize)
+				assert.Equal(len(s.(*storage).buffer), 0)
+
+				if err := s.Clear(); err != nil {
+					t.Fatal(err)
+				}
+			},
+		},
+		{
+			name:    "new storage with maxBackups",
+			baseDir: os.TempDir(),
+			options: []Option{WithMaxBackups(1)},
+			expect: func(t *testing.T, s Storage, err error) {
+				assert := assert.New(t)
+				assert.NoError(err)
+				assert.Equal(reflect.TypeOf(s).Elem().Name(), "storage")
+				assert.Equal(s.(*storage).maxSize, int64(DefaultMaxSize*megabyte))
+				assert.Equal(s.(*storage).maxBackups, 1)
+				assert.Equal(s.(*storage).bufferSize, DefaultBufferSize)
+				assert.Equal(cap(s.(*storage).buffer), DefaultBufferSize)
+				assert.Equal(len(s.(*storage).buffer), 0)
+
+				if err := s.Clear(); err != nil {
+					t.Fatal(err)
+				}
+			},
+		},
+		{
+			name:    "new storage with bufferSize",
+			baseDir: os.TempDir(),
+			options: []Option{WithBufferSize(1)},
+			expect: func(t *testing.T, s Storage, err error) {
+				assert := assert.New(t)
+				assert.NoError(err)
+				assert.Equal(reflect.TypeOf(s).Elem().Name(), "storage")
+				assert.Equal(s.(*storage).maxSize, int64(DefaultMaxSize*megabyte))
+				assert.Equal(s.(*storage).maxBackups, DefaultMaxBackups)
+				assert.Equal(s.(*storage).bufferSize, 1)
+				assert.Equal(cap(s.(*storage).buffer), 1)
+				assert.Equal(len(s.(*storage).buffer), 0)
+
+				if err := s.Clear(); err != nil {
+					t.Fatal(err)
+				}
+			},
+		},
+		{
+			name:    "new storage failed",
+			baseDir: "/foo",
+			options: []Option{WithMaxBackups(100)},
+			expect: func(t *testing.T, s Storage, err error) {
+				assert := assert.New(t)
+				assert.Error(err)
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			s, err := New(tc.baseDir, tc.options...)
+			tc.expect(t, s, err)
+		})
+	}
+}
+
+func TestStorage_Create(t *testing.T) {
+	tests := []struct {
+		name    string
+		baseDir string
+		options []Option
+		mock    func(s Storage)
+		expect  func(t *testing.T, s Storage, baseDir string)
+	}{
+		{
+			name:    "create record",
+			baseDir: os.TempDir(),
+			options: []Option{WithBufferSize(1)},
+			mock:    func(s Storage) {},
+			expect: func(t *testing.T, s Storage, baseDir string) {
+				assert := assert.New(t)
+				err := s.Create(Record{})
+				assert.NoError(err)
+			},
+		},
+		{
+			name:    "open file failed",
+			baseDir: os.TempDir(),
+			options: []Option{WithBufferSize(0)},
+			mock: func(s Storage) {
+				s.(*storage).baseDir = "foo"
+			},
+			expect: func(t *testing.T, s Storage, baseDir string) {
+				assert := assert.New(t)
+				err := s.Create(Record{})
+				assert.Error(err)
+				s.(*storage).baseDir = baseDir
+			},
+		},
+	}
+
+	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(s)
+			tc.expect(t, s, tc.baseDir)
+			if err := s.Clear(); err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}
+
+func TestStorage_List(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:    "empty 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.List()
+				assert.Error(err)
+			},
+		},
+		{
+			name:    "get file infos failed",
+			baseDir: os.TempDir(),
+			options: []Option{},
+			mock: func(t *testing.T, s Storage, baseDir string, record Record) {
+				s.(*storage).baseDir = "bar"
+			},
+			expect: func(t *testing.T, s Storage, baseDir string, record Record) {
+				assert := assert.New(t)
+				_, err := s.List()
+				assert.Error(err)
+				s.(*storage).baseDir = baseDir
+			},
+		},
+		{
+			name:    "open file failed",
+			baseDir: os.TempDir(),
+			options: []Option{},
+			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, 0300)
+				if err != nil {
+					t.Fatal(err)
+				}
+				file.Close()
+			},
+			expect: func(t *testing.T, s Storage, baseDir string, record Record) {
+				assert := assert.New(t)
+				_, err := s.List()
+				assert.Error(err)
+			},
+		},
+		{
+			name:    "list records of a file",
+			baseDir: os.TempDir(),
+			options: []Option{WithBufferSize(1)},
+			record: Record{
+				ID:                   "1",
+				IP:                   "127.0.0.1",
+				Hostname:             "hostname",
+				BizTag:               "biz_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",
+				ParentBizTag:         "parent_biz_tag",
+				ParentPieceCount:     1,
+				ParentSecurityDomain: "parent_security_domain",
+				ParentIDC:            "parent_idc",
+				ParentNetTopology:    "parent_net_topology",
+				ParentLocation:       "parent_location",
+				ParentFreeUploadLoad: 1,
+				ParentIsCDN:          true,
+				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.List()
+				assert.Error(err)
+
+				if err := s.Create(record); err != nil {
+					t.Fatal(err)
+				}
+				records, err := s.List()
+				assert.NoError(err)
+				assert.Equal(len(records), 1)
+				assert.EqualValues(records[0], record)
+			},
+		},
+		{
+			name:    "list 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)
+				records, err := s.List()
+				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
+		baseDir string
+		options []Option
+		mock    func(s Storage)
+		expect  func(t *testing.T, s Storage, baseDir string)
+	}{
+		{
+			name:    "clear file",
+			baseDir: os.TempDir(),
+			options: []Option{},
+			mock:    func(s Storage) {},
+			expect: func(t *testing.T, s Storage, baseDir string) {
+				assert := assert.New(t)
+				assert.NoError(s.Clear())
+				fileInfos, err := ioutil.ReadDir(filepath.Join(baseDir))
+				assert.NoError(err)
+
+				var backups []fs.FileInfo
+				regexp := regexp.MustCompile(RecordFilePrefix)
+				for _, fileInfo := range fileInfos {
+					if !fileInfo.IsDir() && regexp.MatchString(fileInfo.Name()) {
+						backups = append(backups, fileInfo)
+					}
+				}
+				assert.Equal(len(backups), 0)
+			},
+		},
+		{
+			name:    "open file failed",
+			baseDir: os.TempDir(),
+			options: []Option{},
+			mock: func(s Storage) {
+				s.(*storage).baseDir = "baz"
+			},
+			expect: func(t *testing.T, s Storage, baseDir string) {
+				assert := assert.New(t)
+				assert.Error(s.Clear())
+
+				s.(*storage).baseDir = baseDir
+				assert.NoError(s.Clear())
+			},
+		},
+	}
+
+	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(s)
+			tc.expect(t, s, tc.baseDir)
+		})
+	}
+}
+
+func TestStorage_create(t *testing.T) {
+	tests := []struct {
+		name    string
+		baseDir string
+		options []Option
+		mock    func(s Storage)
+		expect  func(t *testing.T, s Storage, baseDir string)
+	}{
+		{
+			name:    "create record",
+			baseDir: os.TempDir(),
+			options: []Option{},
+			mock:    func(s Storage) {},
+			expect: func(t *testing.T, s Storage, baseDir string) {
+				assert := assert.New(t)
+				err := s.(*storage).create(Record{})
+				assert.NoError(err)
+			},
+		},
+		{
+			name:    "open file failed",
+			baseDir: os.TempDir(),
+			options: []Option{},
+			mock: func(s Storage) {
+				s.(*storage).baseDir = "foo"
+			},
+			expect: func(t *testing.T, s Storage, baseDir string) {
+				assert := assert.New(t)
+				err := s.(*storage).create(Record{})
+				assert.Error(err)
+				s.(*storage).baseDir = baseDir
+			},
+		},
+	}
+
+	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(s)
+			tc.expect(t, s, tc.baseDir)
+			if err := s.Clear(); err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}
+
+func TestStorage_openFile(t *testing.T) {
+	tests := []struct {
+		name    string
+		baseDir string
+		options []Option
+		mock    func(t *testing.T, s Storage)
+		expect  func(t *testing.T, s Storage, baseDir string)
+	}{
+		{
+			name:    "open file failed",
+			baseDir: os.TempDir(),
+			options: []Option{},
+			mock: func(t *testing.T, s Storage) {
+				s.(*storage).baseDir = "baz"
+			},
+			expect: func(t *testing.T, s Storage, baseDir string) {
+				assert := assert.New(t)
+				_, err := s.(*storage).openFile()
+				assert.Error(err)
+				s.(*storage).baseDir = baseDir
+			},
+		},
+		{
+			name:    "open new record file",
+			baseDir: os.TempDir(),
+			options: []Option{WithMaxSize(0), WithBufferSize(1)},
+			mock: func(t *testing.T, s Storage) {
+				if err := s.Create(Record{ID: "1"}); err != nil {
+					t.Fatal(err)
+				}
+
+				if err := s.Create(Record{ID: "2"}); err != nil {
+					t.Fatal(err)
+				}
+			},
+			expect: func(t *testing.T, s Storage, baseDir string) {
+				assert := assert.New(t)
+				file, err := s.(*storage).openFile()
+				assert.NoError(err)
+				assert.Equal(file.Name(), filepath.Join(baseDir, fmt.Sprintf("%s.%s", RecordFilePrefix, RecordFileExt)))
+				file.Close()
+
+				fileInfos, err := s.(*storage).backups()
+				assert.NoError(err)
+				assert.Equal(len(fileInfos), 3)
+				assert.Equal(fileInfos[2].Name(), fmt.Sprintf("%s.%s", RecordFilePrefix, RecordFileExt))
+			},
+		},
+		{
+			name:    "remove record file",
+			baseDir: os.TempDir(),
+			options: []Option{WithMaxSize(0), WithMaxBackups(1), WithBufferSize(1)},
+			mock: func(t *testing.T, s Storage) {
+				if err := s.Create(Record{ID: "1"}); err != nil {
+					t.Fatal(err)
+				}
+			},
+			expect: func(t *testing.T, s Storage, baseDir string) {
+				assert := assert.New(t)
+				file, err := s.(*storage).openFile()
+				assert.NoError(err)
+				assert.Equal(file.Name(), filepath.Join(baseDir, fmt.Sprintf("%s.%s", RecordFilePrefix, RecordFileExt)))
+				file.Close()
+
+				fileInfos, err := s.(*storage).backups()
+				assert.NoError(err)
+				assert.Equal(len(fileInfos), 1)
+				assert.Equal(fileInfos[0].Name(), fmt.Sprintf("%s.%s", RecordFilePrefix, RecordFileExt))
+			},
+		},
+	}
+
+	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.expect(t, s, tc.baseDir)
+			if err := s.Clear(); err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}
+
+func TestStorage_backupFilename(t *testing.T) {
+	baseDir := os.TempDir()
+	s, err := New(baseDir)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	filename := s.(*storage).backupFilename()
+	regexp := regexp.MustCompile(fmt.Sprintf("%s-.*.%s$", RecordFilePrefix, RecordFileExt))
+	assert := assert.New(t)
+	assert.True(regexp.MatchString(filename))
+
+	if err := s.Clear(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestStorage_backups(t *testing.T) {
+	tests := []struct {
+		name    string
+		baseDir string
+		options []Option
+		mock    func(t *testing.T, s Storage)
+		expect  func(t *testing.T, s Storage, baseDir string)
+	}{
+		{
+			name:    "open file failed",
+			baseDir: os.TempDir(),
+			options: []Option{},
+			mock: func(t *testing.T, s Storage) {
+				s.(*storage).baseDir = "bar"
+			},
+			expect: func(t *testing.T, s Storage, baseDir string) {
+				assert := assert.New(t)
+				_, err := s.(*storage).backups()
+				assert.Error(err)
+				s.(*storage).baseDir = baseDir
+				if err := s.Clear(); err != nil {
+					t.Fatal(err)
+				}
+			},
+		},
+		{
+			name:    "not found record file",
+			baseDir: os.TempDir(),
+			options: []Option{},
+			mock:    func(t *testing.T, s Storage) {},
+			expect: func(t *testing.T, s Storage, baseDir string) {
+				assert := assert.New(t)
+				if err := s.Clear(); err != nil {
+					t.Fatal(err)
+				}
+
+				_, err := s.(*storage).backups()
+				assert.Error(err)
+			},
+		},
+		{
+			name:    "get backups",
+			baseDir: os.TempDir(),
+			options: []Option{WithMaxSize(0), WithBufferSize(1)},
+			mock: func(t *testing.T, s Storage) {
+				if err := s.Create(Record{ID: "1"}); err != nil {
+					t.Fatal(err)
+				}
+
+				if err := s.Create(Record{ID: "2"}); 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) {
+				assert := assert.New(t)
+				backups, err := s.(*storage).backups()
+				assert.NoError(err)
+				assert.Equal(len(backups), 3)
+				assert.Equal(backups[2].Name(), fmt.Sprintf("%s.%s", RecordFilePrefix, RecordFileExt))
+				if err := s.Clear(); err != nil {
+					t.Fatal(err)
+				}
+			},
+		},
+	}
+
+	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.expect(t, s, tc.baseDir)
+		})
+	}
+}