testing node unstage volume

agent/engine/task_manager.go

@@ -17,6 +17,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"sync"
@@ -34,10 +35,12 @@ import (
 	"github.com/aws/amazon-ecs-agent/agent/statechange"
 	"github.com/aws/amazon-ecs-agent/agent/taskresource"
 	resourcestatus "github.com/aws/amazon-ecs-agent/agent/taskresource/status"
+	taskresourcevolume "github.com/aws/amazon-ecs-agent/agent/taskresource/volume"
 	apicontainerstatus "github.com/aws/amazon-ecs-agent/ecs-agent/api/container/status"
 	apierrors "github.com/aws/amazon-ecs-agent/ecs-agent/api/errors"
 	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/csiclient"
 	"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"
@@ -252,6 +255,12 @@ func (mtask *managedTask) overseeTask() {
 	mtask.engine.wakeUpTaskQueueMonitor()
 	// TODO: make this idempotent on agent restart
 	go mtask.releaseIPInIPAM()
+
+	err := mtask.UnstageVolumes()
+	if err != nil {
+		logger.Error(fmt.Sprintf("Unable to unstage volumes: %v", err))
+	}
+
 	mtask.cleanupTask(retry.AddJitter(mtask.cfg.TaskCleanupWaitDuration, mtask.cfg.TaskCleanupWaitDurationJitter))
 }
 
@@ -1559,3 +1568,38 @@ func (mtask *managedTask) waitForStopReported() bool {
 	}
 	return taskStopped
 }
+
+func (mtask *managedTask) UnstageVolumes() error {
+	task := mtask.Task
+	if task == nil {
+		return fmt.Errorf("task not is not managed")
+	}
+	if !task.IsEBSTaskAttachEnabled() {
+		logger.Debug("Task is not EBS-backed. Skip NodeUnstageVolume.")
+		return nil
+	}
+	csiClient := csiclient.NewCSIClient(filepath.Join(csiclient.SocketHostPath, csiclient.ImageName, csiclient.SocketName))
+	for _, tv := range task.Volumes {
+		switch tv.Volume.(type) {
+		case *taskresourcevolume.EBSTaskVolumeConfig:
+			ebsCfg := tv.Volume.(*taskresourcevolume.EBSTaskVolumeConfig)
+			volumeId := ebsCfg.VolumeId
+			hostPath := ebsCfg.Source()
+			err := mtask.unstageVolumeWithTimeout(&csiClient, volumeId, hostPath)
+			if err != nil {
+				logger.Error("Unable to unstage volume", logger.Fields{
+					"Task":  task.String(),
+					"Error": err,
+				})
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (mtask *managedTask) unstageVolumeWithTimeout(csiClient csiclient.CSIClient, volumeId, hostPath string) error {
+	derivedCtx, cancel := context.WithTimeout(mtask.ctx, time.Second*3)
+	defer cancel()
+	return csiClient.NodeUnstageVolume(derivedCtx, volumeId, hostPath)
+}