*: add version check before start (#311) (#326)

* add version check

* Update pkg/utils/version.go

Co-authored-by: kennytm <kennytm@gmail.com>

* fix test

* add flag to control check

* address comment

* fix ci

* remove DS_Store

Co-authored-by: luancheng <luancheng@pingcap.com>
Co-authored-by: kennytm <kennytm@gmail.com>

Makefile

@@ -25,6 +25,7 @@ build:
 build_for_integration_test: failpoint-enable
 	(GO111MODULE=on go test -c -cover -covermode=count \
 		-coverpkg=$(BR_PKG)/... \
+		-ldflags '$(LDFLAGS)'\
 		-o bin/br.test && \
 	GO111MODULE=on go build ${RACEFLAG} -o bin/locker tests/br_key_locked/*.go && \
 	GO111MODULE=on go build ${RACEFLAG} -o bin/gc tests/br_z_gc_safepoint/*.go && \

go.mod

@@ -6,6 +6,7 @@ require (
 	cloud.google.com/go/storage v1.5.0
 	github.com/aws/aws-sdk-go v1.30.24
 	github.com/cheggaaa/pb/v3 v3.0.4
+	github.com/coreos/go-semver v0.3.0
 	github.com/fsouza/fake-gcs-server v1.17.0
 	github.com/go-sql-driver/mysql v1.5.0
 	github.com/gogo/protobuf v1.3.1

pkg/conn/conn.go

@@ -156,6 +156,7 @@ func NewMgr(
 	tlsConf *tls.Config,
 	securityOption pd.SecurityOption,
 	storeBehavior StoreBehavior,
+	checkRequirements bool,
 ) (*Mgr, error) {
 	addrs := strings.Split(pdAddrs, ",")
 
@@ -178,7 +179,6 @@ func NewMgr(
 		}
 		processedAddrs = append(processedAddrs, addr)
 		_, failure = pdRequest(ctx, addr, clusterVersionPrefix, cli, http.MethodGet, nil)
-		// TODO need check cluster version >= 3.1 when br release
 		if failure == nil {
 			break
 		}
@@ -197,6 +197,12 @@ func NewMgr(
 		log.Error("fail to create pd client", zap.Error(err))
 		return nil, err
 	}
+	if checkRequirements {
+		err = utils.CheckClusterVersion(ctx, pdClient)
+		if err != nil {
+			return nil, err
+		}
+	}
 	log.Info("new mgr", zap.String("pdAddrs", pdAddrs))
 
 	// Check live tikv.

pkg/task/backup.go

@@ -113,7 +113,7 @@ func RunBackup(c context.Context, g glue.Glue, cmdName string, cfg *BackupConfig
 	if err != nil {
 		return err
 	}
-	mgr, err := newMgr(ctx, g, cfg.PD, cfg.TLS, conn.SkipTiFlash)
+	mgr, err := newMgr(ctx, g, cfg.PD, cfg.TLS, conn.SkipTiFlash, cfg.CheckRequirements)
 	if err != nil {
 		return err
 	}

pkg/task/backup_raw.go

@@ -91,7 +91,7 @@ func RunBackupRaw(c context.Context, g glue.Glue, cmdName string, cfg *RawKvConf
 	if err != nil {
 		return err
 	}
-	mgr, err := newMgr(ctx, g, cfg.PD, cfg.TLS, conn.SkipTiFlash)
+	mgr, err := newMgr(ctx, g, cfg.PD, cfg.TLS, conn.SkipTiFlash, cfg.CheckRequirements)
 	if err != nil {
 		return err
 	}

pkg/task/common.go

@@ -45,10 +45,11 @@ const (
 	flagDatabase = "db"
 	flagTable    = "table"
 
-	flagRateLimit     = "ratelimit"
-	flagRateLimitUnit = "ratelimit-unit"
-	flagConcurrency   = "concurrency"
-	flagChecksum      = "checksum"
+	flagRateLimit        = "ratelimit"
+	flagRateLimitUnit    = "ratelimit-unit"
+	flagConcurrency      = "concurrency"
+	flagChecksum         = "checksum"
+	flagCheckRequirement = "check-requirements"
 )
 
 // TLSConfig is the common configuration for TLS connection.
@@ -91,8 +92,9 @@ type Config struct {
 	// LogProgress is true means the progress bar is printed to the log instead of stdout.
 	LogProgress bool `json:"log-progress" toml:"log-progress"`
 
-	CaseSensitive bool         `json:"case-sensitive" toml:"case-sensitive"`
-	Filter        filter.Rules `json:"black-white-list" toml:"black-white-list"`
+	CaseSensitive     bool         `json:"case-sensitive" toml:"case-sensitive"`
+	CheckRequirements bool         `json:"check-requirements" toml:"check-requirements"`
+	Filter            filter.Rules `json:"black-white-list" toml:"black-white-list"`
 }
 
 // DefineCommonFlags defines the flags common to all BRIE commands.
@@ -116,6 +118,9 @@ func DefineCommonFlags(flags *pflag.FlagSet) {
 	flags.Uint64(flagRateLimitUnit, utils.MB, "The unit of rate limit")
 	_ = flags.MarkHidden(flagRateLimitUnit)
 
+	flags.Bool(flagCheckRequirement, true,
+		"Whether start version check before execute command")
+
 	storage.DefineFlags(flags)
 }
 
@@ -203,6 +208,11 @@ func (cfg *Config) ParseFromFlags(flags *pflag.FlagSet) error {
 			cfg.Filter.DoDBs = []string{db}
 		}
 	}
+	checkRequirements, err := flags.GetBool(flagCheckRequirement)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	cfg.CheckRequirements = checkRequirements
 
 	if err := cfg.BackendOptions.ParseFromFlags(flags); err != nil {
 		return err
@@ -217,6 +227,7 @@ func newMgr(
 	pds []string,
 	tlsConfig TLSConfig,
 	storeBehavior conn.StoreBehavior,
+	checkRequirements bool,
 ) (*conn.Mgr, error) {
 	var (
 		tlsConf *tls.Config
@@ -243,7 +254,7 @@ func newMgr(
 	if err != nil {
 		return nil, err
 	}
-	return conn.NewMgr(ctx, g, pdAddress, store.(tikv.Storage), tlsConf, securityOption, storeBehavior)
+	return conn.NewMgr(ctx, g, pdAddress, store.(tikv.Storage), tlsConf, securityOption, storeBehavior, checkRequirements)
 }
 
 // GetStorage gets the storage backend from the config.

pkg/task/restore.go

@@ -99,7 +99,7 @@ func RunRestore(c context.Context, g glue.Glue, cmdName string, cfg *RestoreConf
 	ctx, cancel := context.WithCancel(c)
 	defer cancel()
 
-	mgr, err := newMgr(ctx, g, cfg.PD, cfg.TLS, conn.SkipTiFlash)
+	mgr, err := newMgr(ctx, g, cfg.PD, cfg.TLS, conn.SkipTiFlash, cfg.CheckRequirements)
 	if err != nil {
 		return err
 	}
@@ -533,7 +533,7 @@ func RunRestoreTiflashReplica(c context.Context, g glue.Glue, cmdName string, cf
 	ctx, cancel := context.WithCancel(c)
 	defer cancel()
 
-	mgr, err := newMgr(ctx, g, cfg.PD, cfg.TLS, conn.SkipTiFlash)
+	mgr, err := newMgr(ctx, g, cfg.PD, cfg.TLS, conn.SkipTiFlash, cfg.CheckRequirements)
 	if err != nil {
 		return err
 	}

pkg/task/restore_raw.go

@@ -51,7 +51,7 @@ func RunRestoreRaw(c context.Context, g glue.Glue, cmdName string, cfg *RestoreR
 	ctx, cancel := context.WithCancel(c)
 	defer cancel()
 
-	mgr, err := newMgr(ctx, g, cfg.PD, cfg.TLS, conn.ErrorOnTiFlash)
+	mgr, err := newMgr(ctx, g, cfg.PD, cfg.TLS, conn.ErrorOnTiFlash, cfg.CheckRequirements)
 	if err != nil {
 		return err
 	}

pkg/utils/version.go

@@ -4,10 +4,15 @@ package utils
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"runtime"
+	"strings"
 
+	"github.com/coreos/go-semver/semver"
+	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
+	pd "github.com/pingcap/pd/v4/client"
 	"github.com/pingcap/tidb/util/israce"
 	"go.uber.org/zap"
 )
@@ -43,3 +48,62 @@ func BRInfo() string {
 	fmt.Fprintf(&buf, "Race Enabled: %t", israce.RaceEnabled)
 	return buf.String()
 }
+
+var minTiKVVersion *semver.Version = semver.New("3.1.0-beta.2")
+var incompatibleTiKVMajor3 *semver.Version = semver.New("3.1.0")
+var incompatibleTiKVMajor4 *semver.Version = semver.New("4.0.0-rc.1")
+
+func removeV(v string) string {
+	return strings.TrimPrefix(v, "v")
+}
+
+// CheckClusterVersion check TiKV version.
+func CheckClusterVersion(ctx context.Context, client pd.Client) error {
+	BRVersion, err := semver.NewVersion(removeV(BRReleaseVersion))
+	if err != nil {
+		return err
+	}
+	stores, err := client.GetAllStores(ctx, pd.WithExcludeTombstone())
+	if err != nil {
+		return err
+	}
+	for _, s := range stores {
+		tikvVersion, err := semver.NewVersion(removeV(s.Version))
+		if err != nil {
+			return err
+		}
+
+		if tikvVersion.Compare(*minTiKVVersion) < 0 {
+			return errors.Errorf("TiKV node %s version %s don't support BR, please upgrade cluster to %s",
+				s.Address, removeV(s.Version), BRReleaseVersion)
+		}
+
+		if tikvVersion.Major != BRVersion.Major {
+			return errors.Errorf("TiKV node %s version %s and BR %s major version mismatch, please use the same version of BR",
+				s.Address, removeV(s.Version), BRReleaseVersion)
+		}
+
+		// BR(https://github.com/pingcap/br/pull/233) and TiKV(https://github.com/tikv/tikv/pull/7241) have breaking changes
+		// if BR include #233 and TiKV not include #7241, BR will panic TiKV during restore
+		// These incompatible version is 3.1.0 and 4.0.0-rc.1
+		if tikvVersion.Major == 3 {
+			if tikvVersion.Compare(*incompatibleTiKVMajor3) < 0 && BRVersion.Compare(*incompatibleTiKVMajor3) >= 0 {
+				return errors.Errorf("TiKV node %s version %s and BR %s version mismatch, please use the same version of BR",
+					s.Address, removeV(s.Version), BRReleaseVersion)
+			}
+		}
+
+		if tikvVersion.Major == 4 {
+			if tikvVersion.Compare(*incompatibleTiKVMajor4) < 0 && BRVersion.Compare(*incompatibleTiKVMajor4) >= 0 {
+				return errors.Errorf("TiKV node %s version %s and BR %s version mismatch, please use the same version of BR",
+					s.Address, removeV(s.Version), BRReleaseVersion)
+			}
+		}
+
+		if tikvVersion.Compare(*BRVersion) > 0 {
+			log.Warn(fmt.Sprintf("BR version is too old, please consider use version %s of BR", removeV(s.Version)))
+			break
+		}
+	}
+	return nil
+}

pkg/utils/version_test.go

@@ -0,0 +1,108 @@
+package utils
+
+import (
+	"context"
+
+	"github.com/coreos/go-semver/semver"
+	"github.com/pingcap/check"
+	"github.com/pingcap/kvproto/pkg/metapb"
+	pd "github.com/pingcap/pd/v4/client"
+)
+
+type versionSuite struct{}
+
+var _ = check.Suite(&versionSuite{})
+
+type mockPDClient struct {
+	pd.Client
+	getAllStores func() []*metapb.Store
+}
+
+func (m *mockPDClient) GetAllStores(ctx context.Context, opts ...pd.GetStoreOption) ([]*metapb.Store, error) {
+	if m.getAllStores != nil {
+		return m.getAllStores(), nil
+	}
+	return []*metapb.Store{}, nil
+}
+
+func (s *versionSuite) TestCheckClusterVersion(c *check.C) {
+	mock := mockPDClient{
+		Client: nil,
+	}
+
+	{
+		BRReleaseVersion = "v3.1.0-beta.2"
+		mock.getAllStores = func() []*metapb.Store {
+			return []*metapb.Store{{Version: minTiKVVersion.String()}}
+		}
+		err := CheckClusterVersion(context.Background(), &mock)
+		c.Assert(err, check.IsNil)
+	}
+
+	{
+		BRReleaseVersion = "v3.1.0-beta.2"
+		mock.getAllStores = func() []*metapb.Store {
+			// TiKV is too lower to support BR
+			return []*metapb.Store{{Version: `v2.1.0`}}
+		}
+		err := CheckClusterVersion(context.Background(), &mock)
+		c.Assert(err, check.ErrorMatches, "TiKV .* don't support BR, please upgrade cluster .*")
+	}
+
+	{
+		BRReleaseVersion = "v3.1.0"
+		mock.getAllStores = func() []*metapb.Store {
+			// TiKV v3.1.0-beta.2 is incompatible with BR v3.1.0
+			return []*metapb.Store{{Version: minTiKVVersion.String()}}
+		}
+		err := CheckClusterVersion(context.Background(), &mock)
+		c.Assert(err, check.ErrorMatches, "TiKV .* mismatch, please .*")
+	}
+
+	{
+		BRReleaseVersion = "v3.1.0"
+		mock.getAllStores = func() []*metapb.Store {
+			// TiKV v4.0.0-rc major version mismatch with BR v3.1.0
+			return []*metapb.Store{{Version: "v4.0.0-rc"}}
+		}
+		err := CheckClusterVersion(context.Background(), &mock)
+		c.Assert(err, check.ErrorMatches, "TiKV .* major version mismatch, please .*")
+	}
+
+	{
+		BRReleaseVersion = "v4.0.0-rc.2"
+		mock.getAllStores = func() []*metapb.Store {
+			// TiKV v4.0.0-rc.2 is incompatible with BR v4.0.0-beta.1
+			return []*metapb.Store{{Version: "v4.0.0-beta.1"}}
+		}
+		err := CheckClusterVersion(context.Background(), &mock)
+		c.Assert(err, check.ErrorMatches, "TiKV .* mismatch, please .*")
+	}
+
+	{
+		BRReleaseVersion = "v4.0.0-rc.2"
+		mock.getAllStores = func() []*metapb.Store {
+			// TiKV v4.0.0-rc.1 with BR v4.0.0-rc.2 is ok
+			return []*metapb.Store{{Version: "v4.0.0-rc.1"}}
+		}
+		err := CheckClusterVersion(context.Background(), &mock)
+		c.Assert(err, check.IsNil)
+	}
+
+	{
+		BRReleaseVersion = "v4.0.0-rc.1"
+		mock.getAllStores = func() []*metapb.Store {
+			// TiKV v4.0.0-rc.2 with BR v4.0.0-rc.1 is ok
+			return []*metapb.Store{{Version: "v4.0.0-rc.2"}}
+		}
+		err := CheckClusterVersion(context.Background(), &mock)
+		c.Assert(err, check.IsNil)
+	}
+}
+
+func (s *versionSuite) TestCompareVersion(c *check.C) {
+	c.Assert(semver.New("4.0.0-rc").Compare(*semver.New("4.0.0-rc.2")), check.Equals, -1)
+	c.Assert(semver.New("4.0.0-beta.3").Compare(*semver.New("4.0.0-rc.2")), check.Equals, -1)
+	c.Assert(semver.New("4.0.0-rc.1").Compare(*semver.New("4.0.0")), check.Equals, -1)
+	c.Assert(semver.New("4.0.0-beta.1").Compare(*semver.New("4.0.0")), check.Equals, -1)
+}