diff --git a/mmv1/third_party/terraform/tpgiamresource/datasource_iam_policy.go b/mmv1/third_party/terraform/tpgiamresource/datasource_iam_policy.go index 8efc95d26d77..d36b00dea32e 100644 --- a/mmv1/third_party/terraform/tpgiamresource/datasource_iam_policy.go +++ b/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())) } diff --git a/mmv1/third_party/terraform/tpgiamresource/iam.go.tmpl b/mmv1/third_party/terraform/tpgiamresource/iam.go.tmpl index b1d6f8e6ea0a..258b6dc13cbc 100644 --- a/mmv1/third_party/terraform/tpgiamresource/iam.go.tmpl +++ b/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()) diff --git a/mmv1/third_party/terraform/tpgiamresource/iam_read_batching_test.go b/mmv1/third_party/terraform/tpgiamresource/iam_read_batching_test.go new file mode 100644 index 000000000000..da89943db190 --- /dev/null +++ b/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) + } + } +} diff --git a/mmv1/third_party/terraform/tpgiamresource/iam_read_modify_write_sa_not_found_test.go b/mmv1/third_party/terraform/tpgiamresource/iam_read_modify_write_sa_not_found_test.go new file mode 100644 index 000000000000..55ff1322a7b6 --- /dev/null +++ b/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") + } +} diff --git a/mmv1/third_party/terraform/tpgiamresource/resource_iam_audit_config.go b/mmv1/third_party/terraform/tpgiamresource/resource_iam_audit_config.go index 3f6482302f7e..e054ca2860b2 100644 --- a/mmv1/third_party/terraform/tpgiamresource/resource_iam_audit_config.go +++ b/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())) } diff --git a/mmv1/third_party/terraform/tpgiamresource/resource_iam_binding.go b/mmv1/third_party/terraform/tpgiamresource/resource_iam_binding.go index 1ea175a743e3..0db10ceef78c 100644 --- a/mmv1/third_party/terraform/tpgiamresource/resource_iam_binding.go +++ b/mmv1/third_party/terraform/tpgiamresource/resource_iam_binding.go @@ -223,7 +223,7 @@ func resourceIamBindingRead(newUpdaterFunc NewResourceIamUpdaterFunc, parentSpec eBinding := getResourceIamBinding(d) eCondition := conditionKeyFromCondition(eBinding.Condition) - p, err := iamPolicyReadWithRetry(updater) + p, err := iamPolicyReadWithRetry(updater, config) if err != nil { return transport_tpg.HandleNotFoundError(err, d, fmt.Sprintf("Resource %q with IAM Binding (Role %q)", updater.DescribeResource(), eBinding.Role)) } @@ -336,7 +336,7 @@ func iamBindingImport(newUpdaterFunc NewResourceIamUpdaterFunc, resourceIdParser if err != nil { return nil, err } - p, err := iamPolicyReadWithRetry(updater) + p, err := iamPolicyReadWithRetry(updater, config) if err != nil { return nil, err } diff --git a/mmv1/third_party/terraform/tpgiamresource/resource_iam_member.go b/mmv1/third_party/terraform/tpgiamresource/resource_iam_member.go index b6f6b89fa491..1f4120acaafa 100644 --- a/mmv1/third_party/terraform/tpgiamresource/resource_iam_member.go +++ b/mmv1/third_party/terraform/tpgiamresource/resource_iam_member.go @@ -152,7 +152,7 @@ func iamMemberImport(newUpdaterFunc NewResourceIamUpdaterFunc, resourceIdParser if err != nil { return nil, err } - p, err := iamPolicyReadWithRetry(updater) + p, err := iamPolicyReadWithRetry(updater, config) if err != nil { return nil, err } @@ -360,7 +360,7 @@ func resourceIamMemberRead(newUpdaterFunc NewResourceIamUpdaterFunc, parentSpeci eMember := getResourceIamMember(d) eCondition := conditionKeyFromCondition(eMember.Condition) - p, err := iamPolicyReadWithRetry(updater) + p, err := iamPolicyReadWithRetry(updater, config) if err != nil { return transport_tpg.HandleNotFoundError(err, d, fmt.Sprintf("Resource %q with IAM Member: Role %q Member %q", updater.DescribeResource(), eMember.Role, eMember.Members[0])) } diff --git a/mmv1/third_party/terraform/tpgiamresource/resource_iam_member_list.go b/mmv1/third_party/terraform/tpgiamresource/resource_iam_member_list.go index 7adcf34f1493..5a9f247c124e 100644 --- a/mmv1/third_party/terraform/tpgiamresource/resource_iam_member_list.go +++ b/mmv1/third_party/terraform/tpgiamresource/resource_iam_member_list.go @@ -241,7 +241,7 @@ func (r *IamMemberListResource) List(ctx context.Context, req list.ListRequest, } continue } - p, err := iamPolicyReadWithRetry(updater) + p, err := iamPolicyReadWithRetry(updater, r.Client) if err != nil { res := req.NewListResult(ctx) res.Diagnostics.AddError("API Error", err.Error()) diff --git a/mmv1/third_party/terraform/tpgiamresource/resource_iam_policy.go b/mmv1/third_party/terraform/tpgiamresource/resource_iam_policy.go index 5369b74dfc62..3456a78fec26 100644 --- a/mmv1/third_party/terraform/tpgiamresource/resource_iam_policy.go +++ b/mmv1/third_party/terraform/tpgiamresource/resource_iam_policy.go @@ -143,7 +143,7 @@ func ResourceIamPolicyRead(newUpdaterFunc NewResourceIamUpdaterFunc, parentSpeci 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())) }