vreplication: change to use new _vt.vreplication

[sougou]

This change deprecates _vt.blp_checkpoint in favor
of vreplication, which stands for Vitess Replication.
The goal is to make vreplication a standalone sub-module
of vitess that other services can use, including the
resharding worflow.

The big change in the model is that vreplication is not owned by the resharding workflow. The workflow instead creates vreplication streams as needed, and controls them individually. The stream id for a replication is now generated by vreplication, which the resharding workflow stores and tracks.

This also means that a vreplication stream can be directly created and managed by anyone as needed. This allows for newer and more flexible workflows in the future.

The following parts of the system underwent major changes:

• A new vreplication package has been added under tabletmanager. This is the core of the vreplication sub-module.
  • The data it needs is completely stored in the _vt.vreplication table. It does not depend on topo.
  • The Engine accepts SQL statements instead of specific RPC commands. Once the statements are validated and applied, the engine makes corresponding changes to its behavior, like starting/stopping replication, etc.
  • It has its own planbuilder to figure out how to handle statements. Design is similar to other plabuilders.
  • There is one controller per stream, and each controller drives a BinlogPlayer.
  • It exports stats just like before, but variables contain the VReplication name.
  • The older BinlogPlayersMap and associated code has been deleted.
  • Simplified handling of context expirty.
• BinlogPlayer
  • Uses vreplication table
  • Saves state on exit if needed. For example, it saves state as 'Stopped' if stopping position has been reached, and won't accidentally restart replication if vttablet is restarted.
  • Directly reads start position from table instead of taking it as input from caller.
  • DBClient code has been cleaned up and improved. We now have fake and mock dbclients, which serve different purposes.
  • New unit tests
• SplitClone and SplitDiff workflows use vreplication
  • The main changes come from the fact that the stream id is generated by vreplication. So, the SourceShard topo records must be created after streams are created.
  • strategy flag has been removed. The flag was obsolete, and hindering with the new changes.
• MigrateServedTypes (and From)
  • Additional to the vreplication changes, MigrateServedTypes has been refactored to be more readable.
• Legacy Blp RPC APIs have been replaced by VReplication functions.
• New vtctl command VReplicationExec has been added.

[derekperkins]

Nit: should it be vtreplication, given that everything else is prefixed with vt?

[sougou]

This follows the VSchema legacy :)

[michael-berlin]

AFAIK, @alainjobart introduced strategy flags early in the implementation such that he could experiment with the code and skip certain parts of it. In general, the code is proven and therefore I don't see a need for these flags.

Sugu, do you plan to make changes to the schema of the vreplication table in the future? That's one probem I hit when I introduced the "max_tps" and "max_replication_lag" columns: When I changed the schema, I had to manually delete the table from all YouTube keyspaces which were previously resharded. I also tried to write some code to automatically detect and update this, but I ran into some limitations how we can run the admin queries through tabletmanager. I don't remember the details.

[sougou]

I'm adding columns to vreplication as we speak. For now, I'm just adding them to create table. Once submitted, subsequent changes must be performed through additional alter table commands that follow the original create table. This will take care of legacy.

[jvaidya]

"BinlogPlayer is handling" -> "BinlogPlayer handles".

[sougou]

Done.

[jvaidya]

This change makes binlog replication a susbsytem in it's own right and resharding the current use case. This has the potential to enable a bunch of exciting use cases.

Overall, I think it will be useful if you described the high level changes to code either in the description of the CR or as comments in key files.

More comments inline.

[jvaidya]

It might make sense to delete the old 'blp_checkpoint' table if it exists.

[sougou]

Done.

[jvaidya]

Is this always gtid? How is file:pos style replication supported? Or is it not supported?

[sougou]

The term gtid made sense when this code was written because both MariaDB and MySQL overloaded the term, which was conveniently reused by us. But now, we're shoehorning other things in this. The ideal future term for this would be VReplicationPosition. The refactor is not worth it at this moment.

[jvaidya]

Is there a comment someplace where we explain what each column stands for and how it is used? I would find such a comment very useful.

[sougou]

Added it here. This seems to be the best place.

[jvaidya]

On line 485 above, "Retrying txn in 1 second" would be a more useful error message.

[sougou]

Done.

[jvaidya]

Could you add comment here describing the logic behind this? May be also state where else the VREngine is started?

[sougou]

Added a comment for now. The action agent has an unusual state machine that I'm still not too familiar with. It's possible that RefreshState is the one where things are first started. So, I'll refrain from commenting until I fully understand.

If at all, it should be a comment describing the tablet manager state machine. Not specifically for VREngine.

[jvaidya]

Is this new code? Seems to be similar to vitess/go/vt/vtgate/planbuilder. What has made this necessary?

[sougou]

It's the same pattern, but the plan is very different. It's only for handling VReplication. I think this is the third planbuilder pattern modeled after tabletserver and vtgate.

[jvaidya]

I really like that this functionality is explicit now. As we discussed, it might make sense to make this a higher level module in a future iteration.

[sougou]

Agree.

[jvaidya]

Would we not need to communicate this in the new way of doing things? Perhaps by giving example queries to run with VReplicationExec?

[sougou]

Yes. I'll add the "user guide" at the end when I polish things up.

[jvaidya]

Could you add comment for step // 2 here?

[sougou]

Done.

[jvaidya]

Could you add comments about why 'strategy' is not needed any longer?

[sougou]

This is in the message of the commit that removes it: 4ea87b9

[sougou]

I'll edit the PR comments when the change is done. All other review comments adressed/answered.

[sougou]

Done.

[sougou]

Done.

[sougou]

Added it here. This seems to be the best place.

[sougou]

Done.

[sougou]

The term gtid made sense when this code was written because both MariaDB and MySQL overloaded the term, which was conveniently reused by us. But now, we're shoehorning other things in this. The ideal future term for this would be VReplicationPosition. The refactor is not worth it at this moment.

[sougou]

It's the same pattern, but the plan is very different. It's only for handling VReplication. I think this is the third planbuilder pattern modeled after tabletserver and vtgate.

[sougou]

Agree.

[sougou]

Yes. I'll add the "user guide" at the end when I polish things up.

[sougou]

Done.

[sougou]

This is in the message of the commit that removes it: 4ea87b9

[sougou]

@demmer @rafael @alainjobart @michael-berlin @jvaidya

This is now ready for review.

Given that everyone is short on time, and this is a big change. I propose the following strategy:

• @jvaidya will do an in-depth review
• @demmer and @rafael will review the high level design, and go style related things
• @alainjobart and @michael-berlin: If you have time, review what you can. If you want to just see what has changed externally, test/resharding.py will show you that.

[rafael]

@sougou - This is really cool. I did the following review:

1. High level review of vt_replication. Most of this part makes sense to me. Didn't look in detail how all the binlog player works, but because the bulk of it didn't change, I didn't want to focus on that. I added few comments in this section.
2. I tried to dig deeper in the MigrateServedTypes changes, most of my comments are in that section.
3. Did not look in detail at the changes in the tests files.

[rafael]

Comments does not seem accurate anymore. startPosition/stopPosition are no longer part of this method.

[rafael]

nit - now that this table is not really tied to what we call workflows in other places, I think this should be something like: stream_creator, stream_manager, stream_owner

[sougou]

This API is still low level, and should not be used by a human directly unless they knew what they're doing. I expect that there will be newer and more flexible workflows that will eventually wrap this.

[rafael]

nit - Maybe: readVRSettings retrieves the throttler settings from vtreplication table.

I think we should avoid references to checkpoint as it could be confusing in the future.

[rafael]

@sougou - This change makes the failure scenario a bit different. I'm not sure if it matters, but wanted to get your thoughts on this.

Before your change:

1. Sets source shards.
2. Gets all tablet alias for destination
3. It refreshes destination tablets.

Now it looks like is doing all these steps one shard at a time in a loop.

[sougou]

Correct. It was necessary because the uid is now returned by the vreplication create command. So, using it right away to create the source shard makes sense. The UID was previously inferred by the order of the shards at a different location. That approach felt too flimsy, and didn't offer any particular advantages.

[rafael]

Ah nice. This is a good validation related to what we were talking about in Slack. With this, current worfklows that merge shards will fail, which is better than current state where it will pass but only checking one side.

[rafael]

out of curiosity why defer cancel() ? In many other places I always see just cancel()

[sougou]

The defer gives you the guarantee that it will get called even if there was a panic in the code. However, it's not a good idea to defer inside a for loop. Also, defer is slightly more expensive than regular function calls. So, it may be worth avoiding in tight loops, like byte-by-byte operations.

[rafael]

There is a risk here that we've been discussing internally at Slack. If enabling query service in destination shards fails, you could end up in a state where source shards won't have any replicas if they get refreshed.

This change does not make it worse, it is the same behavior we have today. But it is something that I think we should change.

[sougou]

Yeah. I want to make a separate PR to fix this.

[rafael]

I think this might be missing in masterMigrateServedType. We need to RefreshState in source destination shards so they restart and don't have query service enabled disabled in replicas.

[sougou]

The refresh of the source shards is done at the beginning of the function as soon as their shard records are updated. This is because we have to first stop serving , then wait for replication to catch-up, and then only enable destinations. The diff tool makes this hard to see. The code reads a lot better if you just looked at it without the diff.

[rafael]

Oh my comment was incorrect. It should have been: refresh destination shards. Before the change when migrating masters:

			if err := si.UpdateServedTypesMap(servedType, cells, reverse); err != nil {
				return err
			}
			if tc := si.GetTabletControl(servedType); !reverse && tc != nil && tc.DisableQueryService {
				// This is a forwards migration, and the
				// destination query service was already in a
				// disabled state. We need to enable and force
				// a refresh, otherwise it's possible that both
				// the source and destination will have query
				// service disabled at the same time, and
				// queries would have nowhere to go.
				if err := si.UpdateDisableQueryService(ctx, servedType, cells, false); err != nil {
					return err
				}
				needToRefreshDestinationTablets = true
			}
        // .... then
	if needToRefreshDestinationTablets {
		event.DispatchUpdate(ev, "refreshing destination shard tablets so they restart their query service")
		for _, si := range destinationShards {
			wr.RefreshTabletsByShard(ctx, si, []topodatapb.TabletType{servedType}, cells)
		}
	}

After the change, I don't see this happening. I'm only seeing refresh destination master (which existent version also does).

[rafael]

We synced on this offline, and got to the conclusion that the slight new behavior makes more sense than the original logic.

[rafael]

Maybe isSource ? We call them source, destination everywhere. It took me a bit of cognitive load to switch context here and think in terms of to, from.

[sougou]

isFrom is used here because the direction can be reversed with the -reverse flag. You can see in replicaMigrateServedTypes that we swap the order based on that flag. Hence the new terminology.

[rafael]

Given, that you are doing all this refactor now. Would you like to take at stab at making this safer and re-enable query service in masters if we can't catch up in filtered replication?

[sougou]

Yes. As mentioned earlier, I'll do this in a new PR.

[sougou]

Comments are all addressed.

[sougou]

This API is still low level, and should not be used by a human directly unless they knew what they're doing. I expect that there will be newer and more flexible workflows that will eventually wrap this.

[sougou]

Correct. It was necessary because the uid is now returned by the vreplication create command. So, using it right away to create the source shard makes sense. The UID was previously inferred by the order of the shards at a different location. That approach felt too flimsy, and didn't offer any particular advantages.

[sougou]

The defer gives you the guarantee that it will get called even if there was a panic in the code. However, it's not a good idea to defer inside a for loop. Also, defer is slightly more expensive than regular function calls. So, it may be worth avoiding in tight loops, like byte-by-byte operations.

[sougou]

Yeah. I want to make a separate PR to fix this.

[sougou]

The refresh of the source shards is done at the beginning of the function as soon as their shard records are updated. This is because we have to first stop serving , then wait for replication to catch-up, and then only enable destinations. The diff tool makes this hard to see. The code reads a lot better if you just looked at it without the diff.

[sougou]

isFrom is used here because the direction can be reversed with the -reverse flag. You can see in replicaMigrateServedTypes that we swap the order based on that flag. Hence the new terminology.

[sougou]

Yes. As mentioned earlier, I'll do this in a new PR.

[jvaidya]

LGTM. This is a pretty exciting feature. Thanks for getting this to the finish line.

[rafael]

Thanks for all the clarifications. This is exciting. LGTM.