cloud: Skip intermediate state snapshots in Terraform Cloud/Enterprise

We've seen some concern about the additional storage usage implied by
creating intermediate state snapshots for particularly long apply phases
that can arise when managing a large number of resource instances together
in a single workspace.

This is an initial coarse approach to solving that concern, just restoring
the original behavior when running inside Terraform Cloud or Enterprise
for now and not creating snapshots at all.

This is here as a solution of last resort in case we cannot find a better
compromise before the v1.5.0 final release. Hopefully a future commit
will implement a more subtle take on this which still gets some of the
benefits when running in a Terraform Enterprise environment but in a way
that will hopefully be less concerning for Terraform Enterprise
administrators.

This does not affect any other state storage implementation except the
Terraform Cloud integration and the "remote" backend's state storage when
running inside a TFC/TFE-driven remote execution environment.

internal/backend/remote/backend.go

@@ -694,7 +694,18 @@ func (b *Remote) StateMgr(name string) (statemgr.Full, error) {
 		runID: os.Getenv("TFE_RUN_ID"),
 	}
 
-	return &remote.State{Client: client}, nil
+	return &remote.State{
+		Client: client,
+
+		// client.runID will be set if we're running a the Terraform Cloud
+		// or Terraform Enterprise remote execution environment, in which
+		// case we'll disable intermediate snapshots to avoid extra storage
+		// costs for Terraform Enterprise customers.
+		// Other implementations of the remote state protocol should not run
+		// in contexts where there's a "TFE Run ID" and so are not affected
+		// by this special case.
+		DisableIntermediateSnapshots: client.runID != "",
+	}, nil
 }
 
 func isLocalExecutionMode(execMode string) bool {

internal/cloud/state.go

@@ -21,6 +21,8 @@ import (
 
 	tfe "github.com/hashicorp/go-tfe"
 	uuid "github.com/hashicorp/go-uuid"
+
+	"github.com/hashicorp/terraform/internal/backend/local"
 	"github.com/hashicorp/terraform/internal/command/jsonstate"
 	"github.com/hashicorp/terraform/internal/states"
 	"github.com/hashicorp/terraform/internal/states/remote"
@@ -66,6 +68,7 @@ remote state version.
 
 var _ statemgr.Full = (*State)(nil)
 var _ statemgr.Migrator = (*State)(nil)
+var _ local.IntermediateStateConditionalPersister = (*State)(nil)
 
 // statemgr.Reader impl.
 func (s *State) State() *states.State {
@@ -223,6 +226,14 @@ func (s *State) PersistState(schemas *terraform.Schemas) error {
 	return nil
 }
 
+// ShouldPersistIntermediateState implements local.IntermediateStateConditionalPersister
+func (*State) ShouldPersistIntermediateState(info *local.IntermediateStatePersistInfo) bool {
+	// We currently don't create intermediate snapshots for Terraform Cloud or
+	// Terraform Enterprise at all, to avoid extra storage costs for Terraform
+	// Enterprise customers.
+	return false
+}
+
 func (s *State) uploadState(lineage string, serial uint64, isForcePush bool, state, jsonState, jsonStateOutputs []byte) error {
 	ctx := context.Background()
 

internal/states/remote/state.go

@@ -11,6 +11,7 @@ import (
 
 	uuid "github.com/hashicorp/go-uuid"
 
+	"github.com/hashicorp/terraform/internal/backend/local"
 	"github.com/hashicorp/terraform/internal/states"
 	"github.com/hashicorp/terraform/internal/states/statefile"
 	"github.com/hashicorp/terraform/internal/states/statemgr"
@@ -39,10 +40,17 @@ type State struct {
 	serial, readSerial   uint64
 	state, readState     *states.State
 	disableLocks         bool
+
+	// If this is set then the state manager will decline to store intermediate
+	// state snapshots created while a Terraform Core apply operation is in
+	// progress. Otherwise (by default) it will accept persistent snapshots
+	// using the default rules defined in the local backend.
+	DisableIntermediateSnapshots bool
 }
 
 var _ statemgr.Full = (*State)(nil)
 var _ statemgr.Migrator = (*State)(nil)
+var _ local.IntermediateStateConditionalPersister = (*State)(nil)
 
 // statemgr.Reader impl.
 func (s *State) State() *states.State {
@@ -219,6 +227,14 @@ func (s *State) PersistState(schemas *terraform.Schemas) error {
 	return nil
 }
 
+// ShouldPersistIntermediateState implements local.IntermediateStateConditionalPersister
+func (s *State) ShouldPersistIntermediateState(info *local.IntermediateStatePersistInfo) bool {
+	if s.DisableIntermediateSnapshots {
+		return false
+	}
+	return local.DefaultIntermediateStatePersistRule(info)
+}
+
 // Lock calls the Client's Lock method if it's implemented.
 func (s *State) Lock(info *statemgr.LockInfo) (string, error) {
 	s.mu.Lock()