Merge #11095

11095: Add `DeletedWith` resource option r=Frassle a=same-id

# Description

In many cases there is no need to delete resources if the container resource is going to be deleted as well.

A few examples:
 * Database object (roles, tables) when database is being deleted
 * Cloud IAM bindings when user itself is being deleted

This helps with:
 * Speeding the deletion process
 * Removing unnecessary calls to providers
 * Avoiding failed deletions when the pulumi user running the plan has access to the container resource but not the contained ones

To avoid deleting contained resources, set the `DeletedWith` resource option to the container resource.

TODO:
 Should we support DeletedWith with PendingDeletes?
 Special case might be when contained resource is marked as pending deletion
 but we now want to delete the container resource, so ultimately there is no
 need to delete the contained anymore

<!--- 
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.
-->


Fixes # (issue)

## Checklist

<!--- Please provide details if the checkbox below is to be left unchecked. -->
- [x] I have added tests that prove my fix is effective or that my feature works
<!--- 
User-facing changes require a CHANGELOG entry.
-->
- [x] 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 Service,
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.

Not sure - maybe
-->
- [ ] Yes, there are changes in this PR that warrants bumping the Pulumi Service API version
  <!-- `@Pulumi` employees: If yes, you must submit corresponding changes in the service repo. -->


Co-authored-by: Sam Eiderman <sameid@gmail.com>
Co-authored-by: Fraser Waters <fraser@pulumi.com>

changelog/pending/20221020--pkg--add-deletedwith-as-a-resource-option.yaml

@@ -0,0 +1,4 @@
+changes:
+- type: feat
+  scope: pkg
+  description: Add `DeletedWith` as a resource option.

pkg/backend/display/json.go

@@ -85,7 +85,7 @@ func stateForJSONOutput(s *resource.State, opts Options) *resource.State {
 	return resource.NewState(s.Type, s.URN, s.Custom, s.Delete, s.ID, inputs,
 		outputs, s.Parent, s.Protect, s.External, s.Dependencies, s.InitErrors, s.Provider,
 		s.PropertyDependencies, s.PendingReplacement, s.AdditionalSecretOutputs, s.Aliases, &s.CustomTimeouts,
-		s.ImportID, s.RetainOnDelete)
+		s.ImportID, s.RetainOnDelete, s.DeletedWith)
 }
 
 // ShowJSONEvents renders incremental engine events to stdout.

pkg/backend/snapshot.go

@@ -231,6 +231,12 @@ func (ssm *sameSnapshotMutation) mustWrite(step *deploy.SameStep) bool {
 		return true
 	}
 
+	// If the DeletedWith attribute of this resource has changed, we must write the checkpoint.
+	if old.DeletedWith != new.DeletedWith {
+		logging.V(9).Infof("SnapshotManager: mustWrite() true because of DeletedWith")
+		return true
+	}
+
 	// If the protection attribute of this resource has changed, we must write the checkpoint.
 	if old.Protect != new.Protect {
 		logging.V(9).Infof("SnapshotManager: mustWrite() true because of Protect")

pkg/backend/snapshot_test.go

@@ -603,7 +603,7 @@ func TestDeletion(t *testing.T) {
 	})
 
 	manager, sp := MockSetup(t, snap)
-	step := deploy.NewDeleteStep(nil, resourceA)
+	step := deploy.NewDeleteStep(nil, map[resource.URN]bool{}, resourceA)
 	mutation, err := manager.BeginMutation(step)
 	if !assert.NoError(t, err) {
 		t.FailNow()
@@ -629,7 +629,7 @@ func TestFailedDelete(t *testing.T) {
 	})
 
 	manager, sp := MockSetup(t, snap)
-	step := deploy.NewDeleteStep(nil, resourceA)
+	step := deploy.NewDeleteStep(nil, map[resource.URN]bool{}, resourceA)
 	mutation, err := manager.BeginMutation(step)
 	if !assert.NoError(t, err) {
 		t.FailNow()
@@ -802,7 +802,7 @@ func TestRecordingDeleteSuccess(t *testing.T) {
 		resourceA,
 	})
 	manager, sp := MockSetup(t, snap)
-	step := deploy.NewDeleteStep(nil, resourceA)
+	step := deploy.NewDeleteStep(nil, map[resource.URN]bool{}, resourceA)
 	mutation, err := manager.BeginMutation(step)
 	if !assert.NoError(t, err) {
 		t.FailNow()
@@ -834,7 +834,7 @@ func TestRecordingDeleteFailure(t *testing.T) {
 		resourceA,
 	})
 	manager, sp := MockSetup(t, snap)
-	step := deploy.NewDeleteStep(nil, resourceA)
+	step := deploy.NewDeleteStep(nil, map[resource.URN]bool{}, resourceA)
 	mutation, err := manager.BeginMutation(step)
 	if !assert.NoError(t, err) {
 		t.FailNow()

pkg/engine/lifecycletest/pulumi_test.go

@@ -3702,6 +3702,215 @@ func TestRetainOnDelete(t *testing.T) {
 	assert.Len(t, snap.Resources, 0)
 }
 
+func TestDeletedWith(t *testing.T) {
+	t.Parallel()
+
+	idCounter := 0
+
+	topURN := resource.URN("")
+
+	loaders := []*deploytest.ProviderLoader{
+		deploytest.NewProviderLoader("pkgA", semver.MustParse("1.0.0"), func() (plugin.Provider, error) {
+			return &deploytest.Provider{
+				DiffF: func(
+					urn resource.URN,
+					id resource.ID,
+					olds, news resource.PropertyMap,
+					ignoreChanges []string) (plugin.DiffResult, error) {
+					if !olds["foo"].DeepEquals(news["foo"]) {
+						// If foo changes do a replace, we use this to check we don't delete on replace
+						return plugin.DiffResult{
+							Changes:     plugin.DiffSome,
+							ReplaceKeys: []resource.PropertyKey{"foo"},
+						}, nil
+					}
+					return plugin.DiffResult{}, nil
+				},
+				CreateF: func(urn resource.URN, news resource.PropertyMap, timeout float64,
+					preview bool) (resource.ID, resource.PropertyMap, resource.Status, error) {
+					resourceID := resource.ID(fmt.Sprintf("created-id-%d", idCounter))
+					idCounter = idCounter + 1
+					return resourceID, news, resource.StatusOK, nil
+				},
+				DeleteF: func(urn resource.URN, id resource.ID, olds resource.PropertyMap,
+					timeout float64) (resource.Status, error) {
+					if urn != topURN {
+						// Only topURN (aURN) should be actually deleted
+						assert.Fail(t, "Delete was called")
+					}
+					return resource.StatusOK, nil
+				},
+			}, nil
+		}, deploytest.WithoutGrpc),
+	}
+
+	ins := resource.NewPropertyMapFromMap(map[string]interface{}{
+		"foo": "bar",
+	})
+
+	createResource := true
+
+	program := deploytest.NewLanguageRuntime(func(_ plugin.RunInfo, monitor *deploytest.ResourceMonitor) error {
+
+		if createResource {
+			aURN, _, _, err := monitor.RegisterResource("pkgA:m:typA", "resA", true, deploytest.ResourceOptions{
+				Inputs: ins,
+			})
+			assert.NoError(t, err)
+			topURN = aURN
+
+			bURN, _, _, err := monitor.RegisterResource("pkgA:m:typA", "resB", true, deploytest.ResourceOptions{
+				Inputs:      ins,
+				DeletedWith: aURN,
+			})
+			assert.NoError(t, err)
+
+			_, _, _, err = monitor.RegisterResource("pkgA:m:typA", "resC", true, deploytest.ResourceOptions{
+				Inputs:      ins,
+				DeletedWith: bURN,
+			})
+			assert.NoError(t, err)
+		}
+
+		return nil
+	})
+	host := deploytest.NewPluginHost(nil, nil, program, loaders...)
+
+	p := &TestPlan{
+		Options: UpdateOptions{Host: host},
+	}
+
+	project := p.GetProject()
+
+	// Run an update to create the resource
+	snap, res := TestOp(Update).Run(project, p.GetTarget(t, nil), p.Options, false, p.BackendClient, nil)
+	assert.Nil(t, res)
+	assert.NotNil(t, snap)
+	assert.Len(t, snap.Resources, 4)
+	assert.Equal(t, "created-id-0", snap.Resources[1].ID.String())
+	assert.Equal(t, "created-id-1", snap.Resources[2].ID.String())
+	assert.Equal(t, "created-id-2", snap.Resources[3].ID.String())
+
+	// Run a new update which will cause a replace, we should only see a provider delete for aURN but should
+	// get a new id for everything
+	ins = resource.NewPropertyMapFromMap(map[string]interface{}{
+		"foo": "baz",
+	})
+	snap, res = TestOp(Update).Run(project, p.GetTarget(t, snap), p.Options, false, p.BackendClient, nil)
+	assert.Nil(t, res)
+	assert.NotNil(t, snap)
+	assert.Len(t, snap.Resources, 4)
+	assert.Equal(t, "created-id-3", snap.Resources[1].ID.String())
+	assert.Equal(t, "created-id-4", snap.Resources[2].ID.String())
+	assert.Equal(t, "created-id-5", snap.Resources[3].ID.String())
+
+	// Run a new update which will cause a delete, we still shouldn't see a provider delete for anything but aURN
+	createResource = false
+	snap, res = TestOp(Update).Run(project, p.GetTarget(t, snap), p.Options, false, p.BackendClient, nil)
+	assert.Nil(t, res)
+	assert.NotNil(t, snap)
+	assert.Len(t, snap.Resources, 0)
+}
+
+func TestDeletedWithCircularDependency(t *testing.T) {
+	// This test should be removed if DeletedWith circular dependency is taken care of.
+	// At the mean time, if there is a circular dependency - none shall be deleted.
+	t.Parallel()
+
+	idCounter := 0
+
+	loaders := []*deploytest.ProviderLoader{
+		deploytest.NewProviderLoader("pkgA", semver.MustParse("1.0.0"), func() (plugin.Provider, error) {
+			return &deploytest.Provider{
+				DiffF: func(
+					urn resource.URN,
+					id resource.ID,
+					olds, news resource.PropertyMap,
+					ignoreChanges []string) (plugin.DiffResult, error) {
+					return plugin.DiffResult{}, nil
+				},
+				CreateF: func(urn resource.URN, news resource.PropertyMap, timeout float64,
+					preview bool) (resource.ID, resource.PropertyMap, resource.Status, error) {
+					resourceID := resource.ID(fmt.Sprintf("created-id-%d", idCounter))
+					idCounter = idCounter + 1
+					return resourceID, news, resource.StatusOK, nil
+				},
+				DeleteF: func(urn resource.URN, id resource.ID, olds resource.PropertyMap,
+					timeout float64) (resource.Status, error) {
+
+					assert.Fail(t, "Delete was called")
+
+					return resource.StatusOK, nil
+				},
+			}, nil
+		}, deploytest.WithoutGrpc),
+	}
+
+	ins := resource.NewPropertyMapFromMap(map[string]interface{}{
+		"foo": "bar",
+	})
+
+	createResource := true
+	cURN := resource.URN("")
+
+	program := deploytest.NewLanguageRuntime(func(_ plugin.RunInfo, monitor *deploytest.ResourceMonitor) error {
+
+		if createResource {
+			aURN, _, _, err := monitor.RegisterResource("pkgA:m:typA", "resA", true, deploytest.ResourceOptions{
+				Inputs:      ins,
+				DeletedWith: cURN,
+			})
+			assert.NoError(t, err)
+
+			bURN, _, _, err := monitor.RegisterResource("pkgA:m:typA", "resB", true, deploytest.ResourceOptions{
+				Inputs:      ins,
+				DeletedWith: aURN,
+			})
+			assert.NoError(t, err)
+
+			cURN, _, _, err = monitor.RegisterResource("pkgA:m:typA", "resC", true, deploytest.ResourceOptions{
+				Inputs:      ins,
+				DeletedWith: bURN,
+			})
+			assert.NoError(t, err)
+		}
+
+		return nil
+	})
+	host := deploytest.NewPluginHost(nil, nil, program, loaders...)
+
+	p := &TestPlan{
+		Options: UpdateOptions{Host: host},
+	}
+
+	project := p.GetProject()
+
+	// Run an update to create the resource
+	snap, res := TestOp(Update).Run(project, p.GetTarget(t, nil), p.Options, false, p.BackendClient, nil)
+	assert.Nil(t, res)
+	assert.NotNil(t, snap)
+	assert.Len(t, snap.Resources, 4)
+	assert.Equal(t, "created-id-0", snap.Resources[1].ID.String())
+	assert.Equal(t, "created-id-1", snap.Resources[2].ID.String())
+	assert.Equal(t, "created-id-2", snap.Resources[3].ID.String())
+
+	// Run again to update DeleteWith for resA
+	snap, res = TestOp(Update).Run(project, p.GetTarget(t, snap), p.Options, false, p.BackendClient, nil)
+	assert.Nil(t, res)
+	assert.NotNil(t, snap)
+	assert.Len(t, snap.Resources, 4)
+	assert.Equal(t, "created-id-0", snap.Resources[1].ID.String())
+	assert.Equal(t, "created-id-1", snap.Resources[2].ID.String())
+	assert.Equal(t, "created-id-2", snap.Resources[3].ID.String())
+
+	// Run a new update which will cause a delete, we still shouldn't see a provider delete
+	createResource = false
+	snap, res = TestOp(Update).Run(project, p.GetTarget(t, snap), p.Options, false, p.BackendClient, nil)
+	assert.Nil(t, res)
+	assert.NotNil(t, snap)
+	assert.Len(t, snap.Resources, 0)
+}
+
 func TestInvalidGetIDReportsUserError(t *testing.T) {
 	t.Parallel()
 

pkg/resource/deploy/deploytest/resourcemonitor.go

@@ -101,6 +101,7 @@ type ResourceOptions struct {
 	ImportID                resource.ID
 	CustomTimeouts          *resource.CustomTimeouts
 	RetainOnDelete          bool
+	DeletedWith             resource.URN
 	SupportsPartialValues   *bool
 	Remote                  bool
 	Providers               map[string]string
@@ -222,6 +223,7 @@ func (rm *ResourceMonitor) RegisterResource(t tokens.Type, name string, custom b
 		RetainOnDelete:             opts.RetainOnDelete,
 		AdditionalSecretOutputs:    additionalSecretOutputs,
 		Aliases:                    aliasObjects,
+		DeletedWith:                string(opts.DeletedWith),
 	}
 
 	// submit request

pkg/resource/deploy/import.go

@@ -169,7 +169,7 @@ func (i *importer) getOrCreateStackResource(ctx context.Context) (resource.URN,
 	typ, name := resource.RootStackType, fmt.Sprintf("%s-%s", projectName, stackName)
 	urn := resource.NewURN(stackName.Q(), projectName, "", typ, tokens.QName(name))
 	state := resource.NewState(typ, urn, false, false, "", resource.PropertyMap{}, nil, "", false, false, nil, nil, "",
-		nil, false, nil, nil, nil, "", false)
+		nil, false, nil, nil, nil, "", false, "")
 	// TODO(seqnum) should stacks be created with 1? When do they ever get recreated/replaced?
 	if !i.executeSerial(ctx, NewCreateStep(i.deployment, noopEvent(0), state)) {
 		return "", false, false
@@ -255,7 +255,7 @@ func (i *importer) registerProviders(ctx context.Context) (map[resource.URN]stri
 		}
 
 		state := resource.NewState(typ, urn, true, false, "", inputs, nil, "", false, false, nil, nil, "", nil, false,
-			nil, nil, nil, "", false)
+			nil, nil, nil, "", false, "")
 		// TODO(seqnum) should default providers be created with 1? When do they ever get recreated/replaced?
 		if issueCheckErrors(i.deployment, state, urn, failures) {
 			return nil, nil, false
@@ -355,7 +355,7 @@ func (i *importer) importResources(ctx context.Context) result.Result {
 
 		// Create the new desired state. Note that the resource is protected.
 		new := resource.NewState(urn.Type(), urn, true, false, imp.ID, resource.PropertyMap{}, nil, parent, imp.Protect,
-			false, nil, nil, provider, nil, false, nil, nil, nil, "", false)
+			false, nil, nil, provider, nil, false, nil, nil, nil, "", false, "")
 		steps = append(steps, newImportDeploymentStep(i.deployment, new, randomSeed))
 	}
 

pkg/resource/deploy/source_eval.go

@@ -330,7 +330,7 @@ func (d *defaultProviders) newRegisterDefaultProviderEvent(
 		goal: resource.NewGoal(
 			providers.MakeProviderType(req.Package()),
 			req.Name(), true, inputs, "", false, nil, "", nil, nil, nil,
-			nil, nil, nil, "", nil, nil, false),
+			nil, nil, nil, "", nil, nil, false, ""),
 		done: done,
 	}
 	return event, done, nil
@@ -639,6 +639,8 @@ func (rm *resmon) SupportsFeature(ctx context.Context,
 		hasSupport = !rm.disableOutputValues
 	case "aliasSpecs":
 		hasSupport = true
+	case "deletedWith":
+		hasSupport = true
 	}
 
 	logging.V(5).Infof("ResourceMonitor.SupportsFeature(id: %s) = %t", req.Id, hasSupport)
@@ -954,6 +956,7 @@ func (rm *resmon) RegisterResource(ctx context.Context,
 	id := resource.ID(req.GetImportId())
 	customTimeouts := req.GetCustomTimeouts()
 	retainOnDelete := req.GetRetainOnDelete()
+	deletedWith := resource.URN(req.GetDeletedWith())
 
 	// Custom resources must have a three-part type so that we can 1) identify if they are providers and 2) retrieve the
 	// provider responsible for managing a particular resource (based on the type's Package).
@@ -1139,9 +1142,9 @@ func (rm *resmon) RegisterResource(ctx context.Context,
 	logging.V(5).Infof(
 		"ResourceMonitor.RegisterResource received: t=%v, name=%v, custom=%v, #props=%v, parent=%v, protect=%v, "+
 			"provider=%v, deps=%v, deleteBeforeReplace=%v, ignoreChanges=%v, aliases=%v, customTimeouts=%v, "+
-			"providers=%v, replaceOnChanges=%v, retainOnDelete=%v",
+			"providers=%v, replaceOnChanges=%v, retainOnDelete=%v, deletedWith=%v",
 		t, name, custom, len(props), parent, protect, providerRef, dependencies, deleteBeforeReplace, ignoreChanges,
-		aliases, timeouts, providerRefs, replaceOnChanges, retainOnDelete)
+		aliases, timeouts, providerRefs, replaceOnChanges, retainOnDelete, deletedWith)
 
 	// If this is a remote component, fetch its provider and issue the construct call. Otherwise, register the resource.
 	var result *RegisterResult
@@ -1187,7 +1190,7 @@ func (rm *resmon) RegisterResource(ctx context.Context,
 		step := &registerResourceEvent{
 			goal: resource.NewGoal(t, name, custom, props, parent, protect, dependencies,
 				providerRef.String(), nil, propertyDependencies, deleteBeforeReplace, ignoreChanges,
-				additionalSecretOutputs, aliases, id, &timeouts, replaceOnChanges, retainOnDelete),
+				additionalSecretOutputs, aliases, id, &timeouts, replaceOnChanges, retainOnDelete, deletedWith),
 			done: make(chan *RegisterResult),
 		}
 
@@ -1249,6 +1252,7 @@ func (rm *resmon) RegisterResource(ctx context.Context,
 	// • additionalSecretOutputs
 	// • replaceOnChanges
 	// • retainOnDelete
+	// • deletedWith
 	// Revisit these semantics in Pulumi v4.0
 	// See this issue for more: https://github.com/pulumi/pulumi/issues/9704
 	if !custom {
@@ -1273,6 +1277,9 @@ func (rm *resmon) RegisterResource(ctx context.Context,
 		rm.checkComponentOption(result.State.URN, "retainOnDelete", func() bool {
 			return retainOnDelete
 		})
+		rm.checkComponentOption(result.State.URN, "deletedWith", func() bool {
+			return deletedWith != ""
+		})
 	}
 
 	logging.V(5).Infof(