*: Introduce version checking mechanism (#1148)

Gopkg.toml

@@ -27,4 +27,3 @@
 [[constraint]]
   name = "github.com/pingcap/kvproto"
   branch = "master"
-

pdctl/command/config_command.go

@@ -25,11 +25,12 @@ import (
 )
 
 var (
-	configPrefix        = "pd/api/v1/config"
-	schedulePrefix      = "pd/api/v1/config/schedule"
-	replicationPrefix   = "pd/api/v1/config/replicate"
-	namespacePrefix     = "pd/api/v1/config/namespace"
-	labelPropertyPrefix = "pd/api/v1/config/label-property"
+	configPrefix         = "pd/api/v1/config"
+	schedulePrefix       = "pd/api/v1/config/schedule"
+	replicationPrefix    = "pd/api/v1/config/replicate"
+	namespacePrefix      = "pd/api/v1/config/namespace"
+	labelPropertyPrefix  = "pd/api/v1/config/label-property"
+	clusterVersionPrefix = "pd/api/v1/config/cluster-version"
 )
 
 // NewConfigCommand return a config subcommand of rootCmd
@@ -55,6 +56,7 @@ func NewShowConfigCommand() *cobra.Command {
 	sc.AddCommand(NewShowNamespaceConfigCommand())
 	sc.AddCommand(NewShowReplicationConfigCommand())
 	sc.AddCommand(NewShowLabelPropertyCommand())
+	sc.AddCommand(NewShowClusterVersionCommand())
 	return sc
 }
 
@@ -98,15 +100,26 @@ func NewShowLabelPropertyCommand() *cobra.Command {
 	return sc
 }
 
+// NewShowClusterVersionCommand returns a cluster version subcommand of show subcommand.
+func NewShowClusterVersionCommand() *cobra.Command {
+	sc := &cobra.Command{
+		Use:   "cluster-version",
+		Short: "show the cluster version",
+		Run:   showClusterVersionCommandFunc,
+	}
+	return sc
+}
+
 // NewSetConfigCommand return a set subcommand of configCmd
 func NewSetConfigCommand() *cobra.Command {
 	sc := &cobra.Command{
-		Use:   "set <option> <value>, set namespace <name> <option> <value>, set label-property <type> <key> <value>",
+		Use:   "set <option> <value>, set namespace <name> <option> <value>, set label-property <type> <key> <value>, set cluster-version <version>",
 		Short: "set the option with value",
 		Run:   setConfigCommandFunc,
 	}
 	sc.AddCommand(NewSetNamespaceConfigCommand())
 	sc.AddCommand(NewSetLabelPropertyCommand())
+	sc.AddCommand(NewSetClusterVersionCommand())
 	return sc
 }
 
@@ -130,6 +143,16 @@ func NewSetLabelPropertyCommand() *cobra.Command {
 	return sc
 }
 
+// NewSetClusterVersionCommand creates a set subcommand of set subcommand
+func NewSetClusterVersionCommand() *cobra.Command {
+	sc := &cobra.Command{
+		Use:   "cluster-version <version>",
+		Short: "set cluster version",
+		Run:   setClusterVersionCommandFunc,
+	}
+	return sc
+}
+
 // NewDeleteConfigCommand a set subcommand of cfgCmd
 func NewDeleteConfigCommand() *cobra.Command {
 	sc := &cobra.Command{
@@ -211,6 +234,15 @@ func showNamespaceConfigCommandFunc(cmd *cobra.Command, args []string) {
 	fmt.Println(r)
 }
 
+func showClusterVersionCommandFunc(cmd *cobra.Command, args []string) {
+	r, err := doRequest(cmd, clusterVersionPrefix, http.MethodGet)
+	if err != nil {
+		fmt.Printf("Failed to get cluster version: %s\n", err)
+		return
+	}
+	fmt.Println(r)
+}
+
 func postConfigDataWithPath(cmd *cobra.Command, key, value, path string) error {
 	var val interface{}
 	data := make(map[string]interface{})
@@ -305,3 +337,14 @@ func postLabelProperty(cmd *cobra.Command, action string, args []string) {
 	prefix := path.Join(labelPropertyPrefix)
 	postJSON(cmd, prefix, input)
 }
+
+func setClusterVersionCommandFunc(cmd *cobra.Command, args []string) {
+	if len(args) != 1 {
+		fmt.Println(cmd.UsageString())
+		return
+	}
+	input := map[string]interface{}{
+		"cluster-version": args[0],
+	}
+	postJSON(cmd, clusterVersionPrefix, input)
+}

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.GetClusterVersion()
+}
+
 func (s *testServer) GetServerID() uint64 {
 	s.RLock()
 	defer s.RUnlock()

pkg/integration_test/version_upgrade_test.go

@@ -0,0 +1,148 @@
+// 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 {
+		if i == 0 {
+			store.Store.State = metapb.StoreState_Tombstone
+		}
+		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/config.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/gorilla/mux"
 	"github.com/juju/errors"
+	"github.com/pingcap/pd/pkg/error_code"
 	"github.com/pingcap/pd/server"
 	"github.com/unrolled/render"
 )
@@ -168,3 +169,25 @@ func (h *confHandler) SetLabelProperty(w http.ResponseWriter, r *http.Request) {
 	}
 	h.rd.JSON(w, http.StatusOK, nil)
 }
+
+func (h *confHandler) GetClusterVersion(w http.ResponseWriter, r *http.Request) {
+	h.rd.JSON(w, http.StatusOK, h.svr.GetClusterVersion())
+}
+
+func (h *confHandler) SetClusterVersion(w http.ResponseWriter, r *http.Request) {
+	input := make(map[string]string)
+	if err := readJSONRespondError(h.rd, w, r.Body, &input); err != nil {
+		return
+	}
+	version, ok := input["cluster-version"]
+	if !ok {
+		errorResp(h.rd, w, errcode.NewInvalidInputErr(errors.New("not set cluster-version")))
+		return
+	}
+	err := h.svr.SetClusterVersion(version)
+	if err != nil {
+		errorResp(h.rd, w, errcode.NewInternalErr(err))
+		return
+	}
+	h.rd.JSON(w, http.StatusOK, nil)
+}

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/router.go

@@ -58,6 +58,8 @@ func createRouter(prefix string, svr *server.Server) *mux.Router {
 	router.HandleFunc("/api/v1/config/namespace/{name}", confHandler.DeleteNamespace).Methods("DELETE")
 	router.HandleFunc("/api/v1/config/label-property", confHandler.GetLabelProperty).Methods("GET")
 	router.HandleFunc("/api/v1/config/label-property", confHandler.SetLabelProperty).Methods("POST")
+	router.HandleFunc("/api/v1/config/cluster-version", confHandler.GetClusterVersion).Methods("GET")
+	router.HandleFunc("/api/v1/config/cluster-version", confHandler.SetClusterVersion).Methods("POST")
 
 	storeHandler := newStoreHandler(svr, rd)
 	router.HandleFunc("/api/v1/store/{id}", storeHandler.Get).Methods("GET")