*: Introduce version checking mechanism

Gopkg.toml

@@ -26,4 +26,5 @@
 
 [[constraint]]
   name = "github.com/pingcap/kvproto"
-  branch = "master"
+  source = "https://github.com/nolouch/kvproto"
+  branch = "version-introduce"

pkg/integration_test/cluster.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"github.com/coreos/etcd/clientv3"
+	"github.com/coreos/go-semver/semver"
 	"github.com/juju/errors"
 	"github.com/pingcap/kvproto/pkg/pdpb"
 	"github.com/pingcap/pd/server"
@@ -116,6 +117,12 @@ func (s *testServer) GetClusterID() uint64 {
 	return s.server.ClusterID()
 }
 
+func (s *testServer) GetClusterVersion() semver.Version {
+	s.RLock()
+	defer s.RUnlock()
+	return s.server.ClusterVersion()
+}
+
 func (s *testServer) GetServerID() uint64 {
 	s.RLock()
 	defer s.RUnlock()

pkg/integration_test/version_upgrade_test.go

@@ -0,0 +1,145 @@
+// 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 (
+	"context"
+
+	"github.com/coreos/go-semver/semver"
+	. "github.com/pingcap/check"
+	"github.com/pingcap/kvproto/pkg/metapb"
+	"github.com/pingcap/kvproto/pkg/pdpb"
+)
+
+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}}},
+	}
+	_, err := server.server.Bootstrap(context.Background(), bootstrapReq)
+	c.Assert(err, IsNil)
+}
+
+func (s *integrationTestSuite) TestStoreRegister(c *C) {
+	c.Parallel()
+	cluster, err := newTestCluster(3)
+	c.Assert(err, IsNil)
+	defer cluster.Destory()
+
+	err = cluster.RunInitialServers()
+	c.Assert(err, IsNil)
+	cluster.WaitLeader()
+	leaderServer := cluster.GetServer(cluster.GetLeader())
+	s.bootstrapCluster(leaderServer, c)
+
+	putStoreRequest := &pdpb.PutStoreRequest{
+		Header: &pdpb.RequestHeader{ClusterId: leaderServer.GetClusterID()},
+		Store: &metapb.Store{
+			Id:      1,
+			Address: "mock-1",
+			Version: "2.0.1",
+		},
+	}
+	_, err = leaderServer.server.PutStore(context.Background(), putStoreRequest)
+	c.Assert(err, IsNil)
+	// FIX ME: read v0.0.0 in sometime
+	cluster.WaitLeader()
+	version := leaderServer.GetClusterVersion()
+	// Restart all PDs.
+	err = cluster.StopAll()
+	c.Assert(err, IsNil)
+	err = cluster.RunInitialServers()
+	c.Assert(err, IsNil)
+	cluster.WaitLeader()
+
+	leaderServer = cluster.GetServer(cluster.GetLeader())
+	newVersion := leaderServer.GetClusterVersion()
+	c.Assert(version, Equals, newVersion)
+
+	// putNewStore with old version
+	putStoreRequest = &pdpb.PutStoreRequest{
+		Header: &pdpb.RequestHeader{ClusterId: leaderServer.GetClusterID()},
+		Store: &metapb.Store{
+			Id:      4,
+			Address: "mock-4",
+			Version: "1.0.1",
+		},
+	}
+	_, err = leaderServer.server.PutStore(context.Background(), putStoreRequest)
+	c.Assert(err, NotNil)
+}
+
+func (s *integrationTestSuite) TestRollingUpgrade(c *C) {
+	c.Parallel()
+	cluster, err := newTestCluster(3)
+	c.Assert(err, IsNil)
+	defer cluster.Destory()
+	err = cluster.RunInitialServers()
+	c.Assert(err, IsNil)
+	cluster.WaitLeader()
+	leaderServer := cluster.GetServer(cluster.GetLeader())
+	s.bootstrapCluster(leaderServer, c)
+
+	stores := []*pdpb.PutStoreRequest{
+		{
+			Header: &pdpb.RequestHeader{ClusterId: leaderServer.GetClusterID()},
+			Store: &metapb.Store{
+				Id:      1,
+				Address: "mock-1",
+				Version: "2.0.1",
+			},
+		},
+		{
+			Header: &pdpb.RequestHeader{ClusterId: leaderServer.GetClusterID()},
+			Store: &metapb.Store{
+				Id:      4,
+				Address: "mock-4",
+				Version: "2.0.1",
+			},
+		},
+		{
+			Header: &pdpb.RequestHeader{ClusterId: leaderServer.GetClusterID()},
+			Store: &metapb.Store{
+				Id:      6,
+				Address: "mock-6",
+				Version: "2.0.1",
+			},
+		},
+		{
+			Header: &pdpb.RequestHeader{ClusterId: leaderServer.GetClusterID()},
+			Store: &metapb.Store{
+				Id:      7,
+				Address: "mock-7",
+				Version: "2.0.1",
+			},
+		},
+	}
+	for _, store := range stores {
+		_, err = leaderServer.server.PutStore(context.Background(), store)
+		c.Assert(err, IsNil)
+	}
+	c.Assert(leaderServer.GetClusterVersion(), Equals, semver.Version{Major: 2, Minor: 0, Patch: 1})
+	// rolling update
+	for i, store := range stores {
+		store.Store.Version = "2.1.0"
+		resp, err := leaderServer.server.PutStore(context.Background(), store)
+		c.Assert(err, IsNil)
+		if i != len(stores)-1 {
+			c.Assert(leaderServer.GetClusterVersion(), Equals, semver.Version{Major: 2, Minor: 0, Patch: 1})
+			c.Assert(resp.GetHeader().GetError(), IsNil)
+		}
+	}
+	c.Assert(leaderServer.GetClusterVersion(), Equals, semver.Version{Major: 2, Minor: 1})
+}

server/api/label_test.go

@@ -46,6 +46,7 @@ func (s *testLabelsStoreSuite) SetUpSuite(c *C) {
 					Value: "ssd",
 				},
 			},
+			Version: "2.0.0",
 		},
 		{
 			Id:      4,
@@ -61,6 +62,7 @@ func (s *testLabelsStoreSuite) SetUpSuite(c *C) {
 					Value: "hdd",
 				},
 			},
+			Version: "2.0.0",
 		},
 		{
 			Id:      6,
@@ -76,6 +78,7 @@ func (s *testLabelsStoreSuite) SetUpSuite(c *C) {
 					Value: "ssd",
 				},
 			},
+			Version: "2.0.0",
 		},
 		{
 			Id:      7,
@@ -95,6 +98,7 @@ func (s *testLabelsStoreSuite) SetUpSuite(c *C) {
 					Value: "test",
 				},
 			},
+			Version: "2.0.0",
 		},
 	}
 

server/api/operator_test.go

@@ -87,6 +87,7 @@ func mustPutStore(c *C, svr *server.Server, id uint64, state metapb.StoreState,
 			Address: fmt.Sprintf("tikv%d", id),
 			State:   state,
 			Labels:  labels,
+			Version: server.MinSupportedVersion(server.Version2_0).String(),
 		},
 	})
 	c.Assert(err, IsNil)

server/api/store_test.go

@@ -44,23 +44,27 @@ func (s *testStoreSuite) SetUpSuite(c *C) {
 			Id:      1,
 			Address: "tikv1",
 			State:   metapb.StoreState_Up,
+			Version: "2.0.0",
 		},
 		{
 			Id:      4,
 			Address: "tikv4",
 			State:   metapb.StoreState_Up,
+			Version: "2.0.0",
 		},
 		{
 			// metapb.StoreState_Offline == 1
 			Id:      6,
 			Address: "tikv6",
 			State:   metapb.StoreState_Offline,
+			Version: "2.0.0",
 		},
 		{
 			// metapb.StoreState_Tombstone == 2
 			Id:      7,
 			Address: "tikv7",
 			State:   metapb.StoreState_Tombstone,
+			Version: "2.0.0",
 		},
 	}
 

server/cluster.go

@@ -290,6 +290,15 @@ func (c *RaftCluster) putStore(store *metapb.Store) error {
 		return errors.Errorf("invalid put store %v", store)
 	}
 
+	v, err := ParseVersion(store.GetVersion())
+	if err != nil {
+		return errors.Errorf("invalid put store %v, error: %s", store, err)
+	}
+	clusterVersion := c.cachedCluster.opt.loadClusterVersion()
+	if !IsCompatible(clusterVersion, *v) {
+		return errors.Errorf("version should compatible with version  %s, got %s", clusterVersion, v)
+	}
+
 	cluster := c.cachedCluster
 
 	// Store address can not be the same as other stores.
@@ -310,16 +319,15 @@ func (c *RaftCluster) putStore(store *metapb.Store) error {
 	} else {
 		// Update an existed store.
 		s.Address = store.Address
+		s.Version = store.Version
 		s.MergeLabels(store.Labels)
 	}
-
 	// Check location labels.
 	for _, k := range c.cachedCluster.GetLocationLabels() {
 		if v := s.GetLabelValue(k); len(v) == 0 {
 			log.Warnf("missing location label %q in store %v", k, s)
 		}
 	}
-
 	return cluster.putStore(s)
 }
 
@@ -518,6 +526,7 @@ func (c *RaftCluster) runBackgroundJobs(interval time.Duration) {
 			c.checkStores()
 			c.collectMetrics()
 			c.coordinator.pruneHistory()
+			c.cachedCluster.OnChangeClusterVersion()
 		}
 	}
 }

server/cluster_info.go

@@ -17,6 +17,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/coreos/go-semver/semver"
 	"github.com/gogo/protobuf/proto"
 	"github.com/juju/errors"
 	"github.com/pingcap/kvproto/pkg/metapb"
@@ -78,6 +79,45 @@ func loadClusterInfo(id core.IDAllocator, kv *core.KV, opt *scheduleOption) (*cl
 	return c, nil
 }
 
+func (c *clusterInfo) OnChangeClusterVersion() {
+	var (
+		minVersion     semver.Version
+		clusterVersion semver.Version
+	)
+
+	clusterVersion = c.opt.loadClusterVersion()
+	stores := c.GetStores()
+	for i, s := range stores {
+		if s.IsTombstone() {
+			continue
+		}
+		if i == 0 {
+			minVersion = *MustParseVersion(s.GetVersion())
+			continue
+		}
+		v := MustParseVersion(s.GetVersion())
+
+		if v.LessThan(minVersion) {
+			minVersion = *v
+		}
+	}
+	if clusterVersion.LessThan(minVersion) {
+		c.opt.SetClusterVersion(minVersion)
+		c.opt.persist(c.kv)
+		log.Infof("cluster version changed from %s to %s", clusterVersion, minVersion)
+	}
+}
+
+// IsSupported check if support the feature.
+func (c *clusterInfo) IsSupported(f Feature) bool {
+	clusterVersion := c.opt.loadClusterVersion()
+	minSupportVersion := MinSupportedVersion(f)
+	if clusterVersion.LessThan(minSupportVersion) {
+		return false
+	}
+	return true
+}
+
 func (c *clusterInfo) allocID() (uint64, error) {
 	return c.id.Alloc()
 }
@@ -618,6 +658,9 @@ func (c *clusterInfo) GetHotRegionLowThreshold() int {
 }
 
 func (c *clusterInfo) IsRaftLearnerEnabled() bool {
+	if !c.IsSupported(RaftLearner) {
+		return false
+	}
 	return c.opt.IsRaftLearnerEnabled()
 }
 

server/config.go

@@ -26,6 +26,7 @@ import (
 	"github.com/BurntSushi/toml"
 	"github.com/coreos/etcd/embed"
 	"github.com/coreos/etcd/pkg/transport"
+	"github.com/coreos/go-semver/semver"
 	"github.com/juju/errors"
 	"github.com/pingcap/pd/pkg/logutil"
 	"github.com/pingcap/pd/pkg/metricutil"
@@ -77,6 +78,8 @@ type Config struct {
 
 	Namespace map[string]NamespaceConfig `json:"namespace"`
 
+	ClusterVersion semver.Version `json:"cluster-version"`
+
 	// QuotaBackendBytes Raise alarms when backend size exceeds the given quota. 0 means use the default quota.
 	// the default size is 2GB, the maximum is 8GB.
 	QuotaBackendBytes typeutil.ByteSize `toml:"quota-backend-bytes" json:"quota-backend-bytes"`