Force resume refreshLoop when engine stat becomes healthy

Signed-off-by: Dani Louca <[email protected]>

Adding test

Signed-off-by: Dani Louca <[email protected]>

Author: Dani Louca

cluster/engine.go

@@ -71,19 +71,22 @@ var (
 	testErrImageNotFound = errors.New("TEST_ERR_IMAGE_NOT_FOUND_SWARM")
 )
 
-// delayer offers a simple API to random delay within a given time range.
+// delayer offers a simple API to random delay within a given time range,
+// and to force resume delays
 type delayer struct {
 	rangeMin time.Duration
 	rangeMax time.Duration
 
-	r *rand.Rand
-	l sync.Mutex
+	resumeCh chan bool
+	r        *rand.Rand
+	l        sync.Mutex
 }
 
 func newDelayer(rangeMin, rangeMax time.Duration) *delayer {
 	return &delayer{
 		rangeMin: rangeMin,
 		rangeMax: rangeMax,
+		resumeCh: make(chan bool, 1),
 		r:        rand.New(rand.NewSource(time.Now().UTC().UnixNano())),
 	}
 }
@@ -417,10 +420,21 @@ func (e *Engine) CheckConnectionErr(err error) {
 	if err == nil {
 		e.setErrMsg("")
 		// If current state is unhealthy, change it to healthy
+		// A lock is needed to avoid race condition, which can lead to block on resumeCh
+		// We should not block on slow operations, ex: emitEvent that ends up calling the engine
+		// instead, we'll check the changed flag which is set to true in the lock block
+		changed := false
+		e.Lock()
 		if e.state == stateUnhealthy {
-			log.WithFields(log.Fields{"name": e.Name, "id": e.ID}).Infof("Engine came back to life after %d retries. Hooray!", e.getFailureCount())
+			changed = true
+			e.state = stateHealthy
+			log.WithFields(log.Fields{"name": e.Name, "id": e.ID}).Infof("Engine came back to life after %d retries. Hooray!", e.failureCount)
+		}
+		e.Unlock()
+		if changed {
+			// signal the refresh loop so we don't wait too long, especially if the failure count was high
+			e.refreshDelayer.resumeCh <- true
 			e.emitEvent("engine_reconnect")
-			e.setState(stateHealthy)
 		}
 		e.resetFailureCount()
 		return
@@ -918,6 +932,7 @@ func (e *Engine) refreshLoop() {
 		}
 		// Wait for the delayer or quit if we get stopped.
 		select {
+		case <-e.refreshDelayer.resumeCh:
 		case <-e.refreshDelayer.Wait(backoffFactor):
 		case <-e.stopCh:
 			return

cluster/engine_test.go

@@ -821,3 +821,88 @@ func TestRemoveImage(t *testing.T) {
 	}
 	apiClient.Mock.AssertExpectations(t)
 }
+
+func TestRefreshLoop(t *testing.T) {
+
+	var (
+		config = &ContainerConfig{containertypes.Config{
+			Image: "busybox",
+			Cmd:   []string{"date"},
+			Tty:   false,
+		}, containertypes.HostConfig{
+			Resources: containertypes.Resources{
+				CPUShares: 1,
+			},
+		}, networktypes.NetworkingConfig{}}
+		state = types.ContainerState{
+			StartedAt:  "2018-05-07T08:33:22.070211457Z",
+			FinishedAt: "0001-01-01T00:00:00Z",
+		}
+		engine    = NewEngine("test", 0, engOpts)
+		apiClient = engineapimock.NewMockClient()
+	)
+
+	engine.apiClient = apiClient
+	engine.setState(stateUnhealthy)
+
+	apiClient.On("Info", mock.Anything).Return(mockInfo, nil)
+	apiClient.On("ServerVersion", mock.Anything).Return(mockVersion, nil)
+	apiClient.On("NetworkList", mock.Anything,
+		mock.AnythingOfType("NetworkListOptions"),
+	).Return([]types.NetworkResource{}, nil)
+	apiClient.On("VolumeList", mock.Anything,
+		mock.AnythingOfType("Args"),
+	).Return(volume.VolumesListOKBody{}, nil)
+	apiClient.On("Events", mock.Anything, mock.AnythingOfType("EventsOptions")).Return(make(chan events.Message), make(chan error))
+	apiClient.On("ImageList", mock.Anything, mock.AnythingOfType("ImageListOptions")).Return([]types.ImageSummary{}, nil)
+	// we need to call ContainerList twice, onece for ConnectWithClient and once for the first iteration when we kick in the refreshLoop
+	apiClient.On("ContainerList", mock.Anything, types.ContainerListOptions{All: true, Size: false}).Return([]types.Container{}, nil).Twice()
+	apiClient.On("NegotiateAPIVersion", mock.Anything).Return()
+
+	assert.NoError(t, engine.ConnectWithClient(apiClient))
+	// Make sure engine is happy
+	assert.True(t, engine.isConnected())
+	assert.True(t, engine.state == stateHealthy)
+	// Stimulate engine failure by increasing the failure count and making it unhealthy
+	engine.failureCount = 900
+	engine.state = stateUnhealthy
+	go engine.refreshLoop()
+	// At this point, the loop should be waiting very long due to high failure count
+	assert.Len(t, engine.Containers(), 0)
+
+	// The below mock methods are used to verify that refresh loop resumed on the next
+	// CheckConnectionErr and added a new container
+	apiClient.On(
+		"ContainerList",
+		mock.Anything,
+		types.ContainerListOptions{
+			All:  true,
+			Size: false,
+		},
+	).Return(
+		[]types.Container{{ID: "100"}},
+		nil,
+	).Once()
+
+	apiClient.On(
+		"ContainerInspect",
+		mock.Anything,
+		"100",
+	).Return(
+		types.ContainerJSON{
+			Config: &config.Config,
+			ContainerJSONBase: &types.ContainerJSONBase{
+				HostConfig: &config.HostConfig,
+				State:      &state,
+			},
+			NetworkSettings: &types.NetworkSettings{
+				Networks: nil,
+			},
+		},
+		nil,
+	).Once()
+	// This forces the refrehLoop to resume and not wait very long
+	engine.CheckConnectionErr(nil)
+	time.Sleep(1 * time.Second)
+	assert.Len(t, engine.Containers(), 1)
+}