*: Add a new way to store metadata of regions

Gopkg.toml

@@ -26,7 +26,8 @@
 
 [[constraint]]
   name = "github.com/pingcap/kvproto"
-  branch = "master"
+  source = "https://github.com/nolouch/kvproto.git"
+  branch = "sync-regions"
 
 [[constraint]]
   name = "github.com/google/btree"

pkg/integration_test/cluster.go

@@ -157,11 +157,14 @@ type testCluster struct {
 	servers map[string]*testServer
 }
 
-func newTestCluster(initialServerCount int) (*testCluster, error) {
+// ConfigOption is used to define customize settings in test.
+type ConfigOption func(conf *server.Config)
+
+func newTestCluster(initialServerCount int, opts ...ConfigOption) (*testCluster, error) {
 	config := newClusterConfig(initialServerCount)
 	servers := make(map[string]*testServer)
 	for _, conf := range config.InitialServers {
-		serverConf, err := conf.Generate()
+		serverConf, err := conf.Generate(opts...)
 		if err != nil {
 			return nil, err
 		}

pkg/integration_test/config.go

@@ -45,7 +45,7 @@ func newServerConfig(name string, cc *clusterConfig, join bool) *serverConfig {
 	}
 }
 
-func (c *serverConfig) Generate() (*server.Config, error) {
+func (c *serverConfig) Generate(opts ...ConfigOption) (*server.Config, error) {
 	arguments := []string{
 		"--name=" + c.Name,
 		"--data-dir=" + c.DataDir,
@@ -65,6 +65,9 @@ func (c *serverConfig) Generate() (*server.Config, error) {
 	if err != nil {
 		return nil, err
 	}
+	for _, opt := range opts {
+		opt(cfg)
+	}
 	return cfg, nil
 }
 

pkg/integration_test/region_syncer_test.go

@@ -0,0 +1,77 @@
+// Copyright 2018 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package integration
+
+import (
+	"fmt"
+	"time"
+
+	. "github.com/pingcap/check"
+	"github.com/pingcap/kvproto/pkg/metapb"
+
+	"github.com/pingcap/pd/server"
+	"github.com/pingcap/pd/server/core"
+)
+
+type idAllocator struct {
+	id uint64
+}
+
+func (alloc *idAllocator) Alloc() uint64 {
+	alloc.id++
+	return alloc.id
+}
+
+func (s *integrationTestSuite) TestRegionSyncer(c *C) {
+	c.Parallel()
+	cluster, err := newTestCluster(3, func(conf *server.Config) { conf.PDServerCfg.EnableRegionStorage = true })
+	c.Assert(err, IsNil)
+
+	err = cluster.RunInitialServers()
+	c.Assert(err, IsNil)
+	cluster.WaitLeader()
+	leaderServer := cluster.GetServer(cluster.GetLeader())
+	s.bootstrapCluster(leaderServer, c)
+	rc := leaderServer.server.GetRaftCluster()
+	c.Assert(rc, NotNil)
+	regionLen := 110
+	id := &idAllocator{}
+	regions := make([]*core.RegionInfo, 0, regionLen)
+	for i := 0; i < regionLen; i++ {
+		r := &metapb.Region{
+			Id: id.Alloc(),
+			RegionEpoch: &metapb.RegionEpoch{
+				ConfVer: 1,
+				Version: 1,
+			},
+			StartKey: []byte{byte(i)},
+			EndKey:   []byte{byte(i + 1)},
+			Peers:    []*metapb.Peer{{Id: id.Alloc(), StoreId: uint64(0)}},
+		}
+		regions = append(regions, core.NewRegionInfo(r, r.Peers[0]))
+	}
+	for _, region := range regions {
+		rc.HandleRegionHeartbeat(region)
+	}
+	// ensure flush to region kv
+	time.Sleep(3 * time.Second)
+	leaderServer.Stop()
+	leader := cluster.WaitLeader()
+	fmt.Println("leader:", leader)
+	leaderServer = cluster.GetServer(cluster.GetLeader())
+	c.Assert(leaderServer, NotNil)
+	loadRegions := leaderServer.server.GetRaftCluster().GetRegions()
+	c.Assert(len(loadRegions), Equals, regionLen)
+	cluster.Destroy()
+}

pkg/integration_test/version_upgrade_test.go

@@ -26,7 +26,7 @@ func (s *integrationTestSuite) bootstrapCluster(server *testServer, c *C) {
 	bootstrapReq := &pdpb.BootstrapRequest{
 		Header: &pdpb.RequestHeader{ClusterId: server.GetClusterID()},
 		Store:  &metapb.Store{Id: 1, Address: "mock://1"},
-		Region: &metapb.Region{Id: 2, Peers: []*metapb.Peer{{3, 1, false}}},
+		Region: &metapb.Region{Id: 2, Peers: []*metapb.Peer{{Id: 3, StoreId: 1, IsLearner: false}}},
 	}
 	_, err := server.server.Bootstrap(context.Background(), bootstrapReq)
 	c.Assert(err, IsNil)

server/cluster.go

@@ -55,8 +55,9 @@ type RaftCluster struct {
 
 	coordinator *coordinator
 
-	wg   sync.WaitGroup
-	quit chan struct{}
+	wg           sync.WaitGroup
+	quit         chan struct{}
+	regionSyncer *regionSyncer
 }
 
 // ClusterStatus saves some state information
@@ -66,10 +67,11 @@ type ClusterStatus struct {
 
 func newRaftCluster(s *Server, clusterID uint64) *RaftCluster {
 	return &RaftCluster{
-		s:           s,
-		running:     false,
-		clusterID:   clusterID,
-		clusterRoot: s.getClusterRootPath(),
+		s:            s,
+		running:      false,
+		clusterID:    clusterID,
+		clusterRoot:  s.getClusterRootPath(),
+		regionSyncer: newRegionSyncer(s),
 	}
 }
 
@@ -115,15 +117,41 @@ func (c *RaftCluster) start() error {
 	c.cachedCluster.regionStats = newRegionStatistics(c.s.scheduleOpt, c.s.classifier)
 	c.quit = make(chan struct{})
 
-	c.wg.Add(2)
+	c.wg.Add(3)
 	go c.runCoordinator()
 	go c.runBackgroundJobs(backgroundJobInterval)
-
+	go c.runSyncRegions()
 	c.running = true
 
 	return nil
 }
 
+func (c *RaftCluster) runSyncRegions() {
+	defer logutil.LogPanic()
+	defer c.wg.Done()
+	var requests []*metapb.Region
+	// grpc has limit on message size
+	for {
+		select {
+		case <-c.quit:
+			return
+		case first := <-c.cachedCluster.getChangedRegions():
+			requests = append(requests, first.GetMeta())
+			pending := len(c.cachedCluster.getChangedRegions())
+			for i := 0; i < pending && i < maxSyncRegionBatchSize; i++ {
+				region := <-c.cachedCluster.getChangedRegions()
+				requests = append(requests, region.GetMeta())
+			}
+			regions := &pdpb.SyncRegionResponse{
+				Header:  &pdpb.ResponseHeader{ClusterId: c.s.clusterID},
+				Regions: requests,
+			}
+			c.regionSyncer.broadcast(regions)
+		}
+		requests = requests[:0]
+	}
+}
+
 func (c *RaftCluster) runCoordinator() {
 	defer logutil.LogPanic()
 

server/cluster_info.go

@@ -38,8 +38,11 @@ type clusterInfo struct {
 	regionStats     *regionStatistics
 	labelLevelStats *labelLevelStatistics
 	prepareChecker  *prepareChecker
+	changedRegions  chan *core.RegionInfo
 }
 
+var defaultChangedRegionsLimit = 10000
+
 func newClusterInfo(id core.IDAllocator, opt *scheduleOption, kv *core.KV) *clusterInfo {
 	return &clusterInfo{
 		core:            schedule.NewBasicCluster(),
@@ -48,6 +51,7 @@ func newClusterInfo(id core.IDAllocator, opt *scheduleOption, kv *core.KV) *clus
 		kv:              kv,
 		labelLevelStats: newLabelLevelStatistics(),
 		prepareChecker:  newPrepareChecker(),
+		changedRegions:  make(chan *core.RegionInfo, defaultChangedRegionsLimit),
 	}
 }
 
@@ -108,6 +112,10 @@ func (c *clusterInfo) OnStoreVersionChange() {
 	}
 }
 
+func (c *clusterInfo) getChangedRegions() <-chan *core.RegionInfo {
+	return c.changedRegions
+}
+
 // IsFeatureSupported checks if the feature is supported by current cluster.
 func (c *clusterInfo) IsFeatureSupported(f Feature) bool {
 	clusterVersion := c.opt.loadClusterVersion()
@@ -511,6 +519,10 @@ func (c *clusterInfo) handleRegionHeartbeat(region *core.RegionInfo) error {
 			// after restart. Here we only log the error then go on updating cache.
 			log.Errorf("[region %d] fail to save region %v: %v", region.GetID(), region, err)
 		}
+		select {
+		case c.changedRegions <- region:
+		default:
+		}
 	}
 	if !isWriteUpdate && !isReadUpdate && !saveCache && !isNew {
 		return nil

server/cluster_info_test.go

@@ -700,6 +700,7 @@ func mustSaveRegions(c *C, kv *core.KV, n int) []*metapb.Region {
 	for _, region := range regions {
 		c.Assert(kv.SaveRegion(region), IsNil)
 	}
+	c.Assert(kv.Flush(), IsNil)
 
 	return regions
 }

server/config.go

@@ -78,6 +78,8 @@ type Config struct {
 
 	Namespace map[string]NamespaceConfig `json:"namespace"`
 
+	PDServerCfg PDServerConfig `toml:"pd-server" json:"pd-server"`
+
 	ClusterVersion semver.Version `json:"cluster-version"`
 
 	// QuotaBackendBytes Raise alarms when backend size exceeds the given quota. 0 means use the default quota.
@@ -633,6 +635,12 @@ func (s SecurityConfig) ToTLSConfig() (*tls.Config, error) {
 	return tlsConfig, nil
 }
 
+// PDServerConfig is the configuration for pd server.
+type PDServerConfig struct {
+	// EnableRegionStorage enables the independent region storage.
+	EnableRegionStorage bool `toml:"enable-region-storage" json:"enable-region-storage"`
+}
+
 // StoreLabel is the config item of LabelPropertyConfig.
 type StoreLabel struct {
 	Key   string `toml:"key" json:"key"`