Batch GetIamPolicy calls for all google_*_iam_* resources

mmv1/third_party/terraform/tpgiamresource/datasource_iam_policy.go

@@ -45,7 +45,7 @@ func DatasourceIamPolicyRead(newUpdaterFunc NewResourceIamUpdaterFunc) schema.Re
 			return err
 		}
 
-		policy, err := iamPolicyReadWithRetry(updater)
+		policy, err := iamPolicyReadWithRetry(updater, config)
 		if err != nil {
 			return transport_tpg.HandleNotFoundError(err, d, fmt.Sprintf("Resource %q with IAM Policy", updater.DescribeResource()))
 		}

mmv1/third_party/terraform/tpgiamresource/iam.go.tmpl

@@ -61,26 +61,50 @@ type (
 	ParentResourceIdFromIdentityParserFunc func(d *schema.ResourceData, identity *schema.IdentityData, config *transport_tpg.Config) (resourceID string, err error)
 )
 
-// Locking wrapper around read-only operation with retries.
-func iamPolicyReadWithRetry(updater ResourceIamUpdater) (*cloudresourcemanager.Policy, error) {
-	mutexKey := updater.GetMutexKey()
-	transport_tpg.MutexStore.Lock(mutexKey)
-	defer transport_tpg.MutexStore.Unlock(mutexKey)
+const batchKeyTmplIamPolicyRead = "%s getIamPolicy"
+
+// iamPolicyReadWithRetry reads a resource's IAM policy, with retries. Concurrent
+// calls for the same resource are batched into a single API call.
+func iamPolicyReadWithRetry(updater ResourceIamUpdater, config *transport_tpg.Config) (*cloudresourcemanager.Policy, error) {
+	batchKey := fmt.Sprintf(batchKeyTmplIamPolicyRead, updater.GetMutexKey())
+
+	req := &transport_tpg.BatchRequest{
+		ResourceName: updater.GetMutexKey(),
+		Body:         nil,
+		// Use empty CombineF since the request is exactly the same no matter how many callers ask for it.
+		CombineF: func(body interface{}, toAdd interface{}) (interface{}, error) { return nil, nil },
+		SendF:    sendIamPolicyRead(updater),
+		DebugId:  fmt.Sprintf("Read IAM policy for %s", updater.DescribeResource()),
+	}
 
-	log.Printf("[DEBUG] Retrieving policy for %s\n", updater.DescribeResource())
-	var policy *cloudresourcemanager.Policy
-	err := transport_tpg.Retry(transport_tpg.RetryOptions{
-		RetryFunc: func() (perr error) {
-			policy, perr = updater.GetResourceIamPolicy()
-			return perr
-		},
-		Timeout: 10 * time.Minute,
-	})
+	result, err := config.RequestBatcherIam.SendRequestWithTimeout(batchKey, req, 10*time.Minute)
 	if err != nil {
 		return nil, err
 	}
-	log.Print(spew.Sprintf("[DEBUG] Retrieved policy for %s: %#v\n", updater.DescribeResource(), policy))
-	return policy, nil
+	return result.(*cloudresourcemanager.Policy), nil
+}
+
+func sendIamPolicyRead(updater ResourceIamUpdater) transport_tpg.BatcherSendFunc {
+	return func(_ string, _ interface{}) (interface{}, error) {
+		mutexKey := updater.GetMutexKey()
+		transport_tpg.MutexStore.Lock(mutexKey)
+		defer transport_tpg.MutexStore.Unlock(mutexKey)
+
+		log.Printf("[DEBUG] Retrieving policy for %s\n", updater.DescribeResource())
+		var policy *cloudresourcemanager.Policy
+		err := transport_tpg.Retry(transport_tpg.RetryOptions{
+			RetryFunc: func() (perr error) {
+				policy, perr = updater.GetResourceIamPolicy()
+				return perr
+			},
+			Timeout: 10 * time.Minute,
+		})
+		if err != nil {
+			return nil, err
+		}
+		log.Print(spew.Sprintf("[DEBUG] Retrieved policy for %s: %#v\n", updater.DescribeResource(), policy))
+		return policy, nil
+	}
 }
 
 // Locking wrapper around read-modify-write cycle for IAM policy.
@@ -162,11 +186,10 @@ func iamPolicyReadModifyWrite(updater ResourceIamUpdater, modify iamPolicyModify
 		// retry in the case that a service account is not found. This can happen when a service account is deleted
 		// out of band.
 		if isServiceAccountNotFoundError, _ := transport_tpg.IamServiceAccountNotFound(err); isServiceAccountNotFoundError {
-			// calling a retryable function within a retry loop is not
-			// strictly the _best_ idea, but this error only happens in
-			// high-traffic projects anyways
-			currentPolicy, rerr := iamPolicyReadWithRetry(updater)
-			if rerr != nil {
+			// We are already holding the mutex here, so must call the updater method directly instead
+			// of iamPolicyReadWithRetry.
+			currentPolicy, rerr := updater.GetResourceIamPolicy()
+			if rerr == nil {
 				if p.Etag != currentPolicy.Etag {
 					// not matching indicates that there is a new state to attempt to apply
 					log.Printf("current and old etag did not match for %s, retrying", updater.DescribeResource())

mmv1/third_party/terraform/tpgiamresource/iam_read_batching_test.go

@@ -0,0 +1,136 @@
+package tpgiamresource
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	transport_tpg "github.com/hashicorp/terraform-provider-google/google/transport"
+	"google.golang.org/api/cloudresourcemanager/v1"
+)
+
+// fakeResourceIamUpdater is a minimal ResourceIamUpdater used to exercise
+// iamPolicyReadWithRetry's batching behavior without any real API calls.
+type fakeResourceIamUpdater struct {
+	mutexKey  string
+	resource  string
+	getCount  int32
+	getPolicy func() (*cloudresourcemanager.Policy, error)
+}
+
+func (u *fakeResourceIamUpdater) GetResourceIamPolicy() (*cloudresourcemanager.Policy, error) {
+	atomic.AddInt32(&u.getCount, 1)
+	return u.getPolicy()
+}
+
+func (u *fakeResourceIamUpdater) SetResourceIamPolicy(policy *cloudresourcemanager.Policy) error {
+	return fmt.Errorf("not implemented")
+}
+
+func (u *fakeResourceIamUpdater) GetMutexKey() string {
+	return u.mutexKey
+}
+
+func (u *fakeResourceIamUpdater) GetResourceId() string {
+	return u.resource
+}
+
+func (u *fakeResourceIamUpdater) DescribeResource() string {
+	return u.resource
+}
+
+func newTestConfigForIamBatching() *transport_tpg.Config {
+	ctx := context.Background()
+	batchingConfig := &transport_tpg.BatchingConfig{
+		SendAfter:      100 * time.Millisecond,
+		EnableBatching: true,
+	}
+	return &transport_tpg.Config{
+		Context:           ctx,
+		BatchingConfig:    batchingConfig,
+		RequestBatcherIam: transport_tpg.NewRequestBatcher("IAM", ctx, batchingConfig),
+	}
+}
+
+func TestIamPolicyReadWithRetry_CollapsesConcurrentCallsIntoOneRequest(t *testing.T) {
+	policy := &cloudresourcemanager.Policy{Etag: "etag-1"}
+	updater := &fakeResourceIamUpdater{
+		mutexKey: "iam-project-my-project",
+		resource: "project my-project",
+		getPolicy: func() (*cloudresourcemanager.Policy, error) {
+			return policy, nil
+		},
+	}
+	config := newTestConfigForIamBatching()
+
+	const numCallers = 10
+	var wg sync.WaitGroup
+	wg.Add(numCallers)
+	errs := make([]error, numCallers)
+	results := make([]*cloudresourcemanager.Policy, numCallers)
+
+	for i := 0; i < numCallers; i++ {
+		go func(idx int) {
+			defer wg.Done()
+			p, err := iamPolicyReadWithRetry(updater, config)
+			errs[idx] = err
+			results[idx] = p
+		}(i)
+	}
+	wg.Wait()
+
+	for i, err := range errs {
+		if err != nil {
+			t.Errorf("caller %d: unexpected error: %v", i, err)
+		}
+		if results[i] != policy {
+			t.Errorf("caller %d: expected shared policy pointer %p, got %p", i, policy, results[i])
+		}
+	}
+
+	if got := atomic.LoadInt32(&updater.getCount); got != 1 {
+		t.Errorf("expected exactly 1 GetResourceIamPolicy call for %d concurrent callers, got %d", numCallers, got)
+	}
+}
+
+func TestIamPolicyReadWithRetry_SeparateResourcesAreNotCombined(t *testing.T) {
+	config := newTestConfigForIamBatching()
+
+	updaters := []*fakeResourceIamUpdater{
+		{
+			mutexKey: "iam-project-project-a",
+			resource: "project project-a",
+			getPolicy: func() (*cloudresourcemanager.Policy, error) {
+				return &cloudresourcemanager.Policy{Etag: "etag-a"}, nil
+			},
+		},
+		{
+			mutexKey: "iam-project-project-b",
+			resource: "project project-b",
+			getPolicy: func() (*cloudresourcemanager.Policy, error) {
+				return &cloudresourcemanager.Policy{Etag: "etag-b"}, nil
+			},
+		},
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(len(updaters))
+	for _, u := range updaters {
+		go func(u *fakeResourceIamUpdater) {
+			defer wg.Done()
+			if _, err := iamPolicyReadWithRetry(u, config); err != nil {
+				t.Errorf("resource %s: unexpected error: %v", u.resource, err)
+			}
+		}(u)
+	}
+	wg.Wait()
+
+	for _, u := range updaters {
+		if got := atomic.LoadInt32(&u.getCount); got != 1 {
+			t.Errorf("resource %s: expected exactly 1 GetResourceIamPolicy call, got %d", u.resource, got)
+		}
+	}
+}

mmv1/third_party/terraform/tpgiamresource/iam_read_modify_write_sa_not_found_test.go

@@ -0,0 +1,152 @@
+package tpgiamresource
+
+import (
+	"fmt"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"google.golang.org/api/cloudresourcemanager/v1"
+	"google.golang.org/api/googleapi"
+)
+
+// serviceAccountNotFoundErr constructs an error matching
+// transport_tpg.IamServiceAccountNotFound's predicate, to trigger
+// iamPolicyReadModifyWrite's "service account not found" retry branch.
+func serviceAccountNotFoundErr() error {
+	return &googleapi.Error{
+		Code: 400,
+		Body: "Service account foo@bar.iam.gserviceaccount.com does not exist.",
+	}
+}
+
+// saNotFoundFakeUpdater is a ResourceIamUpdater whose SetResourceIamPolicy
+// always fails with a "service account not found" error, used to exercise
+// iamPolicyReadModifyWrite's rarely-hit recovery branch for that error.
+type saNotFoundFakeUpdater struct {
+	mutexKey string
+	resource string
+
+	getPolicy func(callNum int32) (*cloudresourcemanager.Policy, error)
+
+	getCount int32
+	setCount int32
+}
+
+func (u *saNotFoundFakeUpdater) GetResourceIamPolicy() (*cloudresourcemanager.Policy, error) {
+	n := atomic.AddInt32(&u.getCount, 1)
+	return u.getPolicy(n)
+}
+
+func (u *saNotFoundFakeUpdater) SetResourceIamPolicy(policy *cloudresourcemanager.Policy) error {
+	atomic.AddInt32(&u.setCount, 1)
+	return serviceAccountNotFoundErr()
+}
+
+func (u *saNotFoundFakeUpdater) GetMutexKey() string {
+	return u.mutexKey
+}
+
+func (u *saNotFoundFakeUpdater) GetResourceId() string {
+	return u.resource
+}
+
+func (u *saNotFoundFakeUpdater) DescribeResource() string {
+	return u.resource
+}
+
+// runWithTimeout runs f in a goroutine and reports whether it completed
+// within the given timeout, to detect a deadlock without hanging the test
+// suite forever if the bug being tested for is present. A panic inside f is
+// recovered (a panic in a spawned goroutine would otherwise crash the whole
+// test binary, bypassing any recover() in the calling goroutine) and fails
+// the test immediately via t.Fatalf.
+func runWithTimeout(t *testing.T, timeout time.Duration, f func() error) (err error, completed bool) {
+	t.Helper()
+	done := make(chan error, 1)
+	panicked := make(chan interface{}, 1)
+	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				panicked <- r
+			}
+		}()
+		done <- f()
+	}()
+	select {
+	case err := <-done:
+		return err, true
+	case r := <-panicked:
+		t.Fatalf("function panicked: %v", r)
+		return nil, true
+	case <-time.After(timeout):
+		return nil, false
+	}
+}
+
+// TestIamPolicyReadModifyWrite_ServiceAccountNotFound_EtagUnchanged verifies
+// that when SetResourceIamPolicy fails with a "service account not found"
+// error and the policy is unchanged since it was first read,
+// iamPolicyReadModifyWrite returns promptly with an error (rather than
+// deadlocking, which is what the unfixed code does: it re-locks the same
+// resource mutex it's already holding).
+func TestIamPolicyReadModifyWrite_ServiceAccountNotFound_EtagUnchanged(t *testing.T) {
+	updater := &saNotFoundFakeUpdater{
+		mutexKey: "iam-project-my-project",
+		resource: "project my-project",
+		getPolicy: func(callNum int32) (*cloudresourcemanager.Policy, error) {
+			// Etag never changes across calls.
+			return &cloudresourcemanager.Policy{Etag: "etag-1"}, nil
+		},
+	}
+
+	err, completed := runWithTimeout(t, 5*time.Second, func() error {
+		return iamPolicyReadModifyWrite(updater, func(p *cloudresourcemanager.Policy) error { return nil })
+	})
+
+	if !completed {
+		t.Fatal("iamPolicyReadModifyWrite did not return within timeout: deadlocked re-acquiring its own resource mutex")
+	}
+	if err == nil {
+		t.Fatal("expected an error to be returned (service account not found, etag unchanged), got nil")
+	}
+	// It should have given up after a single recheck, not retried forever.
+	if got := atomic.LoadInt32(&updater.getCount); got != 2 {
+		t.Errorf("expected exactly 2 GetResourceIamPolicy calls (initial read + one recheck), got %d", got)
+	}
+	if got := atomic.LoadInt32(&updater.setCount); got != 1 {
+		t.Errorf("expected exactly 1 SetResourceIamPolicy call, got %d", got)
+	}
+}
+
+// TestIamPolicyReadModifyWrite_ServiceAccountNotFound_RecheckReadFails
+// verifies that when the recheck read itself fails, iamPolicyReadModifyWrite
+// falls through and returns the original error rather than panicking on a
+// nil policy dereference (the unfixed code's condition is inverted, so it
+// attempts to read .Etag off a nil *cloudresourcemanager.Policy in this
+// case) or deadlocking.
+func TestIamPolicyReadModifyWrite_ServiceAccountNotFound_RecheckReadFails(t *testing.T) {
+	updater := &saNotFoundFakeUpdater{
+		mutexKey: "iam-project-my-project",
+		resource: "project my-project",
+		getPolicy: func(callNum int32) (*cloudresourcemanager.Policy, error) {
+			if callNum == 1 {
+				// initial read at the top of the read-modify-write loop
+				return &cloudresourcemanager.Policy{Etag: "etag-1"}, nil
+			}
+			// the recheck read fails
+			return nil, fmt.Errorf("transient error rechecking policy")
+		},
+	}
+
+	err, completed := runWithTimeout(t, 5*time.Second, func() error {
+		return iamPolicyReadModifyWrite(updater, func(p *cloudresourcemanager.Policy) error { return nil })
+	})
+
+	if !completed {
+		t.Fatal("iamPolicyReadModifyWrite did not return within timeout: deadlocked re-acquiring its own resource mutex")
+	}
+	if err == nil {
+		t.Fatal("expected an error to be returned (service account not found, recheck read failed), got nil")
+	}
+}

mmv1/third_party/terraform/tpgiamresource/resource_iam_audit_config.go

@@ -71,7 +71,7 @@ func resourceIamAuditConfigRead(newUpdaterFunc NewResourceIamUpdaterFunc) schema
 		}
 
 		eAuditConfig := getResourceIamAuditConfig(d)
-		p, err := iamPolicyReadWithRetry(updater)
+		p, err := iamPolicyReadWithRetry(updater, config)
 		if err != nil {
 			return transport_tpg.HandleNotFoundError(err, d, fmt.Sprintf("AuditConfig for %s on %q", eAuditConfig.Service, updater.DescribeResource()))
 		}