core: interpolate the count config during the apply walk

Previously we would interpolate the count config (ResourceConfig.RawCount)
only while preparing to dynamic-expand aggregate resource nodes. This is
problematic because we do not dynamic-expand any resource nodes during the
apply walk, and so previously the count value was not available for
interpolation during apply and would result in an error.

Now we interpolate RawCount once for each resource we visit during the
apply walk -- even though that redundantly interpolates the same config
multiple times when count > 1 -- to ensure that it's available by the
time we interpolate any remaining expressions in the config and any
expressions within "connection" and "provisioner" blocks.

This error was masked by us sharing a single RawConfig instance between
the plan and apply walks when "terraform apply" is run with no explicit
plan file argument, but was exposed by the workflow where the plan is
written first to disk since in that case the interpolation result from
during the plan phase is not present in the deflated plan object. For
this reason, the new context test serializes the plan into an in-memory
buffer and reloads it in order to simulate the effect of the two-step
workflow.

terraform/context_apply_test.go

@@ -2421,6 +2421,73 @@ func TestContext2Apply_countVariableRef(t *testing.T) {
 	}
 }
 
+func TestContext2Apply_provisionerInterpCount(t *testing.T) {
+	// This test ensures that a provisioner can interpolate a resource count
+	// even though the provisioner expression is evaluated during the plan
+	// walk. https://github.com/hashicorp/terraform/issues/16840
+
+	m := testModule(t, "apply-provisioner-interp-count")
+
+	p := testProvider("aws")
+	p.ApplyFn = testApplyFn
+	p.DiffFn = testDiffFn
+
+	pr := testProvisioner()
+
+	providerResolver := ResourceProviderResolverFixed(
+		map[string]ResourceProviderFactory{
+			"aws": testProviderFuncFixed(p),
+		},
+	)
+	provisioners := map[string]ResourceProvisionerFactory{
+		"local-exec": testProvisionerFuncFixed(pr),
+	}
+	ctx := testContext2(t, &ContextOpts{
+		Module:           m,
+		ProviderResolver: providerResolver,
+		Provisioners:     provisioners,
+	})
+
+	plan, err := ctx.Plan()
+	if err != nil {
+		t.Fatalf("plan failed: %s", err)
+	}
+
+	// We'll marshal and unmarshal the plan here, to ensure that we have
+	// a clean new context as would be created if we separately ran
+	// terraform plan -out=tfplan && terraform apply tfplan
+	var planBuf bytes.Buffer
+	err = WritePlan(plan, &planBuf)
+	if err != nil {
+		t.Fatalf("failed to write plan: %s", err)
+	}
+	plan, err = ReadPlan(&planBuf)
+	if err != nil {
+		t.Fatalf("failed to read plan: %s", err)
+	}
+
+	ctx, err = plan.Context(&ContextOpts{
+		// Most options are taken from the plan in this case, but we still
+		// need to provide the plugins.
+		ProviderResolver: providerResolver,
+		Provisioners:     provisioners,
+	})
+	if err != nil {
+		t.Fatalf("failed to create context for plan: %s", err)
+	}
+
+	// Applying the plan should now succeed
+	_, err = ctx.Apply()
+	if err != nil {
+		t.Fatalf("apply failed: %s", err)
+	}
+
+	// Verify apply was invoked
+	if !pr.ApplyCalled {
+		t.Fatalf("provisioner was not applied")
+	}
+}
+
 func TestContext2Apply_mapVariableOverride(t *testing.T) {
 	m := testModule(t, "apply-map-var-override")
 	p := testProvider("aws")

terraform/node_resource_apply.go

@@ -124,6 +124,24 @@ func (n *NodeApplyableResource) evalTreeDataResource(
 				Then: EvalNoop{},
 			},
 
+			// Normally we interpolate count as a preparation step before
+			// a DynamicExpand, but an apply graph has pre-expanded nodes
+			// and so the count would otherwise never be interpolated.
+			//
+			// This is redundant when there are multiple instances created
+			// from the same config (count > 1) but harmless since the
+			// underlying structures have mutexes to make this concurrency-safe.
+			//
+			// In most cases this isn't actually needed because we dealt with
+			// all of the counts during the plan walk, but we do it here
+			// for completeness because other code assumes that the
+			// final count is always available during interpolation.
+			//
+			// Here we are just populating the interpolated value in-place
+			// inside this RawConfig object, like we would in
+			// NodeAbstractCountResource.
+			&EvalInterpolate{Config: n.Config.RawCount},
+
 			// We need to re-interpolate the config here, rather than
 			// just using the diff's values directly, because we've
 			// potentially learned more variable values during the
@@ -236,6 +254,25 @@ func (n *NodeApplyableResource) evalTreeManagedResource(
 				},
 			},
 
+			// Normally we interpolate count as a preparation step before
+			// a DynamicExpand, but an apply graph has pre-expanded nodes
+			// and so the count would otherwise never be interpolated.
+			//
+			// This is redundant when there are multiple instances created
+			// from the same config (count > 1) but harmless since the
+			// underlying structures have mutexes to make this concurrency-safe.
+			//
+			// In most cases this isn't actually needed because we dealt with
+			// all of the counts during the plan walk, but we need to do this
+			// in order to support interpolation of resource counts from
+			// apply-time-interpolated expressions, such as those in
+			// "provisioner" blocks.
+			//
+			// Here we are just populating the interpolated value in-place
+			// inside this RawConfig object, like we would in
+			// NodeAbstractCountResource.
+			&EvalInterpolate{Config: n.Config.RawCount},
+
 			&EvalInterpolate{
 				Config:   n.Config.RawConfig.Copy(),
 				Resource: resource,

terraform/test-fixtures/apply-provisioner-interp-count/provisioner-interp-count.tf

@@ -0,0 +1,17 @@
+variable "count" {
+  default = 3
+}
+
+resource "aws_instance" "a" {
+  count = "${var.count}"
+}
+
+resource "aws_instance" "b" {
+  provisioner "local-exec" {
+    # Since we're in a provisioner block here, this interpolation is
+    # resolved during the apply walk and so the resource count must
+    # be interpolated during that walk, even though apply walk doesn't
+    # do DynamicExpand.
+    command = "echo ${aws_instance.a.count}"
+  }
+}