Search cached images when Agent failed to pull images for essentail c…

…ontainers

and ECS_PULL_DEPENDENT_CONTAINERS_UPFRONT=true. The task will be stopped
if no cached image can be found.

agent/engine/docker_task_engine.go

@@ -958,12 +958,35 @@ func (engine *DockerTaskEngine) pullAndUpdateContainerReference(task *apitask.Ta
 		return metadata
 	}
 	pullSucceeded := metadata.Error == nil
-	if pullSucceeded {
+	findCachedImage := false
+	if !pullSucceeded {
+		// If failed to pull an image for an essential container when DependentContainersPullUpfront is enabled
+		// check local image cache
+		if engine.cfg.DependentContainersPullUpfront.Enabled() && engine.cfg.ImagePullBehavior != config.ImagePullAlwaysBehavior {
+			if _, err := engine.client.InspectImage(container.Image); err != nil {
+				seelog.Errorf("Task engine [%s]: failed to find cached image %s for container %s",
+					task.Arn, container.Image, container.Name)
+				// Stop the task if the essential container failed to pull the image and find the cached image
+				if container.IsEssential() {
+					task.SetKnownStatus(apitaskstatus.TaskStopped)
+					task.SetDesiredStatus(apitaskstatus.TaskStopped)
+					engine.emitTaskEvent(task, fmt.Sprintf("%s: %s", metadata.Error.ErrorName(), metadata.Error.Error()))
+				}
+				return dockerapi.DockerContainerMetadata{Error: metadata.Error}
+			}
+			seelog.Infof("Task engine [%s]: found cached image %s, use it directly for container %s",
+				task.Arn, container.Image, container.Name)
+			findCachedImage = true
+		}
+	}
+
+	if pullSucceeded || findCachedImage {
 		dockerContainer := &apicontainer.DockerContainer{
 			Container: container,
 		}
 		engine.state.AddPulledContainer(dockerContainer, task)
 	}
+
 	engine.updateContainerReference(pullSucceeded, container, task.Arn)
 	return metadata
 }

agent/engine/docker_task_engine_test.go

@@ -1722,6 +1722,43 @@ func TestPullAndUpdateContainerReference(t *testing.T) {
 	assert.Equal(t, dockerapi.DockerContainerMetadata{}, metadata, "expected empty metadata")
 }
 
+// TestPullAndUpdateContainerReferenceWithCachedImage checks whether a container is added to task engine state when
+// the container is an essential container, the image is cached and DependentContainersPullUpfront is enabled.
+func TestPullAndUpdateContainerReferenceWithCachedImage(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.TODO())
+	defer cancel()
+	cfg := &config.Config{
+		DependentContainersPullUpfront: config.BooleanDefaultFalse{Value: config.ExplicitlyEnabled},
+	}
+	ctrl, client, _, privateTaskEngine, _, imageManager, _ := mocks(t, ctx, cfg)
+	defer ctrl.Finish()
+
+	taskEngine, _ := privateTaskEngine.(*DockerTaskEngine)
+	taskEngine._time = nil
+	imageName := "image"
+	taskArn := "taskArn"
+	container := &apicontainer.Container{
+		Type:      apicontainer.ContainerNormal,
+		Image:     imageName,
+		Essential: true,
+	}
+	task := &apitask.Task{
+		Arn:        taskArn,
+		Containers: []*apicontainer.Container{container},
+	}
+
+	client.EXPECT().PullImage(gomock.Any(), imageName, nil,
+		gomock.Any()).Return(dockerapi.DockerContainerMetadata{Error: dockerapi.CannotPullContainerError{fmt.Errorf("error")}})
+	client.EXPECT().InspectImage(imageName).Return(nil, nil)
+	imageManager.EXPECT().RecordContainerReference(container)
+	imageManager.EXPECT().GetImageStateFromImageName(imageName).Return(nil, false)
+	metadata := taskEngine.pullAndUpdateContainerReference(task, container)
+	pulledContainersMap, ok := taskEngine.State().PulledContainerMapByArn(taskArn)
+	require.True(t, ok, "no container found in the agent state")
+	require.Len(t, pulledContainersMap, 1)
+	assert.Equal(t, dockerapi.DockerContainerMetadata{Error: dockerapi.CannotPullContainerError{fmt.Errorf("error")}}, metadata, "expected metadata with error")
+}
+
 // TestMetadataFileUpdatedAgentRestart checks whether metadataManager.Update(...) is
 // invoked in the path DockerTaskEngine.Init() -> .synchronizeState() -> .updateMetadataFile(...)
 // for the following case: