Repository: crossplane/crossplane

Issue:
Installing packages with many CRDs causes reconciler to exceed context deadline
### What happened?

Cannot install a provider package containing 750+ CRDs. 

Trying to install [provider-tf-aws](https://github.com/crossplane-contrib/provider-tf-aws) built from [this commit](https://github.com/crossplane-contrib/provider-tf-aws/tree/daf1e9f7fcbd7959c25ccdcc11e0800d17e904fe), but package is never installed.

I am observing context deadline errors in `providerrevisions.pkg.crossplane.io` events:

```
Spec:
  Desired State:                  Active
  Ignore Crossplane Constraints:  false
  Image:                          turkenh/provider-tf-aws:daf1e9f7-2
  Package Pull Policy:            IfNotPresent
  Revision:                       1
  Skip Dependency Resolution:     false
Status:
  Conditions:
    Last Transition Time:  2021-09-10T09:14:52Z
    Reason:                UnhealthyPackageRevision
    Status:                False
    Type:                  Healthy
  Controller Ref:
    Name:
Events:
  Type     Reason             Age                  From                                         Message
  ----     ------             ----                 ----                                         -------
  Warning  SyncPackage        76s                  packages/providerrevision.pkg.crossplane.io  cannot establish control of object: context deadline exceeded
  Normal   BindClusterRole    16s (x4 over 2m16s)  rbac/providerrevision.pkg.crossplane.io      Bound system ClusterRole to provider ServiceAccount(s)
  Normal   ApplyClusterRoles  16s (x3 over 2m15s)  rbac/providerrevision.pkg.crossplane.io      Applied RBAC ClusterRoles
  Warning  SyncPackage        16s                  packages/providerrevision.pkg.crossplane.io  cannot establish control of object: rate: Wait(n=1) would exceed context deadline
```

### How can we reproduce it?

`kubectl crossplane install provider turkenh/provider-tf-aws:daf1e9f7-2`

### What environment did it happen in?
Crossplane version: v1.3.1
Kubernetes Cluster: Kind - v1.21.1

Possibly related: https://github.com/crossplane-contrib/terrajet/issues/47


Hints from issue comments:
@turkenh: I've updated the issue description, as originally I tried to install provider with controller image 🤦 
@jbw976: This is still occurring for some folks (see #3061) - we should reconsider the hard-coded timeout value and expose it as a configurable value.@negz: Let's make the context deadline configurable via a flag and also bump the default up a little - maybe to 5 mins for now? Ideally we'd avoid having configurable deadlines for only only a subset of reconcilers so maybe we can set one value for all of them? That said I could be convinced that package revisions are a special case.

We might also consider bumping the default Crossplane deployment's CPU resources up - we currently request `150m` cores by default (a small portion of one core) which is quite miniscule.
@jbw976: @haarchri brought this up today in the community meeting, as he and others in the community are seeing similar behavior in #2849 and feeling the pain around it.@negz: Thanks @haarchri and @jbw976! This is going to be a priority ASAP.@negz: I just chatted with @avalanche123, who is going to take a shot at this! Per my comment at https://github.com/crossplane/crossplane/issues/2564#issuecomment-1165007894 the quick fix here is likely going to be to make this deadline configurable. That said, I think it would be interesting to check in with @hasheddan to see whether he can think of any low hanging fruit to speed up the reconcile process - perhaps parallelizing/batching CRD installs?@negz: We've had 1-2 reports of this still happening, so I'm reopening to dig into that.@negz: Specifically seeing this now (I'm using `upbound/crossplane:v1.9.0-up.2`, which should have https://github.com/crossplane/crossplane/pull/3176):


```console
0s          Warning   SyncPackage               providerrevision/crossplane-provider-jet-azure-b7b0db0f74e0   cannot establish control of object: client rate limiter Wait returned an error: rate: Wait(n=1) would exceed context deadline
```@negz: Just confirmed I'm seeing the same with `upbound/crossplane:v1.9.0-up.2` too. Both tests using kind v1.24.0 clusters.@negz: And yet `crossplane/crossplane:v1.10.0-rc.0.31.g781e909f` works as I would expect. 🤔 I guess perhaps this issue is somehow isolated to UXP. Will close and move discussion there.@haarchri: @negz we see the warning as
well in 1.9.0 after a few loops it is working without handmade interactions@Mitsuwa:  I seem to be running into this again trying to do a fresh install of 
 
```
   Image:                          xpkg.upbound.io/upbound/provider-aws:v0.35.0
```

```
  Normal   BindClusterRole    21m (x46 over 20h)     rbac/providerrevision.pkg.crossplane.io      Bound system ClusterRole to provider ServiceAccount(s)
  Normal   ApplyClusterRoles  17m (x68 over 20h)     rbac/providerrevision.pkg.crossplane.io      Applied RBAC ClusterRoles
  Warning  SyncPackage        2m23s (x380 over 19h)  packages/providerrevision.pkg.crossplane.io  cannot establish control of object: Post "https://10.131.0.1:443/apis/apiextensions.k8s.io/v1/customresourcedefinitions?dryRun=All": context deadline exceeded
```@nathkn: We're running into this timeout a lot for provider packages that use the "flattened" filesystem instead of the OCI annotation for the package.yaml file, where all the layers have to be validated by the backend, which can take longer than 3 minutes. It'd be great to get the timeout to be configurable.

Solution patch:
diff --git a/go.mod b/go.mod
index a53669373..02b30d31f 100644
--- a/go.mod
+++ b/go.mod
@@ -13,6 +13,7 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/sirupsen/logrus v1.8.1
 	github.com/spf13/afero v1.8.0
+	golang.org/x/sync v0.0.0-20220513210516-0976fa681c29
 	k8s.io/api v0.24.0
 	k8s.io/apiextensions-apiserver v0.23.0
 	k8s.io/apimachinery v0.24.0
@@ -137,7 +138,6 @@ require (
 	golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3 // indirect
 	golang.org/x/net v0.0.0-20220516155154-20f960328961 // indirect
 	golang.org/x/oauth2 v0.0.0-20220411215720-9780585627b5 // indirect
-	golang.org/x/sync v0.0.0-20220513210516-0976fa681c29 // indirect
 	golang.org/x/sys v0.0.0-20220513210249-45d2b4557a2a // indirect
 	golang.org/x/term v0.0.0-20220411215600-e5f449aeb171 // indirect
 	golang.org/x/text v0.3.7 // indirect
diff --git a/internal/controller/pkg/revision/establisher.go b/internal/controller/pkg/revision/establisher.go
index 274b5b2fe..0a1b0a0ef 100644
--- a/internal/controller/pkg/revision/establisher.go
+++ b/internal/controller/pkg/revision/establisher.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"strings"
 
+	"golang.org/x/sync/errgroup"
 	admv1 "k8s.io/api/admissionregistration/v1"
 	corev1 "k8s.io/api/core/v1"
 	extv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
@@ -39,6 +40,12 @@ import (
 	v1 "github.com/crossplane/crossplane/apis/pkg/v1"
 )
 
+const (
+	// maxConcurrentEstablishers specifies the maximum number of goroutines to use
+	// for establishing resources.
+	maxConcurrentEstablishers = 10
+)
+
 const (
 	errAssertResourceObj            = "cannot assert object to resource.Object"
 	errAssertClientObj              = "cannot assert object to client.Object"
@@ -93,9 +100,20 @@ type currentDesired struct {
 
 // Establish checks that control or ownership of resources can be established by
 // parent, then establishes it.
-func (e *APIEstablisher) Establish(ctx context.Context, objs []runtime.Object, parent v1.PackageRevision, control bool) ([]xpv1.TypedReference, error) { // nolint:gocyclo
-	allObjs := []currentDesired{}
-	resourceRefs := []xpv1.TypedReference{}
+func (e *APIEstablisher) Establish(ctx context.Context, objs []runtime.Object, parent v1.PackageRevision, control bool) ([]xpv1.TypedReference, error) {
+	allObjs, err := e.validate(ctx, objs, parent, control)
+	if err != nil {
+		return nil, err
+	}
+
+	resourceRefs, err := e.establish(ctx, allObjs, parent, control)
+	if err != nil {
+		return nil, err
+	}
+	return resourceRefs, nil
+}
+
+func (e *APIEstablisher) validate(ctx context.Context, objs []runtime.Object, parent v1.PackageRevision, control bool) ([]currentDesired, error) { // nolint:gocyclo
 	var webhookTLSCert []byte
 	if parent.GetWebhookTLSSecretName() != nil {
 		s := &corev1.Secret{}
@@ -108,138 +126,178 @@ func (e *APIEstablisher) Establish(ctx context.Context, objs []runtime.Object, p
 		}
 		webhookTLSCert = s.Data["tls.crt"]
 	}
+	g, ctx := errgroup.WithContext(ctx)
+	g.SetLimit(maxConcurrentEstablishers)
+	out := make(chan currentDesired, len(objs))
 	for _, res := range objs {
-		// Assert desired object to resource.Object so that we can access its
-		// metadata.
-		d, ok := res.(resource.Object)
-		if !ok {
-			return nil, errors.New(errAssertResourceObj)
-		}
-
-		// The generated webhook configurations have a static hard-coded name
-		// that the developers of the providers can't affect. Here, we make sure
-		// to distinguish one from the other by setting the name to the parent
-		// since there is always a single ValidatingWebhookConfiguration and/or
-		// single MutatingWebhookConfiguration object in a provider package.
-		// See https://github.com/kubernetes-sigs/controller-tools/issues/658
-		switch conf := res.(type) {
-		case *admv1.ValidatingWebhookConfiguration:
-			if len(webhookTLSCert) == 0 {
-				continue
-			}
-			if pkgRef, ok := GetPackageOwnerReference(parent); ok {
-				conf.SetName(fmt.Sprintf("crossplane-%s-%s", strings.ToLower(pkgRef.Kind), pkgRef.Name))
+		res := res // Pin the range variable before using it in a Goroutine.
+		g.Go(func() error {
+			// Assert desired object to resource.Object so that we can access its
+			// metadata.
+			d, ok := res.(resource.Object)
+			if !ok {
+				return errors.New(errAssertResourceObj)
 			}
-			for i := range conf.Webhooks {
-				conf.Webhooks[i].ClientConfig.CABundle = webhookTLSCert
-				if conf.Webhooks[i].ClientConfig.Service == nil {
-					conf.Webhooks[i].ClientConfig.Service = &admv1.ServiceReference{}
+
+			// The generated webhook configurations have a static hard-coded name
+			// that the developers of the providers can't affect. Here, we make sure
+			// to distinguish one from the other by setting the name to the parent
+			// since there is always a single ValidatingWebhookConfiguration and/or
+			// single MutatingWebhookConfiguration object in a provider package.
+			// See https://github.com/kubernetes-sigs/controller-tools/issues/658
+			switch conf := res.(type) {
+			case *admv1.ValidatingWebhookConfiguration:
+				if len(webhookTLSCert) == 0 {
+					return nil
 				}
-				conf.Webhooks[i].ClientConfig.Service.Name = parent.GetName()
-				conf.Webhooks[i].ClientConfig.Service.Namespace = e.namespace
-				conf.Webhooks[i].ClientConfig.Service.Port = pointer.Int32(webhookPort)
-			}
-		case *admv1.MutatingWebhookConfiguration:
-			if len(webhookTLSCert) == 0 {
-				continue
-			}
-			if pkgRef, ok := GetPackageOwnerReference(parent); ok {
-				conf.SetName(fmt.Sprintf("crossplane-%s-%s", strings.ToLower(pkgRef.Kind), pkgRef.Name))
-			}
-			for i := range conf.Webhooks {
-				conf.Webhooks[i].ClientConfig.CABundle = webhookTLSCert
-				if conf.Webhooks[i].ClientConfig.Service == nil {
-					conf.Webhooks[i].ClientConfig.Service = &admv1.ServiceReference{}
+				if pkgRef, ok := GetPackageOwnerReference(parent); ok {
+					conf.SetName(fmt.Sprintf("crossplane-%s-%s", strings.ToLower(pkgRef.Kind), pkgRef.Name))
 				}
-				conf.Webhooks[i].ClientConfig.Service.Name = parent.GetName()
-				conf.Webhooks[i].ClientConfig.Service.Namespace = e.namespace
-				conf.Webhooks[i].ClientConfig.Service.Port = pointer.Int32(webhookPort)
-			}
-		case *extv1.CustomResourceDefinition:
-			if conf.Spec.Conversion != nil && conf.Spec.Conversion.Strategy == extv1.WebhookConverter {
+				for i := range conf.Webhooks {
+					conf.Webhooks[i].ClientConfig.CABundle = webhookTLSCert
+					if conf.Webhooks[i].ClientConfig.Service == nil {
+						conf.Webhooks[i].ClientConfig.Service = &admv1.ServiceReference{}
+					}
+					conf.Webhooks[i].ClientConfig.Service.Name = parent.GetName()
+					conf.Webhooks[i].ClientConfig.Service.Namespace = e.namespace
+					conf.Webhooks[i].ClientConfig.Service.Port = pointer.Int32(webhookPort)
+				}
+			case *admv1.MutatingWebhookConfiguration:
 				if len(webhookTLSCert) == 0 {
-					return nil, errors.New(errConversionWithNoWebhookCA)
+					return nil
 				}
-				if conf.Spec.Conversion.Webhook == nil {
-					conf.Spec.Conversion.Webhook = &extv1.WebhookConversion{}
+				if pkgRef, ok := GetPackageOwnerReference(parent); ok {
+					conf.SetName(fmt.Sprintf("crossplane-%s-%s", strings.ToLower(pkgRef.Kind), pkgRef.Name))
 				}
-				if conf.Spec.Conversion.Webhook.ClientConfig == nil {
-					conf.Spec.Conversion.Webhook.ClientConfig = &extv1.WebhookClientConfig{}
+				for i := range conf.Webhooks {
+					conf.Webhooks[i].ClientConfig.CABundle = webhookTLSCert
+					if conf.Webhooks[i].ClientConfig.Service == nil {
+						conf.Webhooks[i].ClientConfig.Service = &admv1.ServiceReference{}
+					}
+					conf.Webhooks[i].ClientConfig.Service.Name = parent.GetName()
+					conf.Webhooks[i].ClientConfig.Service.Namespace = e.namespace
+					conf.Webhooks[i].ClientConfig.Service.Port = pointer.Int32(webhookPort)
 				}
-				if conf.Spec.Conversion.Webhook.ClientConfig.Service == nil {
-					conf.Spec.Conversion.Webhook.ClientConfig.Service = &extv1.ServiceReference{}
+			case *extv1.CustomResourceDefinition:
+				if conf.Spec.Conversion != nil && conf.Spec.Conversion.Strategy == extv1.WebhookConverter {
+					if len(webhookTLSCert) == 0 {
+						return errors.New(errConversionWithNoWebhookCA)
+					}
+					if conf.Spec.Conversion.Webhook == nil {
+						conf.Spec.Conversion.Webhook = &extv1.WebhookConversion{}
+					}
+					if conf.Spec.Conversion.Webhook.ClientConfig == nil {
+						conf.Spec.Conversion.Webhook.ClientConfig = &extv1.WebhookClientConfig{}
+					}
+					if conf.Spec.Conversion.Webhook.ClientConfig.Service == nil {
+						conf.Spec.Conversion.Webhook.ClientConfig.Service = &extv1.ServiceReference{}
+					}
+					conf.Spec.Conversion.Webhook.ClientConfig.CABundle = webhookTLSCert
+					conf.Spec.Conversion.Webhook.ClientConfig.Service.Name = parent.GetName()
+					conf.Spec.Conversion.Webhook.ClientConfig.Service.Namespace = e.namespace
+					conf.Spec.Conversion.Webhook.ClientConfig.Service.Port = pointer.Int32(webhookPort)
 				}
-				conf.Spec.Conversion.Webhook.ClientConfig.CABundle = webhookTLSCert
-				conf.Spec.Conversion.Webhook.ClientConfig.Service.Name = parent.GetName()
-				conf.Spec.Conversion.Webhook.ClientConfig.Service.Namespace = e.namespace
-				conf.Spec.Conversion.Webhook.ClientConfig.Service.Port = pointer.Int32(webhookPort)
 			}
-		}
 
-		// Make a copy of the desired object to be populated with existing
-		// object, if it exists.
-		copy := res.DeepCopyObject()
-		current, ok := copy.(client.Object)
-		if !ok {
-			return nil, errors.New(errAssertClientObj)
-		}
-		err := e.client.Get(ctx, types.NamespacedName{Name: d.GetName(), Namespace: d.GetNamespace()}, current)
-		if resource.IgnoreNotFound(err) != nil {
-			return nil, err
-		}
+			// Make a copy of the desired object to be populated with existing
+			// object, if it exists.
+			copy := res.DeepCopyObject()
+			current, ok := copy.(client.Object)
+			if !ok {
+				return errors.New(errAssertClientObj)
+			}
+			err := e.client.Get(ctx, types.NamespacedName{Name: d.GetName(), Namespace: d.GetNamespace()}, current)
+			if resource.IgnoreNotFound(err) != nil {
+				return err
+			}
 
-		// If resource does not already exist, we must attempt to dry run create
-		// it.
-		if kerrors.IsNotFound(err) {
-			// Add to objects as not existing.
-			allObjs = append(allObjs, currentDesired{
-				Desired: d,
-				Current: nil,
-				Exists:  false,
-			})
-			// We will not create a resource if we are not going to control it,
-			// so we don't need to check with dry run.
-			if control {
-				if err := e.create(ctx, d, parent, client.DryRunAll); err != nil {
-					return nil, err
+			// If resource does not already exist, we must attempt to dry run create
+			// it.
+			if kerrors.IsNotFound(err) {
+				// We will not create a resource if we are not going to control it,
+				// so we don't need to check with dry run.
+				if control {
+					if err := e.create(ctx, d, parent, client.DryRunAll); err != nil {
+						return err
+					}
+				}
+				// Add to objects as not existing.
+				select {
+				case out <- currentDesired{Desired: d, Current: nil, Exists: false}:
+					return nil
+				case <-ctx.Done():
+					return ctx.Err()
 				}
 			}
-			continue
-		}
 
-		c := current.(resource.Object)
-		// Add to objects as existing.
-		allObjs = append(allObjs, currentDesired{
-			Desired: d,
-			Current: c,
-			Exists:  true,
+			c := current.(resource.Object)
+			if err := e.update(ctx, c, d, parent, control, client.DryRunAll); err != nil {
+				return err
+			}
+			// Add to objects as existing.
+			select {
+			case out <- currentDesired{Desired: d, Current: c, Exists: true}:
+				return nil
+			case <-ctx.Done():
+				return ctx.Err()
+			}
 		})
+	}
+	if err := g.Wait(); err != nil {
+		return nil, err
+	}
 
-		if err := e.update(ctx, c, d, parent, control, client.DryRunAll); err != nil {
-			return nil, err
-		}
+	close(out)
+	allObjs := []currentDesired{}
+	for obj := range out {
+		allObjs = append(allObjs, obj)
 	}
+	return allObjs, nil
+}
 
+func (e *APIEstablisher) establish(ctx context.Context, allObjs []currentDesired, parent client.Object, control bool) ([]xpv1.TypedReference, error) { // nolint:gocyclo
+	g, ctx := errgroup.WithContext(ctx)
+	g.SetLimit(maxConcurrentEstablishers)
+	out := make(chan xpv1.TypedReference, len(allObjs))
 	for _, cd := range allObjs {
-		if !cd.Exists {
-			// Only create a missing resource if we are going to control it.
-			// This prevents an inactive revision from racing to create a
-			// resource before an active revision of the same parent.
-			if control {
-				if err := e.create(ctx, cd.Desired, parent); err != nil {
-					return nil, err
+		cd := cd // Pin the loop variable.
+		g.Go(func() error {
+			if !cd.Exists {
+				// Only create a missing resource if we are going to control it.
+				// This prevents an inactive revision from racing to create a
+				// resource before an active revision of the same parent.
+				if control {
+					if err := e.create(ctx, cd.Desired, parent); err != nil {
+						return err
+					}
+				}
+				select {
+				case out <- *meta.TypedReferenceTo(cd.Desired, cd.Desired.GetObjectKind().GroupVersionKind()):
+					return nil
+				case <-ctx.Done():
+					return ctx.Err()
 				}
 			}
-			resourceRefs = append(resourceRefs, *meta.TypedReferenceTo(cd.Desired, cd.Desired.GetObjectKind().GroupVersionKind()))
-			continue
-		}
 
-		if err := e.update(ctx, cd.Current, cd.Desired, parent, control); err != nil {
-			return nil, err
-		}
-		resourceRefs = append(resourceRefs, *meta.TypedReferenceTo(cd.Desired, cd.Desired.GetObjectKind().GroupVersionKind()))
+			if err := e.update(ctx, cd.Current, cd.Desired, parent, control); err != nil {
+				return err
+			}
+			select {
+			case out <- *meta.TypedReferenceTo(cd.Desired, cd.Desired.GetObjectKind().GroupVersionKind()):
+				return nil
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		})
+	}
+	if err := g.Wait(); err != nil {
+		return nil, err
+	}
+	close(out)
+	resourceRefs := []xpv1.TypedReference{}
+	for ref := range out {
+		resourceRefs = append(resourceRefs, ref)
 	}
-
 	return resourceRefs, nil
 }
 



Test patch:
diff --git a/internal/controller/pkg/revision/establisher_test.go b/internal/controller/pkg/revision/establisher_test.go
index 8de62b779..95c70995a 100644
--- a/internal/controller/pkg/revision/establisher_test.go
+++ b/internal/controller/pkg/revision/establisher_test.go
@@ -21,6 +21,7 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	admv1 "k8s.io/api/admissionregistration/v1"
 	corev1 "k8s.io/api/core/v1"
 	extv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
@@ -402,7 +403,10 @@ func TestAPIEstablisherEstablish(t *testing.T) {
 			if diff := cmp.Diff(tc.want.err, err, test.EquateErrors()); diff != "" {
 				t.Errorf("\n%s\ne.Check(...): -want error, +got error:\n%s", tc.reason, diff)
 			}
-			if diff := cmp.Diff(tc.want.refs, refs, test.EquateErrors()); diff != "" {
+			sort := cmpopts.SortSlices(func(x, y xpv1.TypedReference) bool {
+				return x.Name < y.Name
+			})
+			if diff := cmp.Diff(tc.want.refs, refs, test.EquateErrors(), sort); diff != "" {
 				t.Errorf("\n%s\ne.Check(...): -want, +got:\n%s", tc.reason, diff)
 			}
 		})