feat(object): parallelize versions destruction (#1711)

Author: quantumsheep

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/hashicorp/aws-sdk-go-base v1.1.0
 	github.com/hashicorp/awspolicyequivalence v1.6.0
 	github.com/hashicorp/go-cty v1.4.1-0.20200414143053-d3edf31b6320
+	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/go-retryablehttp v0.7.1
 	github.com/hashicorp/terraform-plugin-log v0.7.0
 	github.com/hashicorp/terraform-plugin-sdk/v2 v2.24.1
@@ -33,7 +34,6 @@ require (
 	github.com/hashicorp/go-checkpoint v0.5.0 // indirect
 	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
 	github.com/hashicorp/go-hclog v1.2.1 // indirect
-	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/go-plugin v1.4.6 // indirect
 	github.com/hashicorp/go-uuid v1.0.3 // indirect
 	github.com/hashicorp/go-version v1.6.0 // indirect

internal/workerpool.go

@@ -0,0 +1,88 @@
+package internal
+
+import (
+	"sync"
+)
+
+type WorkerPoolTask func() error
+
+type WorkerPool struct {
+	tasksToDispatch   chan WorkerPoolTask
+	tasksToRun        chan WorkerPoolTask
+	errors            []error
+	errorsMutex       sync.Mutex
+	tasksWaitingGroup sync.WaitGroup
+}
+
+func NewWorkerPool(size int) *WorkerPool {
+	p := &WorkerPool{
+		tasksToDispatch: make(chan WorkerPoolTask),
+		tasksToRun:      make(chan WorkerPoolTask, size),
+	}
+
+	for i := 0; i < size; i++ {
+		go p.worker()
+	}
+
+	go p.dispatcher()
+
+	return p
+}
+
+func (p *WorkerPool) dispatcher() {
+	var pendingTasks []WorkerPoolTask
+
+	for {
+		if len(pendingTasks) > 0 {
+			select {
+			case p.tasksToRun <- pendingTasks[0]:
+				pendingTasks = pendingTasks[1:]
+			default:
+			}
+		}
+
+		select {
+		case task, ok := <-p.tasksToDispatch:
+			if !ok {
+				if len(pendingTasks) > 0 {
+					continue
+				}
+
+				close(p.tasksToRun)
+				return
+			}
+
+			select {
+			case p.tasksToRun <- task:
+			default:
+				pendingTasks = append(pendingTasks, task)
+			}
+		default:
+		}
+	}
+}
+
+func (p *WorkerPool) worker() {
+	for task := range p.tasksToRun {
+		err := task()
+		if err != nil {
+			p.errorsMutex.Lock()
+			p.errors = append(p.errors, err)
+			p.errorsMutex.Unlock()
+		}
+
+		p.tasksWaitingGroup.Done()
+	}
+}
+
+func (p *WorkerPool) AddTask(task WorkerPoolTask) {
+	p.tasksWaitingGroup.Add(1)
+	p.tasksToDispatch <- task
+}
+
+func (p *WorkerPool) CloseAndWait() []error {
+	close(p.tasksToDispatch)
+	p.tasksWaitingGroup.Wait()
+
+	return p.errors
+}

internal/workerpool_test.go

@@ -0,0 +1,92 @@
+package internal
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestWorkerPoolSimple(t *testing.T) {
+	pool := NewWorkerPool(2)
+
+	pool.AddTask(func() error {
+		return nil
+	})
+
+	pool.AddTask(func() error {
+		return errors.New("error")
+	})
+
+	pool.AddTask(func() error {
+		return nil
+	})
+
+	errs := pool.CloseAndWait()
+
+	assert.Equal(t, 1, len(errs))
+	assert.Equal(t, "error", errs[0].Error())
+}
+
+func TestWorkerPoolWaitTime(t *testing.T) {
+	pool := NewWorkerPool(2)
+
+	pool.AddTask(func() error {
+		time.Sleep(50 * time.Millisecond)
+		return nil
+	})
+
+	pool.AddTask(func() error {
+		time.Sleep(50 * time.Millisecond)
+		return errors.New("error")
+	})
+
+	pool.AddTask(func() error {
+		time.Sleep(50 * time.Millisecond)
+		return nil
+	})
+
+	errs := pool.CloseAndWait()
+
+	assert.Equal(t, 1, len(errs))
+	assert.Equal(t, "error", errs[0].Error())
+}
+
+func TestWorkerPoolWaitTimeMultiple(t *testing.T) {
+	pool := NewWorkerPool(5)
+	iterations := 20
+
+	for i := 0; i < iterations; i++ {
+		copyOfI := i
+
+		pool.AddTask(func() error {
+			time.Sleep(100 * time.Millisecond)
+
+			if copyOfI%2 == 0 {
+				return fmt.Errorf("error %d", copyOfI)
+			}
+
+			return nil
+		})
+	}
+
+	errs := pool.CloseAndWait()
+
+	assert.Equal(t, iterations/2, len(errs))
+
+	for i := 0; i < iterations; i++ {
+		if i%2 == 0 {
+			found := false
+			for _, err := range errs {
+				if err.Error() == fmt.Sprintf("error %d", i) {
+					found = true
+					break
+				}
+			}
+
+			assert.True(t, found)
+		}
+	}
+}

scaleway/helpers_object.go

@@ -8,6 +8,7 @@ import (
 	"hash/crc32"
 	"net/http"
 	"os"
+	"runtime"
 	"strings"
 	"time"
 
@@ -18,15 +19,19 @@ import (
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/hashicorp/aws-sdk-go-base/tfawserr"
 	awspolicy "github.com/hashicorp/awspolicyequivalence"
+	"github.com/hashicorp/go-multierror"
 	"github.com/hashicorp/terraform-plugin-log/tflog"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 	"github.com/scaleway/scaleway-sdk-go/scw"
+	"github.com/scaleway/terraform-provider-scaleway/v2/internal"
 )
 
 const (
 	defaultObjectBucketTimeout = 10 * time.Minute
 	retryOnAWSAPI              = 2 * time.Minute
+
+	maxObjectVersionDeletionWorkers = 8
 )
 
 func newS3Client(httpClient *http.Client, region, accessKey, secretKey string) (*s3.S3, error) {
@@ -305,58 +310,94 @@ func removeS3ObjectVersionLegalHold(conn *s3.S3, bucketName string, objectVersio
 }
 
 func deleteS3ObjectVersions(ctx context.Context, conn *s3.S3, bucketName string, force bool) error {
-	var err error
+	var globalErr error
 	listInput := &s3.ListObjectVersionsInput{
 		Bucket: scw.StringPtr(bucketName),
 	}
+
+	deletionWorkers := runtime.NumCPU()
+	if deletionWorkers > maxObjectVersionDeletionWorkers {
+		deletionWorkers = maxObjectVersionDeletionWorkers
+	}
+
 	listErr := conn.ListObjectVersionsPagesWithContext(ctx, listInput, func(page *s3.ListObjectVersionsOutput, lastPage bool) bool {
+		pool := internal.NewWorkerPool(deletionWorkers)
+
 		for _, objectVersion := range page.Versions {
-			objectKey := aws.StringValue(objectVersion.Key)
-			objectVersionID := aws.StringValue(objectVersion.VersionId)
-			err = deleteS3ObjectVersion(conn, bucketName, objectKey, objectVersionID, force)
-
-			if isS3Err(err, ErrCodeAccessDenied, "") && force {
-				legalHoldRemoved, errLegal := removeS3ObjectVersionLegalHold(conn, bucketName, objectVersion)
-				if errLegal != nil {
-					err = fmt.Errorf("failed to remove legal hold: %s", errLegal)
-					return false
+			objectVersion := objectVersion
+
+			pool.AddTask(func() error {
+				objectKey := aws.StringValue(objectVersion.Key)
+				objectVersionID := aws.StringValue(objectVersion.VersionId)
+				err := deleteS3ObjectVersion(conn, bucketName, objectKey, objectVersionID, force)
+
+				if isS3Err(err, ErrCodeAccessDenied, "") && force {
+					legalHoldRemoved, errLegal := removeS3ObjectVersionLegalHold(conn, bucketName, objectVersion)
+					if errLegal != nil {
+						return fmt.Errorf("failed to remove legal hold: %s", errLegal)
+					}
+
+					if legalHoldRemoved {
+						err = deleteS3ObjectVersion(conn, bucketName, objectKey, objectVersionID, force)
+					}
 				}
-				if legalHoldRemoved {
-					err = deleteS3ObjectVersion(conn, bucketName, objectKey, objectVersionID, force)
+
+				if err != nil {
+					return fmt.Errorf("failed to delete S3 object: %s", err)
 				}
-			}
-			if err != nil {
-				err = fmt.Errorf("failed to delete S3 object: %s", err)
-				return false
-			}
+
+				return nil
+			})
 		}
+
+		errors := pool.CloseAndWait()
+		if len(errors) > 0 {
+			globalErr = multierror.Append(nil, errors...)
+			return false
+		}
+
 		return true
 	})
 	if listErr != nil {
-		return fmt.Errorf("error listing S3 objects: %s", err)
+		return fmt.Errorf("error listing S3 objects: %s", globalErr)
 	}
-	if err != nil {
-		return err
+	if globalErr != nil {
+		return globalErr
 	}
+
 	listErr = conn.ListObjectVersionsPagesWithContext(ctx, listInput, func(page *s3.ListObjectVersionsOutput, lastPage bool) bool {
+		pool := internal.NewWorkerPool(deletionWorkers)
+
 		for _, deleteMarkerEntry := range page.DeleteMarkers {
-			deleteMarkerKey := aws.StringValue(deleteMarkerEntry.Key)
-			deleteMarkerVersionsID := aws.StringValue(deleteMarkerEntry.VersionId)
-			err = deleteS3ObjectVersion(conn, bucketName, deleteMarkerKey, deleteMarkerVersionsID, force)
+			deleteMarkerEntry := deleteMarkerEntry
+
+			pool.AddTask(func() error {
+				deleteMarkerKey := aws.StringValue(deleteMarkerEntry.Key)
+				deleteMarkerVersionsID := aws.StringValue(deleteMarkerEntry.VersionId)
+				err := deleteS3ObjectVersion(conn, bucketName, deleteMarkerKey, deleteMarkerVersionsID, force)
+				if err != nil {
+					return fmt.Errorf("failed to delete S3 object delete marker: %s", err)
+				}
 
-			if err != nil {
-				err = fmt.Errorf("failed to delete S3 object delete marker: %s", err)
-				return false
-			}
+				return nil
+			})
+		}
+
+		errors := pool.CloseAndWait()
+		if len(errors) > 0 {
+			globalErr = multierror.Append(nil, errors...)
+			return false
 		}
+
 		return true
 	})
 	if listErr != nil {
-		return fmt.Errorf("error listing S3 objects for delete markers: %s", err)
+		return fmt.Errorf("error listing S3 objects for delete markers: %s", globalErr)
 	}
-	if err != nil {
-		return err
+	if globalErr != nil {
+		return globalErr
 	}
+
 	return nil
 }