kvserver: improve below-Raft migrations

[erikgrinaker]

Long-running migrations can send a MigrateRequest for migrations that must be applied below Raft. This request is special in that it only succeeds once it has been applied to all known replicas of the range -- it is not sufficient simply to commit it to the Raft log following acknowledgement from a quorum of replicas.

cockroach/pkg/kv/kvserver/replica_write.go

Lines 254 to 257 in 455cddd

	applicationErr := waitForApplication(
	ctx, r.store.cfg.NodeDialer, desc.RangeID, desc.Replicas().Descriptors(),
	// We wait for an index >= that of the migration command.
	r.GetLeaseAppliedIndex())

This requirement is in order to guarantee that no state machine replicas rely on legacy, unmigrated state. However, this requires all replicas for all ranges in a cluster to be available and up-to-date, with a 5-second timeout before giving up. Any retries are currently left to the migration code itself. For example, the postSeparatedIntentsMigration uses 5 retries for a given range and then fails the entire migration, having to restart:

cockroach/pkg/migration/migrations/separated_intents.go

Lines 557 to 563 in 4df8ac2

	err := retry.WithMaxAttempts(ctx, base.DefaultRetryOptions(), 5, func() error {
	err := deps.DB.Migrate(ctx, start, end, cv.Version)
	if err != nil {
	log.Infof(ctx, "[batch %d/??] error when running no-op Migrate on range r%d: %s", batchIdx, desc.RangeID, err)
	}
	return err
	})

This could be improved in several ways:

• Consider whether the requirement that all replicas for all ranges are available and up-to-date is necessary, or even viable in large clusters.
• Introduce migration helpers to do automatic batching, checkpointing and retries of such migrations across ranges, to avoid having to implement this in each separate migration. It should also optimistically continue applying it to additional ranges even when one fails.
• Introduce knob to control fan out for how many ranges are migrated at a time, to allow operators to speed up migrations with acceptable hit on foreground traffic.
• Use smaller txns (with low priority) when iterating over full set of range descriptors to reduce contention over meta2. Since long running migrations are long running, locking up meta2 for the entire period is not ideal.
• Make sure the migration jobs can be paused as necessary, and that they use the regular exponential backoff for job retries.
• Make the application timeout configurable, or simply rely on the passed client context (which would be controlled by the new migration infrastructure mentioned above).
• Improve the UX by failing with an informative error message explaining that the migration cannot be completed because range X replicas Y,Z are unavailable/behind/uninitialized/etc, and why all replicas must go through the migration before the upgrade can be finalized.

Jira issue: CRDB-11351

Epic CRDB-39898

[irfansharif]

Thanks for filing. Here's the internal thread that prompted this issue.

[ajwerner]

We do have an existing mechanism in kvserver to wait for application. The merge protocol relies on it. We could consider doing something like that during the execution and not acknowledging the batch until it's been fully replicated. It wouldn't be too hard to implement.

cockroach/pkg/kv/kvserver/replica_command.go

Lines 790 to 821 in b57af7d

	func waitForApplication(
	ctx context.Context,
	dialer *nodedialer.Dialer,
	rangeID roachpb.RangeID,
	replicas []roachpb.ReplicaDescriptor,
	leaseIndex uint64,
	) error {
	if dialer == nil && len(replicas) == 1 {
	// This early return supports unit tests (testContext{}) that also
	// want to perform merges.
	return nil
	}
	return contextutil.RunWithTimeout(ctx, "wait for application", 5*time.Second, func(ctx context.Context) error {
	g := ctxgroup.WithContext(ctx)
	for _, repl := range replicas {
	repl := repl // copy for goroutine
	g.GoCtx(func(ctx context.Context) error {
	conn, err := dialer.Dial(ctx, repl.NodeID, rpc.DefaultClass)
	if err != nil {
	return errors.Wrapf(err, "could not dial n%d", repl.NodeID)
	}
	_, err = NewPerReplicaClient(conn).WaitForApplication(ctx, &WaitForApplicationRequest{
	StoreRequestHeader: StoreRequestHeader{NodeID: repl.NodeID, StoreID: repl.StoreID},
	RangeID: rangeID,
	LeaseIndex: leaseIndex,
	})
	return err
	})
	}
	return g.Wait()
	})
	}

[ajwerner]

Make sure the migration jobs can be paused as necessary, and that they use the regular exponential backoff for job retries.

This should be the case. Is there evidence that it is not?

[erikgrinaker]

We do have an existing mechanism in kvserver to wait for application. The merge protocol relies on it. We could consider doing something like that during the execution and not acknowledging the batch until it's been fully replicated. It wouldn't be too hard to implement.

Well yes, that's what we currently do here:

cockroach/pkg/kv/kvserver/replica_write.go

Lines 206 to 259 in 455cddd

	if ba.Requests[0].GetMigrate() != nil && propResult.Err == nil {
	// Migrate is special since it wants commands to be durably
	// applied on all peers, which we achieve via waitForApplication.
	//
	// We don't have to worry about extant snapshots creating
	// replicas that start at an index before this Migrate request.
	// Snapshots that don't include the recipient (as specified by
	// replicaID and descriptor in the snap vs. the replicaID of the
	// raft instance) are discarded by the recipient, and we're
	// already checking against all replicas in the descriptor below
	// (which include learner replicas currently in the process of
	// receiving snapshots). Snapshots are also discarded unless
	// they move the LAI forward, so we're not worried about old
	// snapshots (with indexes preceding the MLAI here)
	// instantiating pre-migrated state in anyway. We also have a
	// separate mechanism to ensure replicas with older versions are
	// purged from the system[1]. This is driven by a higher-level
	// orchestration layer[2], these are the replicas that we don't
	// have a handle on here as they're eligible for GC (but may
	// still hit replica evaluation code paths with pre-migrated
	// state, unless explicitly purged).
	//
	// It's possible that between the proposal returning and the
	// call to r.Desc() below, the descriptor has already changed.
	// But the only thing that matters is that r.Desc() is at least
	// as up to date as the descriptor the command applied on
	// previously. If a replica got removed - fine,
	// waitForApplication will fail; we will have to cope with that.
	// If one got added - it was likely already a learner when we
	// migrated (in which case waitForApplication will know about
	// it). If that's not the case, we'll note that the Migrate
	// command also declares a read latch on the range descriptor.
	// The replication change will have thus serialized after the
	// migration, and so the snapshot will also include the
	// post-migration state.
	//
	// TODO(irfansharif): In a cluster that is constantly changing
	// its replica sets, it's possible to get into a situation
	// where a Migrate command never manages to complete - all it
	// takes is a single range in each attempt to throw things off.
	// Perhaps an error in waitForApplication should lead to a retry
	// of just the one RPC instead of propagating an error for the
	// entire migrate invocation.
	//
	// [1]: See PurgeOutdatedReplicas from the Migration service.
	// [2]: pkg/migration
	desc := r.Desc()
	// NB: waitForApplication already has a timeout.
	applicationErr := waitForApplication(
	ctx, r.store.cfg.NodeDialer, desc.RangeID, desc.Replicas().Descriptors(),
	// We wait for an index >= that of the migration command.
	r.GetLeaseAppliedIndex())
	propResult.Err = roachpb.NewError(applicationErr)
	}

The problem is that this command has to succeed for every range on every replica in one go (with a few tight retries), otherwise the entire migration fails and has to restart. This can be a problem in large clusters with many ranges (in this case, 400.000 ranges).

Make sure the migration jobs can be paused as necessary, and that they use the regular exponential backoff for job retries.

This should be the case. Is there evidence that it is not?

No, just something we should verify.

[irfansharif]

One thing we're sorely lacking is an integration/qualification/acceptance test to run these migrations at very large scales. It could've shaken out some of these low timeouts for e.g. or further validated the need for generic checkpointing.

[irfansharif]

Introduce migration helpers to do automatic batching, checkpointing and retries of such migrations across ranges, to avoid having to implement this in each separate migration. It should also optimistically continue applying it to additional ranges even when one fails.

This is worth doing, perhaps as part of #84073.

Introduce knob to control fan out for how many ranges are migrated at a time, to allow operators to speed up migrations with acceptable hit on foreground traffic.

A library in pkg/upgrades to do this sort of thing would prevent us from writing one-range-at-a-time migrations, like we do here:

cockroach/pkg/upgrade/upgrades/raft_applied_index_term.go

Lines 40 to 46 in ea13a52

	func raftAppliedIndexTermMigration(
	ctx context.Context, cv clusterversion.ClusterVersion, deps upgrade.SystemDeps, _ *jobs.Job,
	) error {
	var batchIdx, numMigratedRanges int
	init := func() { batchIdx, numMigratedRanges = 1, 0 }
	if err := deps.Cluster.IterateRangeDescriptors(ctx, defaultPageSize, init, func(descriptors ...roachpb.RangeDescriptor) error {
	for _, desc := range descriptors {

[ajwerner]

A library in pkg/upgrades to do this sort of thing would prevent us from writing one-range-at-a-time migrations, like we do here

That linked library should not be particularly difficult to generalize.

[knz]

Would it make sense to also lower the txn priority inside IterateRangeDescriptors (and possibly the implicit txn used for the MigrateRequest), to reduce the visible contention on meta2 from SQL?

[erikgrinaker]

I wouldn't use a transaction at all, and instead scan smaller batches of up-to-date descriptors with individual scan requests. No objection on doing these with low priority, but we specifically don't want a snapshot of meta2, we want fresh data.

[irfansharif]

Using smaller txns (and lowering their priority if still needed) to fetch batches of range descriptors makes sense.

[irfansharif]

I'll apologize for not having done just all this originally, this simple thing has caused me much grief. Had I kicked the tires more on realistic cluster sizes (100k+ ranges with ambient split/merge activity), all this would've been much more apparent.

[knz]

Irfan, using smaller txns with lower priority is not mentioned on the issue erik linked. Mind filing a new followup issue for that (separate) change?

[irfansharif]

Do you mean this issue? I've added a bullet point here.

[knz]

thanks!

[erikgrinaker]

Since we'll be adding a fair bit of concurrency here to improve throughput for trivial migrations, we should probably integrate this with AC somehow as well, to avoid expensive migrations overloading the cluster.

[erikgrinaker]

Adding O-support, since we've had several escalations about below-Raft migrations stalling upgrades.