Improve resource ref unit tests. (#5960)

- Add component ref coverage to the existing test
- Add coverage for a downlevel SDK communicating with an engine that
  supports resource refs
- Add coverage for a downlevel engine communicating with an SDK that
  supports resource refs

As part of improving coverage, these changes add a knob to explicitly
disable resource refs in the engine without the use of the environment
variable. The environment variable is now only read by the CLI, and has
been restored to its prior polarity (i.e. `PULUMI_ENABLE_RESOURCE_REFERENCES`).

CHANGELOG.md

@@ -3,6 +3,10 @@ CHANGELOG
 
 ## HEAD (Unreleased)
 
+- Fix a bug in the core engine that could cause resources references to marshal improperly
+  during preview.
+  [#5960](https://github.com/pulumi/pulumi/pull/5960)
+
 - [sdk/dotnet] Add collection initializers for smooth support of Union<T, U> as element type
   [#5938](https://github.com/pulumi/pulumi/pull/5938)
 

pkg/cmd/pulumi/destroy.go

@@ -123,13 +123,14 @@ func newDestroyCmd() *cobra.Command {
 			}
 
 			opts.Engine = engine.UpdateOptions{
-				Parallel:               parallel,
-				Debug:                  debug,
-				Refresh:                refresh,
-				DestroyTargets:         targetUrns,
-				TargetDependents:       targetDependents,
-				UseLegacyDiff:          useLegacyDiff(),
-				DisableProviderPreview: disableProviderPreview(),
+				Parallel:                  parallel,
+				Debug:                     debug,
+				Refresh:                   refresh,
+				DestroyTargets:            targetUrns,
+				TargetDependents:          targetDependents,
+				UseLegacyDiff:             useLegacyDiff(),
+				DisableProviderPreview:    disableProviderPreview(),
+				DisableResourceReferences: disableResourceReferences(),
 			}
 
 			_, res := s.Destroy(commandContext(), backend.UpdateOperation{

pkg/cmd/pulumi/preview.go

@@ -144,15 +144,16 @@ func newPreviewCmd() *cobra.Command {
 
 			opts := backend.UpdateOptions{
 				Engine: engine.UpdateOptions{
-					LocalPolicyPacks:       engine.MakeLocalPolicyPacks(policyPackPaths, policyPackConfigPaths),
-					Parallel:               parallel,
-					Debug:                  debug,
-					Refresh:                refresh,
-					ReplaceTargets:         replaceURNs,
-					UseLegacyDiff:          useLegacyDiff(),
-					DisableProviderPreview: disableProviderPreview(),
-					UpdateTargets:          targetURNs,
-					TargetDependents:       targetDependents,
+					LocalPolicyPacks:          engine.MakeLocalPolicyPacks(policyPackPaths, policyPackConfigPaths),
+					Parallel:                  parallel,
+					Debug:                     debug,
+					Refresh:                   refresh,
+					ReplaceTargets:            replaceURNs,
+					UseLegacyDiff:             useLegacyDiff(),
+					DisableProviderPreview:    disableProviderPreview(),
+					DisableResourceReferences: disableResourceReferences(),
+					UpdateTargets:             targetURNs,
+					TargetDependents:          targetDependents,
 				},
 				Display: displayOpts,
 			}

pkg/cmd/pulumi/refresh.go

@@ -122,11 +122,12 @@ func newRefreshCmd() *cobra.Command {
 			}
 
 			opts.Engine = engine.UpdateOptions{
-				Parallel:               parallel,
-				Debug:                  debug,
-				UseLegacyDiff:          useLegacyDiff(),
-				DisableProviderPreview: disableProviderPreview(),
-				RefreshTargets:         targetUrns,
+				Parallel:                  parallel,
+				Debug:                     debug,
+				UseLegacyDiff:             useLegacyDiff(),
+				DisableProviderPreview:    disableProviderPreview(),
+				DisableResourceReferences: disableResourceReferences(),
+				RefreshTargets:            targetUrns,
 			}
 
 			changes, res := s.Refresh(commandContext(), backend.UpdateOperation{

pkg/cmd/pulumi/up.go

@@ -122,16 +122,17 @@ func newUpCmd() *cobra.Command {
 		}
 
 		opts.Engine = engine.UpdateOptions{
-			LocalPolicyPacks:       engine.MakeLocalPolicyPacks(policyPackPaths, policyPackConfigPaths),
-			Parallel:               parallel,
-			Debug:                  debug,
-			Refresh:                refresh,
-			RefreshTargets:         targetURNs,
-			ReplaceTargets:         replaceURNs,
-			UseLegacyDiff:          useLegacyDiff(),
-			DisableProviderPreview: disableProviderPreview(),
-			UpdateTargets:          targetURNs,
-			TargetDependents:       targetDependents,
+			LocalPolicyPacks:          engine.MakeLocalPolicyPacks(policyPackPaths, policyPackConfigPaths),
+			Parallel:                  parallel,
+			Debug:                     debug,
+			Refresh:                   refresh,
+			RefreshTargets:            targetURNs,
+			ReplaceTargets:            replaceURNs,
+			UseLegacyDiff:             useLegacyDiff(),
+			DisableProviderPreview:    disableProviderPreview(),
+			DisableResourceReferences: disableResourceReferences(),
+			UpdateTargets:             targetURNs,
+			TargetDependents:          targetDependents,
 		}
 
 		changes, res := s.Update(commandContext(), backend.UpdateOperation{

pkg/cmd/pulumi/util.go

@@ -71,6 +71,14 @@ func disableProviderPreview() bool {
 	return cmdutil.IsTruthy(os.Getenv("PULUMI_DISABLE_PROVIDER_PREVIEW"))
 }
 
+func disableResourceReferences() bool {
+	// Allow the resource reference feature to be enabled by explicitly setting an env var.
+	if v, ok := os.LookupEnv("PULUMI_ENABLE_RESOURCE_REFERENCES"); ok && cmdutil.IsTruthy(v) {
+		return false
+	}
+	return true
+}
+
 // skipConfirmations returns whether or not confirmation prompts should
 // be skipped. This should be used by pass any requirement that a --yes
 // parameter has been set for non-interactive scenarios.

pkg/cmd/pulumi/watch.go

@@ -114,12 +114,13 @@ func newWatchCmd() *cobra.Command {
 			}
 
 			opts.Engine = engine.UpdateOptions{
-				LocalPolicyPacks:       engine.MakeLocalPolicyPacks(policyPackPaths, policyPackConfigPaths),
-				Parallel:               parallel,
-				Debug:                  debug,
-				Refresh:                refresh,
-				UseLegacyDiff:          useLegacyDiff(),
-				DisableProviderPreview: disableProviderPreview(),
+				LocalPolicyPacks:          engine.MakeLocalPolicyPacks(policyPackPaths, policyPackConfigPaths),
+				Parallel:                  parallel,
+				Debug:                     debug,
+				Refresh:                   refresh,
+				UseLegacyDiff:             useLegacyDiff(),
+				DisableProviderPreview:    disableProviderPreview(),
+				DisableResourceReferences: disableResourceReferences(),
 			}
 
 			res := s.Watch(commandContext(), backend.UpdateOperation{

pkg/engine/deployment.go

@@ -236,17 +236,18 @@ func (deployment *deployment) run(cancelCtx *Context, actions runActions, policy
 	var walkResult result.Result
 	go func() {
 		opts := deploy.Options{
-			Events:            actions,
-			Parallel:          deployment.Options.Parallel,
-			Refresh:           deployment.Options.Refresh,
-			RefreshOnly:       deployment.Options.isRefresh,
-			RefreshTargets:    deployment.Options.RefreshTargets,
-			ReplaceTargets:    deployment.Options.ReplaceTargets,
-			DestroyTargets:    deployment.Options.DestroyTargets,
-			UpdateTargets:     deployment.Options.UpdateTargets,
-			TargetDependents:  deployment.Options.TargetDependents,
-			TrustDependencies: deployment.Options.trustDependencies,
-			UseLegacyDiff:     deployment.Options.UseLegacyDiff,
+			Events:                    actions,
+			Parallel:                  deployment.Options.Parallel,
+			Refresh:                   deployment.Options.Refresh,
+			RefreshOnly:               deployment.Options.isRefresh,
+			RefreshTargets:            deployment.Options.RefreshTargets,
+			ReplaceTargets:            deployment.Options.ReplaceTargets,
+			DestroyTargets:            deployment.Options.DestroyTargets,
+			UpdateTargets:             deployment.Options.UpdateTargets,
+			TargetDependents:          deployment.Options.TargetDependents,
+			TrustDependencies:         deployment.Options.trustDependencies,
+			UseLegacyDiff:             deployment.Options.UseLegacyDiff,
+			DisableResourceReferences: deployment.Options.DisableResourceReferences,
 		}
 		walkResult = deployment.Deployment.Execute(ctx, opts, preview)
 		close(done)

pkg/engine/lifeycletest/pulumi_test.go

@@ -1689,57 +1689,6 @@ func TestSingleComponentDefaultProviderLifecycle(t *testing.T) {
 	p.Run(t, nil)
 }
 
-func TestResourceReferences(t *testing.T) {
-	var urnA resource.URN
-	var idA resource.ID
-
-	loaders := []*deploytest.ProviderLoader{
-		deploytest.NewProviderLoader("pkgA", semver.MustParse("1.0.0"), func() (plugin.Provider, error) {
-			v := &deploytest.Provider{
-				CreateF: func(urn resource.URN, news resource.PropertyMap,
-					timeout float64, preview bool) (resource.ID, resource.PropertyMap, resource.Status, error) {
-
-					if urn.Name() == "resB" {
-						propA, ok := news["resA"]
-						assert.True(t, ok)
-						assert.Equal(t, resource.MakeResourceReference(urnA, idA, true, ""), propA)
-					}
-
-					return "created-id", news, resource.StatusOK, nil
-				},
-				ReadF: func(urn resource.URN, id resource.ID,
-					inputs, state resource.PropertyMap) (plugin.ReadResult, resource.Status, error) {
-					return plugin.ReadResult{Inputs: inputs, Outputs: state}, resource.StatusOK, nil
-				},
-			}
-			return v, nil
-		}),
-	}
-
-	program := deploytest.NewLanguageRuntime(func(_ plugin.RunInfo, monitor *deploytest.ResourceMonitor) error {
-		var err error
-		urnA, idA, _, err = monitor.RegisterResource("pkgA:m:typA", "resA", true)
-		assert.NoError(t, err)
-
-		ins := resource.PropertyMap{
-			"resA": resource.MakeResourceReference(urnA, idA, true, ""),
-		}
-		_, _, props, err := monitor.RegisterResource("pkgA:m:typA", "resB", true, deploytest.ResourceOptions{
-			Inputs: ins,
-		})
-		assert.NoError(t, err)
-		assert.Equal(t, ins, props)
-		return nil
-	})
-	host := deploytest.NewPluginHost(nil, nil, program, loaders...)
-
-	p := &TestPlan{
-		Options: UpdateOptions{Host: host},
-		Steps:   MakeBasicLifecycleSteps(t, 3),
-	}
-	p.Run(t, nil)
-}
-
 type updateContext struct {
 	*deploytest.ResourceMonitor
 

pkg/engine/lifeycletest/resource_reference_test.go

@@ -0,0 +1,221 @@
+//nolint:goconst
+package lifecycletest
+
+import (
+	"testing"
+
+	"github.com/blang/semver"
+	"github.com/stretchr/testify/assert"
+
+	. "github.com/pulumi/pulumi/pkg/v2/engine"
+	"github.com/pulumi/pulumi/pkg/v2/resource/deploy/deploytest"
+	"github.com/pulumi/pulumi/sdk/v2/go/common/resource"
+	"github.com/pulumi/pulumi/sdk/v2/go/common/resource/plugin"
+)
+
+// TestResourceReferences tests that resource references can be marshaled between the engine, language host,
+// resource providers, and statefile if each entity supports resource references.
+func TestResourceReferences(t *testing.T) {
+	var urnA resource.URN
+	var urnB resource.URN
+	var idB resource.ID
+
+	loaders := []*deploytest.ProviderLoader{
+		deploytest.NewProviderLoader("pkgA", semver.MustParse("1.0.0"), func() (plugin.Provider, error) {
+			v := &deploytest.Provider{
+				CreateF: func(urn resource.URN, news resource.PropertyMap,
+					timeout float64, preview bool) (resource.ID, resource.PropertyMap, resource.Status, error) {
+
+					id := "created-id"
+					if preview {
+						id = ""
+					}
+
+					if urn.Name() == "resC" {
+						assert.True(t, news.DeepEquals(resource.PropertyMap{
+							"resA": resource.MakeResourceReference(urnA, "", false, ""),
+							"resB": resource.MakeResourceReference(urnB, idB, true, ""),
+						}))
+					}
+
+					return resource.ID(id), news, resource.StatusOK, nil
+				},
+				ReadF: func(urn resource.URN, id resource.ID,
+					inputs, state resource.PropertyMap) (plugin.ReadResult, resource.Status, error) {
+					return plugin.ReadResult{Inputs: inputs, Outputs: state}, resource.StatusOK, nil
+				},
+			}
+			return v, nil
+		}),
+	}
+
+	program := deploytest.NewLanguageRuntime(func(_ plugin.RunInfo, monitor *deploytest.ResourceMonitor) error {
+		var err error
+		urnA, _, _, err = monitor.RegisterResource("component", "resA", false)
+		assert.NoError(t, err)
+
+		err = monitor.RegisterResourceOutputs(urnA, resource.PropertyMap{})
+		assert.NoError(t, err)
+
+		urnB, idB, _, err = monitor.RegisterResource("pkgA:m:typA", "resB", true)
+		assert.NoError(t, err)
+
+		_, _, props, err := monitor.RegisterResource("pkgA:m:typA", "resC", true, deploytest.ResourceOptions{
+			Inputs: resource.PropertyMap{
+				"resA": resource.MakeResourceReference(urnA, "", false, ""),
+				"resB": resource.MakeResourceReference(urnB, idB, true, ""),
+			},
+		})
+		assert.NoError(t, err)
+
+		assert.True(t, props.DeepEquals(resource.PropertyMap{
+			"resA": resource.MakeResourceReference(urnA, "", false, ""),
+			"resB": resource.MakeResourceReference(urnB, idB, true, ""),
+		}))
+		return nil
+	})
+	host := deploytest.NewPluginHost(nil, nil, program, loaders...)
+
+	p := &TestPlan{
+		Options: UpdateOptions{Host: host},
+		Steps:   MakeBasicLifecycleSteps(t, 4),
+	}
+	p.Run(t, nil)
+}
+
+// TestResourceReferences_DownlevelSDK tests that resource references are properly marshaled as URNs (for references to
+// component resources) or IDs (for references to custom resources) if the SDK does not support resource references.
+func TestResourceReferences_DownlevelSDK(t *testing.T) {
+	var urnA resource.URN
+	var urnB resource.URN
+	var idB resource.ID
+
+	loaders := []*deploytest.ProviderLoader{
+		deploytest.NewProviderLoader("pkgA", semver.MustParse("1.0.0"), func() (plugin.Provider, error) {
+			v := &deploytest.Provider{
+				CreateF: func(urn resource.URN, news resource.PropertyMap,
+					timeout float64, preview bool) (resource.ID, resource.PropertyMap, resource.Status, error) {
+
+					id := "created-id"
+					if preview {
+						id = ""
+					}
+
+					state := resource.PropertyMap{}
+					if urn.Name() == "resC" {
+						state = resource.PropertyMap{
+							"resA": resource.MakeResourceReference(urnA, "", false, ""),
+							"resB": resource.MakeResourceReference(urnB, idB, true, ""),
+						}
+					}
+
+					return resource.ID(id), state, resource.StatusOK, nil
+				},
+				ReadF: func(urn resource.URN, id resource.ID,
+					inputs, state resource.PropertyMap) (plugin.ReadResult, resource.Status, error) {
+					return plugin.ReadResult{Inputs: inputs, Outputs: state}, resource.StatusOK, nil
+				},
+			}
+			return v, nil
+		}),
+	}
+
+	opts := deploytest.ResourceOptions{DisableResourceReferences: true}
+	program := deploytest.NewLanguageRuntime(func(_ plugin.RunInfo, monitor *deploytest.ResourceMonitor) error {
+		var err error
+		urnA, _, _, err = monitor.RegisterResource("component", "resA", false, opts)
+		assert.NoError(t, err)
+
+		err = monitor.RegisterResourceOutputs(urnA, resource.PropertyMap{})
+		assert.NoError(t, err)
+
+		urnB, idB, _, err = monitor.RegisterResource("pkgA:m:typA", "resB", true, opts)
+		assert.NoError(t, err)
+
+		_, _, props, err := monitor.RegisterResource("pkgA:m:typA", "resC", true, opts)
+		assert.NoError(t, err)
+
+		assert.True(t, props.DeepEquals(resource.PropertyMap{
+			"resA": resource.NewStringProperty(string(urnA)),
+			"resB": resource.NewStringProperty(string(idB)),
+		}))
+		return nil
+	})
+	host := deploytest.NewPluginHost(nil, nil, program, loaders...)
+
+	p := &TestPlan{
+		Options: UpdateOptions{Host: host},
+		Steps:   MakeBasicLifecycleSteps(t, 4),
+	}
+	p.Run(t, nil)
+}
+
+// TestResourceReferences_DownlevelEngine tests an SDK that supports resource references communicating with an engine
+// that does not.
+func TestResourceReferences_DownlevelEngine(t *testing.T) {
+	var urnA resource.URN
+	var urnB resource.URN
+	var idB resource.ID
+
+	loaders := []*deploytest.ProviderLoader{
+		deploytest.NewProviderLoader("pkgA", semver.MustParse("1.0.0"), func() (plugin.Provider, error) {
+			v := &deploytest.Provider{
+				CreateF: func(urn resource.URN, news resource.PropertyMap,
+					timeout float64, preview bool) (resource.ID, resource.PropertyMap, resource.Status, error) {
+
+					id := "created-id"
+					if preview {
+						id = ""
+					}
+
+					// If we have resource references here, the engine has not properly disabled them.
+					if urn.Name() == "resC" {
+						assert.Equal(t, resource.NewStringProperty(string(urnA)), news["resA"])
+						assert.Equal(t, resource.NewStringProperty(string(idB)), news["resB"])
+					}
+
+					return resource.ID(id), news, resource.StatusOK, nil
+				},
+				ReadF: func(urn resource.URN, id resource.ID,
+					inputs, state resource.PropertyMap) (plugin.ReadResult, resource.Status, error) {
+					return plugin.ReadResult{Inputs: inputs, Outputs: state}, resource.StatusOK, nil
+				},
+			}
+			return v, nil
+		}),
+	}
+
+	program := deploytest.NewLanguageRuntime(func(_ plugin.RunInfo, monitor *deploytest.ResourceMonitor) error {
+		var err error
+		urnA, _, _, err = monitor.RegisterResource("component", "resA", false)
+		assert.NoError(t, err)
+
+		err = monitor.RegisterResourceOutputs(urnA, resource.PropertyMap{})
+		assert.NoError(t, err)
+
+		urnB, idB, _, err = monitor.RegisterResource("pkgA:m:typA", "resB", true)
+		assert.NoError(t, err)
+
+		_, _, props, err := monitor.RegisterResource("pkgA:m:typA", "resC", true, deploytest.ResourceOptions{
+			Inputs: resource.PropertyMap{
+				"resA": resource.MakeResourceReference(urnA, "", false, ""),
+				"resB": resource.MakeResourceReference(urnB, idB, true, ""),
+			},
+		})
+		assert.NoError(t, err)
+
+		assert.True(t, props.DeepEquals(resource.PropertyMap{
+			"resA": resource.NewStringProperty(string(urnA)),
+			"resB": resource.NewStringProperty(string(idB)),
+		}))
+		return nil
+	})
+
+	host := deploytest.NewPluginHost(nil, nil, program, loaders...)
+
+	p := &TestPlan{
+		Options: UpdateOptions{Host: host, DisableResourceReferences: true},
+		Steps:   MakeBasicLifecycleSteps(t, 4),
+	}
+	p.Run(t, nil)
+}

pkg/engine/update.go

@@ -134,6 +134,9 @@ type UpdateOptions struct {
 	// true if the engine should disable provider previews.
 	DisableProviderPreview bool
 
+	// true if the engine should disable resource reference support.
+	DisableResourceReferences bool
+
 	// true if we should report events for steps that involve default providers.
 	reportDefaultProviderSteps bool
 

pkg/resource/deploy/deployment.go

@@ -49,17 +49,18 @@ type BackendClient interface {
 
 // Options controls the deployment process.
 type Options struct {
-	Events            Events         // an optional events callback interface.
-	Parallel          int            // the degree of parallelism for resource operations (<=1 for serial).
-	Refresh           bool           // whether or not to refresh before executing the deployment.
-	RefreshOnly       bool           // whether or not to exit after refreshing.
-	RefreshTargets    []resource.URN // The specific resources to refresh during a refresh op.
-	ReplaceTargets    []resource.URN // Specific resources to replace.
-	DestroyTargets    []resource.URN // Specific resources to destroy.
-	UpdateTargets     []resource.URN // Specific resources to update.
-	TargetDependents  bool           // true if we're allowing things to proceed, even with unspecified targets
-	TrustDependencies bool           // whether or not to trust the resource dependency graph.
-	UseLegacyDiff     bool           // whether or not to use legacy diffing behavior.
+	Events                    Events         // an optional events callback interface.
+	Parallel                  int            // the degree of parallelism for resource operations (<=1 for serial).
+	Refresh                   bool           // whether or not to refresh before executing the deployment.
+	RefreshOnly               bool           // whether or not to exit after refreshing.
+	RefreshTargets            []resource.URN // The specific resources to refresh during a refresh op.
+	ReplaceTargets            []resource.URN // Specific resources to replace.
+	DestroyTargets            []resource.URN // Specific resources to destroy.
+	UpdateTargets             []resource.URN // Specific resources to update.
+	TargetDependents          bool           // true if we're allowing things to proceed, even with unspecified targets
+	TrustDependencies         bool           // whether or not to trust the resource dependency graph.
+	UseLegacyDiff             bool           // whether or not to use legacy diffing behavior.
+	DisableResourceReferences bool           // true to disable resource reference support.
 }
 
 // DegreeOfParallelism returns the degree of parallelism that should be used during the

pkg/resource/deploy/deploytest/languageruntime.go

@@ -15,6 +15,8 @@
 package deploytest
 
 import (
+	"context"
+
 	"github.com/pulumi/pulumi/sdk/v2/go/common/resource/plugin"
 	"github.com/pulumi/pulumi/sdk/v2/go/common/util/contract"
 	"github.com/pulumi/pulumi/sdk/v2/go/common/workspace"
@@ -43,7 +45,7 @@ func (p *languageRuntime) GetRequiredPlugins(info plugin.ProgInfo) ([]workspace.
 }
 
 func (p *languageRuntime) Run(info plugin.RunInfo) (string, bool, error) {
-	monitor, err := dialMonitor(info.MonitorAddress)
+	monitor, err := dialMonitor(context.Background(), info.MonitorAddress)
 	if err != nil {
 		return "", false, err
 	}

pkg/resource/deploy/deploytest/provider.go

@@ -15,6 +15,7 @@
 package deploytest
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/blang/semver"
@@ -176,7 +177,7 @@ func (prov *Provider) Construct(info plugin.ConstructInfo, typ tokens.Type, name
 	if prov.ConstructF == nil {
 		return plugin.ConstructResult{}, nil
 	}
-	monitor, err := dialMonitor(info.MonitorAddress)
+	monitor, err := dialMonitor(context.Background(), info.MonitorAddress)
 	if err != nil {
 		return plugin.ConstructResult{}, err
 	}

pkg/resource/deploy/deploytest/resourcemonitor.go

@@ -22,6 +22,7 @@ import (
 	"github.com/pulumi/pulumi/sdk/v2/go/common/resource"
 	"github.com/pulumi/pulumi/sdk/v2/go/common/resource/plugin"
 	"github.com/pulumi/pulumi/sdk/v2/go/common/tokens"
+	"github.com/pulumi/pulumi/sdk/v2/go/common/util/contract"
 	"github.com/pulumi/pulumi/sdk/v2/go/common/util/rpcutil"
 	pulumirpc "github.com/pulumi/pulumi/sdk/v2/proto/go"
 	"google.golang.org/grpc"
@@ -30,9 +31,12 @@ import (
 type ResourceMonitor struct {
 	conn   *grpc.ClientConn
 	resmon pulumirpc.ResourceMonitorClient
+
+	supportsSecrets            bool
+	supportsResourceReferences bool
 }
 
-func dialMonitor(endpoint string) (*ResourceMonitor, error) {
+func dialMonitor(ctx context.Context, endpoint string) (*ResourceMonitor, error) {
 	// Connect to the resource monitor and create an appropriate client.
 	conn, err := grpc.Dial(
 		endpoint,
@@ -42,14 +46,37 @@ func dialMonitor(endpoint string) (*ResourceMonitor, error) {
 	if err != nil {
 		return nil, errors.Wrapf(err, "could not connect to resource monitor")
 	}
+	resmon := pulumirpc.NewResourceMonitorClient(conn)
+
+	// Check feature support.
+	supportsSecrets, err := supportsFeature(ctx, resmon, "secrets")
+	if err != nil {
+		contract.IgnoreError(conn.Close())
+		return nil, err
+	}
+	supportsResourceReferences, err := supportsFeature(ctx, resmon, "resourceReferences")
+	if err != nil {
+		contract.IgnoreError(conn.Close())
+		return nil, err
+	}
 
 	// Fire up a resource monitor client and return.
 	return &ResourceMonitor{
-		conn:   conn,
-		resmon: pulumirpc.NewResourceMonitorClient(conn),
+		conn:                       conn,
+		resmon:                     resmon,
+		supportsSecrets:            supportsSecrets,
+		supportsResourceReferences: supportsResourceReferences,
 	}, nil
 }
 
+func supportsFeature(ctx context.Context, resmon pulumirpc.ResourceMonitorClient, id string) (bool, error) {
+	resp, err := resmon.SupportsFeature(ctx, &pulumirpc.SupportsFeatureRequest{Id: id})
+	if err != nil {
+		return false, err
+	}
+	return resp.GetHasSupport(), nil
+}
+
 func (rm *ResourceMonitor) Close() error {
 	return rm.conn.Close()
 }
@@ -73,6 +100,9 @@ type ResourceOptions struct {
 	CustomTimeouts        *resource.CustomTimeouts
 	SupportsPartialValues *bool
 	Remote                bool
+
+	DisableSecrets            bool
+	DisableResourceReferences bool
 }
 
 func (rm *ResourceMonitor) RegisterResource(t tokens.Type, name string, custom bool,
@@ -89,7 +119,8 @@ func (rm *ResourceMonitor) RegisterResource(t tokens.Type, name string, custom b
 	// marshal inputs
 	ins, err := plugin.MarshalProperties(opts.Inputs, plugin.MarshalOptions{
 		KeepUnknowns:  true,
-		KeepResources: true,
+		KeepSecrets:   rm.supportsSecrets,
+		KeepResources: rm.supportsResourceReferences,
 	})
 	if err != nil {
 		return "", "", nil, err
@@ -146,8 +177,8 @@ func (rm *ResourceMonitor) RegisterResource(t tokens.Type, name string, custom b
 		DeleteBeforeReplace:        deleteBeforeReplace,
 		DeleteBeforeReplaceDefined: opts.DeleteBeforeReplace != nil,
 		IgnoreChanges:              opts.IgnoreChanges,
-		AcceptSecrets:              true,
-		AcceptResources:            true,
+		AcceptSecrets:              !opts.DisableSecrets,
+		AcceptResources:            !opts.DisableResourceReferences,
 		Version:                    opts.Version,
 		Aliases:                    aliasStrings,
 		ImportId:                   string(opts.ImportID),
@@ -162,8 +193,13 @@ func (rm *ResourceMonitor) RegisterResource(t tokens.Type, name string, custom b
 		return "", "", nil, err
 	}
 	// unmarshal outputs
+	//
+	// Note that `KeepSecrets` and `KeepResources` are set to `true` so the caller can detect secrets and resource refs
+	// that are erroneously returned (e.g. secrets/resource refs that are returned even though the caller has not set
+	// `AcceptSecrets` or `AcceptResources` to `true` above).
 	outs, err := plugin.UnmarshalProperties(resp.Object, plugin.MarshalOptions{
 		KeepUnknowns:  true,
+		KeepSecrets:   true,
 		KeepResources: true,
 	})
 	if err != nil {

pkg/resource/deploy/source_eval.go

@@ -17,7 +17,6 @@ package deploy
 import (
 	"context"
 	"fmt"
-	"os"
 	"time"
 
 	"github.com/blang/semver"
@@ -32,7 +31,6 @@ import (
 	"github.com/pulumi/pulumi/sdk/v2/go/common/resource/config"
 	"github.com/pulumi/pulumi/sdk/v2/go/common/resource/plugin"
 	"github.com/pulumi/pulumi/sdk/v2/go/common/tokens"
-	"github.com/pulumi/pulumi/sdk/v2/go/common/util/cmdutil"
 	"github.com/pulumi/pulumi/sdk/v2/go/common/util/contract"
 	"github.com/pulumi/pulumi/sdk/v2/go/common/util/logging"
 	"github.com/pulumi/pulumi/sdk/v2/go/common/util/result"
@@ -392,14 +390,15 @@ func (d *defaultProviders) getDefaultProviderRef(req providers.ProviderRequest)
 // resmon implements the pulumirpc.ResourceMonitor interface and acts as the gateway between a language runtime's
 // evaluation of a program and the internal resource planning and deployment logic.
 type resmon struct {
-	providers        ProviderSource                     // the provider source itself.
-	defaultProviders *defaultProviders                  // the default provider manager.
-	constructInfo    plugin.ConstructInfo               // information for construct calls.
-	regChan          chan *registerResourceEvent        // the channel to send resource registrations to.
-	regOutChan       chan *registerResourceOutputsEvent // the channel to send resource output registrations to.
-	regReadChan      chan *readResourceEvent            // the channel to send resource reads to.
-	cancel           chan bool                          // a channel that can cancel the server.
-	done             chan error                         // a channel that resolves when the server completes.
+	providers                 ProviderSource                     // the provider source itself.
+	defaultProviders          *defaultProviders                  // the default provider manager.
+	constructInfo             plugin.ConstructInfo               // information for construct calls.
+	regChan                   chan *registerResourceEvent        // the channel to send resource registrations to.
+	regOutChan                chan *registerResourceOutputsEvent // the channel to send resource output registrations to.
+	regReadChan               chan *readResourceEvent            // the channel to send resource reads to.
+	cancel                    chan bool                          // a channel that can cancel the server.
+	done                      chan error                         // a channel that resolves when the server completes.
+	disableResourceReferences bool                               // true if resource references are disabled.
 }
 
 var _ SourceResourceMonitor = (*resmon)(nil)
@@ -424,12 +423,13 @@ func newResourceMonitor(src *evalSource, provs ProviderSource, regChan chan *reg
 
 	// New up an engine RPC server.
 	resmon := &resmon{
-		providers:        provs,
-		defaultProviders: d,
-		regChan:          regChan,
-		regOutChan:       regOutChan,
-		regReadChan:      regReadChan,
-		cancel:           cancel,
+		providers:                 provs,
+		defaultProviders:          d,
+		regChan:                   regChan,
+		regOutChan:                regOutChan,
+		regReadChan:               regReadChan,
+		cancel:                    cancel,
+		disableResourceReferences: opts.DisableResourceReferences,
 	}
 
 	// Fire up a gRPC server and start listening for incomings.
@@ -531,13 +531,7 @@ func (rm *resmon) SupportsFeature(ctx context.Context,
 	case "secrets":
 		hasSupport = true
 	case "resourceReferences":
-		// TODO: Temporarily disabling resource ref support (https://github.com/pulumi/pulumi-kubernetes/issues/1405)
-		hasSupport = false
-
-		// Allow the resource reference feature to be disabled by explicitly setting an env var.
-		if v, ok := os.LookupEnv("PULUMI_DISABLE_RESOURCE_REFERENCES"); ok && cmdutil.IsTruthy(v) {
-			hasSupport = false
-		}
+		hasSupport = !rm.disableResourceReferences
 	}
 
 	logging.V(5).Infof("ResourceMonitor.SupportsFeature(id: %s) = %t", req.Id, hasSupport)

sdk/go/common/resource/plugin/rpc.go

@@ -166,7 +166,7 @@ func MarshalPropertyValue(v resource.PropertyValue, opts MarshalOptions) (*struc
 		ref := v.ResourceReferenceValue()
 		if !opts.KeepResources {
 			val := string(ref.URN)
-			if ref.ID != "" {
+			if ref.HasID {
 				val = string(ref.ID)
 			}
 			logging.V(5).Infof("marshalling resource value as raw URN or ID as opts.KeepResources is false")