Add batching for IAM binding/member/audit config changes

google/config.go

@@ -158,9 +158,8 @@ type Config struct {
 	ServiceManagementBasePath string
 	clientServiceMan          *servicemanagement.APIService
 
-	ServiceUsageBasePath       string
-	clientServiceUsage         *serviceusage.Service
-	requestBatcherServiceUsage *RequestBatcher
+	ServiceUsageBasePath string
+	clientServiceUsage   *serviceusage.Service
 
 	BigQueryBasePath string
 	clientBigQuery   *bigquery.Service
@@ -187,6 +186,9 @@ type Config struct {
 	// we expose those directly instead of providing the `Service` object
 	// as a factory.
 	clientBigtableProjectsInstances *bigtableadmin.ProjectsInstancesService
+
+	requestBatcherServiceUsage *RequestBatcher
+	requestBatcherIam          *RequestBatcher
 }
 
 var defaultClientScopes = []string{
@@ -399,7 +401,6 @@ func (c *Config) LoadAndValidate() error {
 	}
 	c.clientServiceUsage.UserAgent = userAgent
 	c.clientServiceUsage.BasePath = serviceUsageClientBasePath
-	c.requestBatcherServiceUsage = NewRequestBatcher("Service Usage", context, c.BatchingConfig)
 
 	cloudBillingClientBasePath := removeBasePathVersion(c.CloudBillingBasePath)
 	log.Printf("[INFO] Instantiating Google Cloud Billing client for path %s", cloudBillingClientBasePath)
@@ -544,6 +545,10 @@ func (c *Config) LoadAndValidate() error {
 	c.clientStorageTransfer.BasePath = storageTransferClientBasePath
 
 	c.Region = GetRegionFromRegionSelfLink(c.Region)
+
+	c.requestBatcherServiceUsage = NewRequestBatcher("Service Usage", context, c.BatchingConfig)
+	c.requestBatcherIam = NewRequestBatcher("IAM", context, c.BatchingConfig)
+
 	return nil
 }
 

google/iam_batching.go

@@ -0,0 +1,60 @@
+package google
+
+import (
+	"fmt"
+	"google.golang.org/api/cloudresourcemanager/v1"
+	"time"
+)
+
+const (
+	batchKeyTmplModifyIamPolicy = "%s modifyIamPolicy"
+
+	IamBatchingEnabled  = true
+	IamBatchingDisabled = false
+)
+
+func BatchRequestModifyIamPolicy(updater ResourceIamUpdater, modify iamPolicyModifyFunc, config *Config, reqDesc string) error {
+	batchKey := fmt.Sprintf(batchKeyTmplModifyIamPolicy, updater.GetMutexKey())
+
+	request := &BatchRequest{
+		ResourceName: updater.GetResourceId(),
+		Body:         []iamPolicyModifyFunc{modify},
+		CombineF:     combineBatchIamPolicyModifiers,
+		SendF:        sendBatchModifyIamPolicy(updater),
+		DebugId:      reqDesc,
+	}
+
+	_, err := config.requestBatcherIam.SendRequestWithTimeout(batchKey, request, time.Minute*30)
+	return err
+}
+
+func combineBatchIamPolicyModifiers(currV interface{}, toAddV interface{}) (interface{}, error) {
+	currModifiers, ok := currV.([]iamPolicyModifyFunc)
+	if !ok {
+		return nil, fmt.Errorf("provider error in batch combiner: expected data to be type []iamPolicyModifyFunc, got %v with type %T", currV, currV)
+	}
+
+	newModifiers, ok := toAddV.([]iamPolicyModifyFunc)
+	if !ok {
+		return nil, fmt.Errorf("provider error in batch combiner: expected data to be type []iamPolicyModifyFunc, got %v with type %T", currV, currV)
+	}
+
+	return append(currModifiers, newModifiers...), nil
+}
+
+func sendBatchModifyIamPolicy(updater ResourceIamUpdater) batcherSendFunc {
+	return func(resourceName string, body interface{}) (interface{}, error) {
+		modifiers, ok := body.([]iamPolicyModifyFunc)
+		if !ok {
+			return nil, fmt.Errorf("provider error: expected data to be type []iamPolicyModifyFunc, got %v with type %T", body, body)
+		}
+		return nil, iamPolicyReadModifyWrite(updater, func(policy *cloudresourcemanager.Policy) error {
+			for _, modifyF := range modifiers {
+				if err := modifyF(policy); err != nil {
+					return err
+				}
+			}
+			return nil
+		})
+	}
+}

google/provider.go

@@ -327,9 +327,9 @@ func ResourceMapWithErrors() (map[string]*schema.Resource, error) {
 			"google_organization_policy":                   resourceGoogleOrganizationPolicy(),
 			"google_project":                               resourceGoogleProject(),
 			"google_project_iam_policy":                    resourceGoogleProjectIamPolicy(),
-			"google_project_iam_binding":                   ResourceIamBinding(IamProjectSchema, NewProjectIamUpdater, ProjectIdParseFunc),
-			"google_project_iam_member":                    ResourceIamMember(IamProjectSchema, NewProjectIamUpdater, ProjectIdParseFunc),
-			"google_project_iam_audit_config":              ResourceIamAuditConfig(IamProjectSchema, NewProjectIamUpdater, ProjectIdParseFunc),
+			"google_project_iam_binding":                   ResourceIamBindingWithBatching(IamProjectSchema, NewProjectIamUpdater, ProjectIdParseFunc, IamBatchingEnabled),
+			"google_project_iam_member":                    ResourceIamMemberWithBatching(IamProjectSchema, NewProjectIamUpdater, ProjectIdParseFunc, IamBatchingEnabled),
+			"google_project_iam_audit_config":              ResourceIamAuditConfigWithBatching(IamProjectSchema, NewProjectIamUpdater, ProjectIdParseFunc, IamBatchingEnabled),
 			"google_project_service":                       resourceGoogleProjectService(),
 			"google_project_iam_custom_role":               resourceGoogleProjectIamCustomRole(),
 			"google_project_organization_policy":           resourceGoogleProjectOrganizationPolicy(),

google/resource_google_project_service.go

@@ -76,7 +76,7 @@ func resourceGoogleProjectServiceCreate(d *schema.ResourceData, meta interface{}
 	}
 
 	srv := d.Get("service").(string)
-	err = globalBatchEnableServices([]string{srv}, project, d, config)
+	err = BatchRequestEnableServices([]string{srv}, project, d, config)
 	if err != nil {
 		return err
 	}

google/resource_google_project_services.go

@@ -142,7 +142,7 @@ func setServiceUsageProjectEnabledServices(services []string, project string, d
 		}
 	}
 
-	if err := globalBatchEnableServices(toEnable, project, d, config); err != nil {
+	if err := BatchRequestEnableServices(toEnable, project, d, config); err != nil {
 		return fmt.Errorf("unable to enable Project Services %s (%+v): %s", project, services, err)
 	}
 
@@ -223,7 +223,7 @@ func listCurrentlyEnabledServices(project string, config *Config, timeout time.D
 	return apiServices, nil
 }
 
-// Enables services. WARNING: Use globalBatchEnableServices for better batching if possible.
+// Enables services. WARNING: Use BatchRequestEnableServices for better batching if possible.
 func enableServiceUsageProjectServices(services []string, project string, config *Config, timeout time.Duration) error {
 	// ServiceUsage does not allow more than 20 services to be enabled per
 	// batchEnable API call. See

google/resource_iam_audit_config.go

@@ -39,35 +39,46 @@ var iamAuditConfigSchema = map[string]*schema.Schema{
 }
 
 func ResourceIamAuditConfig(parentSpecificSchema map[string]*schema.Schema, newUpdaterFunc newResourceIamUpdaterFunc, resourceIdParser resourceIdParserFunc) *schema.Resource {
+	return ResourceIamAuditConfigWithBatching(parentSpecificSchema, newUpdaterFunc, resourceIdParser, IamBatchingDisabled)
+}
+
+func ResourceIamAuditConfigWithBatching(parentSpecificSchema map[string]*schema.Schema, newUpdaterFunc newResourceIamUpdaterFunc, resourceIdParser resourceIdParserFunc, enableBatching bool) *schema.Resource {
 	return &schema.Resource{
-		Create: resourceIamAuditConfigCreate(newUpdaterFunc),
+		Create: resourceIamAuditConfigCreate(newUpdaterFunc, enableBatching),
 		Read:   resourceIamAuditConfigRead(newUpdaterFunc),
-		Update: resourceIamAuditConfigUpdate(newUpdaterFunc),
-		Delete: resourceIamAuditConfigDelete(newUpdaterFunc),
+		Update: resourceIamAuditConfigUpdate(newUpdaterFunc, enableBatching),
+		Delete: resourceIamAuditConfigDelete(newUpdaterFunc, enableBatching),
 		Schema: mergeSchemas(iamAuditConfigSchema, parentSpecificSchema),
 		Importer: &schema.ResourceImporter{
 			State: iamAuditConfigImport(resourceIdParser),
 		},
 	}
 }
 
-func resourceIamAuditConfigCreate(newUpdaterFunc newResourceIamUpdaterFunc) schema.CreateFunc {
+func resourceIamAuditConfigCreate(newUpdaterFunc newResourceIamUpdaterFunc, enableBatching bool) schema.CreateFunc {
 	return func(d *schema.ResourceData, meta interface{}) error {
 		config := meta.(*Config)
 		updater, err := newUpdaterFunc(d, config)
 		if err != nil {
 			return err
 		}
 
-		p := getResourceIamAuditConfig(d)
-		err = iamPolicyReadModifyWrite(updater, func(ep *cloudresourcemanager.Policy) error {
-			ep.AuditConfigs = mergeAuditConfigs(append(ep.AuditConfigs, p))
+		ac := getResourceIamAuditConfig(d)
+		modifyF := func(ep *cloudresourcemanager.Policy) error {
+			ep.AuditConfigs = mergeAuditConfigs(append(ep.AuditConfigs, ac))
 			return nil
-		})
+		}
+
+		if enableBatching {
+			err = BatchRequestModifyIamPolicy(updater, modifyF, config, fmt.Sprintf(
+				"Add audit config for service %s on resource %q", ac.Service, updater.DescribeResource()))
+		} else {
+			err = iamPolicyReadModifyWrite(updater, modifyF)
+		}
 		if err != nil {
 			return err
 		}
-		d.SetId(updater.GetResourceId() + "/audit_config/" + p.Service)
+		d.SetId(updater.GetResourceId() + "/audit_config/" + ac.Service)
 		return resourceIamAuditConfigRead(newUpdaterFunc)(d, meta)
 	}
 }
@@ -139,7 +150,7 @@ func iamAuditConfigImport(resourceIdParser resourceIdParserFunc) schema.StateFun
 	}
 }
 
-func resourceIamAuditConfigUpdate(newUpdaterFunc newResourceIamUpdaterFunc) schema.UpdateFunc {
+func resourceIamAuditConfigUpdate(newUpdaterFunc newResourceIamUpdaterFunc, enableBatching bool) schema.UpdateFunc {
 	return func(d *schema.ResourceData, meta interface{}) error {
 		config := meta.(*Config)
 		updater, err := newUpdaterFunc(d, config)
@@ -148,11 +159,17 @@ func resourceIamAuditConfigUpdate(newUpdaterFunc newResourceIamUpdaterFunc) sche
 		}
 
 		ac := getResourceIamAuditConfig(d)
-		err = iamPolicyReadModifyWrite(updater, func(p *cloudresourcemanager.Policy) error {
-			cleaned := removeAllAuditConfigsWithService(p.AuditConfigs, ac.Service)
-			p.AuditConfigs = append(cleaned, ac)
+		modifyF := func(ep *cloudresourcemanager.Policy) error {
+			cleaned := removeAllAuditConfigsWithService(ep.AuditConfigs, ac.Service)
+			ep.AuditConfigs = append(cleaned, ac)
 			return nil
-		})
+		}
+		if enableBatching {
+			err = BatchRequestModifyIamPolicy(updater, modifyF, config, fmt.Sprintf(
+				"Overwrite audit config for service %s on resource %q", ac.Service, updater.DescribeResource()))
+		} else {
+			err = iamPolicyReadModifyWrite(updater, modifyF)
+		}
 		if err != nil {
 			return err
 		}
@@ -161,7 +178,7 @@ func resourceIamAuditConfigUpdate(newUpdaterFunc newResourceIamUpdaterFunc) sche
 	}
 }
 
-func resourceIamAuditConfigDelete(newUpdaterFunc newResourceIamUpdaterFunc) schema.DeleteFunc {
+func resourceIamAuditConfigDelete(newUpdaterFunc newResourceIamUpdaterFunc, enableBatching bool) schema.DeleteFunc {
 	return func(d *schema.ResourceData, meta interface{}) error {
 		config := meta.(*Config)
 		updater, err := newUpdaterFunc(d, config)
@@ -170,10 +187,16 @@ func resourceIamAuditConfigDelete(newUpdaterFunc newResourceIamUpdaterFunc) sche
 		}
 
 		ac := getResourceIamAuditConfig(d)
-		err = iamPolicyReadModifyWrite(updater, func(p *cloudresourcemanager.Policy) error {
-			p.AuditConfigs = removeAllAuditConfigsWithService(p.AuditConfigs, ac.Service)
+		modifyF := func(ep *cloudresourcemanager.Policy) error {
+			ep.AuditConfigs = removeAllAuditConfigsWithService(ep.AuditConfigs, ac.Service)
 			return nil
-		})
+		}
+		if enableBatching {
+			err = BatchRequestModifyIamPolicy(updater, modifyF, config, fmt.Sprintf(
+				"Delete audit config for service %s on resource %q", ac.Service, updater.DescribeResource()))
+		} else {
+			err = iamPolicyReadModifyWrite(updater, modifyF)
+		}
 		if err != nil {
 			return handleNotFoundError(err, d, fmt.Sprintf("Resource %s with IAM audit config %q", updater.DescribeResource(), d.Id()))
 		}

google/resource_iam_binding.go

@@ -34,19 +34,24 @@ var iamBindingSchema = map[string]*schema.Schema{
 }
 
 func ResourceIamBinding(parentSpecificSchema map[string]*schema.Schema, newUpdaterFunc newResourceIamUpdaterFunc, resourceIdParser resourceIdParserFunc) *schema.Resource {
+	return ResourceIamBindingWithBatching(parentSpecificSchema, newUpdaterFunc, resourceIdParser, IamBatchingDisabled)
+}
+
+// Resource that batches requests to the same IAM policy across multiple IAM fine-grained resources
+func ResourceIamBindingWithBatching(parentSpecificSchema map[string]*schema.Schema, newUpdaterFunc newResourceIamUpdaterFunc, resourceIdParser resourceIdParserFunc, enableBatching bool) *schema.Resource {
 	return &schema.Resource{
-		Create: resourceIamBindingCreateUpdate(newUpdaterFunc),
+		Create: resourceIamBindingCreateUpdate(newUpdaterFunc, enableBatching),
 		Read:   resourceIamBindingRead(newUpdaterFunc),
-		Update: resourceIamBindingCreateUpdate(newUpdaterFunc),
-		Delete: resourceIamBindingDelete(newUpdaterFunc),
+		Update: resourceIamBindingCreateUpdate(newUpdaterFunc, enableBatching),
+		Delete: resourceIamBindingDelete(newUpdaterFunc, enableBatching),
 		Schema: mergeSchemas(iamBindingSchema, parentSpecificSchema),
 		Importer: &schema.ResourceImporter{
 			State: iamBindingImport(resourceIdParser),
 		},
 	}
 }
 
-func resourceIamBindingCreateUpdate(newUpdaterFunc newResourceIamUpdaterFunc) func(*schema.ResourceData, interface{}) error {
+func resourceIamBindingCreateUpdate(newUpdaterFunc newResourceIamUpdaterFunc, enableBatching bool) func(*schema.ResourceData, interface{}) error {
 	return func(d *schema.ResourceData, meta interface{}) error {
 		config := meta.(*Config)
 		updater, err := newUpdaterFunc(d, config)
@@ -55,11 +60,18 @@ func resourceIamBindingCreateUpdate(newUpdaterFunc newResourceIamUpdaterFunc) fu
 		}
 
 		binding := getResourceIamBinding(d)
-		err = iamPolicyReadModifyWrite(updater, func(ep *cloudresourcemanager.Policy) error {
+		modifyF := func(ep *cloudresourcemanager.Policy) error {
 			cleaned := removeAllBindingsWithRole(ep.Bindings, binding.Role)
 			ep.Bindings = append(cleaned, binding)
 			return nil
-		})
+		}
+
+		if enableBatching {
+			err = BatchRequestModifyIamPolicy(updater, modifyF, config, fmt.Sprintf(
+				"Set IAM Binding for role %q on %q", binding.Role, updater.DescribeResource()))
+		} else {
+			err = iamPolicyReadModifyWrite(updater, modifyF)
+		}
 		if err != nil {
 			return err
 		}
@@ -142,7 +154,7 @@ func iamBindingImport(resourceIdParser resourceIdParserFunc) schema.StateFunc {
 	}
 }
 
-func resourceIamBindingDelete(newUpdaterFunc newResourceIamUpdaterFunc) schema.DeleteFunc {
+func resourceIamBindingDelete(newUpdaterFunc newResourceIamUpdaterFunc, enableBatching bool) schema.DeleteFunc {
 	return func(d *schema.ResourceData, meta interface{}) error {
 		config := meta.(*Config)
 		updater, err := newUpdaterFunc(d, config)
@@ -151,10 +163,17 @@ func resourceIamBindingDelete(newUpdaterFunc newResourceIamUpdaterFunc) schema.D
 		}
 
 		binding := getResourceIamBinding(d)
-		err = iamPolicyReadModifyWrite(updater, func(p *cloudresourcemanager.Policy) error {
+		modifyF := func(p *cloudresourcemanager.Policy) error {
 			p.Bindings = removeAllBindingsWithRole(p.Bindings, binding.Role)
 			return nil
-		})
+		}
+
+		if enableBatching {
+			err = BatchRequestModifyIamPolicy(updater, modifyF, config, fmt.Sprintf(
+				"Delete IAM Binding for role %q on %q", binding.Role, updater.DescribeResource()))
+		} else {
+			err = iamPolicyReadModifyWrite(updater, modifyF)
+		}
 		if err != nil {
 			return handleNotFoundError(err, d, fmt.Sprintf("Resource %q for IAM binding with role %q", updater.DescribeResource(), binding.Role))
 		}