[prototype] migration: onboard the first long-running migration

[irfansharif]

This PR onboards the first real long-running migration using the
infrastructure we've been building up within pkg/migration. It adds in
the final missing pieces described in our original RFC (#48843). These
components were originally prototyped in #56107.

The migration in question (which happens to be a below-Raft one, first
attempted in #42822) now lets us do the following:

i. Use the RangeAppliedState on all ranges
ii. Use the unreplicated TruncatedState on all ranges

The missing pieces we introduce along side this migration are:

a. The Migrate KV request. This forces all ranges overlapping with
the request spans to execute the (below-raft) migrations corresponding
to the specific version, moving out of any legacy modes they may
currently be in. KV waits for this command to durably apply on all the
replicas before returning, guaranteeing to the caller that all
pre-migration state has been completely purged from the system.

b. IterateRangeDescriptors. This provides a handle on every range
descriptor in the system, which callers can then use to send out
arbitrary KV requests to in order to run arbitrary KV-level
migrations. These requests will typically just be the Migrate
request, with added code next to the Migrate command implementation
to do the specific KV-level things intended for the specified version.

c. The system.migrations table. We use it to store metadata about
ongoing migrations for external visibility/introspection. The schema
(listed below) is designed with an eye towards scriptability. We
want operators to be able programmatically use this table to control
their upgrade processes, if needed.

  CREATE TABLE public.migrations (
      version STRING NOT NULL,
      status STRING NOT NULL,
      description STRING NOT NULL,
      start TIMESTAMP NOT NULL DEFAULT now():::TIMESTAMP,
      completed TIMESTAMP NULL,
      progress STRING NULL,
      CONSTRAINT "primary" PRIMARY KEY (version ASC),
      FAMILY "primary" (version, status, description, start, completed, progress)
  )

Release note(general change): Cluster version upgrades, as initiated by
SET CLUSTER SETTING version = <major>-<minor>, now perform internal
maintenance duties that will delay how long it takes for the command to
complete. The delay is proportional to the amount of data currently
stored in the cluster. The cluster will also experience a small amount
of additional load during this period while the upgrade is being
finalized.

Release note(general change): We introduce a new system.migrations
table for introspection into crdb internal data migrations. These
migrations are the "maintenance duties" mentioned above. The table
surfaces the currently ongoing migrations, the previously completed
migrations, and in the case of failure, the errors from the last failed
attempt.

[cockroach-teamcity]

This change is 

[irfansharif]

Heads up, this is still a [wip] branch. It's a bit far from landing. I have zero tests, and intend to add them here. I also intend to break it up into a few commits: introducing the system table, introducing the request type, introducing IterateRangeDescriptors, and only after all that, adding the migration itself. For now I've just squashed them for an early first-pass look/sanity check before I spend more time on final polish.

I have a few open questions in there as comments (grep for XXX:), but any+all thoughts welcome on the rest of it.

[tbg]

This looks pretty good already! I left a bunch of comments but I think it's basically time to figure out how to write proper tests and how to put this into a number of smaller PRs.

The migration table introduction looks like it could be a good last PR. The migrate command + testing could come first. Then a PR that updates the manager and adds tests and hooks up the migrate command. Just a suggestion.

Reviewed 49 of 49 files at r1.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @irfansharif and @knz)

---

pkg/kv/kvserver/replica_proposal.go, line 781 at r1 (raw file):

			// below. If we're not running 21.1 yet, migrate over as we've done
			// since the introduction of the applied state key.
			if r.ClusterSettings().Version.IsActive(ctx, clusterversion.TruncatedAndRangeAppliedStateMigration) {

This is exactly the kind of thing that could cause spurious fatal errors when there are replicaGC'able replicas left around that didn't get included in the raft command. On these, you can still hit the evaluation path (they'll try to request leases). I don't know if this PR already does anything to avoid this behavior (I'll just leave this comment now and will continue reviewing then), but as we've discussed in the past there are two ingredients:

• track the range version (in the replica state) and disallow snapshots from an old version to any newer version
• force a run of replicaGC on all replicas in the cluster that don't have the updated range version (via EveryNode presumably)

That way we can

1. call Migrate on the whole keyspace
2. force the run

and be assured that all replicas now in the cluster are at the new version, i.e. it's safe to bump the cluster version.

---

pkg/kv/kvserver/replica_write.go, line 223 at r1 (raw file):

				// applied on all peers, which we achieve via waitForApplication.
				//
				// XXX: Could a snapshot below maxLeaseIndex be in-flight to a

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, so this should not be an issue. The particular case you mention will have the learner have a higher replicaID than the incoming snapshot, so the snapshot will also be out for that reason. Finally, a snapshot is discarded unless it moves the LAI forward.

---

pkg/kv/kvserver/batcheval/cmd_migrate.go, line 16 at r1 (raw file):

	"context"

	cv "github.com/cockroachdb/cockroach/pkg/clusterversion"

small nit, but I'm generally -1 on unnecessary import renames.

---

pkg/migration/helper.go, line 39 at r1 (raw file):

// to execute all relevant node-level operations on them. If any of the nodes
// are found to be unavailable, an error is returned.
func (h *Helper) RequiredNodeIDs(ctx context.Context) ([]roachpb.NodeID, error) {

Should this be exported given that you need to wrap it in a stabilizing retry loop, and thus is a footgun if used from a migration work function?

---

pkg/migration/helper.go, line 100 at r1 (raw file):

// (accessible using the `progress` column in `system.migrations`) for the
// ongoing migration.
func (h *Helper) UpdateProgress(ctx context.Context, progress string) error {

"progress" is a bit of a misnomer, how about "status"

---

pkg/migration/helper.go, line 136 at r1 (raw file):

//
// By the time EveryNode returns, we'll have thus invoked the closure against
// every node in the cluster.

That's not true. A node can always join just as EveryNode returns. What we have achieved is that there is now a causal happened-before relation between executing something on every node in the cluster and the next node joining the cluster. We managed to run something on all nodes without a new node joining half-way through, which could have allowed it to pick up some state (off one of the existing nodes that hadn't finished processing yet).

For the example of bumping the cluster version, we'll have achieved proving that by the time EveryNode returns, every node in the cluster, current and future, will use the new cluster version.

---

pkg/migration/helper.go, line 162 at r1 (raw file):

	for {
		// TODO(irfansharif): We can/should send out these RPCs in parallel.

We should do that soon, too. Ideally in the burst of commits coming out of this one!

---

pkg/migration/migrations.go, line 110 at r1 (raw file):

func TruncatedStateMigration(ctx context.Context, h *Helper) error {
	// TODO(irfansharif): What should the default block size be?
	const blockSize = 50

At 50k ranges, with each range taking ~200ms (a rather conservative assumption but in a wide-area cluster could be true due to hops from the manager to the replicas) to process the Migrate, a block size of 1 is >2.7 hours. Block size 50 is ~3.3 minutes. Block size 200 should easily be in the cards given that each range is doing very little work here, so down to 50s in this case which seems reasonable.

---

pkg/migration/migrations.go, line 116 at r1 (raw file):

		for _, desc := range descriptors {
			if bytes.Compare(desc.StartKey, keys.LocalMax) < 0 {
				// XXX: Does it make sense to "migrate" ranges with local keys?

Yeah, this is a bit of a wart. We do want to reach the first range, but we can't address StartKey. However, LocalMax is on r1, so in this case we basically just use LocalMax instead of StartKey.

I wonder if we'd get sanity back if we officially let r1 begin at LocalMax, but that is not worth getting into here or probably ever.

---

pkg/migration/migrations.go, line 128 at r1 (raw file):

		numMigratedRanges += len(descriptors)
		progress := fmt.Sprintf("[batch %d/??] migrated %d ranges", batchIdx, numMigratedRanges)

Doesn't IterateRangeDescriptors have to handle txn retries? The closure here would need to be idempotent, but the counters aren't. Maybe IterateRangeDescriptors should take an init func() as well which can reset them.

---

pkg/migration/migrations.go, line 140 at r1 (raw file):

	// XXX: I think we'll need to make sure all stores have synced once to
	// persist all applications of the `Migrate` raft command.

Very good point. It sounds reasonable to do this as a separate round of EveryNode (as opposed to doing it on each raft command application, which would cause tons of syncs).
It's probably also just generally a good idea to make this happen on every migration that does work. It could even be part of bumping the cluster version, which by definition already requires syncing; we could just extend that to explicitly mandate syncing all engines.

Hmm, thinking about this further, this is not good enough when nodes restart rapidly. I think the best way to address this is to make EveryNode more aggressive about retrying: in addition to the list of nodes stabilizing, their corresponding epochs also need to remain stable (and, of course, everyone live). That way we don't impose a requirement that everything we do needs to be synced immediately, which is super expensive and we're also pretty much guaranteed to get it wrong.

---

pkg/roachpb/api.go, line 1295 at r1 (raw file):

func (*AdminScatterRequest) flags() int                  { return isAdmin | isRange | isAlone }
func (*AdminVerifyProtectedTimestampRequest) flags() int { return isAdmin | isRange | isAlone }
func (*MigrateRequest) flags() int                       { return isWrite | isRange | isAlone }

I wonder if MigrateRange is a better name.

---

pkg/roachpb/api.proto, line 1500 at r1 (raw file):

// is already going to be the one responsible for stepping through all
// intermediate versions, it's probably saner to only carry out the specific
// migration associated with the target version.

Yeah, that's the impression I had as well.

---

pkg/roachpb/api.proto, line 1502 at r1 (raw file):

// migration associated with the target version.
//
// TODO(irfansharif): Should this be an admin request instead? There's no good

No, admin requests are really just a way to invoke random code on the leaseholder. Here we want to properly go through the raft pipeline, which admin requests don't.

---

pkg/roachpb/api.proto, line 1509 at r1 (raw file):

  RequestHeader header = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true];

  // The version that will become active next. A request with the

incomplete

---

pkg/roachpb/method.go, line 161 at r1 (raw file):
This comment could be better, something like

Migrate updates the range state to conform to a specified cluster version. It is our main mechanism for phasing out legacy code below Raft.

---

pkg/sql/catalog/systemschema/system.go, line 347 at r1 (raw file):

	MigrationsTableSchema = `
CREATE TABLE system.migrations (

Is this going to be gossiped? (The answer should be "no", I forget what the rules are).

---

pkg/sql/catalog/systemschema/system.go, line 615 at r1 (raw file):

	// MigrationsTable is the descriptor for the migrations table. It contains
	// details for all past and ongoing migrations.

We'll want someone from SQL to review this part of the PR. This all looks sane, but I'd want someone to sign off that it is.

---

pkg/sql/logictest/testdata/logic_test/distsql_agg, line 578 at r1 (raw file):

SELECT corr(y, x)::decimal, covar_pop(y, x)::decimal, covar_samp(y, x)::decimal FROM statistics_agg_test
----
0.04522896319136315  3.75  3.787878787878788

Huh. What are these about?

[irfansharif]

Thanks for looking. I've addressed all open comments here, and added a bunch of tests. I'm still going to leave this up as an open prototype branch (for this top level review + discussion + running CI) and pull out bits into separate PRs elsewhere.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @knz, @miretskiy, @tbg, and @yuzefovich)

---

pkg/kv/kvserver/replica_proposal.go, line 781 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

This is exactly the kind of thing that could cause spurious fatal errors when there are replicaGC'able replicas left around that didn't get included in the raft command. On these, you can still hit the evaluation path (they'll try to request leases). I don't know if this PR already does anything to avoid this behavior (I'll just leave this comment now and will continue reviewing then), but as we've discussed in the past there are two ingredients:

• track the range version (in the replica state) and disallow snapshots from an old version to any newer version
• force a run of replicaGC on all replicas in the cluster that don't have the updated range version (via EveryNode presumably)

That way we can

1. call Migrate on the whole keyspace
2. force the run

and be assured that all replicas now in the cluster are at the new version, i.e. it's safe to bump the cluster version.

Instead of tracking the range version in the replica state (somewhat out of laziness), I opted to instead introduce a GCReplicas RPC to the Migration service and wire it in using EveryNode. Whenever we're running a ranged Migrate command across the entire keyspace for vX, we'll want to GC previous version replicas immediately after. I think almost always this would mean processing the entire the replica GC queue anyway.
As for disallowing previous version snapshots, I think our MLAI safety checks (discussed below) will already prevent it from happening.

---

pkg/kv/kvserver/replica_write.go, line 223 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

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, so this should not be an issue. The particular case you mention will have the learner have a higher replicaID than the incoming snapshot, so the snapshot will also be out for that reason. Finally, a snapshot is discarded unless it moves the LAI forward.

Done, incorporated into the comment here.

---

pkg/kv/kvserver/batcheval/cmd_migrate.go, line 16 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

small nit, but I'm generally -1 on unnecessary import renames.

Done.

---

pkg/migration/helper.go, line 39 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

Should this be exported given that you need to wrap it in a stabilizing retry loop, and thus is a footgun if used from a migration work function?

good idea, Done.

---

pkg/migration/helper.go, line 100 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

"progress" is a bit of a misnomer, how about "status"

I agree that "status" is better, but for parity with (the now soured upon) system.jobs, I reserved the use of "status" for the overall status of the migration ("ongoing/succeeded/failed"). Progress was the closest next thing here that seemed appropriate.

---

pkg/migration/helper.go, line 136 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

That's not true. A node can always join just as EveryNode returns. What we have achieved is that there is now a causal happened-before relation between executing something on every node in the cluster and the next node joining the cluster. We managed to run something on all nodes without a new node joining half-way through, which could have allowed it to pick up some state (off one of the existing nodes that hadn't finished processing yet).

For the example of bumping the cluster version, we'll have achieved proving that by the time EveryNode returns, every node in the cluster, current and future, will use the new cluster version.

yea, I had tried elaborating on that in the paragraphs, but we're database people, I should talk in terms of causality. Reworked this entire block.

---

pkg/migration/helper.go, line 162 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

We should do that soon, too. Ideally in the burst of commits coming out of this one!

Done.

---

pkg/migration/migrations.go, line 110 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

At 50k ranges, with each range taking ~200ms (a rather conservative assumption but in a wide-area cluster could be true due to hops from the manager to the replicas) to process the Migrate, a block size of 1 is >2.7 hours. Block size 50 is ~3.3 minutes. Block size 200 should easily be in the cards given that each range is doing very little work here, so down to 50s in this case which seems reasonable.

Makes sense, I'll include this back-of-the-envelope calculation in there.

---

pkg/migration/migrations.go, line 116 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

Yeah, this is a bit of a wart. We do want to reach the first range, but we can't address StartKey. However, LocalMax is on r1, so in this case we basically just use LocalMax instead of StartKey.

I wonder if we'd get sanity back if we officially let r1 begin at LocalMax, but that is not worth getting into here or probably ever.

Done.

---

pkg/migration/migrations.go, line 128 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

Doesn't IterateRangeDescriptors have to handle txn retries? The closure here would need to be idempotent, but the counters aren't. Maybe IterateRangeDescriptors should take an init func() as well which can reset them.

Ah, good point. I like the init idea, done.

---

pkg/migration/migrations.go, line 140 at r1 (raw file):

It sounds reasonable to do this as a separate round of EveryNode

Done. I've added a primitive that simply flushes all engines. Seems like a useful property for the migrations infrastructure to provide.

we could just extend that to explicitly mandate syncing all engines.

I prefer dead-simple to-the-point APIs, so keeping BumpClusterVersion RPC as is.

---

pkg/roachpb/api.go, line 1295 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

I wonder if MigrateRange is a better name.

I think I'll keep it as is. We'd technically still be able to send out a ranged Migrate request over arbitrary keyspans, we're only choosing to paginate because it'd be insane not to.

---

pkg/roachpb/api.proto, line 1500 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

Yeah, that's the impression I had as well.

Cool.

---

pkg/roachpb/api.proto, line 1502 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

No, admin requests are really just a way to invoke random code on the leaseholder. Here we want to properly go through the raft pipeline, which admin requests don't.

Makes sense, this TODO was more for aesthetic reasons than anything functional.

---

pkg/roachpb/api.proto, line 1509 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

incomplete

Looks like I forgot to push out some changes. Done.

---

pkg/roachpb/method.go, line 161 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

This comment could be better, something like

Migrate updates the range state to conform to a specified cluster version. It is our main mechanism for phasing out legacy code below Raft.

Done.

---

pkg/sql/catalog/systemschema/system.go, line 347 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

Is this going to be gossiped? (The answer should be "no", I forget what the rules are).

I'm not sure (lazily hoping one of the SQL reviewers can answer).

---

pkg/sql/logictest/testdata/logic_test/distsql_agg, line 578 at r1 (raw file):

Previously, tbg (Tobias Grieger) wrote…

Huh. What are these about?

Who knows, just came up trying to re-write the logictests to include details for the new system table.

make -j1 testlogic TESTFLAGS='--rewrite'

I've previously leaned on @yuzefovich for emotional support during these.

[yuzefovich]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @knz, @miretskiy, and @tbg)

---

pkg/sql/logictest/testdata/logic_test/distsql_agg, line 578 at r1 (raw file):

Previously, irfansharif (irfan sharif) wrote…

Who knows, just came up trying to re-write the logictests to include details for the new system table.

make -j1 testlogic TESTFLAGS='--rewrite'

I've previously leaned on @yuzefovich for emotional support during these.

I think I fixed it on master, if you rebase, these should go away.

[irfansharif]

@ajwerner, would love see what you think of this schema.

[irfansharif]

Most of this prototype was folded into #57650. The contentious opinions here (#57662) were scrapped. As for introspection into in-progress migration state, we're simply making use of logging channels for now. There's something to be said for "jobs-ifying" long running migrations at a per-migration level (which in this prototype we simply re-built and dumped into a system.migrations table), but that's not a thread we're going to pull on for a bit I think.