pump/: Add syn-log option

cmd/pump/pump.toml

@@ -27,6 +27,9 @@ pd-urls = "http://127.0.0.1:2379"
 # Path of file that contains X509 key in PEM format for connection with cluster components.
 # ssl-key = "/path/to/drainer-key.pem"
 #
+# [storage]
+# Set to `true` (default) for best reliability, which prevents data loss when there is a power failure.
+# sync-log = true
 #
 # we suggest using the default config of the embedded LSM DB now, do not change it useless you know what you are doing
 # [storage.kv]
@@ -36,7 +39,7 @@ pd-urls = "http://127.0.0.1:2379"
 # compaction-L0-trigger = 8
 # compaction-table-size = 67108864
 # compaction-total-size = 536870912
-# compaction-total-size-multiplier = 8
+# compaction-total-size-multiplier = 8.0
 # write-buffer = 67108864
 # write-L0-pause-trigger = 24
 # write-L0-slowdown-trigger = 17

pump/config.go

@@ -64,7 +64,7 @@ type Config struct {
 	configFile      string
 	printVersion    bool
 	tls             *tls.Config
-	Storage         *storage.Config `toml:"storage" json:"storage"`
+	Storage         storage.Config `toml:"storage" json:"storage"`
 }
 
 // NewConfig return an instance of configuration

pump/server.go

@@ -122,7 +122,10 @@ func NewServer(cfg *Config) (*Server, error) {
 		return nil, errors.Trace(err)
 	}
 
-	options := storage.DefaultOptions().WithStorage(cfg.Storage)
+	options := storage.DefaultOptions()
+	options = options.WithKVConfig(cfg.Storage.KV)
+	options = options.WithSync(cfg.Storage.GetSyncLog())
+
 	storage, err := storage.NewAppendWithResolver(cfg.DataDir, options, tiStore, lockResolver)
 	if err != nil {
 		return nil, errors.Trace(err)

pump/storage/storage.go

@@ -100,6 +100,8 @@ func NewAppendWithResolver(dir string, options *Options, tiStore kv.Storage, tiL
 		options = DefaultOptions()
 	}
 
+	log.Infof("options: %+v", options)
+
 	valueDir := path.Join(dir, "value")
 	err = os.MkdirAll(valueDir, 0755)
 	if err != nil {
@@ -113,7 +115,7 @@ func NewAppendWithResolver(dir string, options *Options, tiStore kv.Storage, tiL
 	}
 
 	kvDir := path.Join(dir, "kv")
-	metadata, err := openMetadataDB(kvDir, options.Storage)
+	metadata, err := openMetadataDB(kvDir, options.KVConfig)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -992,7 +994,17 @@ func getStorageSize(dir string) (size storageSize, err error) {
 
 // Config holds the configuration of storage
 type Config struct {
-	KVConfig `toml:"kv" json:"kv"`
+	SyncLog *bool     `toml:"sync-log" json:"sync-log"`
+	KV      *KVConfig `toml:"kv" json:"kv"`
+}
+
+// GetSyncLog return sync-log config option
+func (c *Config) GetSyncLog() bool {
+	if c.SyncLog == nil {
+		return true
+	}
+
+	return *c.SyncLog
 }
 
 // KVConfig if the configuration of goleveldb
@@ -1009,57 +1021,55 @@ type KVConfig struct {
 	WriteL0SlowdownTrigger        int     `toml:"write-L0-slowdown-trigger" json:"write-L0-slowdown-trigger"`
 }
 
-var defaultStorageConfig = &Config{
-	KVConfig: KVConfig{
-		BlockCacheCapacity:            8388608,
-		BlockRestartInterval:          16,
-		BlockSize:                     4096,
-		CompactionL0Trigger:           8,
-		CompactionTableSize:           67108864,
-		CompactionTotalSize:           536870912,
-		CompactionTotalSizeMultiplier: 8,
-		WriteBuffer:                   67108864,
-		WriteL0PauseTrigger:           24,
-		WriteL0SlowdownTrigger:        17,
-	},
+var defaultStorageKVConfig = &KVConfig{
+	BlockCacheCapacity:            8388608,
+	BlockRestartInterval:          16,
+	BlockSize:                     4096,
+	CompactionL0Trigger:           8,
+	CompactionTableSize:           67108864,
+	CompactionTotalSize:           536870912,
+	CompactionTotalSizeMultiplier: 8,
+	WriteBuffer:                   67108864,
+	WriteL0PauseTrigger:           24,
+	WriteL0SlowdownTrigger:        17,
 }
 
-func setDefaultStorageConfig(cf *Config) {
+func setDefaultStorageConfig(cf *KVConfig) {
 	if cf.BlockCacheCapacity <= 0 {
-		cf.BlockCacheCapacity = defaultStorageConfig.BlockCacheCapacity
+		cf.BlockCacheCapacity = defaultStorageKVConfig.BlockCacheCapacity
 	}
 	if cf.BlockRestartInterval <= 0 {
-		cf.BlockRestartInterval = defaultStorageConfig.BlockRestartInterval
+		cf.BlockRestartInterval = defaultStorageKVConfig.BlockRestartInterval
 	}
 	if cf.BlockSize <= 0 {
-		cf.BlockSize = defaultStorageConfig.BlockSize
+		cf.BlockSize = defaultStorageKVConfig.BlockSize
 	}
 	if cf.CompactionL0Trigger <= 0 {
-		cf.CompactionL0Trigger = defaultStorageConfig.CompactionL0Trigger
+		cf.CompactionL0Trigger = defaultStorageKVConfig.CompactionL0Trigger
 	}
 	if cf.CompactionTableSize <= 0 {
-		cf.CompactionTableSize = defaultStorageConfig.CompactionTableSize
+		cf.CompactionTableSize = defaultStorageKVConfig.CompactionTableSize
 	}
 	if cf.CompactionTotalSize <= 0 {
-		cf.CompactionTotalSize = defaultStorageConfig.CompactionTotalSize
+		cf.CompactionTotalSize = defaultStorageKVConfig.CompactionTotalSize
 	}
 	if cf.CompactionTotalSizeMultiplier <= 0 {
-		cf.CompactionTotalSizeMultiplier = defaultStorageConfig.CompactionTotalSizeMultiplier
+		cf.CompactionTotalSizeMultiplier = defaultStorageKVConfig.CompactionTotalSizeMultiplier
 	}
 	if cf.WriteBuffer <= 0 {
-		cf.WriteBuffer = defaultStorageConfig.WriteBuffer
+		cf.WriteBuffer = defaultStorageKVConfig.WriteBuffer
 	}
 	if cf.WriteL0PauseTrigger <= 0 {
-		cf.WriteL0PauseTrigger = defaultStorageConfig.WriteL0PauseTrigger
+		cf.WriteL0PauseTrigger = defaultStorageKVConfig.WriteL0PauseTrigger
 	}
 	if cf.WriteL0SlowdownTrigger <= 0 {
-		cf.WriteL0SlowdownTrigger = defaultStorageConfig.WriteL0SlowdownTrigger
+		cf.WriteL0SlowdownTrigger = defaultStorageKVConfig.WriteL0SlowdownTrigger
 	}
 }
 
-func openMetadataDB(kvDir string, cf *Config) (*leveldb.DB, error) {
+func openMetadataDB(kvDir string, cf *KVConfig) (*leveldb.DB, error) {
 	if cf == nil {
-		cf = defaultStorageConfig
+		cf = defaultStorageKVConfig
 	} else {
 		setDefaultStorageConfig(cf)
 	}

pump/storage/storage_test.go

@@ -322,15 +322,13 @@ func (s *OpenDBSuit) TestWhenConfigIsNotProvided(c *check.C) {
 }
 
 func (s *OpenDBSuit) TestProvidedConfigValsNotOverwritten(c *check.C) {
-	cf := Config{
-		KVConfig: KVConfig{
-			BlockRestartInterval: 32,
-			WriteL0PauseTrigger:  12,
-		},
+	cf := KVConfig{
+		BlockRestartInterval: 32,
+		WriteL0PauseTrigger:  12,
 	}
 	_, err := openMetadataDB(s.dir, &cf)
 	c.Assert(err, check.IsNil)
 	c.Assert(cf.BlockRestartInterval, check.Equals, 32)
 	c.Assert(cf.WriteL0PauseTrigger, check.Equals, 12)
-	c.Assert(cf.BlockCacheCapacity, check.Equals, defaultStorageConfig.BlockCacheCapacity)
+	c.Assert(cf.BlockCacheCapacity, check.Equals, defaultStorageKVConfig.BlockCacheCapacity)
 }

pump/storage/vlog.go

@@ -29,7 +29,7 @@ type Options struct {
 	ValueLogFileSize int64
 	Sync             bool
 
-	Storage *Config
+	KVConfig *KVConfig
 }
 
 // DefaultOptions return the default options
@@ -41,8 +41,8 @@ func DefaultOptions() *Options {
 }
 
 // WithStorage set the Config
-func (o *Options) WithStorage(storage *Config) *Options {
-	o.Storage = storage
+func (o *Options) WithKVConfig(kvConfig *KVConfig) *Options {
+	o.KVConfig = kvConfig
 	return o
 }