Merge pull request #43 from supermeng/master

Optimized for timeout

Author: supermeng

apiserver/server.go

@@ -181,7 +181,7 @@ func initOrcEngine(swarmAddr string, etcdAddr string, isDebug bool) (*engine.Orc
 		return nil, err
 	}
 
-	cluster, err := swarm.NewCluster(swarmAddr, 30*time.Second, 90*time.Second)
+	cluster, err := swarm.NewCluster(swarmAddr, 10*time.Second, 20*time.Second)
 	if err != nil {
 		return nil, err
 	}

engine/config.go

@@ -27,6 +27,8 @@ type Guard struct {
 const (
 	EtcdResourcesKey   = "/lain/config/resources"
 	EtcdGuardSwitchKey = "/lain/config/guardswitch"
+	EtcdCloudVolumeRootKey = "/lain/config/cloud_volumes_root"
+	EtcdVolumeRootKey      = "/lain/config/volumes_root"
 
 	EtcdConfigKey = "/lain/deployd/engine/config"
 

engine/engine.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/laincloud/deployd/cluster"
@@ -262,7 +263,7 @@ func (engine *OrcEngine) RescheduleSpec(name string, podSpec PodSpec) error {
 	if pgCtrl, ok := engine.pgCtrls[name]; !ok {
 		return ErrPodGroupNotExists
 	} else {
-		if err := canOperation(pgCtrl, PGOpStateScheduling); err != nil {
+		if err := canOperation(pgCtrl, PGOpStateUpgrading); err != nil {
 			return err
 		}
 		for _, depends := range podSpec.Dependencies {
@@ -567,10 +568,7 @@ func (engine *OrcEngine) initOperationWorker() {
 				rInterval := RefreshInterval / 2 * 1000 / len(engine.pgCtrls)
 				index := 0
 				for _, pgCtrl := range engine.pgCtrls {
-					pgCtrl.RLock()
-					refreshable := pgCtrl.refreshable
-					pgCtrl.RUnlock()
-					if refreshable {
+					if atomic.LoadInt32(&(pgCtrl.refreshable)) == 1 {
 						interval := index * rInterval
 						_pgCtrl := pgCtrl
 						index++
@@ -652,8 +650,8 @@ func (engine *OrcEngine) checkPodGroupRemoveResult(name string, pgCtrl *podGroup
 }
 
 func canOperation(pgCtrl *podGroupController, target PGOpState) error {
-	if opState := pgCtrl.CanOperate(target); opState != PGOpStateIdle {
-		return OperLockedError{info: opState.String()}
+	if canOp := pgCtrl.CanOperate(target); !canOp {
+		return OperLockedError{info: "Scheduling"}
 	}
 	return nil
 }

engine/events.go

@@ -3,6 +3,7 @@ package engine
 import (
 	"strings"
 	"sync"
+	"sync/atomic"
 
 	"github.com/laincloud/deployd/utils/util"
 	"github.com/mijia/adoc"
@@ -76,13 +77,10 @@ func handleDieEvent(engine *OrcEngine, event *adoc.Event) {
 			if !ok {
 				return
 			}
-
 			pgCtrl.RLock()
-			state := pgCtrl.opState
 			spec := pgCtrl.spec.Clone()
 			pgCtrl.RUnlock()
-
-			if state != PGOpStateScheduling {
+			if atomic.LoadInt32((*int32)(&pgCtrl.opState)) != PGOpStateUpgrading {
 				log.Warnf("got %s event from %s, refresh this instance", event.Status, name)
 				pgCtrl.opsChan <- pgOperRefreshInstance{instance, spec}
 			}

engine/pod.go

@@ -22,7 +22,7 @@ var RestartInfoClearInterval time.Duration
 
 const (
 	DefaultHealthInterval = 10
-	DefaultHealthTimeout  = 1
+	DefaultHealthTimeout  = 3
 	DefaultHealthRetries  = 3
 
 	DefaultSetUpTime = 20
@@ -32,7 +32,7 @@ const (
 	CPUMaxLevel     = 8
 	CPUDeafultLevel = 2
 
-	CURL_TMPLT = `echo $(timeout %v curl -s -o /dev/null -w '%%{http_code}\n' %s) | grep -Eq "^[2-3]..$"`
+	CURL_TMPLT = `echo $(curl -m %v -s -o /dev/null -w '%%{http_code}\n' %s) | grep -Eq "^[2-3]..$"`
 )
 
 // podController is controlled by the podGroupController
@@ -522,10 +522,12 @@ func (pc *podController) createHostConfig(index int) adoc.HostConfig {
 			BlkioDeviceWriteIOps = append(BlkioDeviceWriteIOps, iops)
 		}
 	}
+	swappiness := int64(0)
 	hc := adoc.HostConfig{
 		Resources: adoc.Resources{
 			Memory:               spec.MemoryLimit,
 			MemorySwap:           spec.MemoryLimit, // Memory == MemorySwap means disable swap
+			MemorySwappiness: 	  &swappiness,
 			CPUPeriod:            CPUQuota,
 			CPUQuota:             int64(spec.CpuLimit*resource.Cpu*CPUMaxPctg) * CPUQuota / int64(CPUMaxLevel*100),
 			BlkioDeviceReadBps:   BlkioDeviceReadBps,

engine/podgroup.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/laincloud/deployd/cluster"
@@ -39,7 +40,7 @@ type podGroupController struct {
 	podCtrls   []*podController
 	opsChan    chan pgOperation
 
-	refreshable bool
+	refreshable int32
 
 	lastPodSpecKey string
 	storedKey      string
@@ -50,33 +51,32 @@ func (pgCtrl *podGroupController) String() string {
 	return fmt.Sprintf("PodGroupCtrl %s", pgCtrl.spec)
 }
 
-func (pgCtrl *podGroupController) CanOperate(pgops PGOpState) PGOpState {
-	pgCtrl.Lock()
-	defer pgCtrl.Unlock()
-	if pgCtrl.opState == PGOpStateIdle {
-		pgCtrl.opState = pgops
-		return PGOpStateIdle
+func (pgCtrl *podGroupController) CanOperate(pgops PGOpState) bool {
+	if atomic.CompareAndSwapInt32((*int32)(&pgCtrl.opState), PGOpStateIdle, int32(pgops)) {
+		pgCtrl.DisableRefresh()
+		return true
+	} else if atomic.LoadInt32((*int32)(&pgCtrl.opState)) == PGOpStateUpgrading &&
+		pgops == PGOpStateUpgrading {
+		// when pg is in upgreading state so flush old opchans and start new op
+		pgCtrl.DisableRefresh()
+		return true
 	}
-	return pgCtrl.opState
+	return false
 }
 
 func (pgCtrl *podGroupController) DisableRefresh() {
-	pgCtrl.Lock()
-	defer pgCtrl.Unlock()
-	pgCtrl.refreshable = false
+	atomic.StoreInt32(&(pgCtrl.refreshable), int32(0))
 }
 
 func (pgCtrl *podGroupController) EnableRefresh() {
-	pgCtrl.Lock()
-	defer pgCtrl.Unlock()
-	pgCtrl.refreshable = true
+	atomic.StoreInt32(&(pgCtrl.refreshable), int32(1))
 }
 
 // called by signle goroutine
 func (pgCtrl *podGroupController) OperateOver() {
-	pgCtrl.Lock()
-	defer pgCtrl.Unlock()
-	pgCtrl.opState = PGOpStateIdle
+	pgCtrl.emitOperationEvent(OperationOver)
+	atomic.StoreInt32((*int32)(&pgCtrl.opState), PGOpStateIdle)
+	pgCtrl.EnableRefresh()
 }
 
 func (pgCtrl *podGroupController) Inspect() PodGroupWithSpec {
@@ -112,6 +112,7 @@ func (pgCtrl *podGroupController) IsPending() bool {
 }
 
 func (pgCtrl *podGroupController) Deploy() {
+	pgCtrl.flushAllOps()
 	pgCtrl.emitOperationEvent(OperationStart)
 	defer func() {
 		pgCtrl.opsChan <- pgOperOver{}
@@ -139,6 +140,7 @@ func (pgCtrl *podGroupController) Deploy() {
 }
 
 func (pgCtrl *podGroupController) RescheduleInstance(numInstances int, restartPolicy ...RestartPolicy) {
+	pgCtrl.flushAllOps()
 	pgCtrl.emitOperationEvent(OperationStart)
 	defer func() {
 		pgCtrl.opsChan <- pgOperOver{}
@@ -190,17 +192,14 @@ func (pgCtrl *podGroupController) RescheduleInstance(numInstances int, restartPo
 }
 
 func (pgCtrl *podGroupController) RescheduleSpec(podSpec PodSpec) {
+	pgCtrl.flushAllOps()
 	pgCtrl.emitOperationEvent(OperationStart)
 	defer func() {
 		pgCtrl.opsChan <- pgOperOver{}
 	}()
 	pgCtrl.RLock()
 	spec := pgCtrl.spec.Clone()
 	pgCtrl.RUnlock()
-	// pgCtrl.group.Pods[0].NodeName()
-	if spec.Pod.Equals(podSpec) {
-		return
-	}
 	pgCtrl.emptyError()
 	if ok := pgCtrl.updatePodPorts(podSpec); !ok {
 		return
@@ -230,6 +229,7 @@ func (pgCtrl *podGroupController) RescheduleSpec(podSpec PodSpec) {
 }
 
 func (pgCtrl *podGroupController) RescheduleDrift(fromNode, toNode string, instanceNo int, force bool) {
+	pgCtrl.flushAllOps()
 	defer func() {
 		pgCtrl.opsChan <- pgOperOver{}
 	}()
@@ -254,6 +254,7 @@ func (pgCtrl *podGroupController) RescheduleDrift(fromNode, toNode string, insta
 }
 
 func (pgCtrl *podGroupController) Remove() {
+	pgCtrl.flushAllOps()
 	pgCtrl.emitOperationEvent(OperationStart)
 	defer func() {
 		pgCtrl.opsChan <- pgOperOver{}
@@ -273,6 +274,7 @@ func (pgCtrl *podGroupController) Remove() {
 }
 
 func (pgCtrl *podGroupController) ChangeState(op string, instance int) {
+	pgCtrl.flushAllOps()
 	pgCtrl.emitOperationEvent(OperationStart)
 	defer func() {
 		pgCtrl.opsChan <- pgOperOver{}
@@ -295,6 +297,10 @@ func (pgCtrl *podGroupController) Refresh(force bool) {
 	if pgCtrl.IsRemoved() || pgCtrl.IsPending() {
 		return
 	}
+	pgCtrl.DisableRefresh()
+	defer func() {
+		pgCtrl.opsChan <- pgOperOver{}
+	}()
 	pgCtrl.RLock()
 	spec := pgCtrl.spec.Clone()
 	pgCtrl.RUnlock()
@@ -328,11 +334,22 @@ func (pgCtrl *podGroupController) Activate(c cluster.Cluster, store storage.Stor
 	}()
 }
 
+func (pgCtrl *podGroupController) LastSpec() *PodGroupSpec {
+	log.Infof("Fetch LastPodSpec !")
+	var lastSpec PodGroupSpec
+	if err := pgCtrl.engine.store.Get(pgCtrl.lastPodSpecKey, &lastSpec); err != nil {
+		log.Infof("Fetch LastPodSpec with err:%v", err)
+		return nil
+	}
+	log.Infof("Fetch LastPodSpec :%v", lastSpec)
+	return &lastSpec
+}
+
 /*
  * clean all ops in chan synchronously
  *
  */
-func (pgCtrl *podGroupController) FlushAllOps() {
+func (pgCtrl *podGroupController) flushAllOps() {
 	for {
 		if len(pgCtrl.opsChan) == 0 {
 			return
@@ -345,17 +362,6 @@ func (pgCtrl *podGroupController) FlushAllOps() {
 	}
 }
 
-func (pgCtrl *podGroupController) LastSpec() *PodGroupSpec {
-	log.Infof("Fetch LastPodSpec !")
-	var lastSpec PodGroupSpec
-	if err := pgCtrl.engine.store.Get(pgCtrl.lastPodSpecKey, &lastSpec); err != nil {
-		log.Infof("Fetch LastPodSpec with err:%v", err)
-		return nil
-	}
-	log.Infof("Fetch LastPodSpec :%v", lastSpec)
-	return &lastSpec
-}
-
 /*
  * To clean corrupted containers which do not used by cluster app any more
  * Should be called just after refrehsed podgroups or clean will works terrible
@@ -492,9 +498,6 @@ func (pgCtrl *podGroupController) checkPodPorts() bool {
 
 func (pgCtrl *podGroupController) updatePodPorts(podSpec PodSpec) bool {
 	spec := pgCtrl.spec
-	if spec.Pod.Equals(podSpec) {
-		return true
-	}
 	var oldsps, sps StreamPorts
 	if err := json.Unmarshal([]byte(spec.Pod.Annotation), &oldsps); err != nil {
 		log.Errorf("annotation unmarshal error:%v\n", err)
@@ -573,7 +576,7 @@ func (pgCtrl *podGroupController) rollBack() bool {
 		// 1. disable refresh(so no others can produce operation) and flush ops chan
 		log.Infof("Start Rollback!")
 		pgCtrl.DisableRefresh()
-		pgCtrl.FlushAllOps()
+		pgCtrl.flushAllOps()
 		// 2. rollback podgroup podspec info
 		pgCtrl.Lock()
 		spec := pgCtrl.spec.Clone()
@@ -692,7 +695,7 @@ func newPodGroupController(spec PodGroupSpec, states []PodPrevState, pg PodGroup
 		podCtrls: podCtrls,
 		opsChan:  make(chan pgOperation, 500),
 
-		refreshable: true,
+		refreshable: 1,
 
 		lastPodSpecKey: strings.Join([]string{kLainDeploydRootKey, kLainLastPodSpecKey, spec.Namespace, spec.Name}, "/"),
 		storedKey:      strings.Join([]string{kLainDeploydRootKey, kLainPodGroupKey, spec.Namespace, spec.Name}, "/"),

engine/podgroup_ops.go

@@ -158,6 +158,10 @@ func (op pgOperRefreshInstance) Do(pgCtrl *podGroupController, c cluster.Cluster
 		pgCtrl.RUnlock()
 	}()
 
+	if(op.instanceNo > len(pgCtrl.podCtrls)){
+		log.Warnf("Pod is not exists")
+		return false
+	}
 	podCtrl := pgCtrl.podCtrls[op.instanceNo-1]
 
 	podCtrl.Refresh(c)
@@ -604,6 +608,5 @@ type pgOperOver struct {
 
 func (op pgOperOver) Do(pgCtrl *podGroupController, c cluster.Cluster, store storage.Store, ev *RuntimeEagleView) bool {
 	pgCtrl.OperateOver()
-	pgCtrl.emitOperationEvent(OperationOver)
 	return false
 }

engine/runtimes.go

@@ -39,6 +39,7 @@ const (
 
 const (
 	PGOpStateIdle = iota
+	PGOpStateUpgrading
 	PGOpStateScheduling
 	PGOpStateDrifting
 	PGOpStateRemoving
@@ -102,6 +103,8 @@ func (pgos PGOpState) String() string {
 	switch pgos {
 	case PGOpStateIdle:
 		return "Idle"
+	case PGOpStateUpgrading:
+		return "Upgrading"
 	case PGOpStateScheduling:
 		return "Scheduling"
 	case PGOpStateDrifting:

engine/specs.go

@@ -32,8 +32,6 @@ const (
 	MinPodKillTimeout = 10
 	MaxPodKillTimeout = 120
 
-	etcdCloudVolumeRootKey = "/lain/config/cloud_volumes_root"
-	etcdVolumeRootKey      = "/lain/config/volumes_root"
 )
 
 var (
@@ -60,11 +58,11 @@ type ContainerLabel struct {
 }
 
 func configSpecsVars(store storage.Store) error {
-	if v, err := store.GetRaw(etcdCloudVolumeRootKey); err == nil {
+	if v, err := store.GetRaw(EtcdCloudVolumeRootKey); err == nil {
 		kLainCloudVolumeRoot = v
 	}
 
-	if v, err := store.GetRaw(etcdVolumeRootKey); err == nil {
+	if v, err := store.GetRaw(EtcdVolumeRootKey); err == nil {
 		kLainVolumeRoot = v
 	}
 	log.Debugf("cloud_volume_root: %s, volumes_root: %s", kLainCloudVolumeRoot, kLainVolumeRoot)