Merge branch 'new_etcd' into 'master'

New etcd

See merge request !145

Author: CMGS

core.yaml.sample

@@ -28,8 +28,6 @@ docker:
     local_dns: true
 
 scheduler:
-    lock_key: "_scheduler_lock"
-    lock_ttl: 10
     maxshare: -1
     sharebase: 10
 

lock/etcdlock/lock.go

@@ -3,53 +3,52 @@ package etcdlock
 import (
 	"fmt"
 	"strings"
-	"sync"
+	"time"
 
 	"github.com/coreos/etcd/clientv3"
 	"github.com/coreos/etcd/clientv3/concurrency"
 	"golang.org/x/net/context"
 )
 
-const defaultTTL = 60
+const defaultTTL = 30
 
 type Mutex struct {
-	mutex   sync.Mutex
-	lock    *concurrency.Mutex
+	timeout time.Duration
+	mutex   *concurrency.Mutex
 	session *concurrency.Session
 }
 
 func New(cli *clientv3.Client, key string, ttl int) (*Mutex, error) {
 	if key == "" {
-		return nil, fmt.Errorf("Where is lock key")
+		return nil, fmt.Errorf("[etcdlock] no lock key")
 	}
 
 	if !strings.HasPrefix(key, "/") {
 		key = fmt.Sprintf("/%s", key)
 	}
 
-	if ttl < 1 {
+	if ttl <= 0 {
 		ttl = defaultTTL
 	}
 
-	s, err := concurrency.NewSession(
-		cli,
-		concurrency.WithTTL(ttl),
-		concurrency.WithContext(context.Background()),
-	)
+	session, err := concurrency.NewSession(cli, concurrency.WithTTL(ttl))
 	if err != nil {
 		return nil, err
 	}
-	lock := concurrency.NewMutex(s, key)
-	return &Mutex{mutex: sync.Mutex{}, lock: lock, session: s}, nil
+
+	mutex := &Mutex{mutex: concurrency.NewMutex(session, key), session: session}
+	mutex.timeout = time.Duration(ttl) * time.Second
+	return mutex, nil
 }
 
 func (m *Mutex) Lock() error {
-	m.mutex.Lock()
-	return m.lock.Lock(context.TODO())
+	ctx, cancel := context.WithTimeout(context.Background(), m.timeout)
+	defer cancel()
+	return m.mutex.Lock(ctx)
 }
 
 func (m *Mutex) Unlock() error {
-	defer m.mutex.Unlock()
 	defer m.session.Close()
-	return m.lock.Unlock(context.TODO())
+	// 一定要释放
+	return m.mutex.Unlock(context.Background())
 }

rpc/rpc_test.go

@@ -200,8 +200,8 @@ func initConfig(mStore *mockstore.MockStore) (types.Config, *vibranium) {
 			SCMType: "gitlab",
 		},
 		Scheduler: types.SchedConfig{
-			LockKey: "_scheduler_lock",
-			LockTTL: 10,
+			MaxShare:  -1,
+			ShareBase: 10,
 		},
 		Syslog: types.SyslogConfig{
 			Address:  "udp://localhost:5111",

scheduler/complex/potassium_test.go

@@ -23,8 +23,6 @@ func newPotassium() (*potassium, error) {
 		EtcdMachines:   []string{"http://127.0.0.1:2379"},
 		EtcdLockPrefix: "/eru-core/_lock",
 		Scheduler: types.SchedConfig{
-			LockKey:   "/coretest",
-			LockTTL:   1,
 			ShareBase: 10,
 			MaxShare:  -1,
 		},
@@ -186,8 +184,6 @@ func TestComplexNodes(t *testing.T) {
 		EtcdMachines:   []string{"http://127.0.0.1:2379"},
 		EtcdLockPrefix: "/eru-core/_lock",
 		Scheduler: types.SchedConfig{
-			LockKey:   "/coretest",
-			LockTTL:   1,
 			ShareBase: 10,
 			MaxShare:  -1,
 		},

scheduler/scheduler_test.go

@@ -13,8 +13,6 @@ func TestSchedulerInvoke(t *testing.T) {
 		EtcdMachines:   []string{"http://127.0.0.1:2379"},
 		EtcdLockPrefix: "/eru-core/_lock",
 		Scheduler: types.SchedConfig{
-			LockKey:   "/coretest",
-			LockTTL:   1,
 			ShareBase: 10,
 			MaxShare:  -1,
 		},

types/config.go

@@ -43,10 +43,8 @@ type DockerConfig struct {
 
 // SchedConfig holds scheduler config
 type SchedConfig struct {
-	LockKey   string `yaml:"lock_key"`  // key for etcd lock
-	LockTTL   int    `yaml:"lock_ttl"`  // TTL for etcd lock
-	MaxShare  int64  `yaml:"maxshare"`  // comlpex scheduler use maxshare
-	ShareBase int64  `yaml:"sharebase"` // how many pieces for one core
+	MaxShare  int64 `yaml:"maxshare"`  // comlpex scheduler use maxshare
+	ShareBase int64 `yaml:"sharebase"` // how many pieces for one core
 }
 
 // SyslogConfig 用于debug模式容器的日志收集