Add exponential backoff for deregistering task definitions (#4)

* add debug flag, set default parallel=2

* turn on verbose flag when debug flag is on

* implement exponential backoff across goroutines for task definition deregistration

* remove commented code

* fix typo

* break out worker and dispatcher into own functions; replace os.Exit(1) with return

* upload assets as zip files instead of raw binaries

.github/workflows/release.yml

@@ -24,9 +24,10 @@ jobs:
 
       - name: Build artifacts
         run: |
-          CGO_ENABLED=0 GOOS=linux   GOARCH=amd64 go build -ldflags "-s -X main.Version=TEST" -a -o build/Linux/go-ecs-cleaner .
-          CGO_ENABLED=0 GOOS=darwin  GOARCH=amd64 go build -ldflags "-s -X main.Version=TEST" -a -o build/macOS/go-ecs-cleaner .
-          CGO_ENABLED=0 GOOS=windows GOARCH=amd64 go build -ldflags "-s -X main.Version=TEST" -a -o build/Windows/go-ecs-cleaner.exe .
+          CGO_ENABLED=0 GOOS=linux   GOARCH=amd64 go build -ldflags "-s -X main.Version=${{ github.ref }}" -a -o build/Linux/go-ecs-cleaner .
+          CGO_ENABLED=0 GOOS=darwin  GOARCH=amd64 go build -ldflags "-s -X main.Version=${{ github.ref }}" -a -o build/macOS/go-ecs-cleaner .
+          CGO_ENABLED=0 GOOS=windows GOARCH=amd64 go build -ldflags "-s -X main.Version=${{ github.ref }}" -a -o build/Windows/go-ecs-cleaner.exe .
+          for os in Linux macOS Windows ; do cd build && zip -r ${os}.zip ${os} && cd .. ; done
 
       - name: Create release
         id: create_release
@@ -39,8 +40,8 @@ jobs:
           draft: false
           prerelease: false
 
-      - name: Upload Linux binary
-        id: upload-linux-binary
+      - name: Upload Linux assets
+        id: upload-linux-assets
         uses: actions/[email protected]
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
@@ -49,28 +50,28 @@ jobs:
             # This pulls from the `Create release` step above, referencing its ID to get its outputs object,
             # which includes an `upload_url`. See this blog post for more info:
             # https://jasonet.co/posts/new-features-of-github-actions/#passing-data-to-future-steps 
-          asset_path: ./build/Linux/go-ecs-cleaner
-          asset_name: go-ecs-cleaner-linux
+          asset_path: ./build/Linux.zip
+          asset_name: go-ecs-cleaner-linux.zip
           asset_content_type: application/zip
 
-      - name: Upload macOS binary
-        id: upload-macos-binary
+      - name: Upload macOS assets
+        id: upload-macos-assets
         uses: actions/[email protected]
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         with:
           upload_url: ${{ steps.create_release.outputs.upload_url }} 
-          asset_path: ./build/macOS/go-ecs-cleaner
-          asset_name: go-ecs-cleaner-darwin
+          asset_path: ./build/macOS.zip
+          asset_name: go-ecs-cleaner-macos.zip
           asset_content_type: application/zip
 
-      - name: Upload Windows binary
-        id: upload-windows-binary
+      - name: Upload Windows assets
+        id: upload-windows-assets
         uses: actions/[email protected]
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         with:
           upload_url: ${{ steps.create_release.outputs.upload_url }} 
-          asset_path: ./build/Windows/go-ecs-cleaner.exe
-          asset_name: go-ecs-cleaner-windows.exe
+          asset_path: ./build/Windows.zip
+          asset_name: go-ecs-cleaner-windows.zip
           asset_content_type: application/zip

cmd/ecs_task.go

@@ -1,6 +1,9 @@
 package cmd
 
 import (
+	"fmt"
+	"os"
+
 	"github.com/quintilesims/go-ecs-cleaner/ecstask"
 	"github.com/spf13/cobra"
 )
@@ -13,9 +16,19 @@ var ecsTaskCmd = &cobra.Command{
 BEFORE RUNNING: Make sure that you've properly configured your environment with
 the AWS CLI for the AWS account you want to clean up.`,
 	Run: func(cmd *cobra.Command, args []string) {
+		if parallelFlag < 1 {
+			fmt.Println("minimum parallel is 1")
+			os.Exit(1)
+		}
+
+		if debugFlag {
+			verboseFlag = true
+		}
+
 		flags := map[string]interface{}{
 			"apply":    applyFlag,
 			"cutoff":   cutoffFlag,
+			"debug":    debugFlag,
 			"parallel": parallelFlag,
 			"region":   regionFlag,
 			"verbose":  verboseFlag,
@@ -27,14 +40,16 @@ the AWS CLI for the AWS account you want to clean up.`,
 
 var applyFlag bool
 var cutoffFlag int
+var debugFlag bool
 var parallelFlag int
 var regionFlag string
 var verboseFlag bool
 
 func init() {
 	ecsTaskCmd.Flags().BoolVarP(&applyFlag, "apply", "a", false, "actually perform task definition deregistration")
 	ecsTaskCmd.Flags().IntVarP(&cutoffFlag, "cutoff", "c", 5, "how many most-recent task definitions to keep around")
-	ecsTaskCmd.Flags().IntVarP(&parallelFlag, "parallel", "p", 10, "how many concurrent deregistration requests to make")
+	ecsTaskCmd.Flags().BoolVarP(&debugFlag, "debug", "d", false, "enable for all the output")
+	ecsTaskCmd.Flags().IntVarP(&parallelFlag, "parallel", "p", 2, "how many concurrent deregistration requests to make")
 	ecsTaskCmd.Flags().StringVarP(&regionFlag, "region", "r", "us-west-2", "the AWS region in which to operate")
 	ecsTaskCmd.Flags().BoolVarP(&verboseFlag, "verbose", "v", false, "enable for chattier output")
 	rootCmd.AddCommand(ecsTaskCmd)

ecstask/ecs_task.go

@@ -5,10 +5,14 @@ import (
 	"regexp"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/golang-collections/collections/stack"
+	"github.com/jpillora/backoff"
 	"github.com/spf13/cobra"
 )
 
@@ -269,9 +273,9 @@ func Run(cmd *cobra.Command, args []string, flags map[string]interface{}) {
 		fmt.Println("\n`--apply` flag present")
 		fmt.Printf("deregistering %d task definitions...\n", len(allTaskDefinitionArns))
 
-		deregisterTaskDefinitions(svc, allTaskDefinitionArns, flags["parallel"].(int))
+		deregisterTaskDefinitions(svc, allTaskDefinitionArns, flags["parallel"].(int), flags["debug"].(bool))
 
-		fmt.Println("finished")
+		fmt.Println("\nfinished")
 	} else {
 		fmt.Println("\nthis is a dry run")
 		fmt.Println("use the `--apply` flag to deregister these task definitions")
@@ -397,44 +401,121 @@ func removeAFromB(a, b []string) []string {
 	return diff
 }
 
-func deregisterTaskDefinitions(svc *ecs.ECS, taskDefinitionArns []string, parallel int) {
-	arnsChan := make(chan string, len(taskDefinitionArns))
+// Job carries information through a Job channel.
+type Job struct {
+	Arn string
+}
 
-	deregisterTaskDefinition := func(arn string) {
-		_, err := svc.DeregisterTaskDefinition(&ecs.DeregisterTaskDefinitionInput{
-			TaskDefinition: aws.String(arn),
-		})
-		if err != nil {
-			fmt.Println("Error deregistering task definition:", err)
-		}
+// Result carries information through a Result channel.
+type Result struct {
+	Arn string
+	Err error
+}
+
+func deregisterTaskDefinitions(svc *ecs.ECS, taskDefinitionArns []string, parallel int, debug bool) {
+	jobsChan := make(chan Job, parallel)
+	resultsChan := make(chan Result, parallel)
+
+	var wg sync.WaitGroup
+	for i := 0; i < parallel; i++ {
+		wg.Add(1)
+		go worker(svc, &wg, jobsChan, resultsChan)
 	}
 
-	worker := func(wg *sync.WaitGroup) {
-		for arn := range arnsChan {
-			deregisterTaskDefinition(arn)
+	wg.Add(1)
+	go dispatcher(&wg, taskDefinitionArns, parallel, jobsChan, resultsChan, debug)
+
+	wg.Wait()
+}
+
+func worker(svc *ecs.ECS, wg *sync.WaitGroup, jobsChan <-chan Job, resultsChan chan<- Result) {
+	for job := range jobsChan {
+		input := &ecs.DeregisterTaskDefinitionInput{
+			TaskDefinition: aws.String(job.Arn),
 		}
 
-		wg.Done()
+		_, err := svc.DeregisterTaskDefinition(input)
+
+		result := Result{Arn: job.Arn, Err: err}
+		resultsChan <- result
+	}
+
+	wg.Done()
+}
+
+func dispatcher(wg *sync.WaitGroup, arns []string, parallel int, jobsChan chan Job, resultsChan chan Result, debug bool) {
+	jobs := stack.New()
+	for _, arn := range arns {
+		jobs.Push(Job{arn})
+	}
+
+	var completedJobs int
+
+	preload := 1
+	if parallel > 1 {
+		preload = parallel - 1
 	}
 
-	createWorkerPool := func(numWorkers int) {
-		var wg sync.WaitGroup
-		for i := 0; i < numWorkers; i++ {
-			wg.Add(1)
-			go worker(&wg)
+	for i := 0; i < preload; i++ {
+		jobsChan <- jobs.Pop().(Job)
+	}
+
+	b := &backoff.Backoff{
+		Min:    100 * time.Millisecond,
+		Max:    2 * time.Minute,
+		Jitter: true,
+	}
+
+	for result := range resultsChan {
+		if result.Err != nil {
+			if !isThrottlingError(result.Err) {
+				fmt.Println("\nError deregistering task definition,", result.Err)
+				close(jobsChan)
+				close(resultsChan)
+				wg.Done()
+				return
+			}
+
+			t := b.Duration()
+			if debug {
+				fmt.Printf("\nbackoff triggered for %s,", result.Arn)
+				fmt.Printf("\nwaiting for %v\n", t)
+			}
+
+			time.Sleep(t)
+			jobs.Push(Job{Arn: result.Arn})
+		} else {
+			b.Reset()
+			completedJobs++
+			fmt.Printf("\r%d deregistered task definitions", completedJobs)
 		}
 
-		wg.Wait()
+		if jobs.Len() > 0 {
+			jobsChan <- jobs.Pop().(Job)
+		}
+
+		if completedJobs == len(arns) {
+			close(jobsChan)
+			close(resultsChan)
+		}
 	}
 
-	allocate := func(arns []string) {
-		for _, arn := range arns {
-			arnsChan <- arn
+	wg.Done()
+}
+
+func isThrottlingError(err error) bool {
+	if awsErr, ok := err.(awserr.Error); ok {
+		code := awsErr.Code()
+
+		if code == "Throttling" || code == "ThrottlingException" {
+			return true
 		}
 
-		close(arnsChan)
+		message := strings.ToLower(awsErr.Message())
+		if code == "ClientException" && strings.Contains(message, "too many concurrent attempts") {
+			return true
+		}
 	}
 
-	go allocate(taskDefinitionArns)
-	createWorkerPool(parallel)
+	return false
 }

go.mod

@@ -4,6 +4,8 @@ go 1.13
 
 require (
 	github.com/aws/aws-sdk-go v1.25.34
+	github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3
+	github.com/jpillora/backoff v1.0.0
 	github.com/spf13/cobra v0.0.5
 	golang.org/x/net v0.0.0-20191112182307-2180aed22343 // indirect
 )

go.sum

@@ -10,12 +10,16 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3 h1:zN2lZNZRflqFyxVaTIU61KNKQ9C0055u9CAfpmqUvo4=
+github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3/go.mod h1:nPpo7qLxd6XL3hWJG/O60sR8ZKfMCiIoNap5GvD12KU=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af h1:pmfjZENx5imkbgOkpRUYLnmbU7UEFbjtDA2hxJ1ichM=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
+github.com/jpillora/backoff v1.0.0 h1:uvFg412JmmHBHw7iwprIxkPMI+sGQ4kzOWsMeHnm2EA=
+github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
 github.com/magiconair/properties v1.8.0 h1:LLgXmsheXeRoUOBOjtwPQCWIYqM/LU1ayDtDePerRcY=
 github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
 github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=