controller: add the option for sync calls

Signed-off-by: Muvaffak Onus <me@muvaf.com>

go.mod

@@ -3,16 +3,14 @@ module github.com/crossplane-contrib/terrajet
 go 1.16
 
 require (
-	github.com/crossplane/crossplane-runtime v0.15.0
+	github.com/crossplane/crossplane-runtime v0.15.1-0.20210930095326-d5661210733b
 	github.com/google/go-cmp v0.5.6
 	github.com/hashicorp/terraform-plugin-sdk/v2 v2.7.0
 	github.com/iancoleman/strcase v0.2.0
 	github.com/json-iterator/go v1.1.11
 	github.com/muvaf/typewriter v0.0.0-20210910160850-80e49fe1eb32
 	github.com/pkg/errors v0.9.1
-	github.com/spf13/afero v1.6.0
-	go.uber.org/multierr v1.7.0
-	k8s.io/apimachinery v0.21.2
-	k8s.io/client-go v0.21.2
-	sigs.k8s.io/controller-runtime v0.9.2
+	go.uber.org/multierr v1.7.0 // indirect
+	k8s.io/apimachinery v0.21.3
+	sigs.k8s.io/controller-runtime v0.9.6
 )

pkg/controller/external.go

@@ -38,18 +38,32 @@ const (
 	errUnexpectedObject = "the managed resource is not an Terraformed resource"
 )
 
-// SetupFn is a function that returns Terraform setup which contains
-// provider requirement, configuration and Terraform version.
-type SetupFn func(ctx context.Context, client client.Client, mg xpresource.Managed) (tjclient.TerraformSetup, error)
+type Option func(*Connector)
+
+func WithLogger(l logging.Logger) Option {
+	return func(c *Connector) {
+		c.logger = l
+	}
+}
+
+func UseAsync() Option {
+	return func(c *Connector) {
+		c.async = true
+	}
+}
 
 // NewConnector returns a new Connector object.
-func NewConnector(kube client.Client, l logging.Logger, ws *tjclient.WorkspaceStore, sf SetupFn) *Connector {
-	return &Connector{
+func NewConnector(kube client.Client, ws *tjclient.WorkspaceStore, sf tjclient.SetupFn, opts ...Option) *Connector {
+	c := &Connector{
 		kube:           kube,
-		logger:         l,
+		logger:         logging.NewNopLogger(),
 		terraformSetup: sf,
 		store:          ws,
 	}
+	for _, f := range opts {
+		f(c)
+	}
+	return c
 }
 
 // Connector initializes the external client with credentials and other configuration
@@ -58,7 +72,8 @@ type Connector struct {
 	kube           client.Client
 	logger         logging.Logger
 	store          *tjclient.WorkspaceStore
-	terraformSetup SetupFn
+	terraformSetup tjclient.SetupFn
+	async          bool
 }
 
 // Connect makes sure the underlying client is ready to issue requests to the
@@ -80,16 +95,17 @@ func (c *Connector) Connect(ctx context.Context, mg xpresource.Managed) (managed
 	}
 
 	return &external{
-		kube:   c.kube,
-		tf:     tf,
-		log:    c.logger,
-		record: event.NewNopRecorder(),
+		kube:  c.kube,
+		tf:    tf,
+		log:   c.logger,
+		async: c.async,
 	}, nil
 }
 
 type external struct {
-	kube client.Client
-	tf   Client
+	kube  client.Client
+	tf    Client
+	async bool
 
 	log    logging.Logger
 	record event.Recorder
@@ -111,14 +127,6 @@ func (e *external) Observe(ctx context.Context, mg xpresource.Managed) (managed.
 			ResourceUpToDate: true,
 		}, nil
 	}
-
-	plan, err := e.tf.Plan(ctx)
-	if err != nil {
-		return managed.ExternalObservation{}, errors.Wrap(err, "cannot check resource status")
-	}
-	if !plan.Exists {
-		return managed.ExternalObservation{}, nil
-	}
 	// After a successful observation, we now have a state to consume.
 	// We will consume the state by:
 	// - returning "sensitive attributes" as connection details
@@ -148,6 +156,11 @@ func (e *external) Observe(ctx context.Context, mg xpresource.Managed) (managed.
 	// step completed (i.e. resource exists).
 	tr.SetConditions(xpv1.Available())
 
+	plan, err := e.tf.Plan(ctx)
+	if err != nil {
+		return managed.ExternalObservation{}, errors.Wrap(err, "cannot check resource status")
+	}
+
 	// TODO(muvaf): Handle connection details.
 	return managed.ExternalObservation{
 		ResourceExists:          true,
@@ -156,16 +169,38 @@ func (e *external) Observe(ctx context.Context, mg xpresource.Managed) (managed.
 	}, nil
 }
 
-func (e *external) Create(_ context.Context, _ xpresource.Managed) (managed.ExternalCreation, error) {
-	return managed.ExternalCreation{}, errors.Wrap(e.tf.ApplyAsync(), "cannot start async apply")
+func (e *external) Create(ctx context.Context, mg xpresource.Managed) (managed.ExternalCreation, error) {
+	res, err := e.Update(ctx, mg)
+	return managed.ExternalCreation{ConnectionDetails: res.ConnectionDetails}, err
 }
 
-func (e *external) Update(_ context.Context, _ xpresource.Managed) (managed.ExternalUpdate, error) {
-	return managed.ExternalUpdate{}, errors.Wrap(e.tf.ApplyAsync(), "cannot start async apply")
+func (e *external) Update(ctx context.Context, mg xpresource.Managed) (managed.ExternalUpdate, error) {
+	if e.async {
+		return managed.ExternalUpdate{}, errors.Wrap(e.tf.ApplyAsync(), "cannot start async apply")
+	}
+	tr, ok := mg.(resource.Terraformed)
+	if !ok {
+		return managed.ExternalUpdate{}, errors.New(errUnexpectedObject)
+	}
+	res, err := e.tf.Apply(ctx)
+	if err != nil {
+		return managed.ExternalUpdate{}, errors.Wrap(err, "cannot create")
+	}
+	attr := map[string]interface{}{}
+	if err := json.JSParser.Unmarshal(res.State.GetAttributes(), &attr); err != nil {
+		return managed.ExternalUpdate{}, errors.Wrap(err, "cannot unmarshal state attributes")
+	}
+	// NOTE(muvaf): Status is lost after the result of creation. That's why we
+	// do not set observation.
+	_, err = lateInitializeAnnotations(tr, attr, string(res.State.GetPrivateRaw()))
+	return managed.ExternalUpdate{}, errors.Wrap(err, "cannot late initialize annotations")
 }
 
-func (e *external) Delete(_ context.Context, _ xpresource.Managed) error {
-	return errors.Wrap(e.tf.DestroyAsync(), "cannot start async destroy")
+func (e *external) Delete(ctx context.Context, _ xpresource.Managed) error {
+	if e.async {
+		return errors.Wrap(e.tf.DestroyAsync(), "cannot start async destroy")
+	}
+	return errors.Wrap(e.tf.Destroy(ctx), "cannot destroy")
 }
 
 func lateInitializeAnnotations(tr resource.Terraformed, attr map[string]interface{}, privateRaw string) (bool, error) {

pkg/pipeline/templates/controller.go.tmpl

@@ -21,11 +21,16 @@ import (
 )
 
 // Setup adds a controller that reconciles {{ .CRD.Kind }} managed resources.
-func Setup(mgr ctrl.Manager, l logging.Logger, rl workqueue.RateLimiter, ps tjcontroller.SetupFn, ws *terraform.WorkspaceStore, concurrency int) error {
+func Setup(mgr ctrl.Manager, l logging.Logger, rl workqueue.RateLimiter, s terraform.SetupFn, ws *terraform.WorkspaceStore, concurrency int) error {
 	name := managed.ControllerName({{ .TypePackageAlias }}{{ .CRD.Kind }}GroupVersionKind.String())
 	r := managed.NewReconciler(mgr,
 		xpresource.ManagedKind({{ .TypePackageAlias }}{{ .CRD.Kind }}GroupVersionKind),
-		managed.WithExternalConnecter(tjcontroller.NewConnector(mgr.GetClient(), l, ws, ps)),
+		managed.WithExternalConnecter(tjcontroller.NewConnector(mgr.GetClient(), ws, s,
+			tjcontroller.WithLogger(l),
+			{{- if .UseAsync }}
+			tjcontroller.UseAsync(),
+			{{- end}}
+		)),
 		managed.WithLogger(l.WithValues("controller", name)),
 		managed.WithRecorder(event.NewAPIRecorder(mgr.GetEventRecorderFor(name))),
 		managed.WithFinalizer(terraform.NewWorkspaceFinalizer(ws, xpresource.NewAPIFinalizer(mgr.GetClient(), "finalizer.managedresource.crossplane.io"))),

pkg/resource/configuration.go

@@ -101,4 +101,9 @@ type Configuration struct {
 	// the resource. Its default is "id" and in almost all cases, you don't need
 	// to overwrite it.
 	TerraformIDFieldName string
+
+	// UseAsync should be enabled for resource whose creation and/or deletion
+	// takes more than 1 minute to complete such as Kubernetes clusters or
+	// databases.
+	UseAsync bool
 }

pkg/terraform/files.go

@@ -42,7 +42,7 @@ const (
 )
 
 // NewFileProducer returns a new FileProducer.
-func NewFileProducer(tr resource.Terraformed, ts TerraformSetup) (*FileProducer, error) {
+func NewFileProducer(tr resource.Terraformed, ts Setup) (*FileProducer, error) {
 	params, err := tr.GetParameters()
 	if err != nil {
 		return nil, errors.Wrap(err, "cannot get parameters")
@@ -63,7 +63,7 @@ func NewFileProducer(tr resource.Terraformed, ts TerraformSetup) (*FileProducer,
 // every time like parameters and observation maps.
 type FileProducer struct {
 	Resource resource.Terraformed
-	Setup    TerraformSetup
+	Setup    Setup
 
 	parameters  map[string]interface{}
 	observation map[string]interface{}

pkg/terraform/store.go

@@ -23,17 +23,20 @@ import (
 	"path/filepath"
 	"sync"
 
-	"k8s.io/apimachinery/pkg/types"
-
 	"github.com/crossplane/crossplane-runtime/pkg/logging"
+	xpresource "github.com/crossplane/crossplane-runtime/pkg/resource"
+	"github.com/pkg/errors"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/crossplane-contrib/terrajet/pkg/resource"
 	"github.com/crossplane-contrib/terrajet/pkg/resource/json"
-
-	xpresource "github.com/crossplane/crossplane-runtime/pkg/resource"
-	"github.com/pkg/errors"
 )
 
+// SetupFn is a function that returns Terraform setup which contains
+// provider requirement, configuration and Terraform version.
+type SetupFn func(ctx context.Context, client client.Client, mg xpresource.Managed) (Setup, error)
+
 // ProviderRequirement holds values for the Terraform HCL setup requirements
 type ProviderRequirement struct {
 	Source  string
@@ -43,15 +46,15 @@ type ProviderRequirement struct {
 // ProviderConfiguration holds the setup configuration body
 type ProviderConfiguration map[string]interface{}
 
-// TerraformSetup holds values for the Terraform version and setup
+// Setup holds values for the Terraform version and setup
 // requirements and configuration body
-type TerraformSetup struct {
+type Setup struct {
 	Version       string
 	Requirement   ProviderRequirement
 	Configuration ProviderConfiguration
 }
 
-func (p TerraformSetup) validate() error {
+func (p Setup) validate() error {
 	if p.Version == "" {
 		return errors.New(fmtErrValidationVersion)
 	}
@@ -76,7 +79,7 @@ type WorkspaceStore struct {
 
 // TODO(muvaf): Take EnqueueFn as parameter tow WorkspaceStore?
 
-func (ws *WorkspaceStore) Workspace(ctx context.Context, tr resource.Terraformed, ts TerraformSetup, l logging.Logger, _ EnqueueFn) (*Workspace, error) {
+func (ws *WorkspaceStore) Workspace(ctx context.Context, tr resource.Terraformed, ts Setup, l logging.Logger, _ EnqueueFn) (*Workspace, error) {
 	dir := filepath.Join(os.TempDir(), string(tr.GetUID()))
 	fp, err := NewFileProducer(tr, ts)
 	if err != nil {