tests/provider: Add concurrency to sweep (DB instance)

aws/provider_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/iam"
 	"github.com/aws/aws-sdk-go/service/organizations"
 	"github.com/hashicorp/aws-sdk-go-base/tfawserr"
+	multierror "github.com/hashicorp/go-multierror"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
@@ -1005,49 +1006,41 @@ func testAccAwsRegionProviderFunc(region string, providers *[]*schema.Provider)
 	}
 }
 
-func testAccCheckResourceDisappears(provider *schema.Provider, resource *schema.Resource, resourceName string) resource.TestCheckFunc {
-	return func(s *terraform.State) error {
-		resourceState, ok := s.RootModule().Resources[resourceName]
+func testAccDeleteResource(resource *schema.Resource, d *schema.ResourceData, meta interface{}) error {
+	if resource.DeleteContext != nil || resource.DeleteWithoutTimeout != nil {
+		var diags diag.Diagnostics
 
-		if !ok {
-			return fmt.Errorf("resource not found: %s", resourceName)
+		if resource.DeleteContext != nil {
+			diags = resource.DeleteContext(context.Background(), d, meta)
+		} else {
+			diags = resource.DeleteWithoutTimeout(context.Background(), d, meta)
 		}
 
-		if resourceState.Primary.ID == "" {
-			return fmt.Errorf("resource ID missing: %s", resourceName)
+		for i := range diags {
+			if diags[i].Severity == diag.Error {
+				return fmt.Errorf("error deleting resource: %s", diags[i].Summary)
+			}
 		}
 
-		if resource.DeleteContext != nil || resource.DeleteWithoutTimeout != nil {
-			var diags diag.Diagnostics
+		return nil
+	}
 
-			if resource.DeleteContext != nil {
-				diags = resource.DeleteContext(context.Background(), resource.Data(resourceState.Primary), provider.Meta())
-			} else {
-				diags = resource.DeleteWithoutTimeout(context.Background(), resource.Data(resourceState.Primary), provider.Meta())
-			}
+	return resource.Delete(d, meta)
+}
 
-			for i := range diags {
-				if diags[i].Severity == diag.Error {
-					return fmt.Errorf("error deleting resource: %s", diags[i].Summary)
-				}
-			}
+func testAccCheckResourceDisappears(provider *schema.Provider, resource *schema.Resource, resourceName string) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		resourceState, ok := s.RootModule().Resources[resourceName]
 
-			return nil
+		if !ok {
+			return fmt.Errorf("resource not found: %s", resourceName)
 		}
 
-		if resource.DeleteWithoutTimeout != nil {
-			diags := resource.DeleteWithoutTimeout(context.Background(), resource.Data(resourceState.Primary), provider.Meta())
-
-			for i := range diags {
-				if diags[i].Severity == diag.Error {
-					return fmt.Errorf("error deleting resource: %s", diags[i].Summary)
-				}
-			}
-
-			return nil
+		if resourceState.Primary.ID == "" {
+			return fmt.Errorf("resource ID missing: %s", resourceName)
 		}
 
-		return resource.Delete(resource.Data(resourceState.Primary), provider.Meta())
+		return testAccDeleteResource(resource, resource.Data(resourceState.Primary), provider.Meta())
 	}
 }
 
@@ -1190,6 +1183,57 @@ func testAccPreCheckSkipError(err error) bool {
 	return false
 }
 
+type testSweepResource struct {
+	d        *schema.ResourceData
+	meta     interface{}
+	resource *schema.Resource
+}
+
+func NewTestSweepResource(resource *schema.Resource, d *schema.ResourceData, meta interface{}) *testSweepResource {
+	return &testSweepResource{
+		d:        d,
+		meta:     meta,
+		resource: resource,
+	}
+}
+
+func testSweepResourceOrchestrator(sweepResources []*testSweepResource) error {
+	var wg sync.WaitGroup
+	wgDone := make(chan bool)
+	errChan := make(chan error, len(sweepResources))
+
+	for _, sweepResource := range sweepResources {
+		wg.Add(1)
+
+		go func(sweepResource *testSweepResource) {
+			defer wg.Done()
+
+			err := testAccDeleteResource(sweepResource.resource, sweepResource.d, sweepResource.meta)
+			if err != nil {
+				errChan <- err
+			}
+		}(sweepResource)
+	}
+
+	go func() {
+		wg.Wait()
+		close(wgDone)
+		close(errChan)
+	}()
+
+	// Keep in mind: everything above (except the declarations) is executing in separate goroutines than the one that hits this line.
+	// The errChan range waits to exit for errChan to close (which, in turn, won't happen until the wait group finishes).
+	var errors *multierror.Error
+	var mutex = &sync.Mutex{}
+	for err := range errChan {
+		mutex.Lock()
+		errors = multierror.Append(errors, err)
+		mutex.Unlock()
+	}
+
+	return errors.ErrorOrNil()
+}
+
 // Check sweeper API call error for reasons to skip sweeping
 // These include missing API endpoints and unsupported API calls
 func testSweepSkipSweepError(err error) bool {

aws/resource_aws_db_instance_test.go

@@ -5,7 +5,6 @@ import (
 	"log"
 	"regexp"
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/rds"
@@ -28,27 +27,17 @@ func testSweepDbInstances(region string) error {
 	if err != nil {
 		return fmt.Errorf("error getting client: %s", err)
 	}
+
 	conn := client.(*AWSClient).rdsconn
+	sweepResources := make([]*testSweepResource, 0)
 
 	err = conn.DescribeDBInstancesPages(&rds.DescribeDBInstancesInput{}, func(out *rds.DescribeDBInstancesOutput, lastPage bool) bool {
 		for _, dbi := range out.DBInstances {
-			log.Printf("[INFO] Deleting DB instance: %s", *dbi.DBInstanceIdentifier)
-
-			_, err := conn.DeleteDBInstance(&rds.DeleteDBInstanceInput{
-				DBInstanceIdentifier: dbi.DBInstanceIdentifier,
-				SkipFinalSnapshot:    aws.Bool(true),
-			})
-			if err != nil {
-				log.Printf("[ERROR] Failed to delete DB instance %s: %s",
-					*dbi.DBInstanceIdentifier, err)
-				continue
-			}
-
-			err = waitUntilAwsDbInstanceIsDeleted(*dbi.DBInstanceIdentifier, conn, 40*time.Minute)
-			if err != nil {
-				log.Printf("[ERROR] Failure while waiting for DB instance %s to be deleted: %s",
-					*dbi.DBInstanceIdentifier, err)
-			}
+			r := resourceAwsDbInstance()
+			d := r.Data(nil)
+			d.SetId(aws.StringValue(dbi.DBInstanceIdentifier))
+			d.Set("skip_final_snapshot", true)
+			sweepResources = append(sweepResources, NewTestSweepResource(r, d, client))
 		}
 		return !lastPage
 	})
@@ -60,7 +49,7 @@ func testSweepDbInstances(region string) error {
 		return fmt.Errorf("Error retrieving DB instances: %s", err)
 	}
 
-	return nil
+	return testSweepResourceOrchestrator(sweepResources)
 }
 
 func TestAccAWSDBInstance_basic(t *testing.T) {