chore(task): run task in environment

Makefile

@@ -152,6 +152,7 @@ gen-mocks: tools
 	${GOBIN}/mockgen -package=mocks -destination=./internal/pkg/describe/mocks/mock_pipeline_status.go -source=./internal/pkg/describe/pipeline_status.go
 	${GOBIN}/mockgen -package=mocks -destination=./internal/pkg/aws/ecr/mocks/mock_ecr.go -source=./internal/pkg/aws/ecr/ecr.go
 	${GOBIN}/mockgen -package=mocks -destination=./internal/pkg/aws/ecs/mocks/mock_ecs.go -source=./internal/pkg/aws/ecs/ecs.go
+	${GOBIN}/mockgen -package=mocks -destination=./internal/pkg/aws/ec2/mocks/mock_ec2.go -source=./internal/pkg/aws/ec2/ec2.go
 	${GOBIN}/mockgen -package=mocks -destination=./internal/pkg/aws/identity/mocks/mock_identity.go -source=./internal/pkg/aws/identity/identity.go
 	${GOBIN}/mockgen -package=mocks -destination=./internal/pkg/aws/route53/mocks/mock_route53.go -source=./internal/pkg/aws/route53/route53.go
 	${GOBIN}/mockgen -package=mocks -destination=./internal/pkg/aws/secretsmanager/mocks/mock_secretsmanager.go -source=./internal/pkg/aws/secretsmanager/secretsmanager.go

internal/pkg/aws/ec2/ec2.go

@@ -6,21 +6,16 @@ package ec2
 
 import (
 	"fmt"
-	"github.com/aws/copilot-cli/internal/pkg/deploy"
-
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/ec2"
 )
 
 const (
 	defaultForAZFilterName = "default-for-az"
-)
 
-// Names for tag filters
-var (
-	TagFilterNameForApp = fmt.Sprintf("tag:%s", deploy.AppTagKey)
-	TagFilterNameForEnv = fmt.Sprintf("tag:%s", deploy.EnvTagKey)
+	// TagFilterName is the filter name format for tag filters
+	TagFilterName = "tag:%s"
 )
 
 var (

internal/pkg/aws/ec2/ec2_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/copilot-cli/internal/pkg/aws/ec2/mocks"
+	"github.com/aws/copilot-cli/internal/pkg/deploy"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
 	"testing"
@@ -17,11 +18,11 @@ import (
 var (
 	inAppEnvFilters = []Filter{
 		Filter{
-			Name: TagFilterNameForApp,
+			Name: fmt.Sprintf(TagFilterName, deploy.AppTagKey),
 			Values: []string{"my-app"},
 		},
 		Filter{
-			Name:   TagFilterNameForEnv,
+			Name:   fmt.Sprintf(TagFilterName, deploy.EnvTagKey),
 			Values: []string{"my-env"},
 		},
 	}

internal/pkg/task/config_runner.go

@@ -4,7 +4,6 @@
 package task
 
 import (
-	"errors"
 	"fmt"
 
 	"github.com/aws/copilot-cli/internal/pkg/aws/ecs"
@@ -34,7 +33,7 @@ func (r *NetworkConfigRunner) Run() ([]string, error) {
 
 	cluster, err := r.ClusterGetter.DefaultCluster()
 	if err != nil {
-		return nil, fmt.Errorf("get default cluster: %w", err)
+		return nil, fmt.Errorf(fmtErrDefaultCluster, err)
 	}
 
 	arns, err := r.Starter.RunTask(ecs.RunTaskInput{
@@ -46,19 +45,22 @@ func (r *NetworkConfigRunner) Run() ([]string, error) {
 		StartedBy:      startedBy,
 	})
 	if err != nil {
-		return nil, fmt.Errorf("run task %s: %w", r.GroupName, err)
+		return nil, &errRunTask{
+			groupName: r.GroupName,
+			parentErr: err,
+		}
 	}
 
 	return arns, nil
 }
 
 func (r *NetworkConfigRunner) validateDependencies() error {
 	if r.ClusterGetter == nil {
-		return errors.New("cluster getter is not set")
+		return errClusterGetterNil
 	}
 
 	if r.Starter == nil {
-		return errors.New("starter is not set")
+		return errStarterNil
 	}
 
 	return nil

internal/pkg/task/config_runner_test.go

@@ -34,7 +34,7 @@ func TestNetworkConfigRunner_Run(t *testing.T) {
 			mockStarter: func(m *mocks.MockTaskRunner) {
 				m.EXPECT().RunTask(gomock.Any()).Times(0)
 			},
-			wantedError: fmt.Errorf("get default cluster: error getting default cluster"),
+			wantedError: fmt.Errorf(fmtErrDefaultCluster, errors.New("error getting default cluster")),
 		},
 		"failed to kick off task": {
 			count:     1,
@@ -47,17 +47,13 @@ func TestNetworkConfigRunner_Run(t *testing.T) {
 				m.EXPECT().DefaultCluster().Return("cluster-1", nil)
 			},
 			mockStarter: func(m *mocks.MockTaskRunner) {
-				m.EXPECT().RunTask(ecs.RunTaskInput{
-					Cluster:        "cluster-1",
-					Count:          1,
-					Subnets:        []string{"subnet-1", "subnet-2"},
-					SecurityGroups: []string{"sg-1", "sg-2"},
-					TaskFamilyName: taskFamilyName("my-task"),
-					StartedBy:      startedBy,
-				}).Return(nil, errors.New("error running task"))
+				m.EXPECT().RunTask(gomock.Any()).Return(nil, errors.New("error running task"))
 			},
 
-			wantedError: fmt.Errorf("run task my-task: error running task"),
+			wantedError: &errRunTask{
+				groupName: "my-task",
+				parentErr: errors.New("error running task"),
+			},
 		},
 		"successfully kick off task with both input subnets and security groups": {
 			count:     1,

internal/pkg/task/default_runner.go

@@ -4,12 +4,15 @@
 package task
 
 import (
-	"errors"
 	"fmt"
 	"github.com/aws/copilot-cli/internal/pkg/aws/ec2"
 	"github.com/aws/copilot-cli/internal/pkg/aws/ecs"
 )
 
+const (
+	fmtErrDefaultSubnets = "get default subnet IDs: %w"
+)
+
 // DefaultVPCRunner can run an Amazon ECS task in the default VPC and the default cluster.
 type DefaultVPCRunner struct {
 	// Count of the tasks to be launched.
@@ -31,12 +34,12 @@ func (r *DefaultVPCRunner) Run() ([]string, error) {
 
 	cluster, err := r.ClusterGetter.DefaultCluster()
 	if err != nil {
-		return nil, fmt.Errorf("get default cluster: %w", err)
+		return nil, fmt.Errorf(fmtErrDefaultCluster, err)
 	}
 
 	subnets, err := r.VPCGetter.SubnetIDs(ec2.FilterForDefaultVPCSubnets)
 	if err != nil {
-		return nil, fmt.Errorf("get default subnet IDs: %w", err)
+		return nil, fmt.Errorf(fmtErrDefaultSubnets, err)
 	}
 	if len(subnets) == 0 {
 		return nil, errNoSubnetFound
@@ -50,23 +53,26 @@ func (r *DefaultVPCRunner) Run() ([]string, error) {
 		StartedBy:      startedBy,
 	})
 	if err != nil {
-		return nil, fmt.Errorf("run task %s: %w", r.GroupName, err)
+		return nil, &errRunTask{
+			groupName: r.GroupName,
+			parentErr: err,
+		}
 	}
 
 	return arns, nil
 }
 
 func (r *DefaultVPCRunner) validateDependencies() error {
 	if r.VPCGetter == nil {
-		return errors.New("vpc getter is not set")
+		return errVPCGetterNil
 	}
 
 	if r.ClusterGetter == nil {
-		return errors.New("cluster getter is not set")
+		return errClusterGetterNil
 	}
 
 	if r.Starter == nil {
-		return errors.New("starter is not set")
+		return errStarterNil
 	}
 
 	return nil

internal/pkg/task/default_runner_test.go

@@ -36,7 +36,7 @@ func TestDefaultVPCRunner_Run(t *testing.T) {
 			mockStarter: func(m *mocks.MockTaskRunner) {
 				m.EXPECT().RunTask(gomock.Any()).Times(0)
 			},
-			wantedError: fmt.Errorf("get default cluster: error getting cluster"),
+			wantedError: fmt.Errorf(fmtErrDefaultCluster, errors.New("error getting cluster")),
 		},
 		"failed to get subnet": {
 			mockClusterGetter: func(m *mocks.MockDefaultClusterGetter) {
@@ -48,7 +48,7 @@ func TestDefaultVPCRunner_Run(t *testing.T) {
 			mockStarter: func(m *mocks.MockTaskRunner) {
 				m.EXPECT().RunTask(gomock.Any()).Times(0)
 			},
-			wantedError: errors.New("get default subnet IDs: error getting subnets"),
+			wantedError: fmt.Errorf(fmtErrDefaultSubnets, errors.New("error getting subnets")),
 		},
 		"failed to kick off task": {
 			count:     1,
@@ -62,7 +62,10 @@ func TestDefaultVPCRunner_Run(t *testing.T) {
 			mockStarter: func(m *mocks.MockTaskRunner) {
 				m.EXPECT().RunTask(gomock.Any()).Return(nil, errors.New("error running task"))
 			},
-			wantedError: errors.New("run task my-task: error running task"),
+			wantedError: &errRunTask{
+				groupName: "my-task",
+				parentErr: errors.New("error running task"),
+			},
 		},
 	}
 

internal/pkg/task/env_runner.go

@@ -0,0 +1,137 @@
+// Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package task
+
+import (
+	"fmt"
+	"github.com/aws/copilot-cli/internal/pkg/aws/ec2"
+	"github.com/aws/copilot-cli/internal/pkg/aws/ecs"
+	"github.com/aws/copilot-cli/internal/pkg/deploy"
+)
+
+const (
+	clusterResourceType  = "ecs:cluster"
+
+	fmtErrClusterFromEnv        = "get cluster by env %s: %w"
+	fmtErrNoClusterFoundFromEnv = "no cluster found in env %s"
+	fmtErrMoreThanOneClusterFromEnv = "more than one cluster is found in environment %s"
+	fmtErrPublicSubnetsFromEnv  = "get public subnet IDs from environment %s: %w"
+	fmtErrSecurityGroupsFromEnv = "get security groups from environment %s: %w"
+)
+
+// Names for tag filters
+var (
+	tagFilterNameForApp = fmt.Sprintf(ec2.TagFilterName, deploy.AppTagKey)
+	tagFilterNameForEnv = fmt.Sprintf(ec2.TagFilterName, deploy.EnvTagKey)
+)
+
+// EnvRunner can run an Amazon ECS task in the VPC and the cluster of an environment.
+type EnvRunner struct {
+	// Count of the tasks to be launched.
+	Count int
+	// Group Name of the tasks that use the same task definition.
+	GroupName string
+
+	// App and Env in which the tasks will be launched.
+	App string
+	Env string
+
+	// Interfaces to interact with dependencies. Must not be nil.
+	VPCGetter     VPCGetter
+	ClusterGetter ResourceGetter
+	Starter       TaskRunner
+}
+
+// Run runs tasks in the environment of the application, and returns the task ARNs.
+func (r *EnvRunner) Run() ([]string, error) {
+	if err := r.validateDependencies(); err != nil {
+		return nil, err
+	}
+
+	cluster, err := r.cluster(r.App, r.Env)
+	if err != nil {
+		return nil, err
+	}
+
+	filters := r.filtersForVPCFromAppEnv()
+
+	subnets, err := r.VPCGetter.PublicSubnetIDs(filters...)
+	if err != nil {
+		return nil, fmt.Errorf(fmtErrPublicSubnetsFromEnv, r.Env, err)
+	}
+	if len(subnets) == 0 {
+		return nil, errNoSubnetFound
+	}
+
+	securityGroups, err := r.VPCGetter.SecurityGroups(filters...)
+	if err != nil {
+		return nil, fmt.Errorf(fmtErrSecurityGroupsFromEnv, r.Env, err)
+	}
+
+	arns, err := r.Starter.RunTask(ecs.RunTaskInput{
+		Cluster:        cluster,
+		Count:          r.Count,
+		Subnets:        subnets,
+		SecurityGroups: securityGroups,
+		TaskFamilyName: taskFamilyName(r.GroupName),
+		StartedBy:      startedBy,
+	})
+	if err != nil {
+		return nil, &errRunTask{
+			groupName: r.GroupName,
+			parentErr: err,
+		}
+	}
+	return arns, nil
+}
+
+func (r *EnvRunner) cluster(app, env string) (string, error) {
+	clusters, err := r.ClusterGetter.GetResourcesByTags(clusterResourceType, map[string]string{
+		deploy.AppTagKey: app,
+		deploy.EnvTagKey: env,
+	})
+
+	if err != nil {
+		return "", fmt.Errorf(fmtErrClusterFromEnv, env, err)
+	}
+
+	if len(clusters) == 0 {
+		return "", fmt.Errorf(fmtErrNoClusterFoundFromEnv, env)
+	}
+
+	// NOTE: only one cluster is associated with an application and an environment
+	if len(clusters) > 1 {
+		return "", fmt.Errorf(fmtErrMoreThanOneClusterFromEnv, r.Env)
+	}
+	return clusters[0].ARN, nil
+}
+
+func (r *EnvRunner) filtersForVPCFromAppEnv() []ec2.Filter {
+	return []ec2.Filter{
+		ec2.Filter{
+			Name:   tagFilterNameForEnv,
+			Values: []string{r.Env},
+		},
+		ec2.Filter{
+			Name:   tagFilterNameForApp,
+			Values: []string{r.App},
+		},
+	}
+}
+
+func (r *EnvRunner) validateDependencies() error {
+	if r.VPCGetter == nil {
+		return errVPCGetterNil
+	}
+
+	if r.ClusterGetter == nil {
+		return errClusterGetterNil
+	}
+
+	if r.Starter == nil {
+		return errStarterNil
+	}
+
+	return nil
+}