EFS functional testing (#2247)

* Add efs client to Gopkg.*

* Add efs client to vendor directory

* EFS functional test

* Reuse EFS filesystem and mount target(s)

* review fixups

* more code review fixups

agent/functional_tests/testdata/taskdefinitions/task-efs-vol-read/task-definition.json

@@ -0,0 +1,38 @@
+{
+    "family": "ecsftest-task-efs-volume",
+    "containerDefinitions": [
+        {
+            "name": "task-efs-vol-read",
+            "image": "127.0.0.1:51670/busybox:latest",
+            "cpu": 10,
+            "command": ["sh", "-c", "while true; do sleep 1; [ -f /ecs/success ] && if grep -q 'can you read me' /ecs/success; then exit 42; fi done"],
+            "memory": 256,
+            "memoryReservation": 128,
+            "mountPoints": [
+                {
+                  "sourceVolume": "task-efs-shared",
+                  "containerPath": "/ecs/"
+                }
+            ]
+        }
+    ],
+    "volumes":[
+        {
+            "name": "task-efs-shared",
+            "host": null,
+            "dockerVolumeConfiguration" : {
+                "scope": "shared",
+                "autoprovision": true,
+                "driver": "local",
+                "labels": {
+                    "mylabels": "test"
+                },
+                "driverOpts": {
+                    "type": "nfs",
+                    "device": ":/",
+                    "o": "addr=FILESYSTEM_ID.efs.TEST_REGION.amazonaws.com,nfsvers=4.1,rsize=1048576,wsize=1048576,hard,timeo=600,retrans=2,noresvport"
+                }
+            }
+        }
+    ]
+}

agent/functional_tests/testdata/taskdefinitions/task-efs-vol-write/task-definition.json

@@ -0,0 +1,38 @@
+{
+    "family": "ecsftest-task-efs-volume",
+    "containerDefinitions": [
+        {
+            "name": "task-efs-vol-write",
+            "image": "127.0.0.1:51670/busybox:latest",
+            "cpu": 10,
+            "command": ["sh", "-c", "echo 'can you read me'> /ecs/success; [ -f /ecs/success ] && exit 42 || exit 1"],
+            "memory": 256,
+            "memoryReservation": 128,
+            "mountPoints": [
+                {
+                  "sourceVolume": "task-efs-shared",
+                  "containerPath": "/ecs/"
+                }
+            ]
+        }
+    ],
+    "volumes":[
+        {
+            "name": "task-efs-shared",
+            "host": null,
+            "dockerVolumeConfiguration" : {
+                "scope": "shared",
+                "autoprovision": true,
+                "driver": "local",
+                "labels": {
+                    "mylabels": "test"
+                },
+                "driverOpts": {
+                    "type": "nfs",
+                    "device": ":/",
+                    "o": "addr=FILESYSTEM_ID.efs.TEST_REGION.amazonaws.com,nfsvers=4.1,rsize=1048576,wsize=1048576,hard,timeo=600,retrans=2,noresvport"
+                }
+            }
+        }
+    ]
+}

agent/functional_tests/tests/functionaltests_unix_test.go

@@ -37,6 +37,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/cloudwatchlogs"
+	"github.com/aws/aws-sdk-go/service/efs"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/aws/aws-sdk-go/service/s3/s3manager"
 	"github.com/aws/aws-sdk-go/service/secretsmanager"
@@ -1354,6 +1355,134 @@ func TestASMSecretsARN(t *testing.T) {
 	assert.Equal(t, 42, exitCode, fmt.Sprintf("Expected exit code of 42; got %d", exitCode))
 }
 
+// TestRunEFSVolumeTask does the following:
+//   1. creates an EFS filesystem with a mount target.
+//   2. spins up a task to mount and write to the filesystem.
+//   3. spins up a task to mount and read from the filesystem.
+//   4. TODO: do this via EFSVolumeConfiguration instead of via NFS/Docker.
+func TestRunEFSVolumeTask(t *testing.T) {
+	os.Setenv("ECS_FTEST_FORCE_NET_HOST", "true")
+	defer os.Unsetenv("ECS_FTEST_FORCE_NET_HOST")
+	if IsCNPartition() {
+		t.Skip("Skip TestRunEFSVolumeTask in China partition")
+	}
+
+	agent := RunAgent(t, nil)
+	defer agent.Cleanup()
+
+	efsClient := efs.New(session.New(), aws.NewConfig().WithRegion(*ECS.Config.Region))
+	fsID := createEFSFileSystem(t, efsClient)
+	createMountTarget(t, efsClient, fsID)
+
+	// start writer task first
+	overrides := map[string]string{
+		"FILESYSTEM_ID": fsID,
+		"TEST_REGION":   *ECS.Config.Region,
+	}
+	wTask, err := agent.StartTaskWithTaskDefinitionOverrides(t, "task-efs-vol-write", overrides)
+	require.NoError(t, err, "Register task definition failed")
+	// Wait for the first task to create the volume
+	wErr := wTask.WaitStopped(waitTaskStateChangeDuration)
+	require.NoError(t, wErr, "Error waiting for task to transition to STOPPED")
+	wExitCode, ok := wTask.ContainerExitcode("task-efs-vol-write")
+	require.True(t, ok, "Error code for container [task-efs-vol-write] not found, check the logs")
+	require.Equal(t, 42, wExitCode, fmt.Sprintf("Expected exit code of 42; got %d", wExitCode))
+
+	// then reader task try to read from the volume
+	rTask, err := agent.StartTaskWithTaskDefinitionOverrides(t, "task-efs-vol-read", overrides)
+	require.NoError(t, err, "Register task definition failed")
+
+	rErr := rTask.WaitStopped(waitTaskStateChangeDuration)
+	require.NoError(t, rErr, "Error waiting for task to transition to STOPPED")
+	rExitCode, ok := rTask.ContainerExitcode("task-efs-vol-read")
+	require.True(t, ok, "Error code for container [task-efs-vol-read] not found, check the logs")
+	require.Equal(t, 42, rExitCode, fmt.Sprintf("Expected exit code of 42; got %d", rExitCode))
+	return
+}
+
+// createEFSFileSystem creates a new EFS file system and returns the FileSystemId.
+// will ignore already-created filesystems
+// also will wait until filesystem is "available" before returning
+func createEFSFileSystem(t *testing.T, efsClient *efs.EFS) string {
+	creationToken := "efs-func-tests"
+	fs, err := efsClient.CreateFileSystem(&efs.CreateFileSystemInput{
+		CreationToken: aws.String(creationToken),
+	})
+
+	if err != nil {
+		if aerr, ok := err.(awserr.Error); ok {
+			switch aerr.Code() {
+			case efs.ErrCodeFileSystemAlreadyExists:
+				t.Logf("EFS filesystem already exists")
+				out, err := efsClient.DescribeFileSystems(&efs.DescribeFileSystemsInput{
+					CreationToken: aws.String(creationToken),
+				})
+				require.NoError(t, err, "Unexpected error from DescribeFileSystems: %s", err)
+				return *out.FileSystems[0].FileSystemId
+			default:
+				require.NoError(t, err, "Error creating EFS Filesystem")
+			}
+		}
+	}
+
+	// Wait for filesystem to be "available"
+	for i := 0; i < 20; i++ {
+		out, err := efsClient.DescribeFileSystems(&efs.DescribeFileSystemsInput{
+			CreationToken: aws.String(creationToken),
+		})
+		require.NoError(t, err, "Unexpected error from DescribeFileSystems: %s", err)
+		if *out.FileSystems[0].LifeCycleState == efs.LifeCycleStateAvailable {
+			return *out.FileSystems[0].FileSystemId
+		}
+		time.Sleep(time.Second * 30)
+	}
+
+	t.Fatalf("Test timed out waiting for EFS Filesystem [%s] to become 'available'", *fs.FileSystemId)
+	return ""
+}
+
+// createMountTarget attempts to create a mount target on the given filesystem ID
+// if it already exists, the error code (MountTargetConflict) is ignored.
+func createMountTarget(t *testing.T, efsClient *efs.EFS, fsID string) {
+	subnetID, err := GetSubnetID()
+	require.NoError(t, err)
+	sgroups, err := GetSecurityGroupIDs()
+	require.NoError(t, err)
+	sgroupsP := []*string{}
+	for _, sgroup := range sgroups {
+		sgroupsP = append(sgroupsP, aws.String(sgroup))
+	}
+	mt, err := efsClient.CreateMountTarget(&efs.CreateMountTargetInput{
+		FileSystemId:   aws.String(fsID),
+		SubnetId:       aws.String(subnetID),
+		SecurityGroups: sgroupsP,
+	})
+	if err != nil {
+		if aerr, ok := err.(awserr.Error); ok {
+			switch aerr.Code() {
+			case efs.ErrCodeMountTargetConflict:
+				t.Logf("EFS mount target already exists")
+				return
+			default:
+				require.NoError(t, err, "Error creating EFS mount target")
+			}
+		}
+	}
+	// Wait for mount target to be "available"
+	for i := 0; i < 20; i++ {
+		out, err := efsClient.DescribeMountTargets(&efs.DescribeMountTargetsInput{
+			MountTargetId: mt.MountTargetId,
+		})
+		require.NoError(t, err, "Unexpected error from DescribeMountTargets: %s", err)
+		if *out.MountTargets[0].LifeCycleState == efs.LifeCycleStateAvailable {
+			return
+		}
+		time.Sleep(time.Second * 30)
+	}
+
+	t.Fatalf("Test timed out waiting for EFS mount target [%s] to become 'available'", *mt.MountTargetId)
+}
+
 // Note: This functional test requires ECS GPU instance which has at least 1 GPU.
 func TestRunGPUTask(t *testing.T) {
 	gpuInstances := []string{"p2", "p3", "g3", "g4dn"}

agent/functional_tests/util/utils.go

@@ -275,7 +275,7 @@ func (agent *TestAgent) startAWSVPCTask(taskDefinition string) (*TestTask, error
 	agent.t.Logf("Task definition: %s", taskDefinition)
 	// Get the subnet ID, which is a required parameter for starting
 	// tasks in 'awsvpc' network mode
-	subnet, err := getSubnetID()
+	subnet, err := GetSubnetID()
 	if err != nil {
 		return nil, err
 	}
@@ -809,8 +809,8 @@ func AttributesToMap(attributes []*ecs.Attribute) map[string]string {
 	return attributeMap
 }
 
-// getSubnetID gets the subnet id for the instance from ec2 instance metadata
-func getSubnetID() (string, error) {
+// GetSubnetID gets the subnet id for the instance from ec2 instance metadata
+func GetSubnetID() (string, error) {
 	ec2Metadata := ec2metadata.New(session.Must(session.NewSession()))
 	mac, err := ec2Metadata.GetMetadata("mac")
 	if err != nil {
@@ -824,6 +824,21 @@ func getSubnetID() (string, error) {
 	return subnet, nil
 }
 
+// GetSecurityGroupIDs returns all of the security group IDs that the instance is in.
+func GetSecurityGroupIDs() ([]string, error) {
+	ec2Metadata := ec2metadata.New(session.Must(session.NewSession()))
+	mac, err := ec2Metadata.GetMetadata("mac")
+	if err != nil {
+		return []string{}, errors.Wrapf(err, "unable to get mac from ec2 metadata")
+	}
+	sgroups, err := ec2Metadata.GetMetadata("network/interfaces/macs/" + mac + "/security-group-ids")
+	if err != nil {
+		return []string{}, errors.Wrapf(err, "unable to get security group ids from ec2 metadata")
+	}
+
+	return strings.Fields(sgroups), nil
+}
+
 // GetAccountID returns the aws account id from the instance metadata
 func GetAccountID() (string, error) {
 	ec2Metadata := ec2metadata.New(session.Must(session.NewSession()))