*: 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-region"
 
 [[constraint]]
   name = "github.com/google/btree"

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

@@ -19,6 +19,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/gogo/protobuf/proto"
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/kvproto/pkg/pdpb"
 	"github.com/pingcap/pd/pkg/error_code"
@@ -55,8 +56,10 @@ type RaftCluster struct {
 
 	coordinator *coordinator
 
-	wg   sync.WaitGroup
-	quit chan struct{}
+	wg           sync.WaitGroup
+	quit         chan struct{}
+	clients      map[string]pdpb.PDClient
+	regionSyncer *regionSyncer
 }
 
 // ClusterStatus saves some state information
@@ -66,10 +69,12 @@ 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(),
+		clients:      make(map[string]pdpb.PDClient),
+		regionSyncer: newRegionSyncer(s),
 	}
 }
 
@@ -115,15 +120,55 @@ 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
+	maxBatchSize := 100
+	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 < maxBatchSize; i++ {
+				region := <-c.cachedCluster.getChangedRegions()
+				requests = append(requests, region.GetMeta())
+			}
+			msg := &pdpb.MetaRegions{
+				Count:   uint32(len(requests)),
+				Regions: requests}
+			data, err := proto.Marshal(msg)
+			if err != nil {
+				log.Errorf("Report regions meet error: %s", err)
+				continue
+			}
+			req := &pdpb.SyncRegionResponse{
+				Header: &pdpb.ResponseHeader{ClusterId: c.s.clusterID},
+				Data:   data,
+			}
+			for _, stream := range c.regionSyncer.streams {
+				err := stream.Send(req)
+				if err != nil {
+					log.Errorf("Report regions meet error: %s", err)
+				}
+			}
+		}
+		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,9 @@ 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)
 		}
+		if c.opt.loadPDServerConfig().EnableRegionStorage {
+			c.changedRegions <- region
+		}
 	}
 	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.FlushRegion(), 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 enable 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"`