Add Container runtime check and messaging mechanism

Author: fierlion

agent/acs/handler/acs_handler.go

@@ -29,6 +29,8 @@ import (
 	"github.com/aws/amazon-ecs-agent/agent/config"
 	rolecredentials "github.com/aws/amazon-ecs-agent/agent/credentials"
 	"github.com/aws/amazon-ecs-agent/agent/data"
+	"github.com/aws/amazon-ecs-agent/agent/dockerclient/dockerapi"
+	"github.com/aws/amazon-ecs-agent/agent/doctor"
 	"github.com/aws/amazon-ecs-agent/agent/engine"
 	"github.com/aws/amazon-ecs-agent/agent/engine/dockerstate"
 	"github.com/aws/amazon-ecs-agent/agent/eventhandler"
@@ -84,6 +86,7 @@ type session struct {
 	agentConfig                     *config.Config
 	deregisterInstanceEventStream   *eventstream.EventStream
 	taskEngine                      engine.TaskEngine
+	dockerClient                    dockerapi.DockerClient
 	ecsClient                       api.ECSClient
 	state                           dockerstate.TaskEngineState
 	dataClient                      data.Client
@@ -94,6 +97,7 @@ type session struct {
 	backoff                         retry.Backoff
 	resources                       sessionResources
 	latestSeqNumTaskManifest        *int64
+	doctor                          *doctor.Doctor
 	_heartbeatTimeout               time.Duration
 	_heartbeatJitter                time.Duration
 	_inactiveInstanceReconnectDelay time.Duration
@@ -137,17 +141,22 @@ type sessionState interface {
 }
 
 // NewSession creates a new Session object
-func NewSession(ctx context.Context,
+func NewSession(
+	ctx context.Context,
 	config *config.Config,
 	deregisterInstanceEventStream *eventstream.EventStream,
-	containerInstanceArn string,
+	containerInstanceARN string,
 	credentialsProvider *credentials.Credentials,
+	dockerClient dockerapi.DockerClient,
 	ecsClient api.ECSClient,
 	taskEngineState dockerstate.TaskEngineState,
 	dataClient data.Client,
 	taskEngine engine.TaskEngine,
 	credentialsManager rolecredentials.Manager,
-	taskHandler *eventhandler.TaskHandler, latestSeqNumTaskManifest *int64) Session {
+	taskHandler *eventhandler.TaskHandler,
+	latestSeqNumTaskManifest *int64,
+	doctor *doctor.Doctor,
+) Session {
 	resources := newSessionResources(credentialsProvider)
 	backoff := retry.NewExponentialBackoff(connectionBackoffMin, connectionBackoffMax,
 		connectionBackoffJitter, connectionBackoffMultiplier)
@@ -156,9 +165,10 @@ func NewSession(ctx context.Context,
 	return &session{
 		agentConfig:                     config,
 		deregisterInstanceEventStream:   deregisterInstanceEventStream,
-		containerInstanceARN:            containerInstanceArn,
+		containerInstanceARN:            containerInstanceARN,
 		credentialsProvider:             credentialsProvider,
 		ecsClient:                       ecsClient,
+		dockerClient:                    dockerClient,
 		state:                           taskEngineState,
 		dataClient:                      dataClient,
 		taskEngine:                      taskEngine,
@@ -169,6 +179,7 @@ func NewSession(ctx context.Context,
 		backoff:                         backoff,
 		resources:                       resources,
 		latestSeqNumTaskManifest:        latestSeqNumTaskManifest,
+		doctor:                          doctor,
 		_heartbeatTimeout:               heartbeatTimeout,
 		_heartbeatJitter:                heartbeatJitter,
 		_inactiveInstanceReconnectDelay: inactiveInstanceReconnectDelay,
@@ -335,8 +346,7 @@ func (acsSession *session) startACSSession(client wsclient.ClientServer) error {
 
 	client.AddRequestHandler(payloadHandler.handlerFunc())
 
-	// Add HeartbeatHandler to acknowledge ACS heartbeats
-	heartbeatHandler := newHeartbeatHandler(acsSession.ctx, client)
+	heartbeatHandler := newHeartbeatHandler(acsSession.ctx, client, acsSession.doctor)
 	defer heartbeatHandler.clearAcks()
 	heartbeatHandler.start()
 	defer heartbeatHandler.stop()

agent/acs/handler/acs_handler_test.go

@@ -38,6 +38,8 @@ import (
 	rolecredentials "github.com/aws/amazon-ecs-agent/agent/credentials"
 	mock_credentials "github.com/aws/amazon-ecs-agent/agent/credentials/mocks"
 	"github.com/aws/amazon-ecs-agent/agent/data"
+	mock_dockerapi "github.com/aws/amazon-ecs-agent/agent/dockerclient/dockerapi/mocks"
+	"github.com/aws/amazon-ecs-agent/agent/doctor"
 	"github.com/aws/amazon-ecs-agent/agent/engine/dockerstate"
 	mock_engine "github.com/aws/amazon-ecs-agent/agent/engine/mocks"
 	"github.com/aws/amazon-ecs-agent/agent/eventhandler"
@@ -805,6 +807,7 @@ func TestHandlerDoesntLeakGoroutines(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 	taskEngine := mock_engine.NewMockTaskEngine(ctrl)
+	dockerClient := mock_dockerapi.NewMockDockerClient(ctrl)
 	ecsClient := mock_api.NewMockECSClient(ctrl)
 	ctx, cancel := context.WithCancel(context.Background())
 	taskHandler := eventhandler.NewTaskHandler(ctx, data.NewNoopClient(), nil, nil)
@@ -831,6 +834,10 @@ func TestHandlerDoesntLeakGoroutines(t *testing.T) {
 	})
 	taskEngine.EXPECT().Version().Return("Docker: 1.5.0", nil).AnyTimes()
 	taskEngine.EXPECT().AddTask(gomock.Any()).AnyTimes()
+	dockerClient.EXPECT().SystemPing(gomock.Any(), gomock.Any()).AnyTimes()
+
+	emptyHealthchecksList := []doctor.Healthcheck{}
+	emptyDoctor, _ := doctor.NewDoctor(emptyHealthchecksList, "test-cluster", "this:is:an:instance:arn")
 
 	ended := make(chan bool, 1)
 	go func() {
@@ -840,6 +847,7 @@ func TestHandlerDoesntLeakGoroutines(t *testing.T) {
 			credentialsProvider:      testCreds,
 			agentConfig:              testConfig,
 			taskEngine:               taskEngine,
+			dockerClient:             dockerClient,
 			ecsClient:                ecsClient,
 			dataClient:               data.NewNoopClient(),
 			taskHandler:              taskHandler,
@@ -849,6 +857,7 @@ func TestHandlerDoesntLeakGoroutines(t *testing.T) {
 			resources:                newSessionResources(testCreds),
 			credentialsManager:       rolecredentials.NewManager(),
 			latestSeqNumTaskManifest: aws.Int64(12),
+			doctor:                   emptyDoctor,
 		}
 		acsSession.Start()
 		ended <- true
@@ -916,6 +925,10 @@ func TestStartSessionHandlesRefreshCredentialsMessages(t *testing.T) {
 	taskEngine.EXPECT().Version().Return("Docker: 1.5.0", nil).AnyTimes()
 
 	credentialsManager := mock_credentials.NewMockManager(ctrl)
+	dockerClient := mock_dockerapi.NewMockDockerClient(ctrl)
+
+	emptyHealthchecksList := []doctor.Healthcheck{}
+	emptyDoctor, _ := doctor.NewDoctor(emptyHealthchecksList, "test-cluster", "this:is:a:container:arn")
 
 	latestSeqNumberTaskManifest := int64(10)
 	ended := make(chan bool, 1)
@@ -925,12 +938,15 @@ func TestStartSessionHandlesRefreshCredentialsMessages(t *testing.T) {
 			nil,
 			"myArn",
 			testCreds,
+			dockerClient,
 			ecsClient,
 			dockerstate.NewTaskEngineState(),
 			data.NewNoopClient(),
 			taskEngine,
 			credentialsManager,
-			taskHandler, &latestSeqNumberTaskManifest,
+			taskHandler,
+			&latestSeqNumberTaskManifest,
+			emptyDoctor,
 		)
 		acsSession.Start()
 		// StartSession should never return unless the context is canceled
@@ -1017,11 +1033,12 @@ func TestHandlerReconnectsCorrectlySetsSendCredentialsURLParameter(t *testing.T)
 	taskHandler := eventhandler.NewTaskHandler(ctx, data.NewNoopClient(), nil, nil)
 
 	mockWsClient := mock_wsclient.NewMockClientServer(ctrl)
-
 	mockWsClient.EXPECT().SetAnyRequestHandler(gomock.Any()).AnyTimes()
 	mockWsClient.EXPECT().AddRequestHandler(gomock.Any()).AnyTimes()
 	mockWsClient.EXPECT().Close().Return(nil).AnyTimes()
 	mockWsClient.EXPECT().Serve().Return(io.EOF).AnyTimes()
+
+	dockerClient := mock_dockerapi.NewMockDockerClient(ctrl)
 	resources := newSessionResources(testCreds)
 	gomock.InOrder(
 		// When the websocket client connects to ACS for the first
@@ -1041,6 +1058,7 @@ func TestHandlerReconnectsCorrectlySetsSendCredentialsURLParameter(t *testing.T)
 		credentialsProvider:  testCreds,
 		agentConfig:          testConfig,
 		taskEngine:           taskEngine,
+		dockerClient:         dockerClient,
 		ecsClient:            ecsClient,
 		dataClient:           data.NewNoopClient(),
 		taskHandler:          taskHandler,

agent/acs/handler/heartbeat_handler.go

@@ -17,6 +17,7 @@ import (
 	"context"
 
 	"github.com/aws/amazon-ecs-agent/agent/acs/model/ecsacs"
+	"github.com/aws/amazon-ecs-agent/agent/doctor"
 	"github.com/aws/amazon-ecs-agent/agent/wsclient"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/cihub/seelog"
@@ -29,12 +30,11 @@ type heartbeatHandler struct {
 	ctx                       context.Context
 	cancel                    context.CancelFunc
 	acsClient                 wsclient.ClientServer
+	doctor                    *doctor.Doctor
 }
 
 // newHeartbeatHandler returns an instance of the heartbeatHandler struct
-func newHeartbeatHandler(ctx context.Context,
-	acsClient wsclient.ClientServer) heartbeatHandler {
-
+func newHeartbeatHandler(ctx context.Context, acsClient wsclient.ClientServer, heartbeatDoctor *doctor.Doctor) heartbeatHandler {
 	// Create a cancelable context from the parent context
 	derivedContext, cancel := context.WithCancel(ctx)
 	return heartbeatHandler{
@@ -43,6 +43,7 @@ func newHeartbeatHandler(ctx context.Context,
 		ctx:                       derivedContext,
 		cancel:                    cancel,
 		acsClient:                 acsClient,
+		doctor:                    heartbeatDoctor,
 	}
 }
 
@@ -73,7 +74,15 @@ func (heartbeatHandler *heartbeatHandler) handleHeartbeatMessage() {
 }
 
 func (heartbeatHandler *heartbeatHandler) handleSingleHeartbeatMessage(message *ecsacs.HeartbeatMessage) error {
-	// Agent currently has no other action hooked to heartbeat messages, except simple ack
+	// TestHandlerDoesntLeakGoroutines unit test is failing because of this section
+
+	// Agent will run healthchecks triggered by ACS heartbeat
+	// healthcheck results will be sent on to TACS, but for now just to debug logs.
+	go func() {
+		heartbeatHandler.doctor.RunHealthchecks()
+	}()
+
+	// Agent will send simple ack to the heartbeatAckMessageBuffer
 	go func() {
 		response := &ecsacs.HeartbeatAckRequest{
 			MessageId: message.MessageId,

agent/acs/handler/heartbeat_handler_test.go

@@ -20,6 +20,8 @@ import (
 	"testing"
 
 	"github.com/aws/amazon-ecs-agent/agent/acs/model/ecsacs"
+	mock_dockerapi "github.com/aws/amazon-ecs-agent/agent/dockerclient/dockerapi/mocks"
+	"github.com/aws/amazon-ecs-agent/agent/doctor"
 	mock_wsclient "github.com/aws/amazon-ecs-agent/agent/wsclient/mock"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/golang/mock/gomock"
@@ -92,7 +94,14 @@ func validateHeartbeatAck(t *testing.T, heartbeatReceived *ecsacs.HeartbeatMessa
 		cancel()
 	}).Times(1)
 
-	handler := newHeartbeatHandler(ctx, mockWsClient)
+	dockerClient := mock_dockerapi.NewMockDockerClient(ctrl)
+	dockerClient.EXPECT().SystemPing(gomock.Any(), gomock.Any()).AnyTimes()
+
+	emptyHealthchecksList := []doctor.Healthcheck{}
+	emptyDoctor, _ := doctor.NewDoctor(emptyHealthchecksList, "testCluster", "this:is:an:instance:arn")
+
+	handler := newHeartbeatHandler(ctx, mockWsClient, emptyDoctor)
+
 	go handler.sendHeartbeatAck()
 
 	handler.handleSingleHeartbeatMessage(heartbeatReceived)

agent/app/agent.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/aws/amazon-ecs-agent/agent/doctor"
 	"github.com/aws/amazon-ecs-agent/agent/eni/watcher"
 	"github.com/aws/aws-sdk-go/aws/awserr"
 
@@ -358,6 +359,15 @@ func (agent *ecsAgent) doStart(containerChangeEventStream *eventstream.EventStre
 		agent.saveMetadata(data.EC2InstanceIDKey, currentEC2InstanceID)
 	}
 
+	// now that we know the container instance ARN, we can build out the doctor
+	// and pass it on to ACS and TACS
+	doctor, doctorCreateErr := agent.newDoctorWithHealthchecks(agent.cfg.Cluster, agent.containerInstanceARN)
+	if doctorCreateErr != nil {
+		seelog.Warnf("Error starting doctor, healthchecks won't be running: %v", err)
+	} else {
+		seelog.Debug("Doctor healthchecks set up properly.")
+	}
+
 	// Begin listening to the docker daemon and saving changes
 	taskEngine.SetDataClient(agent.dataClient)
 	imageManager.SetDataClient(agent.dataClient)
@@ -370,11 +380,11 @@ func (agent *ecsAgent) doStart(containerChangeEventStream *eventstream.EventStre
 	taskHandler := eventhandler.NewTaskHandler(agent.ctx, agent.dataClient, state, client)
 	attachmentEventHandler := eventhandler.NewAttachmentEventHandler(agent.ctx, agent.dataClient, client)
 	agent.startAsyncRoutines(containerChangeEventStream, credentialsManager, imageManager,
-		taskEngine, deregisterInstanceEventStream, client, taskHandler, attachmentEventHandler, state)
+		taskEngine, deregisterInstanceEventStream, client, taskHandler, attachmentEventHandler, state, doctor)
 
 	// Start the acs session, which should block doStart
 	return agent.startACSSession(credentialsManager, taskEngine,
-		deregisterInstanceEventStream, client, state, taskHandler)
+		deregisterInstanceEventStream, client, state, taskHandler, doctor)
 }
 
 // newTaskEngine creates a new docker task engine object. It tries to load the
@@ -447,6 +457,21 @@ func (agent *ecsAgent) initMetricsEngine() {
 	metrics.PublishMetrics()
 }
 
+// newDoctorWithHealthchecks creates a new doctor and also configures
+// the healthchecks that the doctor should be running
+func (agent *ecsAgent) newDoctorWithHealthchecks(cluster, containerInstanceARN string) (*doctor.Doctor, error) {
+	// configure the required healthchecks
+	runtimeHealthCheck := doctor.NewDockerRuntimeHealthcheck(agent.dockerClient)
+
+	// put the healthechecks in a list
+	healthcheckList := []doctor.Healthcheck{
+		runtimeHealthCheck,
+	}
+
+	// set up the doctor and return it
+	return doctor.NewDoctor(healthcheckList, cluster, containerInstanceARN)
+}
+
 // setClusterInConfig sets the cluster name in the config object based on
 // previous state. It returns an error if there's a mismatch between the
 // the current cluster name with what's restored from the cluster state
@@ -648,7 +673,9 @@ func (agent *ecsAgent) startAsyncRoutines(
 	client api.ECSClient,
 	taskHandler *eventhandler.TaskHandler,
 	attachmentEventHandler *eventhandler.AttachmentEventHandler,
-	state dockerstate.TaskEngineState) {
+	state dockerstate.TaskEngineState,
+	doctor *doctor.Doctor,
+) {
 
 	// Start of the periodic image cleanup process
 	if !agent.cfg.ImageCleanupDisabled.Enabled() {
@@ -687,6 +714,7 @@ func (agent *ecsAgent) startAsyncRoutines(
 		ECSClient:                     client,
 		TaskEngine:                    taskEngine,
 		StatsEngine:                   statsEngine,
+		Doctor:                        doctor,
 	}
 
 	// Start metrics session in a go routine
@@ -748,21 +776,24 @@ func (agent *ecsAgent) startACSSession(
 	deregisterInstanceEventStream *eventstream.EventStream,
 	client api.ECSClient,
 	state dockerstate.TaskEngineState,
-	taskHandler *eventhandler.TaskHandler) int {
+	taskHandler *eventhandler.TaskHandler,
+	doctor *doctor.Doctor) int {
 
 	acsSession := acshandler.NewSession(
 		agent.ctx,
 		agent.cfg,
 		deregisterInstanceEventStream,
 		agent.containerInstanceARN,
 		agent.credentialProvider,
+		agent.dockerClient,
 		client,
 		state,
 		agent.dataClient,
 		taskEngine,
 		credentialsManager,
 		taskHandler,
 		agent.latestSeqNumberTaskManifest,
+		doctor,
 	)
 	seelog.Info("Beginning Polling for updates")
 	err := acsSession.Start()

agent/dockerclient/dockerapi/docker_client.go

@@ -569,6 +569,7 @@ func (dg *dockerGoClient) createContainer(ctx context.Context,
 	config *dockercontainer.Config,
 	hostConfig *dockercontainer.HostConfig,
 	name string) DockerContainerMetadata {
+
 	client, err := dg.sdkDockerClient()
 	if err != nil {
 		return DockerContainerMetadata{Error: CannotGetDockerClientError{version: dg.version, err: err}}
@@ -610,8 +611,8 @@ func (dg *dockerGoClient) startContainer(ctx context.Context, id string) DockerC
 	if err != nil {
 		return DockerContainerMetadata{Error: CannotGetDockerClientError{version: dg.version, err: err}}
 	}
-
 	err = client.ContainerStart(ctx, id, types.ContainerStartOptions{})
+
 	metadata := dg.containerMetadata(ctx, id)
 	if err != nil {
 		metadata.Error = CannotStartContainerError{err}