Reopen Docker event stream when getting EOF error

agent/dockerclient/dockerapi/docker_client.go

@@ -863,22 +863,26 @@ func (dg *dockerGoClient) ContainerEvents(ctx context.Context) (<-chan DockerCon
 		for {
 			select {
 			case err := <-eventErr:
-				if err == io.EOF || err == io.ErrUnexpectedEOF {
-					seelog.Info("DockerGoClient: All events have been received")
-					cancel()
+				// If parent ctx has been canceled, stop listening and return. Otherwise reopen the stream.
+				if ctx.Err() != nil {
 					return
+				}
+
+				if err == io.EOF {
+					seelog.Infof("DockerGoClient: Docker events stream closed with: %v", err)
 				} else {
-					// If an error is returned, we need to reopen channel to continue listening
-					seelog.Errorf("DockerGoClient: error while listening to Docker Events : %v", err)
-					nextCtx, nextCancel := context.WithCancel(ctx)
-					dockerEvents, eventErr = client.Events(nextCtx, types.EventsOptions{})
-					// Cache the event from docker client.
-					go buffer.StartListening(nextCtx, dockerEvents)
-					// Close previous stream after starting to listen on new one
-					cancel()
-					// Reassign cancel variable next Cancel function to setup next iteration of loop.
-					cancel = nextCancel
+					seelog.Errorf("DockerGoClient: Docker events stream closed with error: %v", err)
 				}
+
+				// Reopen a new event stream to continue listening.
+				nextCtx, nextCancel := context.WithCancel(ctx)
+				dockerEvents, eventErr = client.Events(nextCtx, types.EventsOptions{})
+				// Cache the event from docker client.
+				go buffer.StartListening(nextCtx, dockerEvents)
+				// Close previous stream after starting to listen on new one
+				cancel()
+				// Reassign cancel variable next Cancel function to setup next iteration of loop.
+				cancel = nextCancel
 			case <-ctx.Done():
 				return
 			}

agent/dockerclient/dockerapi/docker_client_test.go

@@ -761,15 +761,18 @@ func TestContainerEventsEOFError(t *testing.T) {
 
 	eventsChan := make(chan events.Message, dockerEventBufferSize)
 	errChan := make(chan error)
-	mockDockerSDK.EXPECT().Events(gomock.Any(), gomock.Any()).Return(eventsChan, errChan)
+	mockDockerSDK.EXPECT().Events(gomock.Any(), gomock.Any()).Return(eventsChan, errChan).MinTimes(1)
 
 	dockerEvents, err := client.ContainerEvents(context.TODO())
 	require.NoError(t, err, "Could not get container events")
 	go func() {
 		errChan <- io.EOF
+		eventsChan <- events.Message{Type: "container", ID: "containerId", Status: "create"}
 	}()
 
-	assert.Equal(t, len(dockerEvents), 0, "Wrong number of docker events")
+	event := <-dockerEvents
+	assert.Equal(t, event.DockerID, "containerId", "Wrong docker id")
+	assert.Equal(t, event.Status, apicontainerstatus.ContainerCreated, "Wrong status")
 }
 
 func TestContainerEventsStreamError(t *testing.T) {