[ECS] Fix remove all previous active tasksets on QuickSync (#4600)

* Remove previous ACTIVE tasksets if present on quicksync

Signed-off-by: khanhtc1202 <khanhtc1202@gmail.com>

* Remove GetPrimaryTaskSet interface

Signed-off-by: khanhtc1202 <khanhtc1202@gmail.com>

---------

Signed-off-by: khanhtc1202 <khanhtc1202@gmail.com>

pkg/app/piped/executor/ecs/ecs.go

@@ -26,7 +26,6 @@ import (
 
 	"github.com/pipe-cd/pipecd/pkg/app/piped/deploysource"
 	"github.com/pipe-cd/pipecd/pkg/app/piped/executor"
-	"github.com/pipe-cd/pipecd/pkg/app/piped/platformprovider"
 	provider "github.com/pipe-cd/pipecd/pkg/app/piped/platformprovider/ecs"
 	"github.com/pipe-cd/pipecd/pkg/config"
 	"github.com/pipe-cd/pipecd/pkg/model"
@@ -214,10 +213,9 @@ func runStandaloneTask(
 }
 
 func createPrimaryTaskSet(ctx context.Context, client provider.Client, service types.Service, taskDef types.TaskDefinition, targetGroup *types.LoadBalancer) error {
-	// Get current PRIMARY task set.
-	prevPrimaryTaskSet, err := client.GetPrimaryTaskSet(ctx, service)
-	// Ignore error in case it's not found error, the prevPrimaryTaskSet doesn't exist for newly created Service.
-	if err != nil && !errors.Is(err, platformprovider.ErrNotFound) {
+	// Get current PRIMARY/ACTIVE task sets.
+	prevTaskSets, err := client.GetServiceTaskSets(ctx, service)
+	if err != nil {
 		return err
 	}
 
@@ -234,9 +232,9 @@ func createPrimaryTaskSet(ctx context.Context, client provider.Client, service t
 		return err
 	}
 
-	// Remove old taskSet if existed.
-	if prevPrimaryTaskSet != nil {
-		if err = client.DeleteTaskSet(ctx, *prevPrimaryTaskSet); err != nil {
+	// Remove old taskSets if existed.
+	for _, prevTaskSet := range prevTaskSets {
+		if err = client.DeleteTaskSet(ctx, *prevTaskSet); err != nil {
 			return err
 		}
 	}

pkg/app/piped/platformprovider/ecs/client.go

@@ -251,29 +251,6 @@ func (c *client) CreateTaskSet(ctx context.Context, service types.Service, taskD
 	return output.TaskSet, nil
 }
 
-func (c *client) GetPrimaryTaskSet(ctx context.Context, service types.Service) (*types.TaskSet, error) {
-	input := &ecs.DescribeServicesInput{
-		Cluster: service.ClusterArn,
-		Services: []string{
-			*service.ServiceArn,
-		},
-	}
-	output, err := c.ecsClient.DescribeServices(ctx, input)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get primary task set of service %s: %w", *service.ServiceName, err)
-	}
-	if len(output.Services) == 0 {
-		return nil, fmt.Errorf("failed to get primary task set of service %s: services empty", *service.ServiceName)
-	}
-	taskSets := output.Services[0].TaskSets
-	for _, taskSet := range taskSets {
-		if aws.ToString(taskSet.Status) == "PRIMARY" {
-			return &taskSet, nil
-		}
-	}
-	return nil, platformprovider.ErrNotFound
-}
-
 func (c *client) GetServiceTaskSets(ctx context.Context, service types.Service) ([]*types.TaskSet, error) {
 	input := &ecs.DescribeServicesInput{
 		Cluster: service.ClusterArn,
@@ -294,6 +271,9 @@ func (c *client) GetServiceTaskSets(ctx context.Context, service types.Service)
 		if aws.ToString(svc.TaskSets[i].Status) == "DRAINING" {
 			continue
 		}
+		if !IsPipeCDManagedTaskSet(&svc.TaskSets[i]) {
+			continue
+		}
 		taskSets = append(taskSets, &svc.TaskSets[i])
 	}
 	return taskSets, nil

pkg/app/piped/platformprovider/ecs/ecs.go

@@ -48,7 +48,6 @@ type ECS interface {
 	WaitServiceStable(ctx context.Context, service types.Service) error
 	RegisterTaskDefinition(ctx context.Context, taskDefinition types.TaskDefinition) (*types.TaskDefinition, error)
 	RunTask(ctx context.Context, taskDefinition types.TaskDefinition, clusterArn string, launchType string, awsVpcConfiguration *config.ECSVpcConfiguration, tags []types.Tag) error
-	GetPrimaryTaskSet(ctx context.Context, service types.Service) (*types.TaskSet, error)
 	GetServiceTaskSets(ctx context.Context, service types.Service) ([]*types.TaskSet, error)
 	CreateTaskSet(ctx context.Context, service types.Service, taskDefinition types.TaskDefinition, targetGroup *types.LoadBalancer, scale int) (*types.TaskSet, error)
 	DeleteTaskSet(ctx context.Context, taskSet types.TaskSet) error

pkg/app/piped/platformprovider/ecs/task_set.go

@@ -0,0 +1,26 @@
+// Copyright 2023 The PipeCD Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ecs
+
+import "github.com/aws/aws-sdk-go-v2/service/ecs/types"
+
+func IsPipeCDManagedTaskSet(ts *types.TaskSet) bool {
+	for _, tag := range ts.Tags {
+		if *tag.Key == LabelManagedBy && *tag.Value == ManagedByPiped {
+			return true
+		}
+	}
+	return false
+}

pkg/app/piped/platformprovider/ecs/task_set_test.go

@@ -0,0 +1,61 @@
+// Copyright 2023 The PipeCD Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ecs
+
+import (
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/ecs/types"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsPipeCDManagedTaskSet(t *testing.T) {
+	t.Parallel()
+
+	testcases := []struct {
+		name     string
+		ts       *types.TaskSet
+		expected bool
+	}{
+		{
+			name: "managed by piped",
+			ts: &types.TaskSet{Tags: []types.Tag{
+				{Key: aws.String(LabelManagedBy), Value: aws.String(ManagedByPiped)},
+			}},
+			expected: true,
+		},
+		{
+			name:     "nil tags",
+			ts:       &types.TaskSet{},
+			expected: false,
+		},
+		{
+			name: "not managed by piped",
+			ts: &types.TaskSet{Tags: []types.Tag{
+				{Key: aws.String(LabelManagedBy), Value: aws.String("other")},
+				{Key: aws.String("hoge"), Value: aws.String("fuga")},
+			}},
+			expected: false,
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := IsPipeCDManagedTaskSet(tc.ts)
+			assert.Equal(t, tc.expected, got)
+		})
+	}
+}