Merge branch 'aws:dev' into dev

agent/acs/session/payload_responder_test.go

@@ -722,7 +722,7 @@ func TestHandlePayloadMessageAddedEBSToTask(t *testing.T) {
 							Value: aws.String(taskresourcevolume.TestFileSystem),
 						},
 					},
-					AttachmentType: aws.String(apiresource.AmazonElasticBlockStorage),
+					AttachmentType: aws.String(apiresource.EBSTaskAttach),
 				},
 			},
 		},

agent/api/task/task_attachment_handler.go

@@ -83,7 +83,7 @@ func handleTaskAttachments(acsTask *ecsacs.Task, task *Task) error {
 			switch aws.StringValue(attachment.AttachmentType) {
 			case serviceConnectAttachmentType:
 				serviceConnectAttachment = attachment
-			case apiresource.AmazonElasticBlockStorage:
+			case apiresource.EBSTaskAttach:
 				ebsVolumeAttachments = append(ebsVolumeAttachments, attachment)
 			default:
 				logger.Debug("Received an attachment type", logger.Fields{
@@ -117,7 +117,7 @@ func handleTaskAttachments(acsTask *ecsacs.Task, task *Task) error {
 				}
 				taskVolume := TaskVolume{
 					Name:   ebs.VolumeName,
-					Type:   apiresource.AmazonElasticBlockStorage,
+					Type:   apiresource.EBSTaskAttach,
 					Volume: ebs,
 				}
 				task.Volumes = append(task.Volumes, taskVolume)

agent/api/task/task_attachment_handler_test.go

@@ -271,7 +271,7 @@ func TestHandleTaskAttachmentWithEBSVolumeAttachment(t *testing.T) {
 								Value: stringToPointer(tc.testFileSystem),
 							},
 						},
-						AttachmentType: stringToPointer(apiresource.AmazonElasticBlockStorage),
+						AttachmentType: stringToPointer(apiresource.EBSTaskAttach),
 					},
 				},
 			}

agent/api/task/task_test.go

@@ -4688,7 +4688,7 @@ func TestTaskWithEBSVolumeAttachment(t *testing.T) {
 						Value: strptr(taskresourcevolume.TestFileSystem),
 					},
 				},
-				AttachmentType: strptr(apiresource.AmazonElasticBlockStorage),
+				AttachmentType: strptr(apiresource.EBSTaskAttach),
 			},
 		},
 	}

agent/api/task/taskvolume.go

@@ -76,7 +76,7 @@ func (tv *TaskVolume) UnmarshalJSON(b []byte) error {
 		return tv.unmarshalEFSVolume(intermediate["efsVolumeConfiguration"])
 	case FSxWindowsFileServerVolumeType:
 		return tv.unmarshalFSxWindowsFileServerVolume(intermediate["fsxWindowsFileServerVolumeConfiguration"])
-	case apiresource.AmazonElasticBlockStorage:
+	case apiresource.EBSTaskAttach:
 		return tv.unmarshalEBSVolume(intermediate["ebsVolumeConfiguration"])
 	default:
 		return errors.Errorf("unrecognized volume type: %q", tv.Type)
@@ -103,7 +103,7 @@ func (tv *TaskVolume) MarshalJSON() ([]byte, error) {
 		result["efsVolumeConfiguration"] = tv.Volume
 	case FSxWindowsFileServerVolumeType:
 		result["fsxWindowsFileServerVolumeConfiguration"] = tv.Volume
-	case apiresource.AmazonElasticBlockStorage:
+	case apiresource.EBSTaskAttach:
 		result["ebsVolumeConfiguration"] = tv.Volume
 	default:
 		return nil, errors.Errorf("unrecognized volume type: %q", tv.Type)

agent/api/task/taskvolume_test.go

@@ -195,7 +195,7 @@ func TestMarshalEBSVolumes(t *testing.T) {
 		Volumes: []TaskVolume{
 			{
 				Name: "1",
-				Type: apiresource.AmazonElasticBlockStorage,
+				Type: apiresource.EBSTaskAttach,
 				Volume: &taskresourcevolume.EBSTaskVolumeConfig{
 					VolumeId:             "vol-12345",
 					VolumeName:           "test-volume",
@@ -230,7 +230,7 @@ func TestMarshalEBSVolumes(t *testing.T) {
 				"dockerVolumeName": ""
 			},
 			"name": "1",
-			"type": "AmazonElasticBlockStorage"
+			"type": "amazonebs"
 		  }
 		],
 		"DesiredStatus": "NONE",
@@ -275,7 +275,7 @@ func TestUnmarshalEBSVolumes(t *testing.T) {
 				"dockerVolumeName": ""
 			},
 			"name": "1",
-			"type": "AmazonElasticBlockStorage"
+			"type": "amazonebs"
 		  }
 		],
 		"DesiredStatus": "NONE",
@@ -306,7 +306,7 @@ func TestUnmarshalEBSVolumes(t *testing.T) {
 	require.NoError(t, err, "Could not unmarshal task")
 
 	require.Len(t, task.Volumes, 1)
-	assert.Equal(t, apiresource.AmazonElasticBlockStorage, task.Volumes[0].Type)
+	assert.Equal(t, apiresource.EBSTaskAttach, task.Volumes[0].Type)
 	assert.Equal(t, "1", task.Volumes[0].Name)
 	ebsConfig, ok := task.Volumes[0].Volume.(*taskresourcevolume.EBSTaskVolumeConfig)
 	require.True(t, ok)

agent/app/agent.go

@@ -33,6 +33,7 @@ import (
 	"github.com/aws/amazon-ecs-agent/agent/dockerclient/dockerapi"
 	"github.com/aws/amazon-ecs-agent/agent/dockerclient/sdkclientfactory"
 	dockerdoctor "github.com/aws/amazon-ecs-agent/agent/doctor" // for Docker specific container instance health checks
+	ebs "github.com/aws/amazon-ecs-agent/agent/ebs"
 	"github.com/aws/amazon-ecs-agent/agent/ec2"
 	"github.com/aws/amazon-ecs-agent/agent/ecscni"
 	"github.com/aws/amazon-ecs-agent/agent/engine"
@@ -64,7 +65,6 @@ import (
 	"github.com/aws/amazon-ecs-agent/ecs-agent/eventstream"
 	"github.com/aws/amazon-ecs-agent/ecs-agent/logger"
 	"github.com/aws/amazon-ecs-agent/ecs-agent/logger/field"
-	md "github.com/aws/amazon-ecs-agent/ecs-agent/manageddaemon"
 	metricsfactory "github.com/aws/amazon-ecs-agent/ecs-agent/metrics"
 	"github.com/aws/amazon-ecs-agent/ecs-agent/tcs/model/ecstcs"
 	"github.com/aws/amazon-ecs-agent/ecs-agent/utils/retry"
@@ -153,6 +153,7 @@ type ecsAgent struct {
 	serviceconnectManager       engineserviceconnect.Manager
 	daemonManagers              map[string]dm.DaemonManager
 	eniWatcher                  *watcher.ENIWatcher
+	ebsWatcher                  *ebs.EBSWatcher
 	cniClient                   ecscni.CNIClient
 	vpc                         string
 	subnet                      string
@@ -341,29 +342,6 @@ func (agent *ecsAgent) doStart(containerChangeEventStream *eventstream.EventStre
 		StringSetValue: aws.StringSlice(gpuIDs),
 	}
 
-	daemonDefinitions, err := md.ImportAll()
-	// we will not panic when daemons fail to import; agent should continue running
-	// container instance health will act as the mechanism to advertise daemon failures
-	if err != nil {
-		seelog.Errorf("Daemon import mountpoint failure: %s", err)
-	}
-	if len(daemonDefinitions) == 0 {
-		seelog.Infof("daemonDefinitions is empty/nil after import")
-	}
-	// load and add daemons to agent
-	for _, md := range daemonDefinitions {
-		thisDaemon := dm.NewDaemonManager(md)
-		if _, err := thisDaemon.LoadImage(agent.ctx, agent.dockerClient); err != nil {
-			seelog.Errorf("Managed Daemon Load failure %v", err)
-		}
-		if loaded, err := thisDaemon.IsLoaded(agent.dockerClient); loaded {
-			imageManager.AddImageToCleanUpExclusionList(md.GetLoadedDaemonImageRef())
-			agent.daemonManagers[md.GetImageName()] = thisDaemon
-		} else {
-			seelog.Errorf("Unable to load Managed Daemon: %s, err: %s", md.GetImageName(), err)
-		}
-	}
-
 	// Create the task engine
 	taskEngine, currentEC2InstanceID, err := agent.newTaskEngine(
 		containerChangeEventStream, credentialsManager, state, imageManager, hostResources, execCmdMgr,
@@ -461,6 +439,13 @@ func (agent *ecsAgent) doStart(containerChangeEventStream *eventstream.EventStre
 		imageManager.AddImageToCleanUpExclusionList(agent.serviceconnectManager.GetLoadedImageName())
 	}
 
+	// exclude all daemon images from cleanup
+	for _, csiDM := range agent.daemonManagers {
+		if loaded, _ := csiDM.IsLoaded(agent.dockerClient); loaded {
+			imageManager.AddImageToCleanUpExclusionList(csiDM.GetManagedDaemon().GetLoadedDaemonImageRef())
+		}
+	}
+
 	// Add container instance ARN to metadata manager
 	if agent.cfg.ContainerMetadataEnabled.Enabled() {
 		agent.metadataManager.SetContainerInstanceARN(agent.containerInstanceARN)
@@ -499,7 +484,8 @@ func (agent *ecsAgent) doStart(containerChangeEventStream *eventstream.EventStre
 	attachmentEventHandler := eventhandler.NewAttachmentEventHandler(agent.ctx, agent.dataClient, client)
 	agent.startAsyncRoutines(containerChangeEventStream, credentialsManager, imageManager,
 		taskEngine, deregisterInstanceEventStream, client, taskHandler, attachmentEventHandler, state, doctor)
-
+	// TODO add EBS watcher to async routines
+	agent.startEBSWatcher(state, taskEngine)
 	// Start the acs session, which should block doStart
 	return agent.startACSSession(credentialsManager, taskEngine,
 		deregisterInstanceEventStream, client, state, taskHandler, doctor)
@@ -1150,6 +1136,10 @@ func (agent *ecsAgent) saveMetadata(key, val string) {
 	}
 }
 
+func (agent *ecsAgent) setDaemonManager(key string, val dm.DaemonManager) {
+	agent.daemonManagers[key] = val
+}
+
 // setVPCSubnet sets the vpc and subnet ids for the agent by querying the
 // instance metadata service
 func (agent *ecsAgent) setVPCSubnet() (error, bool) {

agent/app/agent_capability.go

@@ -22,9 +22,11 @@ import (
 
 	"github.com/aws/amazon-ecs-agent/agent/config"
 	"github.com/aws/amazon-ecs-agent/agent/dockerclient"
+	dm "github.com/aws/amazon-ecs-agent/agent/engine/daemonmanager"
 	"github.com/aws/amazon-ecs-agent/ecs-agent/ecs_client/model/ecs"
 	"github.com/aws/amazon-ecs-agent/ecs-agent/logger"
 	"github.com/aws/amazon-ecs-agent/ecs-agent/logger/field"
+	md "github.com/aws/amazon-ecs-agent/ecs-agent/manageddaemon"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/cihub/seelog"
 	"github.com/pkg/errors"
@@ -78,6 +80,7 @@ const (
 	capabilityExternal                                     = "external"
 	capabilityServiceConnect                               = "service-connect-v1"
 	capabilityGpuDriverVersion                             = "gpu-driver-version"
+	capabilityEBSTaskAttach                                = "storage.ebs-task-volume-attach"
 
 	// network capabilities, going forward, please append "network." prefix to any new networking capability we introduce
 	networkCapabilityPrefix      = "network."
@@ -126,6 +129,7 @@ var (
 		attributePrefix + taskEIAAttributeSuffix,
 		attributePrefix + taskEIAWithOptimizedCPU,
 		attributePrefix + capabilityServiceConnect,
+		attributePrefix + capabilityEBSTaskAttach,
 	}
 	// List of capabilities that are only supported on external capaciity. Currently only one but keep as a list
 	// for future proof and also align with externalUnsupportedCapabilities.
@@ -291,6 +295,9 @@ func (agent *ecsAgent) capabilities() ([]*ecs.Attribute, error) {
 	// add service-connect capabilities if applicable
 	capabilities = agent.appendServiceConnectCapabilities(capabilities)
 
+	// add ebs-task-attach attribute if applicable
+	capabilities = agent.appendEBSTaskAttachCapabilities(capabilities)
+
 	if agent.cfg.External.Enabled() {
 		// Add external specific capability; remove external unsupported capabilities.
 		for _, cap := range externalSpecificCapabilities {
@@ -499,6 +506,36 @@ func (agent *ecsAgent) appendServiceConnectCapabilities(capabilities []*ecs.Attr
 	return capabilities
 }
 
+func (agent *ecsAgent) appendEBSTaskAttachCapabilities(capabilities []*ecs.Attribute) []*ecs.Attribute {
+	// todo update to import multiple daemons and append capabilities
+	// for now load only EBS CSI Driver daemon
+	daemonDefinitions, err := md.ImportAll()
+	if err != nil {
+		logger.Error(fmt.Sprintf("Daemon import failure: %s", err))
+		return capabilities
+	}
+	if len(daemonDefinitions) == 0 {
+		logger.Warn("daemonDefinitions is empty/nil after import")
+		return capabilities
+	}
+	for _, daemonDef := range daemonDefinitions {
+		if daemonDef.GetImageName() == md.EbsCsiDriver {
+			csiDaemonManager := dm.NewDaemonManager(daemonDef)
+			agent.setDaemonManager(md.EbsCsiDriver, csiDaemonManager)
+			if _, err := csiDaemonManager.LoadImage(agent.ctx, agent.dockerClient); err != nil {
+				logger.Error("Failed to load the EBS CSI Driver. This container instance will not be able to support EBS Task Attach",
+					logger.Fields{
+						field.Error: err,
+					},
+				)
+				return capabilities
+			}
+		}
+	}
+	capabilities = appendNameOnlyAttribute(capabilities, attributePrefix+capabilityEBSTaskAttach)
+	return capabilities
+}
+
 func defaultGetSubDirectories(path string) ([]string, error) {
 	var subDirectories []string