vreplication: Reshard command

go.mod

@@ -53,6 +53,7 @@ require (
 	github.com/opentracing-contrib/go-grpc v0.0.0-20180928155321-4b5a12d3ff02
 	github.com/opentracing/opentracing-go v1.1.0
 	github.com/pborman/uuid v0.0.0-20160824210600-b984ec7fa9ff
+	github.com/pkg/errors v0.8.1
 	github.com/prometheus/client_golang v1.1.0
 	github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4 // indirect
 	github.com/prometheus/common v0.7.0 // indirect

go/vt/key/key.go

@@ -120,6 +120,21 @@ func EvenShardsKeyRange(i, n int) (*topodatapb.KeyRange, error) {
 	return &topodatapb.KeyRange{Start: startBytes, End: endBytes}, nil
 }
 
+// KeyRangeAdd adds two adjacent keyranges into a single value.
+// If the values are not adjacent, it returns false.
+func KeyRangeAdd(first, second *topodatapb.KeyRange) (*topodatapb.KeyRange, bool) {
+	if first == nil || second == nil {
+		return nil, false
+	}
+	if len(first.End) != 0 && bytes.Equal(first.End, second.Start) {
+		return &topodatapb.KeyRange{Start: first.Start, End: second.End}, true
+	}
+	if len(second.End) != 0 && bytes.Equal(second.End, first.Start) {
+		return &topodatapb.KeyRange{Start: second.Start, End: first.End}, true
+	}
+	return nil, false
+}
+
 // KeyRangeContains returns true if the provided id is in the keyrange.
 func KeyRangeContains(kr *topodatapb.KeyRange, id []byte) bool {
 	if kr == nil {

go/vt/key/key_test.go

@@ -139,6 +139,107 @@ func TestEvenShardsKeyRange(t *testing.T) {
 	}
 }
 
+func TestKeyRangeAdd(t *testing.T) {
+	testcases := []struct {
+		first  string
+		second string
+		out    string
+		ok     bool
+	}{{
+		first:  "",
+		second: "",
+		out:    "",
+		ok:     false,
+	}, {
+		first:  "",
+		second: "-80",
+		out:    "",
+		ok:     false,
+	}, {
+		first:  "-80",
+		second: "",
+		out:    "",
+		ok:     false,
+	}, {
+		first:  "",
+		second: "80-",
+		out:    "",
+		ok:     false,
+	}, {
+		first:  "80-",
+		second: "",
+		out:    "",
+		ok:     false,
+	}, {
+		first:  "80-",
+		second: "-40",
+		out:    "",
+		ok:     false,
+	}, {
+		first:  "-40",
+		second: "80-",
+		out:    "",
+		ok:     false,
+	}, {
+		first:  "-80",
+		second: "80-",
+		out:    "-",
+		ok:     true,
+	}, {
+		first:  "80-",
+		second: "-80",
+		out:    "-",
+		ok:     true,
+	}, {
+		first:  "-40",
+		second: "40-80",
+		out:    "-80",
+		ok:     true,
+	}, {
+		first:  "40-80",
+		second: "-40",
+		out:    "-80",
+		ok:     true,
+	}, {
+		first:  "40-80",
+		second: "80-c0",
+		out:    "40-c0",
+		ok:     true,
+	}, {
+		first:  "80-c0",
+		second: "40-80",
+		out:    "40-c0",
+		ok:     true,
+	}}
+	stringToKeyRange := func(spec string) *topodatapb.KeyRange {
+		if spec == "" {
+			return nil
+		}
+		parts := strings.Split(spec, "-")
+		if len(parts) != 2 {
+			panic("invalid spec")
+		}
+		kr, err := ParseKeyRangeParts(parts[0], parts[1])
+		if err != nil {
+			panic(err)
+		}
+		return kr
+	}
+	keyRangeToString := func(kr *topodatapb.KeyRange) string {
+		if kr == nil {
+			return ""
+		}
+		return KeyRangeString(kr)
+	}
+	for _, tcase := range testcases {
+		first := stringToKeyRange(tcase.first)
+		second := stringToKeyRange(tcase.second)
+		out, ok := KeyRangeAdd(first, second)
+		assert.Equal(t, tcase.out, keyRangeToString(out))
+		assert.Equal(t, tcase.ok, ok)
+	}
+}
+
 func TestEvenShardsKeyRange_Error(t *testing.T) {
 	testCases := []struct {
 		i, n      int

go/vt/topo/shard.go

@@ -279,7 +279,7 @@ func (ts *Server) CreateShard(ctx context.Context, keyspace, shard string) (err
 	defer unlock(&err)
 
 	// validate parameters
-	name, keyRange, err := ValidateShardName(shard)
+	_, keyRange, err := ValidateShardName(shard)
 	if err != nil {
 		return err
 	}
@@ -288,27 +288,20 @@ func (ts *Server) CreateShard(ctx context.Context, keyspace, shard string) (err
 		KeyRange: keyRange,
 	}
 
-	isMasterServing := true
-
-	// start the shard IsMasterServing. If it overlaps with
-	// other shards for some serving types, remove them.
-
-	if IsShardUsingRangeBasedSharding(name) {
-		// if we are using range-based sharding, we don't want
-		// overlapping shards to all serve and confuse the clients.
-		sis, err := ts.FindAllShardsInKeyspace(ctx, keyspace)
-		if err != nil && !IsErrType(err, NoNode) {
-			return err
-		}
-		for _, si := range sis {
-			if si.KeyRange == nil || key.KeyRangesIntersect(si.KeyRange, keyRange) {
-				isMasterServing = false
-			}
+	// Set master as serving only if its keyrange doesn't overlap
+	// with other shards. This applies to unsharded keyspaces also
+	value.IsMasterServing = true
+	sis, err := ts.FindAllShardsInKeyspace(ctx, keyspace)
+	if err != nil && !IsErrType(err, NoNode) {
+		return err
+	}
+	for _, si := range sis {
+		if si.KeyRange == nil || key.KeyRangesIntersect(si.KeyRange, keyRange) {
+			value.IsMasterServing = false
+			break
 		}
 	}
 
-	value.IsMasterServing = isMasterServing
-
 	// Marshal and save.
 	data, err := proto.Marshal(value)
 	if err != nil {

go/vt/topotools/shard_test.go

@@ -55,9 +55,11 @@ func TestCreateShard(t *testing.T) {
 	}
 }
 
-// TestCreateShardCustomSharding checks ServedTypes is set correctly
-// when creating multiple custom sharding shards
-func TestCreateShardCustomSharding(t *testing.T) {
+// TestCreateShardMultiUnsharded checks ServedTypes is set
+// only for the first created shard.
+// TODO(sougou): we should eventually disallow multiple shards
+// for unsharded keyspaces.
+func TestCreateShardMultiUnsharded(t *testing.T) {
 	ctx := context.Background()
 
 	// Set up topology.
@@ -90,7 +92,7 @@ func TestCreateShardCustomSharding(t *testing.T) {
 	if si, err := ts.GetShard(ctx, keyspace, shard1); err != nil {
 		t.Fatalf("GetShard(shard1) failed: %v", err)
 	} else {
-		if !si.IsMasterServing {
+		if si.IsMasterServing {
 			t.Fatalf("shard1 should have all 3 served types")
 		}
 	}

go/vt/topotools/split.go

@@ -17,14 +17,66 @@ limitations under the License.
 package topotools
 
 import (
+	"errors"
 	"fmt"
 	"sort"
 
 	"golang.org/x/net/context"
 	"vitess.io/vitess/go/vt/key"
+	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	"vitess.io/vitess/go/vt/topo"
 )
 
+// ValidateForReshard returns an error if sourceShards cannot reshard into
+// targetShards.
+func ValidateForReshard(sourceShards, targetShards []*topo.ShardInfo) error {
+	for _, source := range sourceShards {
+		for _, target := range targetShards {
+			if key.KeyRangeEqual(source.KeyRange, target.KeyRange) {
+				return fmt.Errorf("same keyrange is present in source and target: %v", key.KeyRangeString(source.KeyRange))
+			}
+		}
+	}
+	sourcekr, err := combineKeyRanges(sourceShards)
+	if err != nil {
+		return err
+	}
+	targetkr, err := combineKeyRanges(targetShards)
+	if err != nil {
+		return err
+	}
+	if !key.KeyRangeEqual(sourcekr, targetkr) {
+		return fmt.Errorf("source and target keyranges don't match: %v vs %v", key.KeyRangeString(sourcekr), key.KeyRangeString(targetkr))
+	}
+	return nil
+}
+
+func combineKeyRanges(shards []*topo.ShardInfo) (*topodatapb.KeyRange, error) {
+	if len(shards) == 0 {
+		return nil, fmt.Errorf("there are no shards to combine")
+	}
+	result := shards[0].KeyRange
+	krmap := make(map[string]*topodatapb.KeyRange)
+	for _, si := range shards[1:] {
+		krmap[si.ShardName()] = si.KeyRange
+	}
+	for len(krmap) != 0 {
+		foundOne := false
+		for k, kr := range krmap {
+			newkr, ok := key.KeyRangeAdd(result, kr)
+			if ok {
+				foundOne = true
+				result = newkr
+				delete(krmap, k)
+			}
+		}
+		if !foundOne {
+			return nil, errors.New("shards don't form a contiguous keyrange")
+		}
+	}
+	return result, nil
+}
+
 // OverlappingShards contains sets of shards that overlap which each-other.
 // With this library, there is no guarantee of which set will be left or right.
 type OverlappingShards struct {

go/vt/topotools/split_test.go

@@ -20,6 +20,7 @@ import (
 	"encoding/hex"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"vitess.io/vitess/go/vt/topo"
 
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
@@ -94,6 +95,64 @@ func compareResultLists(t *testing.T, os []*OverlappingShards, expected []expect
 	}
 }
 
+func TestValidateForReshard(t *testing.T) {
+	testcases := []struct {
+		sources []string
+		targets []string
+		out     string
+	}{{
+		sources: []string{"-80", "80-"},
+		targets: []string{"-40", "40-"},
+		out:     "",
+	}, {
+		sources: []string{"80-", "-80"},
+		targets: []string{"-40", "40-"},
+		out:     "",
+	}, {
+		sources: []string{"-40", "40-80", "80-"},
+		targets: []string{"-30", "30-"},
+		out:     "",
+	}, {
+		sources: []string{"0"},
+		targets: []string{"-40", "40-"},
+		out:     "",
+	}, {
+		sources: []string{"-40", "40-80", "80-"},
+		targets: []string{"-40", "40-"},
+		out:     "same keyrange is present in source and target: -40",
+	}, {
+		sources: []string{"-30", "30-80"},
+		targets: []string{"-40", "40-"},
+		out:     "source and target keyranges don't match: -80 vs -",
+	}, {
+		sources: []string{"-30", "20-80"},
+		targets: []string{"-40", "40-"},
+		out:     "shards don't form a contiguous keyrange",
+	}}
+	buildShards := func(shards []string) []*topo.ShardInfo {
+		sis := make([]*topo.ShardInfo, 0, len(shards))
+		for _, shard := range shards {
+			_, kr, err := topo.ValidateShardName(shard)
+			if err != nil {
+				panic(err)
+			}
+			sis = append(sis, topo.NewShardInfo("", shard, &topodatapb.Shard{KeyRange: kr}, nil))
+		}
+		return sis
+	}
+
+	for _, tcase := range testcases {
+		sources := buildShards(tcase.sources)
+		targets := buildShards(tcase.targets)
+		err := ValidateForReshard(sources, targets)
+		if tcase.out == "" {
+			assert.NoError(t, err)
+		} else {
+			assert.EqualError(t, err, tcase.out)
+		}
+	}
+}
+
 func TestFindOverlappingShardsNoOverlap(t *testing.T) {
 	var shardMap map[string]*topo.ShardInfo
 	var os []*OverlappingShards

go/vt/vtctl/vtctl.go

@@ -307,6 +307,9 @@ var commands = []commandGroup{
 			{"ValidateKeyspace", commandValidateKeyspace,
 				"[-ping-tablets] <keyspace name>",
 				"Validates that all nodes reachable from the specified keyspace are consistent."},
+			{"Reshard", commandReshard,
+				"[-skip_schema_copy] <keyspace.workflow> <source_shards> <target_shards>",
+				"Start a Resharding process. Example: Reshard ks.workflow001 '0' '-80,80-'"},
 			{"SplitClone", commandSplitClone,
 				"<keyspace> <from_shards> <to_shards>",
 				"Start the SplitClone process to perform horizontal resharding. Example: SplitClone ks '0' '-80,80-'"},
@@ -1784,6 +1787,23 @@ func commandValidateKeyspace(ctx context.Context, wr *wrangler.Wrangler, subFlag
 	return wr.ValidateKeyspace(ctx, keyspace, *pingTablets)
 }
 
+func commandReshard(ctx context.Context, wr *wrangler.Wrangler, subFlags *flag.FlagSet, args []string) error {
+	skipSchemaCopy := subFlags.Bool("skip_schema_copy", false, "Skip copying of schema to targets")
+	if err := subFlags.Parse(args); err != nil {
+		return err
+	}
+	if subFlags.NArg() != 3 {
+		return fmt.Errorf("three arguments are required: <keyspace.workflow>, source_shards, target_shards")
+	}
+	keyspace, workflow, err := splitKeyspaceWorkflow(subFlags.Arg(0))
+	if err != nil {
+		return err
+	}
+	source := strings.Split(subFlags.Arg(1), ",")
+	target := strings.Split(subFlags.Arg(2), ",")
+	return wr.Reshard(ctx, keyspace, workflow, source, target, *skipSchemaCopy)
+}
+
 func commandSplitClone(ctx context.Context, wr *wrangler.Wrangler, subFlags *flag.FlagSet, args []string) error {
 	if err := subFlags.Parse(args); err != nil {
 		return err