chore: render env stack updates from the env controller (#1897)

https://user-images.githubusercontent.com/879348/106506161-2edfae00-647e-11eb-9aee-42d1e5d81b1e.mov

_By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice._

Author: efekarakus

internal/pkg/deploy/cloudformation/cloudformation.go

@@ -17,6 +17,7 @@ import (
 	"github.com/aws/copilot-cli/internal/pkg/aws/cloudformation"
 	"github.com/aws/copilot-cli/internal/pkg/aws/cloudformation/stackset"
 	"github.com/aws/copilot-cli/internal/pkg/aws/ecs"
+	"github.com/aws/copilot-cli/internal/pkg/deploy"
 	"github.com/aws/copilot-cli/internal/pkg/stream"
 	"github.com/aws/copilot-cli/internal/pkg/term/log"
 	"github.com/aws/copilot-cli/internal/pkg/term/progress"
@@ -30,7 +31,8 @@ const (
 	waitForStackTimeout = 1*time.Hour + 30*time.Minute
 
 	// CloudFormation resource types.
-	ecsServiceResourceType = "AWS::ECS::Service"
+	ecsServiceResourceType    = "AWS::ECS::Service"
+	envControllerResourceType = "Custom::EnvControllerFunction"
 )
 
 // StackConfiguration represents the set of methods needed to deploy a cloudformation stack.
@@ -174,7 +176,7 @@ func (cf CloudFormation) renderStackChanges(in *renderStackChangesInput) error {
 	defer cancelWait()
 	g, ctx := errgroup.WithContext(waitCtx)
 
-	renderer, err := cf.createStackRenderer(g, ctx, changeSetID, in.stackName, in.stackDescription, progress.RenderOptions{})
+	renderer, err := cf.createChangeSetRenderer(g, ctx, changeSetID, in.stackName, in.stackDescription, progress.RenderOptions{})
 	if err != nil {
 		return err
 	}
@@ -190,7 +192,7 @@ func (cf CloudFormation) renderStackChanges(in *renderStackChangesInput) error {
 	return nil
 }
 
-func (cf CloudFormation) createStackRenderer(group *errgroup.Group, ctx context.Context, changeSetID, stackName, description string, opts progress.RenderOptions) (progress.DynamicRenderer, error) {
+func (cf CloudFormation) createChangeSetRenderer(group *errgroup.Group, ctx context.Context, changeSetID, stackName, description string, opts progress.RenderOptions) (progress.DynamicRenderer, error) {
 	changeSet, err := cf.cfnClient.DescribeChangeSet(changeSetID, stackName)
 	if err != nil {
 		return nil, err
@@ -206,30 +208,34 @@ func (cf CloudFormation) createStackRenderer(group *errgroup.Group, ctx context.
 
 	streamer := stream.NewStackStreamer(cf.cfnClient, stackName, changeSet.CreationTime)
 	children, err := cf.changeRenderers(changeRenderersInput{
-		g:             group,
-		ctx:           ctx,
-		stackStreamer: streamer,
-		changes:       changeSet.Changes,
-		descriptions:  descriptions,
-		opts:          progress.NestedRenderOptions(opts),
+		g:                  group,
+		ctx:                ctx,
+		stackName:          stackName,
+		stackStreamer:      streamer,
+		changes:            changeSet.Changes,
+		changeSetTimestamp: changeSet.CreationTime,
+		descriptions:       descriptions,
+		opts:               progress.NestedRenderOptions(opts),
 	})
 	if err != nil {
 		return nil, err
 	}
-	renderer := progress.ListeningStackRenderer(streamer, stackName, description, children, opts)
+	renderer := progress.ListeningChangeSetRenderer(streamer, stackName, description, children, opts)
 	group.Go(func() error {
 		return stream.Stream(ctx, streamer)
 	})
 	return renderer, nil
 }
 
 type changeRenderersInput struct {
-	g             *errgroup.Group             // Group that all goroutines belong.
-	ctx           context.Context             // Context associated with the group.
-	stackStreamer progress.StackSubscriber    // Streamer for the stack where changes belong.
-	changes       []*sdkcloudformation.Change // List of changes that will be applied to the stack.
-	descriptions  map[string]string           // Descriptions for the logical IDs of the changes.
-	opts          progress.RenderOptions      // Display options that should be applied to the changes.
+	g                  *errgroup.Group             // Group that all goroutines belong.
+	ctx                context.Context             // Context associated with the group.
+	stackName          string                      // Name of the stack.
+	stackStreamer      progress.StackSubscriber    // Streamer for the stack where changes belong.
+	changes            []*sdkcloudformation.Change // List of changes that will be applied to the stack.
+	changeSetTimestamp time.Time                   // ChangeSet creation time.
+	descriptions       map[string]string           // Descriptions for the logical IDs of the changes.
+	opts               progress.RenderOptions      // Display options that should be applied to the changes.
 }
 
 // changeRenderers filters changes by resources that have a description and returns the appropriate progress.Renderer for each resource type.
@@ -243,6 +249,21 @@ func (cf CloudFormation) changeRenderers(in changeRenderersInput) ([]progress.Re
 		}
 		var renderer progress.Renderer
 		switch {
+		case aws.StringValue(change.ResourceChange.ResourceType) == envControllerResourceType:
+			r, err := cf.createEnvControllerRenderer(&envControllerRendererInput{
+				g:                 in.g,
+				ctx:               in.ctx,
+				workloadStackName: in.stackName,
+				workloadTimestamp: in.changeSetTimestamp,
+				change:            change,
+				description:       description,
+				serviceStack:      in.stackStreamer,
+				renderOpts:        in.opts,
+			})
+			if err != nil {
+				return nil, err
+			}
+			renderer = r
 		case aws.StringValue(change.ResourceChange.ResourceType) == ecsServiceResourceType:
 			renderer = progress.ListeningECSServiceResourceRenderer(in.stackStreamer, cf.ecsClient, logicalID, description, progress.ECSServiceRendererOpts{
 				Group:      in.g,
@@ -254,19 +275,63 @@ func (cf CloudFormation) changeRenderers(in changeRenderersInput) ([]progress.Re
 			changeSetID := aws.StringValue(change.ResourceChange.ChangeSetId)
 			stackName := parseStackNameFromARN(aws.StringValue(change.ResourceChange.PhysicalResourceId))
 
-			r, err := cf.createStackRenderer(in.g, in.ctx, changeSetID, stackName, description, in.opts)
+			r, err := cf.createChangeSetRenderer(in.g, in.ctx, changeSetID, stackName, description, in.opts)
 			if err != nil {
 				return nil, err
 			}
 			renderer = r
 		default:
-			renderer = progress.ListeningResourceRenderer(in.stackStreamer, logicalID, description, in.opts)
+			renderer = progress.ListeningResourceRenderer(in.stackStreamer, logicalID, description, progress.ResourceRendererOpts{
+				RenderOpts: in.opts,
+			})
 		}
 		resources = append(resources, renderer)
 	}
 	return resources, nil
 }
 
+type envControllerRendererInput struct {
+	g                 *errgroup.Group
+	ctx               context.Context
+	workloadStackName string
+	workloadTimestamp time.Time
+	change            *sdkcloudformation.Change
+	description       string
+	serviceStack      progress.StackSubscriber
+	renderOpts        progress.RenderOptions
+}
+
+func (cf CloudFormation) createEnvControllerRenderer(in *envControllerRendererInput) (progress.DynamicRenderer, error) {
+	switch {
+	case isEnvControllerInAction(in.change):
+		// We can't use the workloadStackName to retrieve the envStackName because app names and env names can contain
+		// dashes. So we fetch the tags belonging to the workload stack and generate the stack name from there.
+		workload, err := cf.cfnClient.Describe(in.workloadStackName)
+		if err != nil {
+			return nil, err
+		}
+		envStackName := fmt.Sprintf("%s-%s", parseAppNameFromTags(workload.Tags), parseEnvNameFromTags(workload.Tags))
+		body, err := cf.cfnClient.TemplateBody(envStackName)
+		if err != nil {
+			return nil, err
+		}
+		descriptions, err := cloudformation.ParseTemplateDescriptions(body)
+		if err != nil {
+			return nil, fmt.Errorf("parse cloudformation template for resource descriptions: %w", err)
+		}
+		streamer := stream.NewStackStreamer(cf.cfnClient, envStackName, in.workloadTimestamp)
+		in.g.Go(func() error {
+			return stream.Stream(in.ctx, streamer)
+		})
+		return progress.ListeningStackRenderer(streamer, envStackName, in.description, descriptions, in.renderOpts), nil
+	default:
+		logicalID := aws.StringValue(in.change.ResourceChange.LogicalResourceId)
+		return progress.ListeningResourceRenderer(in.serviceStack, logicalID, in.description, progress.ResourceRendererOpts{
+			RenderOpts: in.renderOpts,
+		}), nil
+	}
+}
+
 func (cf CloudFormation) errOnFailedStack(stackName string) error {
 	stack, err := cf.cfnClient.Describe(stackName)
 	if err != nil {
@@ -307,6 +372,42 @@ func parseStackNameFromARN(stackARN string) string {
 	return strings.Split(stackARN, "/")[1]
 }
 
+func isEnvControllerInAction(controller *sdkcloudformation.Change) bool {
+	switch action := aws.StringValue(controller.ResourceChange.Action); action {
+	case sdkcloudformation.ChangeActionAdd:
+		return true
+	case sdkcloudformation.ChangeActionModify:
+		for _, detail := range controller.ResourceChange.Details {
+			attribute := aws.StringValue(detail.Target.Name)
+			if attribute == "Parameters" {
+				// The env controller modifies the env stack only if the "Parameters" field is modified.
+				return true
+			}
+		}
+		return false
+	default:
+		return false
+	}
+}
+
+func parseAppNameFromTags(tags []*sdkcloudformation.Tag) string {
+	for _, t := range tags {
+		if aws.StringValue(t.Key) == deploy.AppTagKey {
+			return aws.StringValue(t.Value)
+		}
+	}
+	return ""
+}
+
+func parseEnvNameFromTags(tags []*sdkcloudformation.Tag) string {
+	for _, t := range tags {
+		if aws.StringValue(t.Key) == deploy.EnvTagKey {
+			return aws.StringValue(t.Value)
+		}
+	}
+	return ""
+}
+
 func stopSpinner(spinner *progress.Spinner, err error, label string) {
 	if err == nil {
 		spinner.Stop(log.Ssuccessf("%s\n", label))

internal/pkg/deploy/cloudformation/cloudformation_test.go

@@ -252,6 +252,98 @@ Resources:
 	require.Contains(t, buf.String(), "[completed]", "Rollout state of service should be rendered")
 }
 
+func testDeployWorkload_WithEnvControllerRenderer(t *testing.T, svcStackName string, when func(w progress.FileWriter, cf CloudFormation) error) {
+	// GIVEN
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	mockCFN := mocks.NewMockcfnClient(ctrl)
+	deploymentTime := time.Date(2020, time.November, 23, 18, 0, 0, 0, time.UTC)
+
+	mockCFN.EXPECT().Create(gomock.Any()).Return("1234", nil)
+	mockCFN.EXPECT().DescribeChangeSet("1234", svcStackName).Return(&cloudformation.ChangeSetDescription{
+		Changes: []*sdkcloudformation.Change{
+			{
+				ResourceChange: &sdkcloudformation.ResourceChange{
+					LogicalResourceId: aws.String("EnvControllerAction"),
+					ResourceType:      aws.String("Custom::EnvControllerFunction"),
+					Action:            aws.String(sdkcloudformation.ChangeActionAdd),
+				},
+			},
+		},
+	}, nil)
+	mockCFN.EXPECT().TemplateBodyFromChangeSet("1234", svcStackName).Return(`
+Resources:
+  EnvControllerAction:
+    Metadata:
+      'aws:copilot:description': "Updating environment"
+`, nil)
+	mockCFN.EXPECT().Describe(svcStackName).Return(&cloudformation.StackDescription{
+		Tags: []*sdkcloudformation.Tag{
+			{
+				Key:   aws.String("copilot-application"),
+				Value: aws.String("my-app"),
+			},
+			{
+				Key:   aws.String("copilot-environment"),
+				Value: aws.String("my-env"),
+			},
+		},
+	}, nil)
+	mockCFN.EXPECT().TemplateBody("my-app-my-env").Return(`
+Resources:
+  PublicLoadBalancer:
+    Metadata:
+      'aws:copilot:description': "Updating ALB"
+`, nil)
+	mockCFN.EXPECT().DescribeStackEvents(&sdkcloudformation.DescribeStackEventsInput{
+		StackName: aws.String(svcStackName),
+	}).Return(&sdkcloudformation.DescribeStackEventsOutput{
+		StackEvents: []*sdkcloudformation.StackEvent{
+			{
+				EventId:           aws.String("1"),
+				LogicalResourceId: aws.String(svcStackName),
+				ResourceType:      aws.String("AWS::CloudFormation::Stack"),
+				ResourceStatus:    aws.String("CREATE_COMPLETE"),
+				Timestamp:         aws.Time(deploymentTime),
+			},
+		},
+	}, nil).AnyTimes()
+	mockCFN.EXPECT().DescribeStackEvents(&sdkcloudformation.DescribeStackEventsInput{
+		StackName: aws.String("my-app-my-env"),
+	}).Return(&sdkcloudformation.DescribeStackEventsOutput{
+		StackEvents: []*sdkcloudformation.StackEvent{
+			{
+				EventId:           aws.String("1"),
+				LogicalResourceId: aws.String("PublicLoadBalancer"),
+				ResourceType:      aws.String("AWS::ElasticLoadBalancingV2::LoadBalancer"),
+				ResourceStatus:    aws.String("CREATE_COMPLETE"),
+				Timestamp:         aws.Time(deploymentTime),
+			},
+			{
+				EventId:           aws.String("2"),
+				LogicalResourceId: aws.String("my-app-my-env"),
+				ResourceType:      aws.String("AWS::CloudFormation::Stack"),
+				ResourceStatus:    aws.String("CREATE_COMPLETE"),
+				Timestamp:         aws.Time(deploymentTime),
+			},
+		},
+	}, nil).AnyTimes()
+
+	mockCFN.EXPECT().Describe(svcStackName).Return(&cloudformation.StackDescription{
+		StackStatus: aws.String("CREATE_COMPLETE"),
+	}, nil)
+	client := CloudFormation{cfnClient: mockCFN}
+	buf := new(strings.Builder)
+
+	// WHEN
+	err := when(mockFileWriter{Writer: buf}, client)
+
+	// THEN
+	require.NoError(t, err)
+	require.Contains(t, buf.String(), "Updating environment", "env stack description is rendered")
+	require.Contains(t, buf.String(), "Updating ALB", "resource in the env stack should be rendered")
+}
+
 func testDeployWorkload_RenderNewlyCreatedStackWithAddons(t *testing.T, stackName string, when func(w progress.FileWriter, cf CloudFormation) error) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()

internal/pkg/deploy/cloudformation/workload_test.go

@@ -54,7 +54,7 @@ func (m *mockStackConfig) Tags() []*sdkcloudformation.Tag {
 
 func TestCloudFormation_DeployService(t *testing.T) {
 	serviceConfig := &mockStackConfig{
-		name:     "hello",
+		name:     "myapp-myenv-mysvc",
 		template: "template",
 		parameters: map[string]string{
 			"port": "80",
@@ -83,13 +83,16 @@ func TestCloudFormation_DeployService(t *testing.T) {
 		testDeployWorkload_StackStreamerFailureShouldCancelRenderer(t, when)
 	})
 	t.Run("returns an error if stack creation fails", func(t *testing.T) {
-		testDeployWorkload_StreamUntilStackCreationFails(t, "hello", when)
+		testDeployWorkload_StreamUntilStackCreationFails(t, "myapp-myenv-mysvc", when)
+	})
+	t.Run("renders a stack with an EnvController", func(t *testing.T) {
+		testDeployWorkload_WithEnvControllerRenderer(t, "myapp-myenv-mysvc", when)
 	})
 	t.Run("renders a stack with an ECS service", func(t *testing.T) {
-		testDeployWorkload_RenderNewlyCreatedStackWithECSService(t, "hello", when)
+		testDeployWorkload_RenderNewlyCreatedStackWithECSService(t, "myapp-myenv-mysvc", when)
 	})
 	t.Run("renders a stack with addons template if stack creation is successful", func(t *testing.T) {
-		testDeployWorkload_RenderNewlyCreatedStackWithAddons(t, "hello", when)
+		testDeployWorkload_RenderNewlyCreatedStackWithAddons(t, "myapp-myenv-mysvc", when)
 	})
 }
 

internal/pkg/stream/cloudformation.go

@@ -5,6 +5,7 @@ package stream
 
 import (
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -37,6 +38,7 @@ type StackStreamer struct {
 	done          chan struct{}
 	pastEventIDs  map[string]bool
 	eventsToFlush []StackEvent
+	mu            sync.Mutex
 }
 
 // NewStackStreamer creates a StackStreamer from a cloudformation client, stack name, and the change set creation timestamp.
@@ -52,6 +54,8 @@ func NewStackStreamer(cfn StackEventsDescriber, stackName string, csCreationTime
 
 // Subscribe returns a read-only channel that will receive stack events from the StackStreamer.
 func (s *StackStreamer) Subscribe() <-chan StackEvent {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	c := make(chan StackEvent)
 	s.subscribers = append(s.subscribers, c)
 	return c
@@ -116,8 +120,15 @@ func (s *StackStreamer) Fetch() (next time.Time, err error) {
 
 // Notify flushes all new events to the streamer's subscribers.
 func (s *StackStreamer) Notify() {
+	// Copy current list of subscribers over, so that we can we add more subscribers while
+	// notifying previous subscribers of older events.
+	s.mu.Lock()
+	var subs []chan StackEvent
+	subs = append(subs, s.subscribers...)
+	s.mu.Unlock()
+
 	for _, event := range s.eventsToFlush {
-		for _, sub := range s.subscribers {
+		for _, sub := range subs {
 			sub <- event
 		}
 	}
@@ -126,6 +137,9 @@ func (s *StackStreamer) Notify() {
 
 // Close closes all subscribed channels notifying them that no more events will be sent.
 func (s *StackStreamer) Close() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
 	for _, sub := range s.subscribers {
 		close(sub)
 	}