Merge cockroachdb#107145 cockroachdb#107187 cockroachdb#107203 cockroachdb#107240

107145: backupccl: add RESTORE job to wait for background downloads r=dt a=dt

After an online restore completes its initial data mounting phase, we
actually go download the data in the background, both so that it will be
faster to access once it is stored locally but also to make it
permanently owned by the cluster, rather than remaining linked to the
backup, which could be moved, expire, etc.

Users need to be able to see, observe and control this process, in
particular they need to know when it completes or will complete, so that
they know when to expect normal performance or when they can lift
modification/expiration blocks on their backups.

This change introduces new RESTORE job, created by the initial restore
job after it has linked in the external sstables, which polls the stores
to determine how much data remains to be downloaded, and completes when
this number reaches zero.

A future change will likely extend this job to also explicitly invoke a
compaction specifically configured to download these files, to make this
happen more immediately and to make it controllable by the job, but for
now this initial implementation only waits and polls to track the state
of the compaction, rather than actively compacting itself.

Release note: none.
Epic: none.

107187: cloud: fix incorrect rebase of ResumingReader open at size patch r=rhu713 a=rhu713

Recently cockroachdb#103462 was merged after incorrectly rebasing and did not correctly incorporate the addition of the NoFileSize read option for external storages. This patch fixes the rebase in GCS storage and in cloud unit tests.

Fixes: cockroachdb#107136
Fixes: cockroachdb#107138

Cloud Unit Test Manual Run: https://teamcity.cockroachdb.com/buildConfiguration/Cockroach_Nightlies_CloudUnitTests/10977689

Release note: None

107203: kvserver: enable `raft.Config.StepDownOnRemoval` r=erikgrinaker a=erikgrinaker

**kvserver: tweak `TestRaftLeaderRemovesItself`**

**kvserver: campaign when leader is demoted to learner**

This patch campaigns when the leader demotes itself to a learner during an atomic conf change, instead of waiting until it is completely removed from the range. Relying on the old leader to commit the final conf change while it's a learner appears unfortunate, and is not compatible with an upcoming etcd/raft change that makes the learner step down in this case.

This is backwards compatible with 23.1, because the same follower replica is responsible for campaigning, and it does not matter if it campaigns during the demotion or removal -- in particular because it forces an immediate election via `forceCampaignLocked()` which immediately bumps the term.

**kvserver: enable `raft.Config.StepDownOnRemoval`**

This causes the Raft leader to step down when it removes itself from the range or demotes itself to a learner. This doesn't make any difference functionally, since we're careful to no longer tick the replica or otherwise use it, but in principle it could cause a leader that was no longer part of the range to continue to assert leadership, stalling the range since it wouldn't be able to propose anything. Stepping down appears safer and cleaner.

Epic: none
Release note: None

107240: upgrade/upgrades: skip TestJSONForwardingIndexes r=mgartner a=DrewKimball

Refs: cockroachdb#107169

Reason: flaky test

Generated by bin/skip-test.

Release justification: non-production code changes
Release note: None
Epic: None

Co-authored-by: David Taylor <[email protected]>
Co-authored-by: Rui Hu <[email protected]>
Co-authored-by: Erik Grinaker <[email protected]>
Co-authored-by: Drew Kimball <[email protected]>

Author: 5 people

pkg/ccl/backupccl/restore_job.go

@@ -66,6 +66,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/envutil"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/humanizeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/interval"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/log/eventpb"
@@ -407,7 +408,7 @@ func restore(
 			return sendAddRemoteSSTs(
 				ctx,
 				execCtx,
-				job.ID(),
+				job,
 				dataToRestore,
 				endTime,
 				encryption,
@@ -1541,14 +1542,19 @@ func (r *restoreResumer) Resume(ctx context.Context, execCtx interface{}) error
 }
 
 func (r *restoreResumer) doResume(ctx context.Context, execCtx interface{}) error {
-	details := r.job.Details().(jobspb.RestoreDetails)
 	p := execCtx.(sql.JobExecContext)
 	r.execCfg = p.ExecCfg()
 
+	details := r.job.Details().(jobspb.RestoreDetails)
+
 	if err := maybeRelocateJobExecution(ctx, r.job.ID(), p, details.ExecutionLocality, "RESTORE"); err != nil {
 		return err
 	}
 
+	if len(details.DownloadSpans) > 0 {
+		return r.doDownloadFiles(ctx, p)
+	}
+
 	mem := p.ExecCfg().RootMemoryMonitor.MakeBoundAccount()
 	defer mem.Close(ctx)
 
@@ -2401,6 +2407,13 @@ func (r *restoreResumer) OnFailOrCancel(
 	p := execCtx.(sql.JobExecContext)
 	r.execCfg = p.ExecCfg()
 
+	details := r.job.Details().(jobspb.RestoreDetails)
+
+	// If this is a download-only job, there's no cleanup to do on cancel.
+	if len(details.DownloadSpans) > 0 {
+		return nil
+	}
+
 	// Emit to the event log that the job has started reverting.
 	emitRestoreJobEvent(ctx, p, jobs.StatusReverting, r.job)
 
@@ -2417,7 +2430,6 @@ func (r *restoreResumer) OnFailOrCancel(
 		return err
 	}
 
-	details := r.job.Details().(jobspb.RestoreDetails)
 	logutil.LogJobCompletion(ctx, restoreJobEventType, r.job.ID(), false, jobErr, r.restoreStats.Rows)
 
 	execCfg := execCtx.(sql.JobExecContext).ExecCfg()
@@ -3142,7 +3154,7 @@ var onlineRestoreGate = envutil.EnvOrDefaultBool("COCKROACH_UNSAFE_RESTORE", fal
 func sendAddRemoteSSTs(
 	ctx context.Context,
 	execCtx sql.JobExecContext,
-	jobID jobspb.JobID,
+	job *jobs.Job,
 	dataToRestore restorationData,
 	restoreTime hlc.Timestamp,
 	encryption *jobspb.BackupEncryptionOptions,
@@ -3284,7 +3296,23 @@ func sendAddRemoteSSTs(
 			}
 		}
 	}
-	return nil
+
+	downloadSpans := dataToRestore.getSpans()
+
+	log.Infof(ctx, "creating job to track downloads in %d spans", len(downloadSpans))
+	downloadJobRecord := jobs.Record{
+		Description: fmt.Sprintf("Background Data Download for %s", job.Payload().Description),
+		Username:    job.Payload().UsernameProto.Decode(),
+		Details:     jobspb.RestoreDetails{DownloadSpans: downloadSpans},
+		Progress:    jobspb.RestoreProgress{},
+	}
+
+	return execCtx.ExecCfg().InternalDB.DescsTxn(ctx, func(
+		ctx context.Context, txn descs.Txn,
+	) error {
+		_, err := execCtx.ExecCfg().JobRegistry.CreateJobWithTxn(ctx, downloadJobRecord, job.ID()+1, txn)
+		return err
+	})
 }
 
 var _ jobs.Resumer = &restoreResumer{}
@@ -3301,3 +3329,94 @@ func init() {
 		jobs.UsesTenantCostControl,
 	)
 }
+
+func (r *restoreResumer) doDownloadFiles(ctx context.Context, execCtx sql.JobExecContext) error {
+	details := r.job.Details().(jobspb.RestoreDetails)
+	total := r.job.Progress().Details.(*jobspb.Progress_Restore).Restore.TotalDownloadRequired
+
+	// If this is the first resumption of this job, we need to find out the total
+	// amount we expect to download and persist it so that we can indiciate our
+	// progress as that number goes down later.
+	if total == 0 {
+		log.Infof(ctx, "calculating total download size (across all stores) to complete restore")
+		if err := r.job.NoTxn().RunningStatus(ctx, func(_ context.Context, _ jobspb.Details) (jobs.RunningStatus, error) {
+			return jobs.RunningStatus("Calculating total download size..."), nil
+		}); err != nil {
+			return errors.Wrapf(err, "failed to update running status of job %d", errors.Safe(r.job.ID()))
+		}
+
+		for _, span := range details.DownloadSpans {
+			resp, err := execCtx.ExecCfg().TenantStatusServer.SpanStats(ctx, &roachpb.SpanStatsRequest{
+				Spans: []roachpb.Span{span},
+			})
+			if err != nil {
+				return err
+			}
+			for _, stats := range resp.SpanToStats {
+				total += stats.ExternalFileBytes
+			}
+		}
+
+		if total == 0 {
+			return nil
+		}
+
+		if err := r.job.NoTxn().FractionProgressed(ctx, func(ctx context.Context, details jobspb.ProgressDetails) float32 {
+			prog := details.(*jobspb.Progress_Restore).Restore
+			prog.TotalDownloadRequired = total
+			return 0.0
+		}); err != nil {
+			return err
+		}
+
+		if err := r.job.NoTxn().RunningStatus(ctx, func(_ context.Context, _ jobspb.Details) (jobs.RunningStatus, error) {
+			return jobs.RunningStatus(fmt.Sprintf("Downloading %s of restored data...", sz(total))), nil
+		}); err != nil {
+			return errors.Wrapf(err, "failed to update running status of job %d", errors.Safe(r.job.ID()))
+		}
+	}
+
+	var lastProgressUpdate time.Time
+	for rt := retry.StartWithCtx(
+		ctx, retry.Options{InitialBackoff: time.Second * 10, Multiplier: 1.2, MaxBackoff: time.Minute * 5},
+	); ; rt.Next() {
+
+		var remaining uint64
+		for _, span := range details.DownloadSpans {
+			resp, err := execCtx.ExecCfg().TenantStatusServer.SpanStats(ctx, &roachpb.SpanStatsRequest{
+				Spans: []roachpb.Span{span},
+			})
+			if err != nil {
+				return err
+			}
+			for _, stats := range resp.SpanToStats {
+				remaining += stats.ExternalFileBytes
+			}
+		}
+
+		fractionComplete := float32(total-remaining) / float32(total)
+		log.Infof(ctx, "restore download phase, %s downloaded, %s remaining of %s total (%.1f complete)",
+			sz(total-remaining), sz(remaining), sz(total), fractionComplete,
+		)
+
+		if remaining == 0 {
+			return nil
+		}
+
+		if timeutil.Since(lastProgressUpdate) > time.Minute {
+			if err := r.job.NoTxn().FractionProgressed(ctx, func(ctx context.Context, details jobspb.ProgressDetails) float32 {
+				return fractionComplete
+			}); err != nil {
+				return err
+			}
+			lastProgressUpdate = timeutil.Now()
+		}
+	}
+}
+
+type sz int64
+
+func (b sz) String() string { return string(humanizeutil.IBytes(int64(b))) }
+
+// TODO(dt): move this to humanizeutil and allow-list it there.
+//func (b sz) SafeValue()     {}

pkg/cloud/amazon/s3_storage_test.go

@@ -631,7 +631,7 @@ func TestReadFileAtReturnsSize(t *testing.T) {
 	_, err = w.Write(data)
 	require.NoError(t, err)
 	require.NoError(t, w.Close())
-	reader, _, err := s.ReadFile(ctx, file, cloud.ReadOptions{NoFileSize: true})
+	reader, _, err := s.ReadFile(ctx, file, cloud.ReadOptions{})
 	require.NoError(t, err)
 
 	rr, ok := reader.(*cloud.ResumingReader)

pkg/cloud/gcp/gcs_storage.go

@@ -294,7 +294,7 @@ func (g *gcsStorage) ReadFile(
 					return nil, 0, io.EOF
 				}
 			}
-			r, err := g.bucket.Object(object).NewRangeReader(ctx, pos, -1)
+			r, err := g.bucket.Object(object).NewRangeReader(ctx, pos, length)
 			if err != nil {
 				return nil, 0, err
 			}

pkg/cloud/gcp/gcs_storage_test.go

@@ -480,7 +480,7 @@ func TestReadFileAtReturnsSize(t *testing.T) {
 	_, err = w.Write(data)
 	require.NoError(t, err)
 	require.NoError(t, w.Close())
-	reader, _, err := s.ReadFile(ctx, file, cloud.ReadOptions{NoFileSize: true})
+	reader, _, err := s.ReadFile(ctx, file, cloud.ReadOptions{})
 	require.NoError(t, err)
 
 	rr, ok := reader.(*cloud.ResumingReader)

pkg/cloud/httpsink/http_storage_test.go

@@ -486,7 +486,7 @@ func TestReadFileAtReturnsSize(t *testing.T) {
 	_, err = w.Write(data)
 	require.NoError(t, err)
 	require.NoError(t, w.Close())
-	reader, _, err := s.ReadFile(ctx, file, cloud.ReadOptions{NoFileSize: true})
+	reader, _, err := s.ReadFile(ctx, file, cloud.ReadOptions{})
 	require.NoError(t, err)
 
 	rr, ok := reader.(*cloud.ResumingReader)

pkg/jobs/jobspb/jobs.proto

@@ -479,8 +479,12 @@ message RestoreDetails {
   roachpb.Locality execution_locality = 30 [(gogoproto.nullable) = false];
 
   bool experimental_online = 31;
+  
+  // DownloadSpans indicates this job is the download-step of a multi-step
+  // online restore.
+  repeated roachpb.Span download_spans = 32 [(gogoproto.nullable) = false]; 
 
-  // NEXT ID: 31.
+  // NEXT ID: 33.
 }
 
 
@@ -496,6 +500,11 @@ message RestoreProgress {
   }
 
   repeated FrontierEntry checkpoint = 2 [(gogoproto.nullable) = false];
+
+  // TotalDownloadRequired is set in the download job for an online restore, and
+  // reflects the total amount that was initially found to be needing to be 
+  // downloaded. 
+  uint64 total_download_required = 3;
 }
 
 message ImportDetails {

pkg/kv/kvserver/client_raft_test.go

@@ -6547,7 +6547,7 @@ func TestRaftLeaderRemovesItself(t *testing.T) {
 				// Set a large election timeout. We don't want replicas to call
 				// elections due to timeouts, we want them to campaign and obtain
 				// votes despite PreVote+CheckQuorum.
-				RaftElectionTimeoutTicks: 200,
+				RaftElectionTimeoutTicks: 300,
 			},
 			Knobs: base.TestingKnobs{
 				Store: &kvserver.StoreTestingKnobs{
@@ -6601,6 +6601,9 @@ func TestRaftLeaderRemovesItself(t *testing.T) {
 	tc.RemoveVotersOrFatal(t, key, tc.Target(0))
 	t.Logf("n1 removed from range")
 
+	// Make sure we didn't time out on the above.
+	require.NoError(t, ctx.Err())
+
 	require.Eventually(t, func() bool {
 		logStatus(repl2.RaftStatus())
 		logStatus(repl3.RaftStatus())
@@ -6610,6 +6613,8 @@ func TestRaftLeaderRemovesItself(t *testing.T) {
 		}
 		return false
 	}, 10*time.Second, 500*time.Millisecond)
+
+	require.NoError(t, ctx.Err())
 }
 
 // TestRaftUnquiesceLeaderNoProposal tests that unquiescing a Raft leader does

pkg/kv/kvserver/replica_init.go

@@ -255,12 +255,14 @@ func (r *Replica) initRaftMuLockedReplicaMuLocked(s kvstorage.LoadedReplicaState
 // initRaftGroupRaftMuLockedReplicaMuLocked initializes a Raft group for the
 // replica, replacing the existing Raft group if any.
 func (r *Replica) initRaftGroupRaftMuLockedReplicaMuLocked() error {
+	ctx := r.AnnotateCtx(context.Background())
 	rg, err := raft.NewRawNode(newRaftConfig(
+		ctx,
 		raft.Storage((*replicaRaftStorage)(r)),
 		uint64(r.replicaID),
 		r.mu.state.RaftAppliedIndex,
 		r.store.cfg,
-		&raftLogger{ctx: r.AnnotateCtx(context.Background())},
+		&raftLogger{ctx: ctx},
 	))
 	if err != nil {
 		return err

pkg/kv/kvserver/replica_raft.go

@@ -2603,9 +2603,9 @@ func ComputeRaftLogSize(
 }
 
 // shouldCampaignAfterConfChange returns true if the current replica should
-// campaign after a conf change. If the leader replica is removed, the
-// leaseholder should campaign. We don't want to campaign on multiple replicas,
-// since that would cause ties.
+// campaign after a conf change. If the leader replica is demoted or removed,
+// the leaseholder should campaign. We don't want to campaign on multiple
+// replicas, since that would cause ties.
 //
 // If there is no current leaseholder we'll have to wait out the election
 // timeout before someone campaigns, but that's ok -- either we'll have to wait
@@ -2622,7 +2622,7 @@ func shouldCampaignAfterConfChange(
 	raftStatus raft.BasicStatus,
 	leaseStatus kvserverpb.LeaseStatus,
 ) bool {
-	if raftStatus.Lead == 0 {
+	if raftStatus.Lead == raft.None {
 		// Leader unknown. We can't know if it was removed by the conf change, and
 		// because we force an election without prevote we don't want to risk
 		// throwing spurious elections.
@@ -2637,9 +2637,11 @@ func shouldCampaignAfterConfChange(
 		// don't expect to hit this, but let's be defensive.
 		return false
 	}
-	if _, ok := desc.GetReplicaDescriptorByID(roachpb.ReplicaID(raftStatus.Lead)); ok {
-		// The leader is still in the descriptor.
-		return false
+	if replDesc, ok := desc.GetReplicaDescriptorByID(roachpb.ReplicaID(raftStatus.Lead)); ok {
+		if replDesc.IsAnyVoter() {
+			// The leader is still a voter in the descriptor.
+			return false
+		}
 	}
 	// Prior to 23.2, the first voter in the descriptor campaigned, so we do
 	// the same in mixed-version clusters to avoid ties.

pkg/kv/kvserver/store.go

@@ -68,6 +68,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/spanconfig"
 	"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigstore"
 	"github.com/cockroachdb/cockroach/pkg/storage"
+	"github.com/cockroachdb/cockroach/pkg/util"
 	"github.com/cockroachdb/cockroach/pkg/util/admission"
 	"github.com/cockroachdb/cockroach/pkg/util/admission/admissionpb"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
@@ -257,6 +258,12 @@ var ExportRequestsLimit = settings.RegisterIntSetting(
 	settings.PositiveInt,
 )
 
+// raftStepDownOnRemoval is a metamorphic test parameter that makes Raft leaders
+// step down on demotion or removal. Following an upgrade, clusters may have
+// replicas with mixed settings, because it's only changed when initializing
+// replicas. Varying it makes sure we handle this state.
+var raftStepDownOnRemoval = util.ConstantWithMetamorphicTestBool("raft-step-down-on-removal", true)
+
 // TestStoreConfig has some fields initialized with values relevant in tests.
 func TestStoreConfig(clock *hlc.Clock) StoreConfig {
 	return testStoreConfig(clock, clusterversion.TestingBinaryVersion)
@@ -299,6 +306,7 @@ func testStoreConfig(clock *hlc.Clock, version roachpb.Version) StoreConfig {
 }
 
 func newRaftConfig(
+	ctx context.Context,
 	strg raft.Storage,
 	id uint64,
 	appliedIndex kvpb.RaftIndex,
@@ -320,6 +328,20 @@ func newRaftConfig(
 		Storage:                     strg,
 		Logger:                      logger,
 
+		// StepDownOnRemoval requires 23.2. Otherwise, in a mixed-version cluster, a
+		// 23.2 leader may step down when it demotes itself to learner, but a
+		// designated follower (first in the range) running 23.1 will only campaign
+		// once the leader is entirely removed from the range descriptor (see
+		// shouldCampaignOnConfChange). This would leave the range without a leader,
+		// having to wait out an election timeout.
+		//
+		// We only set this on replica initialization, so replicas without
+		// StepDownOnRemoval may remain on 23.2 nodes until they restart. That's
+		// totally fine, we just can't rely on this behavior until 24.1, but
+		// we currently don't either.
+		StepDownOnRemoval: storeCfg.Settings.Version.IsActive(ctx, clusterversion.V23_2) &&
+			raftStepDownOnRemoval,
+
 		PreVote:     true,
 		CheckQuorum: storeCfg.RaftEnableCheckQuorum,
 	}