Merge pull request #66 from zilliztech/dev

Support fallback to distributed lock for object storage services that do not support condition write

Author: tinswzy

common/config/configuration.go

@@ -17,9 +17,10 @@
 package config
 
 import (
-	"gopkg.in/yaml.v3"
 	"os"
 	"strings"
+
+	"gopkg.in/yaml.v3"
 )
 
 // MetaConfig stores the metadata storage configuration.
@@ -80,6 +81,31 @@ type RetentionPolicyConfig struct {
 	TTL int `yaml:"ttl"` // Time to live for truncated segments before eligible for GC
 }
 
+// FencePolicyConfig stores the fence policy configuration.
+type FencePolicyConfig struct {
+	ConditionWrite string `yaml:"conditionWrite"`
+}
+
+func (f *FencePolicyConfig) IsConditionWriteEnabled() bool {
+	return strings.EqualFold(f.ConditionWrite, "enable")
+}
+
+func (f *FencePolicyConfig) IsConditionWriteDisabled() bool {
+	return strings.EqualFold(f.ConditionWrite, "disable")
+}
+
+func (f *FencePolicyConfig) IsConditionWriteAuto() bool {
+	return strings.EqualFold(f.ConditionWrite, "auto")
+}
+
+func (f *FencePolicyConfig) SetConditionWriteEnableOrNot(enable bool) {
+	if enable {
+		f.ConditionWrite = "enable"
+	} else {
+		f.ConditionWrite = "disable"
+	}
+}
+
 // LogFileConfig stores the log file configuration.
 type LogFileConfig struct {
 	RootPath   string `yaml:"rootPath"`
@@ -203,6 +229,7 @@ type LogstoreConfig struct {
 	SegmentCompactionPolicy SegmentCompactionPolicy `yaml:"segmentCompactionPolicy"`
 	SegmentReadPolicy       SegmentReadPolicyConfig `yaml:"segmentReadPolicy"`
 	RetentionPolicy         RetentionPolicyConfig   `yaml:"retentionPolicy"`
+	FencePolicy             FencePolicyConfig       `yaml:"fencePolicy"`
 }
 
 type StorageConfig struct {
@@ -316,6 +343,9 @@ func getDefaultWoodpeckerConfig() WoodpeckerConfig {
 			RetentionPolicy: RetentionPolicyConfig{
 				TTL: 259200, // 72 hours
 			},
+			FencePolicy: FencePolicyConfig{
+				ConditionWrite: "auto",
+			},
 		},
 		Storage: StorageConfig{
 			Type:     "default",

common/config/configuration_test.go

@@ -56,6 +56,7 @@ func TestNewConfiguration(t *testing.T) {
 	assert.Equal(t, int64(16000000), config.Woodpecker.Logstore.SegmentReadPolicy.MaxBatchSize)
 	assert.Equal(t, 32, config.Woodpecker.Logstore.SegmentReadPolicy.MaxFetchThreads)
 	assert.Equal(t, 259200, config.Woodpecker.Logstore.RetentionPolicy.TTL) // 72h = 259200s
+	assert.Equal(t, "auto", config.Woodpecker.Logstore.FencePolicy.ConditionWrite)
 	assert.Equal(t, "minio", config.Woodpecker.Storage.Type)
 	assert.Equal(t, "/var/lib/woodpecker", config.Woodpecker.Storage.RootPath)
 	assert.Equal(t, "info", config.Log.Level)
@@ -127,6 +128,7 @@ func TestNewConfiguration(t *testing.T) {
 	assert.Equal(t, int64(16000000), defaultConfig.Woodpecker.Logstore.SegmentReadPolicy.MaxBatchSize)
 	assert.Equal(t, 32, defaultConfig.Woodpecker.Logstore.SegmentReadPolicy.MaxFetchThreads)
 	assert.Equal(t, 259200, defaultConfig.Woodpecker.Logstore.RetentionPolicy.TTL) // 72h = 259200s
+	assert.Equal(t, "auto", defaultConfig.Woodpecker.Logstore.FencePolicy.ConditionWrite)
 	assert.Equal(t, "default", defaultConfig.Woodpecker.Storage.Type)
 	assert.Equal(t, "/tmp/woodpecker", defaultConfig.Woodpecker.Storage.RootPath)
 	assert.Equal(t, "info", defaultConfig.Log.Level)

common/etcd/etcd_server.go

@@ -49,27 +49,7 @@ func InitEtcdServer(
 	if useEmbedEtcd {
 		var initError error
 		initOnce.Do(func() {
-			path := configPath
-			var cfg *embed.Config
-			if len(path) > 0 {
-				cfgFromFile, err := embed.ConfigFromFile(path)
-				if err != nil {
-					initError = err
-				}
-				cfg = cfgFromFile
-			} else {
-				cfg = embed.NewConfig()
-			}
-			cfg.Dir = dataDir
-			cfg.LogOutputs = []string{logPath}
-			cfg.LogLevel = logLevel
-			e, err := embed.StartEtcd(cfg)
-			if err != nil {
-				log.Printf("failed to init embedded Etcd server %v", err)
-				initError = err
-			}
-			etcdServer = e
-			log.Printf("finish init Etcd config path:%s, dataDir:%s", path, dataDir)
+			initError = StartEtcdServerUnsafe(useEmbedEtcd, configPath, dataDir, logPath, logLevel)
 		})
 		return initError
 	}
@@ -88,3 +68,37 @@ func StopEtcdServer() {
 		})
 	}
 }
+
+func StartEtcdServerUnsafe(useEmbedEtcd bool,
+	configPath string,
+	dataDir string,
+	logPath string,
+	logLevel string) error {
+	var initError error
+	path := configPath
+	var cfg *embed.Config
+	if len(path) > 0 {
+		cfgFromFile, err := embed.ConfigFromFile(path)
+		if err != nil {
+			initError = err
+		}
+		cfg = cfgFromFile
+	} else {
+		cfg = embed.NewConfig()
+	}
+	cfg.Dir = dataDir
+	cfg.LogOutputs = []string{logPath}
+	cfg.LogLevel = logLevel
+	e, err := embed.StartEtcd(cfg)
+	if err != nil {
+		log.Printf("failed to init embedded Etcd server %v", err)
+		initError = err
+	}
+	etcdServer = e
+	log.Printf("finish init Etcd config path:%s, dataDir:%s", path, dataDir)
+	return initError
+}
+
+func ShutdownEtcdServerUnsafe() {
+	etcdServer.Close()
+}