Allow component providers to return outputs to the engine (#15408)

<!--- 
Thanks so much for your contribution! If this is your first time
contributing, please ensure that you have read the
[CONTRIBUTING](https://github.com/pulumi/pulumi/blob/master/CONTRIBUTING.md)
documentation.
-->

# Description

<!--- Please include a summary of the change and which issue is fixed.
Please also include relevant motivation and context. -->

This adds a new flag to provider Construct and Call requests to indicate
that the engine will accept output values in the response.

If a component provider sees that flag set they can return Output
PropertyValues and optionally skip filling in the dependency map in the
response.

The engine will use the combined information of the dependency map and
any dependencies found on any output values in the property to build the
final dependency map that is returned to programs.

## Checklist

- [x] I have run `make tidy` to update any new dependencies
- [x] I have run `make lint` to verify my code passes the lint check
  - [ ] I have formatted my code using `gofumpt`

<!--- Please provide details if the checkbox below is to be left
unchecked. -->
- [ ] I have added tests that prove my fix is effective or that my
feature works
<!--- 
User-facing changes require a CHANGELOG entry.
-->
- [ ] I have run `make changelog` and committed the
`changelog/pending/<file>` documenting my change
<!--
If the change(s) in this PR is a modification of an existing call to the
Pulumi Cloud,
then the service should honor older versions of the CLI where this
change would not exist.
You must then bump the API version in
/pkg/backend/httpstate/client/api.go, as well as add
it to the service.
-->
- [ ] Yes, there are changes in this PR that warrants bumping the Pulumi
Cloud API version
<!-- @pulumi employees: If yes, you must submit corresponding changes in
the service repo. -->

changelog/pending/20240208--components--component-providers-can-return-output-values-in-their-response-to-the-engine.yaml

@@ -0,0 +1,4 @@
+changes:
+- type: feat
+  scope: components
+  description: Component providers can return output values in their response to the engine.

pkg/engine/lifecycletest/pulumi_test.go

@@ -5001,3 +5001,141 @@ func TestConstructCallReturnDependencies(t *testing.T) {
 		test(t, deploytest.WithoutGrpc)
 	})
 }
+
+// Test that the engine can receive OutputValues for Construct and Call
+func TestConstructCallReturnOutputs(t *testing.T) {
+	t.Parallel()
+
+	test := func(t *testing.T, opt deploytest.PluginOption) {
+		loaders := []*deploytest.ProviderLoader{
+			deploytest.NewProviderLoader("pkgA", semver.MustParse("1.0.0"), func() (plugin.Provider, error) {
+				return &deploytest.Provider{
+					CreateF: func(urn resource.URN, inputs resource.PropertyMap, timeout float64,
+						preview bool,
+					) (resource.ID, resource.PropertyMap, resource.Status, error) {
+						return "created-id", inputs, 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
+					},
+					ConstructF: func(monitor *deploytest.ResourceMonitor, typ, name string, parent resource.URN,
+						inputs resource.PropertyMap, info plugin.ConstructInfo, options plugin.ConstructOptions,
+					) (plugin.ConstructResult, error) {
+						urn, _, _, _, err := monitor.RegisterResource(tokens.Type(typ), name, false, deploytest.ResourceOptions{})
+						assert.NoError(t, err)
+
+						urnA, _, _, _, err := monitor.RegisterResource("pkgA:m:typA", name+"-a", true, deploytest.ResourceOptions{
+							Parent: urn,
+						})
+						assert.NoError(t, err)
+
+						// Return a secret and unknown output depending on some internal resource
+						deps := []resource.URN{urnA}
+						return plugin.ConstructResult{
+							URN: urn,
+							Outputs: resource.PropertyMap{
+								"foo": resource.NewOutputProperty(resource.Output{
+									Element:      resource.NewStringProperty("foo"),
+									Known:        true,
+									Secret:       true,
+									Dependencies: deps,
+								}),
+								"bar": resource.NewOutputProperty(resource.Output{
+									Dependencies: deps,
+								}),
+							},
+							OutputDependencies: nil, // Left blank on purpose because AcceptsOutputs is true
+						}, nil
+					},
+					CallF: func(monitor *deploytest.ResourceMonitor,
+						tok tokens.ModuleMember, args resource.PropertyMap,
+						info plugin.CallInfo, options plugin.CallOptions,
+					) (plugin.CallResult, error) {
+						// Assume a single output arg that this call depends on
+						arg := args["arg"]
+						deps := arg.OutputValue().Dependencies
+
+						return plugin.CallResult{
+							Return: resource.PropertyMap{
+								"foo": resource.NewOutputProperty(resource.Output{
+									Element:      resource.NewStringProperty("foo"),
+									Known:        true,
+									Secret:       true,
+									Dependencies: deps,
+								}),
+								"bar": resource.NewOutputProperty(resource.Output{
+									Dependencies: deps,
+								}),
+							},
+							ReturnDependencies: nil, // Left blank on purpose because AcceptsOutputs is true
+						}, nil
+					},
+				}, nil
+			}, opt),
+		}
+
+		programF := deploytest.NewLanguageRuntimeF(func(_ plugin.RunInfo, monitor *deploytest.ResourceMonitor) error {
+			_, _, state, deps, err := monitor.RegisterResource("pkgA:m:typA", "resA", false, deploytest.ResourceOptions{
+				Remote: true,
+			})
+			assert.NoError(t, err)
+
+			// The urn of the internal resource the component created
+			urn := resource.URN("urn:pulumi:test::test::pkgA:m:typA$pkgA:m:typA::resA-a")
+
+			// Assert that the outputs are received as just plain values because SDKs don't yet support output
+			// values returned from RegisterResource.
+			assert.Equal(t, resource.PropertyMap{
+				"foo": resource.MakeSecret(resource.NewStringProperty("foo")),
+				"bar": resource.MakeComputed(resource.NewStringProperty("")),
+			}, state)
+			assert.Equal(t, map[resource.PropertyKey][]resource.URN{
+				"foo": {urn},
+				"bar": {urn},
+			}, deps)
+
+			result, deps, _, err := monitor.Call("pkgA:m:typA", resource.PropertyMap{
+				// Send this as an output value using the dependencies returned.
+				"arg": resource.NewOutputProperty(resource.Output{
+					Element:      state["foo"].SecretValue().Element,
+					Known:        true,
+					Secret:       true,
+					Dependencies: []resource.URN{urn},
+				}),
+			}, "", "")
+			assert.NoError(t, err)
+
+			// Assert that the outputs are received as just plain values because SDKs don't yet support output
+			// values returned from Call.
+			assert.Equal(t, resource.PropertyMap{
+				"foo": resource.MakeSecret(resource.NewStringProperty("foo")),
+				"bar": resource.MakeComputed(resource.NewStringProperty("")),
+			}, result)
+			assert.Equal(t, map[resource.PropertyKey][]resource.URN{
+				"foo": {urn},
+				"bar": {urn},
+			}, deps)
+
+			return nil
+		})
+		hostF := deploytest.NewPluginHostF(nil, nil, programF, loaders...)
+
+		p := &TestPlan{
+			Options: TestUpdateOptions{HostF: hostF},
+		}
+
+		project := p.GetProject()
+		_, err := TestOp(Update).Run(project, p.GetTarget(t, nil), p.Options, true, p.BackendClient, nil)
+		assert.NoError(t, err)
+	}
+	t.Run("WithGrpc", func(t *testing.T) {
+		t.Parallel()
+		test(t, deploytest.WithGrpc)
+	})
+	t.Run("WithoutGrpc", func(t *testing.T) {
+		t.Parallel()
+		test(t, deploytest.WithoutGrpc)
+	})
+}

pkg/resource/deploy/source_eval.go

@@ -865,14 +865,12 @@ func (rm *resmon) Call(ctx context.Context, req *pulumirpc.ResourceCallRequest)
 	if err != nil {
 		return nil, fmt.Errorf("call of %v returned an error: %w", tok, err)
 	}
-	mret, err := plugin.MarshalProperties(ret.Return, plugin.MarshalOptions{
-		Label:         label,
-		KeepUnknowns:  true,
-		KeepSecrets:   true,
-		KeepResources: true,
-	})
-	if err != nil {
-		return nil, fmt.Errorf("failed to marshal %v return: %w", tok, err)
+
+	if ret.ReturnDependencies == nil {
+		ret.ReturnDependencies = map[resource.PropertyKey][]resource.URN{}
+	}
+	for k, v := range ret.Return {
+		ret.ReturnDependencies[k] = addOutputDependencies(ret.ReturnDependencies[k], v)
 	}
 
 	returnDependencies := map[string]*pulumirpc.CallResponse_ReturnDependencies{}
@@ -884,6 +882,16 @@ func (rm *resmon) Call(ctx context.Context, req *pulumirpc.ResourceCallRequest)
 		returnDependencies[string(name)] = &pulumirpc.CallResponse_ReturnDependencies{Urns: urns}
 	}
 
+	mret, err := plugin.MarshalProperties(ret.Return, plugin.MarshalOptions{
+		Label:         label,
+		KeepUnknowns:  true,
+		KeepSecrets:   true,
+		KeepResources: true,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal %v return: %w", tok, err)
+	}
+
 	chkfails := slice.Prealloc[*pulumirpc.CheckFailure](len(ret.Failures))
 	for _, failure := range ret.Failures {
 		chkfails = append(chkfails, &pulumirpc.CheckFailure{
@@ -1492,6 +1500,15 @@ func (rm *resmon) RegisterResource(ctx context.Context,
 
 		result = &RegisterResult{State: &resource.State{URN: constructResult.URN, Outputs: constructResult.Outputs}}
 
+		// The provider may have returned OutputValues in "Outputs", we need to downgrade them to Computed or
+		// Secret but also add them to the outputDeps map.
+		if constructResult.OutputDependencies == nil {
+			constructResult.OutputDependencies = map[resource.PropertyKey][]resource.URN{}
+		}
+		for k, v := range result.State.Outputs {
+			constructResult.OutputDependencies[k] = addOutputDependencies(constructResult.OutputDependencies[k], v)
+		}
+
 		outputDeps = map[string]*pulumirpc.RegisterResourceResponse_PropertyDependencies{}
 		for k, deps := range constructResult.OutputDependencies {
 			urns := make([]string, len(deps))
@@ -1500,6 +1517,7 @@ func (rm *resmon) RegisterResource(ctx context.Context,
 			}
 			outputDeps[string(k)] = &pulumirpc.RegisterResourceResponse_PropertyDependencies{Urns: urns}
 		}
+
 	} else {
 		additionalSecretKeys := slice.Prealloc[resource.PropertyKey](len(additionalSecretOutputs))
 		for _, name := range additionalSecretOutputs {
@@ -1884,3 +1902,31 @@ func downgradeOutputValues(v resource.PropertyMap) resource.PropertyMap {
 	}
 	return result
 }
+
+func addOutputDependencies(deps []resource.URN, v resource.PropertyValue) []resource.URN {
+	if v.IsOutput() {
+		output := v.OutputValue()
+		if output.Known {
+			deps = addOutputDependencies(deps, output.Element)
+		}
+		deps = append(deps, output.Dependencies...)
+	}
+	if v.IsResourceReference() {
+		ref := v.ResourceReferenceValue()
+		deps = addOutputDependencies(deps, ref.ID)
+	}
+	if v.IsObject() {
+		for _, elem := range v.ObjectValue() {
+			deps = addOutputDependencies(deps, elem)
+		}
+	}
+	if v.IsArray() {
+		for _, elem := range v.ArrayValue() {
+			deps = addOutputDependencies(deps, elem)
+		}
+	}
+	if v.IsSecret() {
+		deps = addOutputDependencies(deps, v.SecretValue().Element)
+	}
+	return deps
+}

proto/.checksum.txt

@@ -17,7 +17,7 @@
 3421371250 793 proto/pulumi/errors.proto
 829031483 10232 proto/pulumi/language.proto
 2893249402 1992 proto/pulumi/plugin.proto
-2457704516 24303 proto/pulumi/provider.proto
+545212422 24604 proto/pulumi/provider.proto
 2903108406 13833 proto/pulumi/resource.proto
 607478140 1008 proto/pulumi/source.proto
 2565199107 2157 proto/pulumi/testing/language.proto

proto/pulumi/provider.proto

@@ -164,6 +164,8 @@ message CallRequest {
     int32 parallel = 11;            // the degree of parallelism for resource operations (<=1 for serial).
     string monitorEndpoint = 12;    // the address for communicating back to the resource monitor.
     string organization = 14;       // the organization of the stack being deployed into.
+
+    bool accepts_output_values = 17; // the engine can be passed output values back, returnDependencies can be left blank if returning output values.
 }
 
 message CallResponse {
@@ -378,6 +380,8 @@ message ConstructRequest {
     repeated string ignoreChanges           = 22; // properties that should be ignored during a diff
     repeated string replaceOnChanges        = 23; // properties that, when changed, trigger a replacement
     bool retainOnDelete                     = 24; // whether to retain the resource in the cloud provider when it is deleted
+
+    bool accepts_output_values = 25; // the engine can be passed output values back, stateDependencies can be left blank if returning output values.
 }
 
 message ConstructResponse {

sdk/go/common/resource/plugin/provider_plugin.go

@@ -1435,6 +1435,7 @@ func (p *provider) Construct(info ConstructInfo, typ tokens.Type, name string, p
 		IgnoreChanges:           options.IgnoreChanges,
 		ReplaceOnChanges:        options.ReplaceOnChanges,
 		RetainOnDelete:          options.RetainOnDelete,
+		AcceptsOutputValues:     true,
 	}
 	if ct := options.CustomTimeouts; ct != nil {
 		req.CustomTimeouts = &pulumirpc.ConstructRequest_CustomTimeouts{
@@ -1450,10 +1451,11 @@ func (p *provider) Construct(info ConstructInfo, typ tokens.Type, name string, p
 	}
 
 	outputs, err := UnmarshalProperties(resp.GetState(), MarshalOptions{
-		Label:         label + ".outputs",
-		KeepUnknowns:  info.DryRun,
-		KeepSecrets:   true,
-		KeepResources: true,
+		Label:            label + ".outputs",
+		KeepUnknowns:     info.DryRun,
+		KeepSecrets:      true,
+		KeepResources:    true,
+		KeepOutputValues: true,
 	})
 	if err != nil {
 		return ConstructResult{}, err
@@ -1669,15 +1671,16 @@ func (p *provider) Call(tok tokens.ModuleMember, args resource.PropertyMap, info
 	}
 
 	resp, err := client.Call(p.requestContext(), &pulumirpc.CallRequest{
-		Tok:             string(tok),
-		Args:            margs,
-		ArgDependencies: argDependencies,
-		Project:         info.Project,
-		Stack:           info.Stack,
-		Config:          config,
-		DryRun:          info.DryRun,
-		Parallel:        int32(info.Parallel),
-		MonitorEndpoint: info.MonitorAddress,
+		Tok:                 string(tok),
+		Args:                margs,
+		ArgDependencies:     argDependencies,
+		Project:             info.Project,
+		Stack:               info.Stack,
+		Config:              config,
+		DryRun:              info.DryRun,
+		Parallel:            int32(info.Parallel),
+		MonitorEndpoint:     info.MonitorAddress,
+		AcceptsOutputValues: true,
 	})
 	if err != nil {
 		rpcError := rpcerror.Convert(err)
@@ -1687,10 +1690,11 @@ func (p *provider) Call(tok tokens.ModuleMember, args resource.PropertyMap, info
 
 	// Unmarshal any return values.
 	ret, err := UnmarshalProperties(resp.GetReturn(), MarshalOptions{
-		Label:         label + ".returns",
-		KeepUnknowns:  info.DryRun,
-		KeepSecrets:   true,
-		KeepResources: true,
+		Label:            label + ".returns",
+		KeepUnknowns:     info.DryRun,
+		KeepSecrets:      true,
+		KeepResources:    true,
+		KeepOutputValues: true,
 	})
 	if err != nil {
 		return CallResult{}, err

sdk/go/common/resource/plugin/provider_plugin_test.go

@@ -479,6 +479,7 @@ func TestProvider_ConstructOptions(t *testing.T) {
 			tt.want.Name = "name"
 			tt.want.Config = make(map[string]string)
 			tt.want.Inputs = &structpb.Struct{Fields: make(map[string]*structpb.Value)}
+			tt.want.AcceptsOutputValues = true
 
 			var got *pulumirpc.ConstructRequest
 			client := &stubClient{

sdk/go/common/resource/plugin/provider_server.go

@@ -509,7 +509,9 @@ func (p *providerServer) Construct(ctx context.Context,
 		return nil, err
 	}
 
-	outputs, err := MarshalProperties(result.Outputs, p.marshalOptions("outputs"))
+	opts := p.marshalOptions("outputs")
+	opts.KeepOutputValues = req.AcceptsOutputValues
+	outputs, err := MarshalProperties(result.Outputs, opts)
 	if err != nil {
 		return nil, err
 	}
@@ -628,12 +630,9 @@ func (p *providerServer) Call(ctx context.Context, req *pulumirpc.CallRequest) (
 		return nil, err
 	}
 
-	rpcResult, err := MarshalProperties(result.Return, MarshalOptions{
-		Label:         "result",
-		KeepUnknowns:  true,
-		KeepSecrets:   true,
-		KeepResources: true,
-	})
+	opts := p.marshalOptions("return")
+	opts.KeepOutputValues = req.AcceptsOutputValues
+	rpcResult, err := MarshalProperties(result.Return, opts)
 	if err != nil {
 		return nil, err
 	}