feat: scheduler supports storage config (#1864)

Signed-off-by: Gaius <gaius.qi@gmail.com>

deploy/helm-charts

@@ -1 +1 @@
-Subproject commit b824a28e704776f0675409ac5dc997bb436acae5
+Subproject commit 91870893b4df549f0666bb35e280285be571c27c

scheduler/config/config.go

@@ -27,7 +27,6 @@ import (
 	"d7y.io/dragonfly/v2/pkg/net/ip"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/types"
-	"d7y.io/dragonfly/v2/scheduler/storage"
 )
 
 type Config struct {
@@ -335,9 +334,9 @@ func New() *Config {
 			},
 		},
 		Storage: StorageConfig{
-			MaxSize:    storage.DefaultMaxSize,
+			MaxSize:    DefaultStorageMaxSize,
-			MaxBackups: storage.DefaultMaxBackups,
+			MaxBackups: DefaultStorageMaxBackups,
-			BufferSize: storage.DefaultBufferSize,
+			BufferSize: DefaultStorageBufferSize,
 		},
 		Metrics: MetricsConfig{
 			Enable:         false,

scheduler/config/constants.go

@@ -121,3 +121,14 @@ var (
 	// DefaultNetworkEnableIPv6 is default value of enableIPv6.
 	DefaultNetworkEnableIPv6 = false
 )
+
+const (
+	// DefaultStorageMaxSize is the default maximum size of record file.
+	DefaultStorageMaxSize = 100
+
+	// DefaultStorageMaxBackups is the default maximum count of backup.
+	DefaultStorageMaxBackups = 10
+
+	// DefaultStorageBufferSize is the default size of buffer container.
+	DefaultStorageBufferSize = 100
+)

scheduler/scheduler.go

@@ -176,7 +176,12 @@ func New(ctx context.Context, cfg *config.Config, d dfpath.Dfpath) (*Server, err
 	scheduler := scheduler.New(&cfg.Scheduler, dynconfig, d.PluginDir())
 
 	// Initialize Storage.
-	storage, err := storage.New(d.DataDir())
+	storage, err := storage.New(
+		d.DataDir(),
+		cfg.Storage.MaxSize,
+		cfg.Storage.MaxBackups,
+		cfg.Storage.BufferSize,
+	)
 	if err != nil {
 		return nil, err
 	}

scheduler/storage/storage.go

@@ -37,17 +37,6 @@ import (
 	pkgio "d7y.io/dragonfly/v2/pkg/io"
 )
 
-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"
@@ -108,48 +97,18 @@ type storage struct {
 	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) {
+func New(baseDir string, maxSize, maxBackups, bufferSize int) (Storage, error) {
 	s := &storage{
 		baseDir:    baseDir,
 		filename:   filepath.Join(baseDir, fmt.Sprintf("%s.%s", RecordFilePrefix, RecordFileExt)),
-		maxSize:    DefaultMaxSize * megabyte,
+		maxSize:    int64(maxSize * megabyte),
-		maxBackups: DefaultMaxBackups,
+		maxBackups: maxBackups,
-		buffer:     make([]Record, 0, DefaultBufferSize),
+		buffer:     make([]Record, 0, bufferSize),
-		bufferSize: DefaultBufferSize,
+		bufferSize: bufferSize,
 		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

scheduler/storage/storage_test.go

@@ -29,6 +29,8 @@ import (
 
 	"github.com/gocarina/gocsv"
 	"github.com/stretchr/testify/assert"
+
+	"d7y.io/dragonfly/v2/scheduler/config"
 )
 
 var (
@@ -146,21 +148,19 @@ 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).maxSize, int64(config.DefaultStorageMaxSize*megabyte))
-				assert.Equal(s.(*storage).maxBackups, DefaultMaxBackups)
+				assert.Equal(s.(*storage).maxBackups, config.DefaultStorageMaxBackups)
-				assert.Equal(s.(*storage).bufferSize, DefaultBufferSize)
+				assert.Equal(s.(*storage).bufferSize, config.DefaultStorageBufferSize)
-				assert.Equal(cap(s.(*storage).buffer), DefaultBufferSize)
+				assert.Equal(cap(s.(*storage).buffer), config.DefaultStorageBufferSize)
 				assert.Equal(len(s.(*storage).buffer), 0)
 				assert.Equal(s.(*storage).count, int64(0))
 
@@ -169,68 +169,9 @@ func TestStorage_New(t *testing.T) {
 				}
 			},
 		},
-		{
-			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)
-				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)
@@ -240,7 +181,7 @@ func TestStorage_New(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			s, err := New(tc.baseDir, tc.options...)
+			s, err := New(tc.baseDir, config.DefaultStorageMaxSize, config.DefaultStorageMaxBackups, config.DefaultStorageBufferSize)
 			tc.expect(t, s, err)
 		})
 	}
@@ -250,14 +191,14 @@ func TestStorage_Create(t *testing.T) {
 	tests := []struct {
 		name       string
 		baseDir    string
-		options []Option
+		bufferSize int
 		mock       func(s Storage)
 		expect     func(t *testing.T, s Storage, baseDir string)
 	}{
 		{
 			name:       "create record",
 			baseDir:    os.TempDir(),
-			options: []Option{WithBufferSize(1)},
+			bufferSize: 1,
 			mock:       func(s Storage) {},
 			expect: func(t *testing.T, s Storage, baseDir string) {
 				assert := assert.New(t)
@@ -269,7 +210,7 @@ func TestStorage_Create(t *testing.T) {
 		{
 			name:       "create record without buffer",
 			baseDir:    os.TempDir(),
-			options: []Option{WithBufferSize(0)},
+			bufferSize: 0,
 			mock: func(s Storage) {
 			},
 			expect: func(t *testing.T, s Storage, baseDir string) {
@@ -282,7 +223,7 @@ func TestStorage_Create(t *testing.T) {
 		{
 			name:       "write record to file",
 			baseDir:    os.TempDir(),
-			options: []Option{WithBufferSize(1)},
+			bufferSize: 1,
 			mock: func(s Storage) {
 			},
 			expect: func(t *testing.T, s Storage, baseDir string) {
@@ -297,7 +238,7 @@ func TestStorage_Create(t *testing.T) {
 		{
 			name:       "open file failed",
 			baseDir:    os.TempDir(),
-			options: []Option{WithBufferSize(0)},
+			bufferSize: 0,
 			mock: func(s Storage) {
 				s.(*storage).baseDir = "foo"
 			},
@@ -312,7 +253,7 @@ func TestStorage_Create(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			s, err := New(tc.baseDir, tc.options...)
+			s, err := New(tc.baseDir, config.DefaultStorageMaxSize, config.DefaultStorageMaxBackups, tc.bufferSize)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -330,7 +271,7 @@ func TestStorage_List(t *testing.T) {
 	tests := []struct {
 		name       string
 		baseDir    string
-		options []Option
+		bufferSize int
 		record     Record
 		mock       func(t *testing.T, s Storage, baseDir string, record Record)
 		expect     func(t *testing.T, s Storage, baseDir string, record Record)
@@ -338,7 +279,7 @@ func TestStorage_List(t *testing.T) {
 		{
 			name:       "empty csv file given",
 			baseDir:    os.TempDir(),
-			options: []Option{},
+			bufferSize: config.DefaultStorageBufferSize,
 			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)
@@ -349,7 +290,7 @@ func TestStorage_List(t *testing.T) {
 		{
 			name:       "get file infos failed",
 			baseDir:    os.TempDir(),
-			options: []Option{},
+			bufferSize: config.DefaultStorageBufferSize,
 			mock: func(t *testing.T, s Storage, baseDir string, record Record) {
 				s.(*storage).baseDir = "bae"
 			},
@@ -363,7 +304,7 @@ func TestStorage_List(t *testing.T) {
 		{
 			name:       "open file failed",
 			baseDir:    os.TempDir(),
-			options: []Option{},
+			bufferSize: config.DefaultStorageBufferSize,
 			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 {
@@ -380,7 +321,7 @@ func TestStorage_List(t *testing.T) {
 		{
 			name:       "list records of a file",
 			baseDir:    os.TempDir(),
-			options: []Option{WithBufferSize(1)},
+			bufferSize: 1,
 			record:     mockRecord,
 			mock: func(t *testing.T, s Storage, baseDir string, record Record) {
 				if err := s.Create(record); err != nil {
@@ -404,7 +345,7 @@ func TestStorage_List(t *testing.T) {
 		{
 			name:       "list records of multi files",
 			baseDir:    os.TempDir(),
-			options: []Option{WithBufferSize(1)},
+			bufferSize: 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)
@@ -438,7 +379,7 @@ func TestStorage_List(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			s, err := New(tc.baseDir, tc.options...)
+			s, err := New(tc.baseDir, config.DefaultStorageMaxSize, config.DefaultStorageMaxBackups, tc.bufferSize)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -456,7 +397,7 @@ func TestStorage_Open(t *testing.T) {
 	tests := []struct {
 		name       string
 		baseDir    string
-		options []Option
+		bufferSize int
 		record     Record
 		mock       func(t *testing.T, s Storage, baseDir string, record Record)
 		expect     func(t *testing.T, s Storage, baseDir string, record Record)
@@ -464,7 +405,7 @@ func TestStorage_Open(t *testing.T) {
 		{
 			name:       "open storage withempty csv file given",
 			baseDir:    os.TempDir(),
-			options: []Option{},
+			bufferSize: config.DefaultStorageBufferSize,
 			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)
@@ -475,7 +416,7 @@ func TestStorage_Open(t *testing.T) {
 		{
 			name:       "open file infos failed",
 			baseDir:    os.TempDir(),
-			options: []Option{},
+			bufferSize: config.DefaultStorageBufferSize,
 			mock: func(t *testing.T, s Storage, baseDir string, record Record) {
 				s.(*storage).baseDir = "bas"
 			},
@@ -489,7 +430,7 @@ func TestStorage_Open(t *testing.T) {
 		{
 			name:       "open storage with records of a file",
 			baseDir:    os.TempDir(),
-			options: []Option{WithBufferSize(1)},
+			bufferSize: 1,
 			record:     mockRecord,
 			mock: func(t *testing.T, s Storage, baseDir string, record Record) {
 				if err := s.Create(record); err != nil {
@@ -518,7 +459,7 @@ func TestStorage_Open(t *testing.T) {
 		{
 			name:       "open storage with records of multi files",
 			baseDir:    os.TempDir(),
-			options: []Option{WithBufferSize(1)},
+			bufferSize: 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)
@@ -556,7 +497,7 @@ func TestStorage_Open(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			s, err := New(tc.baseDir, tc.options...)
+			s, err := New(tc.baseDir, config.DefaultStorageMaxSize, config.DefaultStorageMaxBackups, tc.bufferSize)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -574,14 +515,12 @@ 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)
@@ -602,7 +541,6 @@ func TestStorage_Clear(t *testing.T) {
 		{
 			name:    "open file failed",
 			baseDir: os.TempDir(),
-			options: []Option{},
 			mock: func(s Storage) {
 				s.(*storage).baseDir = "baz"
 			},
@@ -618,7 +556,7 @@ func TestStorage_Clear(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			s, err := New(tc.baseDir, tc.options...)
+			s, err := New(tc.baseDir, config.DefaultStorageMaxSize, config.DefaultStorageMaxBackups, config.DefaultStorageBufferSize)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -633,14 +571,12 @@ 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)
@@ -651,7 +587,6 @@ func TestStorage_create(t *testing.T) {
 		{
 			name:    "open file failed",
 			baseDir: os.TempDir(),
-			options: []Option{},
 			mock: func(s Storage) {
 				s.(*storage).baseDir = "foo"
 			},
@@ -666,7 +601,7 @@ func TestStorage_create(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			s, err := New(tc.baseDir, tc.options...)
+			s, err := New(tc.baseDir, config.DefaultStorageMaxSize, config.DefaultStorageMaxBackups, config.DefaultStorageBufferSize)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -684,14 +619,19 @@ func TestStorage_openFile(t *testing.T) {
 	tests := []struct {
 		name       string
 		baseDir    string
-		options []Option
+		maxSize    int
+		maxBackups int
+
+		bufferSize int
 		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{},
+			maxSize:    config.DefaultStorageMaxSize,
+			maxBackups: config.DefaultStorageMaxBackups,
+			bufferSize: config.DefaultStorageBufferSize,
 			mock: func(t *testing.T, s Storage) {
 				s.(*storage).baseDir = "bat"
 			},
@@ -705,7 +645,9 @@ func TestStorage_openFile(t *testing.T) {
 		{
 			name:       "open new record file",
 			baseDir:    os.TempDir(),
-			options: []Option{WithMaxSize(0), WithBufferSize(1)},
+			maxSize:    0,
+			maxBackups: config.DefaultStorageMaxBackups,
+			bufferSize: 1,
 			mock: func(t *testing.T, s Storage) {
 				if err := s.Create(Record{ID: "1"}); err != nil {
 					t.Fatal(err)
@@ -726,7 +668,9 @@ func TestStorage_openFile(t *testing.T) {
 		{
 			name:       "remove record file",
 			baseDir:    os.TempDir(),
-			options: []Option{WithMaxSize(0), WithMaxBackups(1), WithBufferSize(1)},
+			maxSize:    0,
+			maxBackups: 1,
+			bufferSize: 1,
 			mock: func(t *testing.T, s Storage) {
 				if err := s.Create(Record{ID: "1"}); err != nil {
 					t.Fatal(err)
@@ -744,7 +688,7 @@ func TestStorage_openFile(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			s, err := New(tc.baseDir, tc.options...)
+			s, err := New(tc.baseDir, tc.maxSize, tc.maxBackups, tc.bufferSize)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -760,7 +704,7 @@ func TestStorage_openFile(t *testing.T) {
 
 func TestStorage_backupFilename(t *testing.T) {
 	baseDir := os.TempDir()
-	s, err := New(baseDir)
+	s, err := New(baseDir, config.DefaultStorageMaxSize, config.DefaultStorageMaxBackups, config.DefaultStorageBufferSize)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -779,14 +723,12 @@ 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"
 			},
@@ -803,7 +745,6 @@ func TestStorage_backups(t *testing.T) {
 		{
 			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)
@@ -819,7 +760,7 @@ func TestStorage_backups(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			s, err := New(tc.baseDir, tc.options...)
+			s, err := New(tc.baseDir, config.DefaultStorageMaxSize, config.DefaultStorageMaxBackups, config.DefaultStorageBufferSize)
 			if err != nil {
 				t.Fatal(err)
 			}