Merge pull request #30900 from hashicorp/jbardin/replace-triggered-by

Configurable instance replacement via lifecycle `replace_triggered_by`

go.mod

@@ -45,7 +45,7 @@ require (
 	github.com/hashicorp/go-uuid v1.0.2
 	github.com/hashicorp/go-version v1.3.0
 	github.com/hashicorp/hcl v0.0.0-20170504190234-a4b07c25de5f
-	github.com/hashicorp/hcl/v2 v2.11.1
+	github.com/hashicorp/hcl/v2 v2.11.2-0.20220408161043-2ef09d129d96
 	github.com/hashicorp/terraform-config-inspect v0.0.0-20210209133302-4fd17a0faac2
 	github.com/hashicorp/terraform-svchost v0.0.0-20200729002733-f050f53b9734
 	github.com/jmespath/go-jmespath v0.4.0

go.sum

@@ -434,8 +434,8 @@ github.com/hashicorp/hcl v0.0.0-20170504190234-a4b07c25de5f h1:UdxlrJz4JOnY8W+Db
 github.com/hashicorp/hcl v0.0.0-20170504190234-a4b07c25de5f/go.mod h1:oZtUIOe8dh44I2q6ScRibXws4Ajl+d+nod3AaR9vL5w=
 github.com/hashicorp/hcl/v2 v2.0.0/go.mod h1:oVVDG71tEinNGYCxinCYadcmKU9bglqW9pV3txagJ90=
 github.com/hashicorp/hcl/v2 v2.3.0/go.mod h1:d+FwDBbOLvpAM3Z6J7gPj/VoAGkNe/gm352ZhjJ/Zv8=
-github.com/hashicorp/hcl/v2 v2.11.1 h1:yTyWcXcm9XB0TEkyU/JCRU6rYy4K+mgLtzn2wlrJbcc=
-github.com/hashicorp/hcl/v2 v2.11.1/go.mod h1:FwWsfWEjyV/CMj8s/gqAuiviY72rJ1/oayI9WftqcKg=
+github.com/hashicorp/hcl/v2 v2.11.2-0.20220408161043-2ef09d129d96 h1:RO/o1b/ZxMUCIgQiKF7qdk0YRwkILQF4KwO39mm9itA=
+github.com/hashicorp/hcl/v2 v2.11.2-0.20220408161043-2ef09d129d96/go.mod h1:FwWsfWEjyV/CMj8s/gqAuiviY72rJ1/oayI9WftqcKg=
 github.com/hashicorp/jsonapi v0.0.0-20210826224640-ee7dae0fb22d h1:9ARUJJ1VVynB176G1HCwleORqCaXm/Vx0uUi0dL26I0=
 github.com/hashicorp/jsonapi v0.0.0-20210826224640-ee7dae0fb22d/go.mod h1:Yog5+CPEM3c99L1CL2CFCYoSzgWm5vTU58idbRUaLik=
 github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64=

internal/command/format/diff.go

@@ -87,6 +87,8 @@ func ResourceChange(
 			buf.WriteString(fmt.Sprintf(color.Color("[bold]  # %s[reset] is tainted, so must be [bold][red]replaced"), dispAddr))
 		case plans.ResourceInstanceReplaceByRequest:
 			buf.WriteString(fmt.Sprintf(color.Color("[bold]  # %s[reset] will be [bold][red]replaced[reset], as requested"), dispAddr))
+		case plans.ResourceInstanceReplaceByTriggers:
+			buf.WriteString(fmt.Sprintf(color.Color("[bold]  # %s[reset] will be [bold][red]replaced[reset] due to changes in replace_triggered_by"), dispAddr))
 		default:
 			buf.WriteString(fmt.Sprintf(color.Color("[bold]  # %s[reset] must be [bold][red]replaced"), dispAddr))
 		}

internal/command/jsonplan/plan.go

@@ -392,6 +392,8 @@ func (p *plan) marshalResourceChanges(resources []*plans.ResourceInstanceChangeS
 			r.ActionReason = "replace_because_tainted"
 		case plans.ResourceInstanceReplaceByRequest:
 			r.ActionReason = "replace_by_request"
+		case plans.ResourceInstanceReplaceByTriggers:
+			r.ActionReason = "replace_by_triggers"
 		case plans.ResourceInstanceDeleteBecauseNoResourceConfig:
 			r.ActionReason = "delete_because_no_resource_config"
 		case plans.ResourceInstanceDeleteBecauseWrongRepetition:

internal/command/views/json/change.go

@@ -68,11 +68,12 @@ func changeAction(action plans.Action) ChangeAction {
 type ChangeReason string
 
 const (
-	ReasonNone         ChangeReason = ""
-	ReasonTainted      ChangeReason = "tainted"
-	ReasonRequested    ChangeReason = "requested"
-	ReasonCannotUpdate ChangeReason = "cannot_update"
-	ReasonUnknown      ChangeReason = "unknown"
+	ReasonNone               ChangeReason = ""
+	ReasonTainted            ChangeReason = "tainted"
+	ReasonRequested          ChangeReason = "requested"
+	ReasonReplaceTriggeredBy ChangeReason = "replace_triggered_by"
+	ReasonCannotUpdate       ChangeReason = "cannot_update"
+	ReasonUnknown            ChangeReason = "unknown"
 
 	ReasonDeleteBecauseNoResourceConfig ChangeReason = "delete_because_no_resource_config"
 	ReasonDeleteBecauseWrongRepetition  ChangeReason = "delete_because_wrong_repetition"
@@ -91,6 +92,8 @@ func changeReason(reason plans.ResourceInstanceChangeActionReason) ChangeReason
 		return ReasonRequested
 	case plans.ResourceInstanceReplaceBecauseCannotUpdate:
 		return ReasonCannotUpdate
+	case plans.ResourceInstanceReplaceByTriggers:
+		return ReasonReplaceTriggeredBy
 	case plans.ResourceInstanceDeleteBecauseNoResourceConfig:
 		return ReasonDeleteBecauseNoResourceConfig
 	case plans.ResourceInstanceDeleteBecauseWrongRepetition:

internal/configs/resource.go

@@ -6,8 +6,11 @@ import (
 	"github.com/hashicorp/hcl/v2"
 	"github.com/hashicorp/hcl/v2/gohcl"
 	"github.com/hashicorp/hcl/v2/hclsyntax"
+	hcljson "github.com/hashicorp/hcl/v2/json"
 
 	"github.com/hashicorp/terraform/internal/addrs"
+	"github.com/hashicorp/terraform/internal/lang"
+	"github.com/hashicorp/terraform/internal/tfdiags"
 )
 
 // Resource represents a "resource" or "data" block in a module or file.
@@ -27,6 +30,8 @@ type Resource struct {
 
 	DependsOn []hcl.Traversal
 
+	TriggersReplacement []hcl.Expression
+
 	// Managed is populated only for Mode = addrs.ManagedResourceMode,
 	// containing the additional fields that apply to managed resources.
 	// For all other resource modes, this field is nil.
@@ -177,6 +182,13 @@ func decodeResourceBlock(block *hcl.Block, override bool) (*Resource, hcl.Diagno
 				r.Managed.PreventDestroySet = true
 			}
 
+			if attr, exists := lcContent.Attributes["replace_triggered_by"]; exists {
+				exprs, hclDiags := decodeReplaceTriggeredBy(attr.Expr)
+				diags = diags.Extend(hclDiags)
+
+				r.TriggersReplacement = append(r.TriggersReplacement, exprs...)
+			}
+
 			if attr, exists := lcContent.Attributes["ignore_changes"]; exists {
 
 				// ignore_changes can either be a list of relative traversals
@@ -237,7 +249,6 @@ func decodeResourceBlock(block *hcl.Block, override bool) (*Resource, hcl.Diagno
 					}
 
 				}
-
 			}
 
 			for _, block := range lcContent.Blocks {
@@ -481,6 +492,115 @@ func decodeDataBlock(block *hcl.Block, override bool) (*Resource, hcl.Diagnostic
 	return r, diags
 }
 
+// decodeReplaceTriggeredBy decodes and does basic validation of the
+// replace_triggered_by expressions, ensuring they only contains references to
+// a single resource, and the only extra variables are count.index or each.key.
+func decodeReplaceTriggeredBy(expr hcl.Expression) ([]hcl.Expression, hcl.Diagnostics) {
+	// Since we are manually parsing the replace_triggered_by argument, we
+	// need to specially handle json configs, in which case the values will
+	// be json strings rather than hcl. To simplify parsing however we will
+	// decode the individual list elements, rather than the entire expression.
+	isJSON := hcljson.IsJSONExpression(expr)
+
+	exprs, diags := hcl.ExprList(expr)
+
+	for i, expr := range exprs {
+		if isJSON {
+			// We can abuse the hcl json api and rely on the fact that calling
+			// Value on a json expression with no EvalContext will return the
+			// raw string. We can then parse that as normal hcl syntax, and
+			// continue with the decoding.
+			v, ds := expr.Value(nil)
+			diags = diags.Extend(ds)
+			if diags.HasErrors() {
+				continue
+			}
+
+			expr, ds = hclsyntax.ParseExpression([]byte(v.AsString()), "", expr.Range().Start)
+			diags = diags.Extend(ds)
+			if diags.HasErrors() {
+				continue
+			}
+			// make sure to swap out the expression we're returning too
+			exprs[i] = expr
+		}
+
+		refs, refDiags := lang.ReferencesInExpr(expr)
+		for _, diag := range refDiags {
+			severity := hcl.DiagError
+			if diag.Severity() == tfdiags.Warning {
+				severity = hcl.DiagWarning
+			}
+
+			desc := diag.Description()
+
+			diags = append(diags, &hcl.Diagnostic{
+				Severity: severity,
+				Summary:  desc.Summary,
+				Detail:   desc.Detail,
+				Subject:  expr.Range().Ptr(),
+			})
+		}
+
+		if refDiags.HasErrors() {
+			continue
+		}
+
+		resourceCount := 0
+		for _, ref := range refs {
+			switch sub := ref.Subject.(type) {
+			case addrs.Resource, addrs.ResourceInstance:
+				resourceCount++
+
+			case addrs.ForEachAttr:
+				if sub.Name != "key" {
+					diags = append(diags, &hcl.Diagnostic{
+						Severity: hcl.DiagError,
+						Summary:  "Invalid each reference in replace_triggered_by expression",
+						Detail:   "Only each.key may be used in replace_triggered_by.",
+						Subject:  expr.Range().Ptr(),
+					})
+				}
+			case addrs.CountAttr:
+				if sub.Name != "index" {
+					diags = append(diags, &hcl.Diagnostic{
+						Severity: hcl.DiagError,
+						Summary:  "Invalid count reference in replace_triggered_by expression",
+						Detail:   "Only count.index may be used in replace_triggered_by.",
+						Subject:  expr.Range().Ptr(),
+					})
+				}
+			default:
+				// everything else should be simple traversals
+				diags = append(diags, &hcl.Diagnostic{
+					Severity: hcl.DiagError,
+					Summary:  "Invalid reference in replace_triggered_by expression",
+					Detail:   "Only resources, count.index, and each.key may be used in replace_triggered_by.",
+					Subject:  expr.Range().Ptr(),
+				})
+			}
+		}
+
+		switch {
+		case resourceCount == 0:
+			diags = append(diags, &hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "Invalid replace_triggered_by expression",
+				Detail:   "Missing resource reference in replace_triggered_by expression.",
+				Subject:  expr.Range().Ptr(),
+			})
+		case resourceCount > 1:
+			diags = append(diags, &hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "Invalid replace_triggered_by expression",
+				Detail:   "Multiple resource references in replace_triggered_by expression.",
+				Subject:  expr.Range().Ptr(),
+			})
+		}
+	}
+	return exprs, diags
+}
+
 type ProviderConfigRef struct {
 	Name       string
 	NameRange  hcl.Range
@@ -640,6 +760,9 @@ var resourceLifecycleBlockSchema = &hcl.BodySchema{
 		{
 			Name: "ignore_changes",
 		},
+		{
+			Name: "replace_triggered_by",
+		},
 	},
 	Blocks: []hcl.BlockHeaderSchema{
 		{Type: "precondition"},

internal/configs/testdata/invalid-files/triggered-invalid-each.tf

@@ -0,0 +1,7 @@
+resource "test_resource" "a" {
+  for_each = var.input
+  lifecycle {
+    // cannot use each.val
+    replace_triggered_by = [ test_resource.b[each.val] ]
+  }
+}

internal/configs/testdata/invalid-files/triggered-invalid-expression.tf

@@ -0,0 +1,6 @@
+resource "test_resource" "a" {
+  count = 1
+  lifecycle {
+    replace_triggered_by = [ not_a_reference ]
+  }
+}

internal/configs/testdata/valid-files/resources.tf

@@ -25,6 +25,7 @@ resource "aws_security_group" "firewall" {
 }
 
 resource "aws_instance" "web" {
+  count = 2
   ami = "ami-1234"
   security_groups = [
     "foo",
@@ -40,3 +41,9 @@ resource "aws_instance" "web" {
     aws_security_group.firewall,
   ]
 }
+
+resource "aws_instance" "depends" {
+  lifecycle {
+    replace_triggered_by = [ aws_instance.web[1], aws_security_group.firewall.id ]
+  }
+}

internal/configs/testdata/valid-files/resources.tf.json

@@ -0,0 +1,18 @@
+{
+  "resource": {
+    "test_object": {
+      "a": {
+        "count": 1,
+        "test_string": "new"
+      },
+      "b": {
+        "count": 1,
+        "lifecycle": {
+          "replace_triggered_by": [
+            "test_object.a[count.index].test_string"
+          ]
+        }
+      }
+    }
+  }
+}

internal/plans/changes.go

@@ -61,6 +61,21 @@ func (c *Changes) ResourceInstance(addr addrs.AbsResourceInstance) *ResourceInst
 
 }
 
+// InstancesForAbsResource returns the planned change for the current objects
+// of the resource instances of the given address, if any. Returns nil if no
+// changes are planned.
+func (c *Changes) InstancesForAbsResource(addr addrs.AbsResource) []*ResourceInstanceChangeSrc {
+	var changes []*ResourceInstanceChangeSrc
+	for _, rc := range c.Resources {
+		resAddr := rc.Addr.ContainingResource()
+		if resAddr.Equal(addr) && rc.DeposedKey == states.NotDeposed {
+			changes = append(changes, rc)
+		}
+	}
+
+	return changes
+}
+
 // InstancesForConfigResource returns the planned change for the current objects
 // of the resource instances of the given address, if any. Returns nil if no
 // changes are planned.
@@ -345,6 +360,11 @@ const (
 	// planning option.)
 	ResourceInstanceReplaceByRequest ResourceInstanceChangeActionReason = 'R'
 
+	// ResourceInstanceReplaceByTriggers indicates that the resource instance
+	// is planned to be replaced because of a corresponding change in a
+	// replace_triggered_by reference.
+	ResourceInstanceReplaceByTriggers ResourceInstanceChangeActionReason = 'D'
+
 	// ResourceInstanceReplaceBecauseCannotUpdate indicates that the resource
 	// instance is planned to be replaced because the provider has indicated
 	// that a requested change cannot be applied as an update.

internal/plans/changes_sync.go

@@ -80,7 +80,7 @@ func (cs *ChangesSync) GetResourceInstanceChange(addr addrs.AbsResourceInstance,
 	panic(fmt.Sprintf("unsupported generation value %#v", gen))
 }
 
-// GetChangesForConfigResource searched the set of resource instance
+// GetChangesForConfigResource searches the set of resource instance
 // changes and returns all changes related to a given configuration address.
 // This is be used to find possible changes related to a configuration
 // reference.
@@ -103,6 +103,27 @@ func (cs *ChangesSync) GetChangesForConfigResource(addr addrs.ConfigResource) []
 	return changes
 }
 
+// GetChangesForAbsResource searches the set of resource instance
+// changes and returns all changes related to a given configuration address.
+//
+// If no such changes exist, nil is returned.
+//
+// The returned objects are a deep copy of the change recorded in the plan, so
+// callers may mutate them although it's generally better (less confusing) to
+// treat planned changes as immutable after they've been initially constructed.
+func (cs *ChangesSync) GetChangesForAbsResource(addr addrs.AbsResource) []*ResourceInstanceChangeSrc {
+	if cs == nil {
+		panic("GetChangesForAbsResource on nil ChangesSync")
+	}
+	cs.lock.Lock()
+	defer cs.lock.Unlock()
+	var changes []*ResourceInstanceChangeSrc
+	for _, c := range cs.changes.InstancesForAbsResource(addr) {
+		changes = append(changes, c.DeepCopy())
+	}
+	return changes
+}
+
 // RemoveResourceInstanceChange searches the set of resource instance changes
 // for one matching the given address and generation, and removes it from the
 // set if it exists.

internal/plans/internal/planproto/planfile.pb.go

@@ -149,6 +149,7 @@ const (
 	ResourceInstanceActionReason_DELETE_BECAUSE_COUNT_INDEX        ResourceInstanceActionReason = 6
 	ResourceInstanceActionReason_DELETE_BECAUSE_EACH_KEY           ResourceInstanceActionReason = 7
 	ResourceInstanceActionReason_DELETE_BECAUSE_NO_MODULE          ResourceInstanceActionReason = 8
+	ResourceInstanceActionReason_REPLACE_BY_TRIGGERS               ResourceInstanceActionReason = 9
 )
 
 // Enum value maps for ResourceInstanceActionReason.
@@ -163,6 +164,7 @@ var (
 		6: "DELETE_BECAUSE_COUNT_INDEX",
 		7: "DELETE_BECAUSE_EACH_KEY",
 		8: "DELETE_BECAUSE_NO_MODULE",
+		9: "REPLACE_BY_TRIGGERS",
 	}
 	ResourceInstanceActionReason_value = map[string]int32{
 		"NONE":                              0,
@@ -174,6 +176,7 @@ var (
 		"DELETE_BECAUSE_COUNT_INDEX":        6,
 		"DELETE_BECAUSE_EACH_KEY":           7,
 		"DELETE_BECAUSE_NO_MODULE":          8,
+		"REPLACE_BY_TRIGGERS":               9,
 	}
 )
 
@@ -1297,7 +1300,7 @@ var file_planfile_proto_rawDesc = []byte{
 	0x45, 0x10, 0x05, 0x12, 0x16, 0x0a, 0x12, 0x44, 0x45, 0x4c, 0x45, 0x54, 0x45, 0x5f, 0x54, 0x48,
 	0x45, 0x4e, 0x5f, 0x43, 0x52, 0x45, 0x41, 0x54, 0x45, 0x10, 0x06, 0x12, 0x16, 0x0a, 0x12, 0x43,
 	0x52, 0x45, 0x41, 0x54, 0x45, 0x5f, 0x54, 0x48, 0x45, 0x4e, 0x5f, 0x44, 0x45, 0x4c, 0x45, 0x54,
-	0x45, 0x10, 0x07, 0x2a, 0xa7, 0x02, 0x0a, 0x1c, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
+	0x45, 0x10, 0x07, 0x2a, 0xc0, 0x02, 0x0a, 0x1c, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
 	0x49, 0x6e, 0x73, 0x74, 0x61, 0x6e, 0x63, 0x65, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65,
 	0x61, 0x73, 0x6f, 0x6e, 0x12, 0x08, 0x0a, 0x04, 0x4e, 0x4f, 0x4e, 0x45, 0x10, 0x00, 0x12, 0x1b,
 	0x0a, 0x17, 0x52, 0x45, 0x50, 0x4c, 0x41, 0x43, 0x45, 0x5f, 0x42, 0x45, 0x43, 0x41, 0x55, 0x53,
@@ -1315,19 +1318,20 @@ var file_planfile_proto_rawDesc = []byte{
 	0x10, 0x06, 0x12, 0x1b, 0x0a, 0x17, 0x44, 0x45, 0x4c, 0x45, 0x54, 0x45, 0x5f, 0x42, 0x45, 0x43,
 	0x41, 0x55, 0x53, 0x45, 0x5f, 0x45, 0x41, 0x43, 0x48, 0x5f, 0x4b, 0x45, 0x59, 0x10, 0x07, 0x12,
 	0x1c, 0x0a, 0x18, 0x44, 0x45, 0x4c, 0x45, 0x54, 0x45, 0x5f, 0x42, 0x45, 0x43, 0x41, 0x55, 0x53,
-	0x45, 0x5f, 0x4e, 0x4f, 0x5f, 0x4d, 0x4f, 0x44, 0x55, 0x4c, 0x45, 0x10, 0x08, 0x2a, 0x6c, 0x0a,
-	0x0d, 0x43, 0x6f, 0x6e, 0x64, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0b,
-	0x0a, 0x07, 0x49, 0x4e, 0x56, 0x41, 0x4c, 0x49, 0x44, 0x10, 0x00, 0x12, 0x19, 0x0a, 0x15, 0x52,
-	0x45, 0x53, 0x4f, 0x55, 0x52, 0x43, 0x45, 0x5f, 0x50, 0x52, 0x45, 0x43, 0x4f, 0x4e, 0x44, 0x49,
-	0x54, 0x49, 0x4f, 0x4e, 0x10, 0x01, 0x12, 0x1a, 0x0a, 0x16, 0x52, 0x45, 0x53, 0x4f, 0x55, 0x52,
-	0x43, 0x45, 0x5f, 0x50, 0x4f, 0x53, 0x54, 0x43, 0x4f, 0x4e, 0x44, 0x49, 0x54, 0x49, 0x4f, 0x4e,
-	0x10, 0x02, 0x12, 0x17, 0x0a, 0x13, 0x4f, 0x55, 0x54, 0x50, 0x55, 0x54, 0x5f, 0x50, 0x52, 0x45,
-	0x43, 0x4f, 0x4e, 0x44, 0x49, 0x54, 0x49, 0x4f, 0x4e, 0x10, 0x03, 0x42, 0x42, 0x5a, 0x40, 0x67,
-	0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63,
-	0x6f, 0x72, 0x70, 0x2f, 0x74, 0x65, 0x72, 0x72, 0x61, 0x66, 0x6f, 0x72, 0x6d, 0x2f, 0x69, 0x6e,
-	0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x70, 0x6c, 0x61, 0x6e, 0x73, 0x2f, 0x69, 0x6e, 0x74,
-	0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x70, 0x6c, 0x61, 0x6e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62,
-	0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x45, 0x5f, 0x4e, 0x4f, 0x5f, 0x4d, 0x4f, 0x44, 0x55, 0x4c, 0x45, 0x10, 0x08, 0x12, 0x17, 0x0a,
+	0x13, 0x52, 0x45, 0x50, 0x4c, 0x41, 0x43, 0x45, 0x5f, 0x42, 0x59, 0x5f, 0x54, 0x52, 0x49, 0x47,
+	0x47, 0x45, 0x52, 0x53, 0x10, 0x09, 0x2a, 0x6c, 0x0a, 0x0d, 0x43, 0x6f, 0x6e, 0x64, 0x69, 0x74,
+	0x69, 0x6f, 0x6e, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0b, 0x0a, 0x07, 0x49, 0x4e, 0x56, 0x41, 0x4c,
+	0x49, 0x44, 0x10, 0x00, 0x12, 0x19, 0x0a, 0x15, 0x52, 0x45, 0x53, 0x4f, 0x55, 0x52, 0x43, 0x45,
+	0x5f, 0x50, 0x52, 0x45, 0x43, 0x4f, 0x4e, 0x44, 0x49, 0x54, 0x49, 0x4f, 0x4e, 0x10, 0x01, 0x12,
+	0x1a, 0x0a, 0x16, 0x52, 0x45, 0x53, 0x4f, 0x55, 0x52, 0x43, 0x45, 0x5f, 0x50, 0x4f, 0x53, 0x54,
+	0x43, 0x4f, 0x4e, 0x44, 0x49, 0x54, 0x49, 0x4f, 0x4e, 0x10, 0x02, 0x12, 0x17, 0x0a, 0x13, 0x4f,
+	0x55, 0x54, 0x50, 0x55, 0x54, 0x5f, 0x50, 0x52, 0x45, 0x43, 0x4f, 0x4e, 0x44, 0x49, 0x54, 0x49,
+	0x4f, 0x4e, 0x10, 0x03, 0x42, 0x42, 0x5a, 0x40, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63,
+	0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x74, 0x65, 0x72,
+	0x72, 0x61, 0x66, 0x6f, 0x72, 0x6d, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f,
+	0x70, 0x6c, 0x61, 0x6e, 0x73, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x70,
+	0x6c, 0x61, 0x6e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (

internal/plans/internal/planproto/planfile.proto

@@ -146,6 +146,7 @@ enum ResourceInstanceActionReason {
     DELETE_BECAUSE_COUNT_INDEX = 6;
     DELETE_BECAUSE_EACH_KEY = 7;
     DELETE_BECAUSE_NO_MODULE = 8;
+    REPLACE_BY_TRIGGERS = 9;
 }
 
 message ResourceInstanceChange {

internal/plans/planfile/tfplan.go

@@ -266,6 +266,8 @@ func resourceChangeFromTfplan(rawChange *planproto.ResourceInstanceChange) (*pla
 		ret.ActionReason = plans.ResourceInstanceReplaceBecauseTainted
 	case planproto.ResourceInstanceActionReason_REPLACE_BY_REQUEST:
 		ret.ActionReason = plans.ResourceInstanceReplaceByRequest
+	case planproto.ResourceInstanceActionReason_REPLACE_BY_TRIGGERS:
+		ret.ActionReason = plans.ResourceInstanceReplaceByTriggers
 	case planproto.ResourceInstanceActionReason_DELETE_BECAUSE_NO_RESOURCE_CONFIG:
 		ret.ActionReason = plans.ResourceInstanceDeleteBecauseNoResourceConfig
 	case planproto.ResourceInstanceActionReason_DELETE_BECAUSE_WRONG_REPETITION:
@@ -611,6 +613,8 @@ func resourceChangeToTfplan(change *plans.ResourceInstanceChangeSrc) (*planproto
 		ret.ActionReason = planproto.ResourceInstanceActionReason_REPLACE_BECAUSE_TAINTED
 	case plans.ResourceInstanceReplaceByRequest:
 		ret.ActionReason = planproto.ResourceInstanceActionReason_REPLACE_BY_REQUEST
+	case plans.ResourceInstanceReplaceByTriggers:
+		ret.ActionReason = planproto.ResourceInstanceActionReason_REPLACE_BY_TRIGGERS
 	case plans.ResourceInstanceDeleteBecauseNoResourceConfig:
 		ret.ActionReason = planproto.ResourceInstanceActionReason_DELETE_BECAUSE_NO_RESOURCE_CONFIG
 	case plans.ResourceInstanceDeleteBecauseWrongRepetition:

internal/plans/resourceinstancechangeactionreason_string.go

@@ -11,6 +11,7 @@ func _() {
 	_ = x[ResourceInstanceChangeNoReason-0]
 	_ = x[ResourceInstanceReplaceBecauseTainted-84]
 	_ = x[ResourceInstanceReplaceByRequest-82]
+	_ = x[ResourceInstanceReplaceByTriggers-68]
 	_ = x[ResourceInstanceReplaceBecauseCannotUpdate-70]
 	_ = x[ResourceInstanceDeleteBecauseNoResourceConfig-78]
 	_ = x[ResourceInstanceDeleteBecauseWrongRepetition-87]
@@ -21,37 +22,34 @@ func _() {
 
 const (
 	_ResourceInstanceChangeActionReason_name_0 = "ResourceInstanceChangeNoReason"
-	_ResourceInstanceChangeActionReason_name_1 = "ResourceInstanceDeleteBecauseCountIndex"
-	_ResourceInstanceChangeActionReason_name_2 = "ResourceInstanceDeleteBecauseEachKeyResourceInstanceReplaceBecauseCannotUpdate"
-	_ResourceInstanceChangeActionReason_name_3 = "ResourceInstanceDeleteBecauseNoModuleResourceInstanceDeleteBecauseNoResourceConfig"
-	_ResourceInstanceChangeActionReason_name_4 = "ResourceInstanceReplaceByRequest"
-	_ResourceInstanceChangeActionReason_name_5 = "ResourceInstanceReplaceBecauseTainted"
-	_ResourceInstanceChangeActionReason_name_6 = "ResourceInstanceDeleteBecauseWrongRepetition"
+	_ResourceInstanceChangeActionReason_name_1 = "ResourceInstanceDeleteBecauseCountIndexResourceInstanceReplaceByTriggersResourceInstanceDeleteBecauseEachKeyResourceInstanceReplaceBecauseCannotUpdate"
+	_ResourceInstanceChangeActionReason_name_2 = "ResourceInstanceDeleteBecauseNoModuleResourceInstanceDeleteBecauseNoResourceConfig"
+	_ResourceInstanceChangeActionReason_name_3 = "ResourceInstanceReplaceByRequest"
+	_ResourceInstanceChangeActionReason_name_4 = "ResourceInstanceReplaceBecauseTainted"
+	_ResourceInstanceChangeActionReason_name_5 = "ResourceInstanceDeleteBecauseWrongRepetition"
 )
 
 var (
-	_ResourceInstanceChangeActionReason_index_2 = [...]uint8{0, 36, 78}
-	_ResourceInstanceChangeActionReason_index_3 = [...]uint8{0, 37, 82}
+	_ResourceInstanceChangeActionReason_index_1 = [...]uint8{0, 39, 72, 108, 150}
+	_ResourceInstanceChangeActionReason_index_2 = [...]uint8{0, 37, 82}
 )
 
 func (i ResourceInstanceChangeActionReason) String() string {
 	switch {
 	case i == 0:
 		return _ResourceInstanceChangeActionReason_name_0
-	case i == 67:
-		return _ResourceInstanceChangeActionReason_name_1
-	case 69 <= i && i <= 70:
-		i -= 69
-		return _ResourceInstanceChangeActionReason_name_2[_ResourceInstanceChangeActionReason_index_2[i]:_ResourceInstanceChangeActionReason_index_2[i+1]]
+	case 67 <= i && i <= 70:
+		i -= 67
+		return _ResourceInstanceChangeActionReason_name_1[_ResourceInstanceChangeActionReason_index_1[i]:_ResourceInstanceChangeActionReason_index_1[i+1]]
 	case 77 <= i && i <= 78:
 		i -= 77
-		return _ResourceInstanceChangeActionReason_name_3[_ResourceInstanceChangeActionReason_index_3[i]:_ResourceInstanceChangeActionReason_index_3[i+1]]
+		return _ResourceInstanceChangeActionReason_name_2[_ResourceInstanceChangeActionReason_index_2[i]:_ResourceInstanceChangeActionReason_index_2[i+1]]
 	case i == 82:
-		return _ResourceInstanceChangeActionReason_name_4
+		return _ResourceInstanceChangeActionReason_name_3
 	case i == 84:
-		return _ResourceInstanceChangeActionReason_name_5
+		return _ResourceInstanceChangeActionReason_name_4
 	case i == 87:
-		return _ResourceInstanceChangeActionReason_name_6
+		return _ResourceInstanceChangeActionReason_name_5
 	default:
 		return "ResourceInstanceChangeActionReason(" + strconv.FormatInt(int64(i), 10) + ")"
 	}

internal/terraform/context_plan2_test.go

@@ -2964,6 +2964,76 @@ output "a" {
 	}
 }
 
+func TestContext2Plan_triggeredBy(t *testing.T) {
+	m := testModuleInline(t, map[string]string{
+		"main.tf": `
+resource "test_object" "a" {
+  count = 1
+  test_string = "new"
+}
+resource "test_object" "b" {
+  count = 1
+  test_string = test_object.a[count.index].test_string
+  lifecycle {
+    # the change to test_string in the other resource should trigger replacement
+    replace_triggered_by = [ test_object.a[count.index].test_string ]
+  }
+}
+`,
+	})
+
+	p := simpleMockProvider()
+
+	ctx := testContext2(t, &ContextOpts{
+		Providers: map[addrs.Provider]providers.Factory{
+			addrs.NewDefaultProvider("test"): testProviderFuncFixed(p),
+		},
+	})
+
+	state := states.BuildState(func(s *states.SyncState) {
+		s.SetResourceInstanceCurrent(
+			mustResourceInstanceAddr("test_object.a[0]"),
+			&states.ResourceInstanceObjectSrc{
+				AttrsJSON: []byte(`{"test_string":"old"}`),
+				Status:    states.ObjectReady,
+			},
+			mustProviderConfig(`provider["registry.terraform.io/hashicorp/test"]`),
+		)
+		s.SetResourceInstanceCurrent(
+			mustResourceInstanceAddr("test_object.b[0]"),
+			&states.ResourceInstanceObjectSrc{
+				AttrsJSON: []byte(`{}`),
+				Status:    states.ObjectReady,
+			},
+			mustProviderConfig(`provider["registry.terraform.io/hashicorp/test"]`),
+		)
+	})
+
+	plan, diags := ctx.Plan(m, state, &PlanOpts{
+		Mode: plans.NormalMode,
+	})
+	if diags.HasErrors() {
+		t.Fatalf("unexpected errors\n%s", diags.Err().Error())
+	}
+	for _, c := range plan.Changes.Resources {
+		switch c.Addr.String() {
+		case "test_object.a[0]":
+			if c.Action != plans.Update {
+				t.Fatalf("unexpected %s change for %s\n", c.Action, c.Addr)
+			}
+		case "test_object.b[0]":
+			if c.Action != plans.DeleteThenCreate {
+				t.Fatalf("unexpected %s change for %s\n", c.Action, c.Addr)
+			}
+			if c.ActionReason != plans.ResourceInstanceReplaceByTriggers {
+				t.Fatalf("incorrect reason for change: %s\n", c.ActionReason)
+			}
+		default:
+			t.Fatal("unexpected change", c.Addr, c.Action)
+		}
+	}
+}
+
 func TestContext2Plan_dataSchemaChange(t *testing.T) {
 	// We can't decode the prior state when a data source upgrades the schema
 	// in an incompatible way. Since prior state for data sources is purely

internal/terraform/eval_context.go

@@ -117,6 +117,11 @@ type EvalContext interface {
 	// evaluating. Set this to nil if the "self" object should not be available.
 	EvaluateExpr(expr hcl.Expression, wantType cty.Type, self addrs.Referenceable) (cty.Value, tfdiags.Diagnostics)
 
+	// EvaluateReplaceTriggeredBy takes the raw reference expression from the
+	// config, and returns the evaluated *addrs.Reference along with a boolean
+	// indicating if that reference forces replacement.
+	EvaluateReplaceTriggeredBy(expr hcl.Expression, repData instances.RepetitionData) (*addrs.Reference, bool, tfdiags.Diagnostics)
+
 	// EvaluationScope returns a scope that can be used to evaluate reference
 	// addresses in this context.
 	EvaluationScope(self addrs.Referenceable, keyData InstanceKeyEvalData) *lang.Scope

internal/terraform/eval_context_builtin.go

@@ -282,7 +282,121 @@ func (ctx *BuiltinEvalContext) EvaluateExpr(expr hcl.Expression, wantType cty.Ty
 	return scope.EvalExpr(expr, wantType)
 }
 
-func (ctx *BuiltinEvalContext) EvaluationScope(self addrs.Referenceable, keyData InstanceKeyEvalData) *lang.Scope {
+func (ctx *BuiltinEvalContext) EvaluateReplaceTriggeredBy(expr hcl.Expression, repData instances.RepetitionData) (*addrs.Reference, bool, tfdiags.Diagnostics) {
+
+	// get the reference to lookup changes in the plan
+	ref, diags := evalReplaceTriggeredByExpr(expr, repData)
+	if diags.HasErrors() {
+		return nil, false, diags
+	}
+
+	var changes []*plans.ResourceInstanceChangeSrc
+	// store the address once we get it for validation
+	var resourceAddr addrs.Resource
+
+	// The reference is either a resource or resource instance
+	switch sub := ref.Subject.(type) {
+	case addrs.Resource:
+		resourceAddr = sub
+		rc := sub.Absolute(ctx.Path())
+		changes = ctx.Changes().GetChangesForAbsResource(rc)
+	case addrs.ResourceInstance:
+		resourceAddr = sub.ContainingResource()
+		rc := sub.Absolute(ctx.Path())
+		change := ctx.Changes().GetResourceInstanceChange(rc, states.CurrentGen)
+		if change != nil {
+			// we'll generate an error below if there was no change
+			changes = append(changes, change)
+		}
+	}
+
+	// Do some validation to make sure we are expecting a change at all
+	cfg := ctx.Evaluator.Config.Descendent(ctx.Path().Module())
+	resCfg := cfg.Module.ResourceByAddr(resourceAddr)
+	if resCfg == nil {
+		diags = diags.Append(&hcl.Diagnostic{
+			Severity: hcl.DiagError,
+			Summary:  `Reference to undeclared resource`,
+			Detail:   fmt.Sprintf(`A resource %s has not been declared in %s`, ref.Subject, moduleDisplayAddr(ctx.Path())),
+			Subject:  expr.Range().Ptr(),
+		})
+		return nil, false, diags
+	}
+
+	if len(changes) == 0 {
+		// If the resource is valid there should always be at least one change.
+		diags = diags.Append(fmt.Errorf("no change found for %s in %s", ref.Subject, moduleDisplayAddr(ctx.Path())))
+		return nil, false, diags
+	}
+
+	// If we don't have a traversal beyond the resource, then we can just look
+	// for any change.
+	if len(ref.Remaining) == 0 {
+		for _, c := range changes {
+			switch c.ChangeSrc.Action {
+			// Only immediate changes to the resource will trigger replacement.
+			case plans.Update, plans.DeleteThenCreate, plans.CreateThenDelete:
+				return ref, true, diags
+			}
+		}
+
+		// no change triggered
+		return nil, false, diags
+	}
+
+	// This must be an instances to have a remaining traversal, which means a
+	// single change.
+	change := changes[0]
+
+	// Make sure the change is actionable. A create or delete action will have
+	// a change in value, but are not valid for our purposes here.
+	switch change.ChangeSrc.Action {
+	case plans.Update, plans.DeleteThenCreate, plans.CreateThenDelete:
+		// OK
+	default:
+		return nil, false, diags
+	}
+
+	// Since we have a traversal after the resource reference, we will need to
+	// decode the changes, which means we need a schema.
+	providerAddr := change.ProviderAddr
+	schema, err := ctx.ProviderSchema(providerAddr)
+	if err != nil {
+		diags = diags.Append(err)
+		return nil, false, diags
+	}
+
+	resAddr := change.Addr.ContainingResource().Resource
+	resSchema, _ := schema.SchemaForResourceType(resAddr.Mode, resAddr.Type)
+	ty := resSchema.ImpliedType()
+
+	before, err := change.ChangeSrc.Before.Decode(ty)
+	if err != nil {
+		diags = diags.Append(err)
+		return nil, false, diags
+	}
+
+	after, err := change.ChangeSrc.After.Decode(ty)
+	if err != nil {
+		diags = diags.Append(err)
+		return nil, false, diags
+	}
+
+	path := traversalToPath(ref.Remaining)
+	attrBefore, _ := path.Apply(before)
+	attrAfter, _ := path.Apply(after)
+
+	if attrBefore == cty.NilVal || attrAfter == cty.NilVal {
+		replace := attrBefore != attrAfter
+		return ref, replace, diags
+	}
+
+	replace := !attrBefore.RawEquals(attrAfter)
+
+	return ref, replace, diags
+}
+
+func (ctx *BuiltinEvalContext) EvaluationScope(self addrs.Referenceable, keyData instances.RepetitionData) *lang.Scope {
 	if !ctx.pathSet {
 		panic("context path not set")
 	}

internal/terraform/eval_context_mock.go

@@ -261,6 +261,10 @@ func (c *MockEvalContext) EvaluateExpr(expr hcl.Expression, wantType cty.Type, s
 	return c.EvaluateExprResult, c.EvaluateExprDiags
 }
 
+func (c *MockEvalContext) EvaluateReplaceTriggeredBy(hcl.Expression, instances.RepetitionData) (*addrs.Reference, bool, tfdiags.Diagnostics) {
+	return nil, false, nil
+}
+
 // installSimpleEval is a helper to install a simple mock implementation of
 // both EvaluateBlock and EvaluateExpr into the receiver.
 //

internal/terraform/evaluate_triggers.go

@@ -0,0 +1,143 @@
+package terraform
+
+import (
+	"strings"
+
+	"github.com/hashicorp/hcl/v2"
+	"github.com/hashicorp/hcl/v2/hclsyntax"
+	"github.com/hashicorp/terraform/internal/addrs"
+	"github.com/hashicorp/terraform/internal/instances"
+	"github.com/hashicorp/terraform/internal/tfdiags"
+	"github.com/zclconf/go-cty/cty"
+)
+
+func evalReplaceTriggeredByExpr(expr hcl.Expression, keyData instances.RepetitionData) (*addrs.Reference, tfdiags.Diagnostics) {
+	var ref *addrs.Reference
+	var diags tfdiags.Diagnostics
+
+	traversal, diags := triggersExprToTraversal(expr, keyData)
+	if diags.HasErrors() {
+		return nil, diags
+	}
+
+	// We now have a static traversal, so we can just turn it into an addrs.Reference.
+	ref, ds := addrs.ParseRef(traversal)
+	diags = diags.Append(ds)
+
+	return ref, diags
+}
+
+// trggersExprToTraversal takes an hcl expression limited to the syntax allowed
+// in replace_triggered_by, and converts it to a static traversal. The
+// RepetitionData contains the data necessary to evaluate the only allowed
+// variables in the expression, count.index and each.key.
+func triggersExprToTraversal(expr hcl.Expression, keyData instances.RepetitionData) (hcl.Traversal, tfdiags.Diagnostics) {
+	var trav hcl.Traversal
+	var diags tfdiags.Diagnostics
+
+	switch e := expr.(type) {
+	case *hclsyntax.RelativeTraversalExpr:
+		t, d := triggersExprToTraversal(e.Source, keyData)
+		diags = diags.Append(d)
+		trav = append(trav, t...)
+		trav = append(trav, e.Traversal...)
+
+	case *hclsyntax.ScopeTraversalExpr:
+		// a static reference, we can just append the traversal
+		trav = append(trav, e.Traversal...)
+
+	case *hclsyntax.IndexExpr:
+		// Get the collection from the index expression
+		t, d := triggersExprToTraversal(e.Collection, keyData)
+		diags = diags.Append(d)
+		if diags.HasErrors() {
+			return nil, diags
+		}
+		trav = append(trav, t...)
+
+		// The index key is the only place where we could have variables that
+		// reference count and each, so we need to parse those independently.
+		idx, hclDiags := parseIndexKeyExpr(e.Key, keyData)
+		diags = diags.Append(hclDiags)
+
+		trav = append(trav, idx)
+
+	default:
+		// Something unexpected got through config validation. We're not sure
+		// what it is, but we'll point it out in the diagnostics for the user
+		// to fix.
+		diags = diags.Append(&hcl.Diagnostic{
+			Severity: hcl.DiagError,
+			Summary:  "Invalid replace_triggered_by expression",
+			Detail:   "Unexpected expression found in replace_triggered_by.",
+			Subject:  e.Range().Ptr(),
+		})
+	}
+
+	return trav, diags
+}
+
+// parseIndexKeyExpr takes an hcl.Expression and parses it as an index key, while
+// evaluating any references to count.index or each.key.
+func parseIndexKeyExpr(expr hcl.Expression, keyData instances.RepetitionData) (hcl.TraverseIndex, hcl.Diagnostics) {
+	idx := hcl.TraverseIndex{
+		SrcRange: expr.Range(),
+	}
+
+	trav, diags := hcl.RelTraversalForExpr(expr)
+	if diags.HasErrors() {
+		return idx, diags
+	}
+
+	keyParts := []string{}
+
+	for _, t := range trav {
+		attr, ok := t.(hcl.TraverseAttr)
+		if !ok {
+			diags = append(diags, &hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "Invalid index expression",
+				Detail:   "Only constant values, count.index or each.key are allowed in index expressions.",
+				Subject:  expr.Range().Ptr(),
+			})
+			return idx, diags
+		}
+		keyParts = append(keyParts, attr.Name)
+	}
+
+	switch strings.Join(keyParts, ".") {
+	case "count.index":
+		if keyData.CountIndex == cty.NilVal {
+			diags = diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  `Reference to "count" in non-counted context`,
+				Detail:   `The "count" object can only be used in "resource" blocks when the "count" argument is set.`,
+				Subject:  expr.Range().Ptr(),
+			})
+		}
+		idx.Key = keyData.CountIndex
+
+	case "each.key":
+		if keyData.EachKey == cty.NilVal {
+			diags = diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  `Reference to "each" in context without for_each`,
+				Detail:   `The "each" object can be used only in "resource" blocks when the "for_each" argument is set.`,
+				Subject:  expr.Range().Ptr(),
+			})
+		}
+		idx.Key = keyData.EachKey
+	default:
+		// Something may have slipped through validation, probably from a json
+		// configuration.
+		diags = append(diags, &hcl.Diagnostic{
+			Severity: hcl.DiagError,
+			Summary:  "Invalid index expression",
+			Detail:   "Only constant values, count.index or each.key are allowed in index expressions.",
+			Subject:  expr.Range().Ptr(),
+		})
+	}
+
+	return idx, diags
+
+}

internal/terraform/evaluate_triggers_test.go

@@ -0,0 +1,94 @@
+package terraform
+
+import (
+	"testing"
+
+	"github.com/hashicorp/hcl/v2"
+	"github.com/hashicorp/hcl/v2/hclsyntax"
+	"github.com/hashicorp/terraform/internal/addrs"
+	"github.com/hashicorp/terraform/internal/instances"
+	"github.com/zclconf/go-cty/cty"
+)
+
+func TestEvalReplaceTriggeredBy(t *testing.T) {
+	tests := map[string]struct {
+		// Raw config expression from within replace_triggered_by list.
+		// If this does not contains any count or each references, it should
+		// directly parse into the same *addrs.Reference.
+		expr string
+
+		// If the expression contains count or each, then we need to add
+		// repetition data, and the static string to parse into the desired
+		// *addrs.Reference
+		repData   instances.RepetitionData
+		reference string
+	}{
+		"single resource": {
+			expr: "test_resource.a",
+		},
+
+		"resource instance attr": {
+			expr: "test_resource.a.attr",
+		},
+
+		"resource instance index attr": {
+			expr: "test_resource.a[0].attr",
+		},
+
+		"resource instance count": {
+			expr: "test_resource.a[count.index]",
+			repData: instances.RepetitionData{
+				CountIndex: cty.NumberIntVal(0),
+			},
+			reference: "test_resource.a[0]",
+		},
+		"resource instance for_each": {
+			expr: "test_resource.a[each.key].attr",
+			repData: instances.RepetitionData{
+				EachKey: cty.StringVal("k"),
+			},
+			reference: `test_resource.a["k"].attr`,
+		},
+		"resource instance for_each map attr": {
+			expr: "test_resource.a[each.key].attr[each.key]",
+			repData: instances.RepetitionData{
+				EachKey: cty.StringVal("k"),
+			},
+			reference: `test_resource.a["k"].attr["k"]`,
+		},
+	}
+
+	for name, tc := range tests {
+		pos := hcl.Pos{Line: 1, Column: 1}
+		t.Run(name, func(t *testing.T) {
+			expr, hclDiags := hclsyntax.ParseExpression([]byte(tc.expr), "", pos)
+			if hclDiags.HasErrors() {
+				t.Fatal(hclDiags)
+			}
+
+			got, diags := evalReplaceTriggeredByExpr(expr, tc.repData)
+			if diags.HasErrors() {
+				t.Fatal(diags.Err())
+			}
+
+			want := tc.reference
+			if want == "" {
+				want = tc.expr
+			}
+
+			// create the desired reference
+			traversal, travDiags := hclsyntax.ParseTraversalAbs([]byte(want), "", pos)
+			if travDiags.HasErrors() {
+				t.Fatal(travDiags)
+			}
+			ref, diags := addrs.ParseRef(traversal)
+			if diags.HasErrors() {
+				t.Fatal(diags.Err())
+			}
+
+			if got.DisplayString() != ref.DisplayString() {
+				t.Fatalf("expected %q: got %q", ref.DisplayString(), got.DisplayString())
+			}
+		})
+	}
+}

internal/terraform/node_resource_abstract.go

@@ -143,12 +143,17 @@ func (n *NodeAbstractResource) References() []*addrs.Reference {
 		refs, _ = lang.ReferencesInExpr(c.ForEach)
 		result = append(result, refs...)
 
+		for _, expr := range c.TriggersReplacement {
+			refs, _ = lang.ReferencesInExpr(expr)
+			result = append(result, refs...)
+		}
+
 		// ReferencesInBlock() requires a schema
 		if n.Schema != nil {
 			refs, _ = lang.ReferencesInBlock(c.Config, n.Schema)
+			result = append(result, refs...)
 		}
 
-		result = append(result, refs...)
 		if c.Managed != nil {
 			if c.Managed.Connection != nil {
 				refs, _ = lang.ReferencesInBlock(c.Managed.Connection.Config, connectionBlockSupersetSchema)

internal/terraform/node_resource_plan_instance.go

@@ -5,9 +5,11 @@ import (
 	"log"
 	"sort"
 
+	"github.com/hashicorp/terraform/internal/instances"
 	"github.com/hashicorp/terraform/internal/plans"
 	"github.com/hashicorp/terraform/internal/states"
 	"github.com/hashicorp/terraform/internal/tfdiags"
+	"github.com/zclconf/go-cty/cty"
 
 	"github.com/hashicorp/terraform/internal/addrs"
 )
@@ -31,6 +33,10 @@ type NodePlannableResourceInstance struct {
 	// it might contain addresses that have nothing to do with the resource
 	// that this node represents, which the node itself must therefore ignore.
 	forceReplace []addrs.AbsResourceInstance
+
+	// replaceTriggeredBy stores references from replace_triggered_by which
+	// triggered this instance to be replaced.
+	replaceTriggeredBy []*addrs.Reference
 }
 
 var (
@@ -192,6 +198,23 @@ func (n *NodePlannableResourceInstance) managedResourceExecute(ctx EvalContext)
 
 	// Plan the instance, unless we're in the refresh-only mode
 	if !n.skipPlanChanges {
+
+		// add this instance to n.forceReplace if replacement is triggered by
+		// another change
+		repData := instances.RepetitionData{}
+		switch k := addr.Resource.Key.(type) {
+		case addrs.IntKey:
+			repData.CountIndex = k.Value()
+		case addrs.StringKey:
+			repData.EachKey = k.Value()
+			repData.EachValue = cty.DynamicVal
+		}
+
+		diags = diags.Append(n.replaceTriggered(ctx, repData))
+		if diags.HasErrors() {
+			return diags
+		}
+
 		change, instancePlanState, repeatData, planDiags := n.plan(
 			ctx, change, instanceRefreshState, n.ForceCreateBeforeDestroy, n.forceReplace,
 		)
@@ -200,6 +223,13 @@ func (n *NodePlannableResourceInstance) managedResourceExecute(ctx EvalContext)
 			return diags
 		}
 
+		// FIXME: here we udpate the change to reflect the reason for
+		// replacement, but we still overload forceReplace to get the correct
+		// change planned.
+		if len(n.replaceTriggeredBy) > 0 {
+			change.ActionReason = plans.ResourceInstanceReplaceByTriggers
+		}
+
 		diags = diags.Append(n.checkPreventDestroy(change))
 		if diags.HasErrors() {
 			return diags
@@ -299,6 +329,36 @@ func (n *NodePlannableResourceInstance) managedResourceExecute(ctx EvalContext)
 	return diags
 }
 
+// replaceTriggered checks if this instance needs to be replace due to a change
+// in a replace_triggered_by reference. If replacement is required, the
+// instance address is added to forceReplace
+func (n *NodePlannableResourceInstance) replaceTriggered(ctx EvalContext, repData instances.RepetitionData) tfdiags.Diagnostics {
+	var diags tfdiags.Diagnostics
+
+	for _, expr := range n.Config.TriggersReplacement {
+		ref, replace, evalDiags := ctx.EvaluateReplaceTriggeredBy(expr, repData)
+		diags = diags.Append(evalDiags)
+		if diags.HasErrors() {
+			continue
+		}
+
+		if replace {
+			// FIXME: forceReplace accomplishes the same goal, however we may
+			// want to communicate more information about which resource
+			// triggered the replacement in the plan.
+			// Rather than further complicating the plan method with more
+			// options, we can refactor both of these features later.
+			n.forceReplace = append(n.forceReplace, n.Addr)
+			log.Printf("[DEBUG] ReplaceTriggeredBy forcing replacement of %s due to change in %s", n.Addr, ref.DisplayString())
+
+			n.replaceTriggeredBy = append(n.replaceTriggeredBy, ref)
+			break
+		}
+	}
+
+	return diags
+}
+
 // mergeDeps returns the union of 2 sets of dependencies
 func mergeDeps(a, b []addrs.ConfigResource) []addrs.ConfigResource {
 	switch {