Merge pull request #19 from supermeng/master

deal with swarm connection error

Author: supermeng

engine/engine.go

@@ -31,6 +31,10 @@ var (
 	ErrNotifyNotExists        = errors.New("Notify uri not existed")
 )
 
+const (
+	ClusterFailedThreadSold = 20
+)
+
 type OrcEngine struct {
 	sync.RWMutex
 
@@ -43,6 +47,7 @@ type OrcEngine struct {
 	rmDepCtrls   map[string]*dependsController
 	opsChan      chan orcOperation
 	stop         chan struct{}
+	clstrFailCnt int
 }
 
 const (
@@ -334,6 +339,18 @@ func (engine *OrcEngine) initPodGroupCtrl(spec PodGroupSpec, states []PodPrevSta
 	return pgCtrl
 }
 
+func (engine *OrcEngine) clusterRequestFailed() {
+	engine.clstrFailCnt++
+	if engine.clstrFailCnt > ClusterFailedThreadSold && engine.clstrFailCnt%ClusterFailedThreadSold == 0 {
+		ntfController.Send(NewNotifySpec("Cluster", "Cluster-Manager",
+			1, time.Now(), NotifyClusterUnHealthy))
+	}
+}
+
+func (engine *OrcEngine) clusterRequestSucceed() {
+	engine.clstrFailCnt = 0
+}
+
 // This will be running inside the go routine
 func (engine *OrcEngine) initOperationWorker() {
 	tick := time.Tick(time.Duration(RefreshInterval) * time.Second)
@@ -493,9 +510,11 @@ func (engine *OrcEngine) startClusterMonitor() {
 	downCount := 0
 	eventMonitorId := engine.cluster.MonitorEvents("", func(event adoc.Event, err error) {
 		if err != nil {
-			log.Warnf("Error during the cluster event monitor, will try to restart the monitor, %s", err)
+			// log.Warnf("Error during the cluster event monitor, will try to restart the monitor, %s", err)
+			engine.clusterRequestFailed()
 			restart <- true
 		} else {
+			engine.clusterRequestSucceed()
 			log.Debugf("Cluster event: %v", event)
 			if strings.HasPrefix(event.From, "swarm") {
 				switch event.Status {
@@ -559,6 +578,7 @@ func New(cluster cluster.Cluster, store storage.Store) (*OrcEngine, error) {
 		rmDepCtrls:   make(map[string]*dependsController),
 		opsChan:      make(chan orcOperation, 500),
 		stop:         nil,
+		clstrFailCnt: 0,
 	}
 
 	eagleView := NewRuntimeEagleView()

engine/notify.go

@@ -22,6 +22,8 @@ var (
 	NotifyLetPodGo     = "LAIN found pod restart too many times in a short period, will let it go"
 	NotifyPodIPLost    = "LAIN found pod lost IP, please inform the SA team"
 	NotifyPodUnHealthy = "LAIN found pod Unhealthy, please check your service"
+
+	NotifyClusterUnHealthy = "LAIN found Cluster Manager Unhealthy, please check your cluster"
 )
 
 type notifyController struct {

engine/pod.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/laincloud/deployd/cluster"
+	"github.com/laincloud/deployd/utils/util"
 	"github.com/mijia/adoc"
 	"github.com/mijia/go-generics"
 	"github.com/mijia/sweb/log"
@@ -69,14 +70,18 @@ func (pc *podController) Deploy(cluster cluster.Cluster) {
 		id, err := pc.createContainer(cluster, filters, i)
 		if err != nil {
 			log.Warnf("%s Cannot create container, error=%q, spec=%+v", pc, err, cSpec)
-			pc.pod.State = RunStateFail
+			if !util.IsConnectionError(err) {
+				pc.pod.State = RunStateFail
+			}
 			pc.pod.LastError = fmt.Sprintf("Cannot create container, %s", err)
 			return
 		}
 
 		if err := cluster.StartContainer(id); err != nil {
 			log.Warnf("%s Cannot start container %s, %s", pc, id, err)
-			pc.pod.State = RunStateFail
+			if !util.IsConnectionError(err) {
+				pc.pod.State = RunStateFail
+			}
 			pc.pod.LastError = fmt.Sprintf("Cannot start container, %s", err)
 		}
 
@@ -177,7 +182,9 @@ func (pc *podController) Stop(cluster cluster.Cluster) {
 	for i, container := range pc.pod.Containers {
 		if err := cluster.StopContainer(container.Id, pc.spec.GetKillTimeout()); err != nil {
 			log.Warnf("%s Cannot stop the container %s, %s", pc, container.Id, err)
-			pc.pod.State = RunStateFail
+			if !util.IsConnectionError(err) {
+				pc.pod.State = RunStateFail
+			}
 			pc.pod.LastError = fmt.Sprintf("Cannot stop container, %s", err)
 		} else {
 			pc.refreshContainer(cluster, i)
@@ -200,7 +207,9 @@ func (pc *podController) Start(cluster cluster.Cluster) {
 	for i, container := range pc.pod.Containers {
 		if err := cluster.StartContainer(container.Id); err != nil {
 			log.Warnf("%s Cannot start the container %s, %s", pc, container.Id, err)
-			pc.pod.State = RunStateFail
+			if !util.IsConnectionError(err) {
+				pc.pod.State = RunStateFail
+			}
 			pc.pod.LastError = fmt.Sprintf("Cannot start container, %s", err)
 		} else {
 			pc.refreshContainer(cluster, i)
@@ -273,7 +282,9 @@ func (pc *podController) refreshContainer(kluster cluster.Cluster, index int) {
 			pc.pod.LastError = fmt.Sprintf("Missing container %q, %s", id, err)
 		} else {
 			log.Warnf("%s Failed to inspect container %s, %s", pc, id, err)
-			pc.pod.State = RunStateFail
+			if !util.IsConnectionError(err) {
+				pc.pod.State = RunStateFail
+			}
 			pc.pod.LastError = fmt.Sprintf("Cannot inspect the container, %s", err)
 		}
 	} else {
@@ -282,7 +293,6 @@ func (pc *podController) refreshContainer(kluster cluster.Cluster, index int) {
 			network = pc.spec.Namespace
 		}
 		prevIP, nowIP := pc.spec.PrevState.IPs[index], info.NetworkSettings.Networks[network].IPAddress
-
 		// NOTE: if the container's ip is not equal to prev ip, try to correct it; if failed, accpet new ip
 		if prevIP != "" && prevIP != nowIP {
 			log.Warnf("%s find the IP changed, prev is %s, but now is %s, try to correct it", pc, prevIP, nowIP)

engine/podgroup_ops.go

@@ -180,8 +180,9 @@ func (op pgOperRefreshInstance) Do(pgCtrl *podGroupController, c cluster.Cluster
 			}
 		} else {
 			log.Warnf("PodGroupCtrl %s, we found pod missing, just redeploy it", op.spec)
-			ntfController.Send(NewNotifySpec(podCtrl.spec.Namespace, podCtrl.spec.Name,
-				op.instanceNo, container.Runtime.State.FinishedAt, NotifyPodMissing))
+			//pod missing usually happended when agent was down, so no need to notify app users
+			// ntfController.Send(NewNotifySpec(podCtrl.spec.Namespace, podCtrl.spec.Name,
+			// 	op.instanceNo, container.Runtime.State.FinishedAt, NotifyPodMissing))
 			newPodSpec := podCtrl.spec.Clone()
 			prevNodeName := newPodSpec.PrevState.NodeName
 			if newPodSpec.IsHardStateful() {

utils/util/util.go

@@ -2,6 +2,8 @@ package util
 
 import (
 	"errors"
+	"net"
+	"net/url"
 	"strconv"
 
 	"github.com/laincloud/deployd/utils/regex"
@@ -22,3 +24,20 @@ func ParseNameInstanceNo(containerName string) (string, int, error) {
 		return g.Group(1), instance, nil
 	}
 }
+
+func IsConnectionError(err error) bool {
+	p := regex.MustCompile(`getsockopt: connection refused`)
+	g := p.Match(err.Error())
+	if g != nil {
+		return true
+	}
+	switch err := err.(type) {
+	case net.Error:
+		return err.Timeout()
+	case *url.Error:
+		if err, ok := err.Err.(net.Error); ok {
+			return err.Timeout()
+		}
+	}
+	return false
+}

utils/util/util_test.go

@@ -1,19 +1,27 @@
 package util
 
 import (
+	"errors"
 	"fmt"
 	"reflect"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func Test_ParseNameInstanceNo(t *testing.T) {
 	containerName := "webrouter.worker.worker.v0-i1-d0"
-	if name, incetance, err := ParseNameInstanceNo(containerName); err == nil {
-		fmt.Printf("name:%v,incetance:%v\n", name, incetance)
-	} else {
-		fmt.Println("err:%v\n", err)
-	}
+
+	name, incetance, err := ParseNameInstanceNo(containerName)
+	assert.Equal(t, nil, err)
+	assert.Equal(t, "webrouter.worker.worker", name)
+	assert.Equal(t, 1, incetance)
+}
+
+func Test_IsConnectionError(t *testing.T) {
+	err := errors.New("dial tcp 192.168.77.21:2376: getsockopt: connection refused")
+	assert.Equal(t, true, IsConnectionError(err))
 }
 
 func Test_deepEqual(t *testing.T) {