Feature ebs

agent/acs/session/payload_responder.go

@@ -14,6 +14,7 @@
 package session
 
 import (
+	"encoding/json"
 	"fmt"
 
 	"github.com/aws/amazon-ecs-agent/agent/api"
@@ -98,6 +99,16 @@ func (pmHandler *payloadMessageHandler) addPayloadTasks(payload *ecsacs.PayloadM
 	allTasksOK := true
 
 	validTasks := make([]*apitask.Task, 0, len(payload.Tasks))
+
+	data, err := json.Marshal(payload)
+	if err != nil {
+		logger.Debug("Unable to marshal")
+	} else {
+		logger.Debug("Here's the task payload", logger.Fields{
+			"taskPayload": string(data),
+		})
+	}
+
 	for _, task := range payload.Tasks {
 		if task == nil {
 			logger.Critical("Received nil task for message", logger.Fields{
@@ -106,6 +117,7 @@ func (pmHandler *payloadMessageHandler) addPayloadTasks(payload *ecsacs.PayloadM
 			allTasksOK = false
 			continue
 		}
+
 		apiTask, err := apitask.TaskFromACS(task, payload)
 		if err != nil {
 			pmHandler.handleInvalidTask(task, err, payload)

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/ecsclient/client.go

@@ -658,6 +658,10 @@ func (client *APIECSClient) SubmitContainerStateChange(change api.ContainerState
 func (client *APIECSClient) SubmitAttachmentStateChange(change api.AttachmentStateChange) error {
 	attachmentStatus := change.Attachment.Status.String()
 
+	seelog.Infof("attachment arn in ecs client %s", change.Attachment.AttachmentARN)
+	seelog.Infof("attachment in ecs client %v", change.Attachment)
+	seelog.Infof("stringified attachment in ecs client %v", aws.String(change.Attachment.AttachmentARN))
+
 	req := ecs.SubmitAttachmentStateChangesInput{
 		Cluster: &client.config.Cluster,
 		Attachments: []*ecs.AttachmentStateChange{
@@ -668,6 +672,8 @@ func (client *APIECSClient) SubmitAttachmentStateChange(change api.AttachmentSta
 		},
 	}
 
+	seelog.Infof("attachment req in ecs client %v", req)
+
 	_, err := client.submitStateChangeClient.SubmitAttachmentStateChanges(&req)
 	if err != nil {
 		seelog.Warnf("Could not submit attachment state change [%s]: %v", change.String(), err)

agent/api/task/task.go

@@ -43,6 +43,7 @@ import (
 	"github.com/aws/amazon-ecs-agent/ecs-agent/acs/model/ecsacs"
 	apicontainerstatus "github.com/aws/amazon-ecs-agent/ecs-agent/api/container/status"
 	apierrors "github.com/aws/amazon-ecs-agent/ecs-agent/api/errors"
+	apiresource "github.com/aws/amazon-ecs-agent/ecs-agent/api/resource"
 	apitaskstatus "github.com/aws/amazon-ecs-agent/ecs-agent/api/task/status"
 	"github.com/aws/amazon-ecs-agent/ecs-agent/credentials"
 	"github.com/aws/amazon-ecs-agent/ecs-agent/ecs_client/model/ecs"
@@ -331,7 +332,37 @@ func TaskFromACS(acsTask *ecsacs.Task, envelope *ecsacs.PayloadMessage) (*Task,
 	return task, nil
 }
 
+func (task *Task) RemoveVolume(index int) {
+	task.lock.Lock()
+	defer task.lock.Unlock()
+	task.removeVolumeUnsafe(index)
+}
+
+func (task *Task) removeVolumeUnsafe(index int) {
+	if index < 0 || index >= len(task.Volumes) {
+		return
+	}
+	// temp := task.Volumes[:1]
+	out := make([]TaskVolume, 0)
+	out = append(out, task.Volumes[:index]...)
+	out = append(out, task.Volumes[index+1:]...)
+	task.Volumes = out
+}
+
 func (task *Task) initializeVolumes(cfg *config.Config, dockerClient dockerapi.DockerClient, ctx context.Context) error {
+
+	// TODO: Have EBS volumes use the DockerVolumeConfig to create the mountpoint
+	if task.IsEBSTaskAttachEnabled() {
+		ebsVolumes := task.GetEBSVolumeNames()
+		for index, tv := range task.Volumes {
+			volumeName := tv.Name
+			volumeType := tv.Type
+			if ebsVolumes[volumeName] && volumeType != apiresource.EBSTaskAttach {
+				task.RemoveVolume(index)
+			}
+		}
+	}
+
 	err := task.initializeDockerLocalVolumes(dockerClient, ctx)
 	if err != nil {
 		return apierrors.NewResourceInitError(task.Arn, err)
@@ -2571,7 +2602,10 @@ func (task *Task) dockerHostBinds(container *apicontainer.Container) ([]string,
 			return []string{}, errors.Errorf("Unable to resolve volume mounts; invalid path: %s %s; %s -> %s",
 				container.Name, mountPoint.SourceVolume, hv.Source(), mountPoint.ContainerPath)
 		}
-
+		logger.Debug("Volume Binds are the following", logger.Fields{
+			"hostPath":      hv.Source(),
+			"containerPath": mountPoint.ContainerPath,
+		})
 		bind := hv.Source() + ":" + mountPoint.ContainerPath
 		if mountPoint.ReadOnly {
 			bind += ":ro"
@@ -3435,9 +3469,46 @@ func (task *Task) IsServiceConnectEnabled() bool {
 // Is EBS Task Attach enabled returns true if this task has EBS volume configuration in its ACS payload.
 // TODO as more daemons come online, we'll want a generic handler these bool checks and payload handling
 func (task *Task) IsEBSTaskAttachEnabled() bool {
+	task.lock.RLock()
+	defer task.lock.RUnlock()
+	return task.isEBSTaskAttachEnabledUnsafe()
+}
+
+func (task *Task) isEBSTaskAttachEnabledUnsafe() bool {
+	logger.Debug("Checking if there are any ebs volume configs")
+	for _, tv := range task.Volumes {
+		switch tv.Volume.(type) {
+		case *taskresourcevolume.EBSTaskVolumeConfig:
+			logger.Debug("found ebs volume config")
+			return true
+		default:
+			continue
+		}
+	}
 	return false
 }
 
+func (task *Task) GetEBSVolumeNames() map[string]bool {
+	task.lock.RLock()
+	defer task.lock.RUnlock()
+	return task.getEBSVolumeNamesUnsafe()
+}
+
+func (task *Task) getEBSVolumeNamesUnsafe() map[string]bool {
+	volNames := map[string]bool{}
+	for _, tv := range task.Volumes {
+		switch tv.Volume.(type) {
+		case *taskresourcevolume.EBSTaskVolumeConfig:
+			logger.Debug("found ebs volume config")
+			ebsCfg := tv.Volume.(*taskresourcevolume.EBSTaskVolumeConfig)
+			volNames[ebsCfg.VolumeName] = true
+		default:
+			continue
+		}
+	}
+	return volNames
+}
+
 func (task *Task) IsServiceConnectBridgeModeApplicationContainer(container *apicontainer.Container) bool {
 	return container.GetNetworkModeFromHostConfig() == "container" && task.IsServiceConnectEnabled()
 }

agent/api/task/task_attachment_handler.go

@@ -14,6 +14,7 @@
 package task
 
 import (
+	"encoding/json"
 	"fmt"
 
 	"github.com/aws/amazon-ecs-agent/agent/api/serviceconnect"
@@ -83,7 +84,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{
@@ -115,9 +116,19 @@ func handleTaskAttachments(acsTask *ecsacs.Task, task *Task) error {
 				if err != nil {
 					return fmt.Errorf("unable to parse and validate EBS volume: %w", err)
 				}
+
+				data, err := json.Marshal(ebs)
+				if err != nil {
+					logger.Debug("Unable to marshal")
+				} else {
+					logger.Debug("Here's the ebs volume config", logger.Fields{
+						"ebs volume config": string(data),
+					})
+				}
+
 				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,15 @@ func (agent *ecsAgent) doStart(containerChangeEventStream *eventstream.EventStre
 		imageManager.AddImageToCleanUpExclusionList(agent.serviceconnectManager.GetLoadedImageName())
 	}
 
+	if csiDM, ok := agent.daemonManagers["ebs-csi-driver"]; ok {
+		seelog.Debug("CSI daemon manager is not empty")
+		if loaded, _ := csiDM.IsLoaded(agent.dockerClient); loaded {
+			imageManager.AddImageToCleanUpExclusionList(csiDM.GetManagedDaemon().GetLoadedDaemonImageRef())
+		}
+	} else {
+		seelog.Debug("CSI daemon manager is empty...")
+	}
+
 	// Add container instance ARN to metadata manager
 	if agent.cfg.ContainerMetadataEnabled.Enabled() {
 		agent.metadataManager.SetContainerInstanceARN(agent.containerInstanceARN)
@@ -500,6 +487,11 @@ func (agent *ecsAgent) doStart(containerChangeEventStream *eventstream.EventStre
 	agent.startAsyncRoutines(containerChangeEventStream, credentialsManager, imageManager,
 		taskEngine, deregisterInstanceEventStream, client, taskHandler, attachmentEventHandler, state, doctor)
 
+	if err := agent.startEBSWatcher(state, taskEngine); err != nil {
+		seelog.Criticalf("Unable to start EBS watcher")
+		return exitcodes.ExitTerminal
+	}
+
 	// Start the acs session, which should block doStart
 	return agent.startACSSession(credentialsManager, taskEngine,
 		deregisterInstanceEventStream, client, state, taskHandler, doctor)
@@ -1055,7 +1047,7 @@ func (agent *ecsAgent) startACSSession(
 		taskComparer,
 		sequenceNumberAccessor,
 		taskStopper,
-		nil,
+		agent.ebsWatcher,
 		updater.NewUpdater(agent.cfg, state, agent.dataClient, taskEngine).AddAgentUpdateHandlers,
 	)
 	logger.Info("Beginning Polling for updates")