Add DeletedWithParent resource option

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 `Parent` resource
option to the container resource and set the `DeletedWithParent`
resource option flag to `true`.

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

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

@@ -0,0 +1,4 @@
+changes:
+- type: feat
+  scope: pkg
+  description: Add `DeletedWithParent` 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.DeletedWithParent)
 }
 
 // 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 DeletedWithParent attribute of this resource has changed, we must write the checkpoint.
+	if old.DeletedWithParent != new.DeletedWithParent {
+		logging.V(9).Infof("SnapshotManager: mustWrite() true because of DeletedWithParent")
+		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,117 @@ func TestRetainOnDelete(t *testing.T) {
 	assert.Len(t, snap.Resources, 0)
 }
 
+func TestDeletedWithParent(t *testing.T) {
+	t.Parallel()
+
+	idCounter := 0
+
+	topParentURN := 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 != topParentURN {
+						// Only top parent 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)
+			topParentURN = aURN
+
+			bURN, _, _, err := monitor.RegisterResource("pkgA:m:typA", "resB", true, deploytest.ResourceOptions{
+				Inputs:            ins,
+				Parent:            aURN,
+				DeletedWithParent: true,
+			})
+			assert.NoError(t, err)
+
+			_, _, _, err = monitor.RegisterResource("pkgA:m:typA", "resC", true, deploytest.ResourceOptions{
+				Inputs:            ins,
+				Parent:            bURN,
+				DeletedWithParent: true,
+			})
+			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 shouldn't see a provider delete but should get a new id
+	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
+	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

@@ -86,6 +86,7 @@ func NewResourceMonitor(resmon pulumirpc.ResourceMonitorClient) *ResourceMonitor
 
 type ResourceOptions struct {
 	Parent                  resource.URN
+	DeletedWithParent       bool
 	Protect                 bool
 	Dependencies            []resource.URN
 	Provider                string
@@ -222,6 +223,7 @@ func (rm *ResourceMonitor) RegisterResource(t tokens.Type, name string, custom b
 		RetainOnDelete:             opts.RetainOnDelete,
 		AdditionalSecretOutputs:    additionalSecretOutputs,
 		Aliases:                    aliasObjects,
+		DeletedWithParent:          opts.DeletedWithParent,
 	}
 
 	// 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, 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, 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, false)
 		steps = append(steps, newImportDeploymentStep(i.deployment, new, randomSeed))
 	}
 

pkg/resource/deploy/source_eval.go

@@ -329,7 +329,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, false),
 		done: done,
 	}
 	return event, done, nil
@@ -944,6 +944,7 @@ func (rm *resmon) RegisterResource(ctx context.Context,
 	id := resource.ID(req.GetImportId())
 	customTimeouts := req.GetCustomTimeouts()
 	retainOnDelete := req.GetRetainOnDelete()
+	deletedWithParent := req.GetDeletedWithParent()
 
 	// 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).
@@ -1110,9 +1111,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, deletedWithParent=%v",
 		t, name, custom, len(props), parent, protect, providerRef, dependencies, deleteBeforeReplace, ignoreChanges,
-		aliases, timeouts, providerRefs, replaceOnChanges, retainOnDelete)
+		aliases, timeouts, providerRefs, replaceOnChanges, retainOnDelete, deletedWithParent)
 
 	// If this is a remote component, fetch its provider and issue the construct call. Otherwise, register the resource.
 	var result *RegisterResult
@@ -1153,7 +1154,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, deletedWithParent),
 			done: make(chan *RegisterResult),
 		}
 
@@ -1207,6 +1208,7 @@ func (rm *resmon) RegisterResource(ctx context.Context,
 	// • additionalSecretOutputs
 	// • replaceOnChanges
 	// • retainOnDelete
+	// • deletedWithParent
 	// Revisit these semantics in Pulumi v4.0
 	// See this issue for more: https://github.com/pulumi/pulumi/issues/9704
 	if !custom {
@@ -1231,6 +1233,9 @@ func (rm *resmon) RegisterResource(ctx context.Context,
 		rm.checkComponentOption(result.State.URN, "retainOnDelete", func() bool {
 			return retainOnDelete
 		})
+		rm.checkComponentOption(result.State.URN, "deletedWithParent", func() bool {
+			return deletedWithParent
+		})
 	}
 
 	logging.V(5).Infof(