br/restore: add version check for backup schema_version (pingcap#929)

pkg/conn/conn.go

@@ -197,7 +197,7 @@ func NewMgr(
 		return nil, errors.Trace(err)
 	}
 	if checkRequirements {
-		err = version.CheckClusterVersion(ctx, controller.GetPDClient())
+		err = version.CheckClusterVersion(ctx, controller.GetPDClient(), version.CheckVersionForBR)
 		if err != nil {
 			return nil, errors.Annotate(err, "running BR in incompatible version of cluster, "+
 				"if you believe it's OK, use --check-requirements=false to skip.")

pkg/task/restore.go

@@ -6,6 +6,8 @@ import (
 	"context"
 	"time"
 
+	"github.com/pingcap/br/pkg/version"
+
 	"github.com/opentracing/opentracing-go"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
@@ -209,6 +211,12 @@ func RunRestore(c context.Context, g glue.Glue, cmdName string, cfg *RestoreConf
 		return errors.Trace(err)
 	}
 	g.Record("Size", utils.ArchiveSize(backupMeta))
+	backupVersion := version.NormalizeBackupVersion(backupMeta.ClusterVersion)
+	if cfg.CheckRequirements && backupVersion != nil {
+		if versionErr := version.CheckClusterVersion(ctx, mgr.GetPDClient(), version.CheckVersionForBackup(backupVersion)); versionErr != nil {
+			return errors.Trace(versionErr)
+		}
+	}
 
 	if err = client.InitBackupMeta(backupMeta, u); err != nil {
 		return errors.Trace(err)

pkg/version/version.go

@@ -6,12 +6,12 @@ import (
 	"context"
 	"fmt"
 	"regexp"
+	"strconv"
 	"strings"
 
-	"github.com/pingcap/kvproto/pkg/metapb"
-
 	"github.com/coreos/go-semver/semver"
 	"github.com/pingcap/errors"
+	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/log"
 	pd "github.com/tikv/pd/client"
 	"go.uber.org/zap"
@@ -74,12 +74,12 @@ func IsTiFlash(store *metapb.Store) bool {
 	return false
 }
 
+// VerChecker is a callback for the CheckClusterVersion, decides whether the cluster is suitable to execute restore.
+// See also: CheckVersionForBackup and CheckVersionForBR.
+type VerChecker func(store *metapb.Store, ver *semver.Version) error
+
 // CheckClusterVersion check TiKV version.
-func CheckClusterVersion(ctx context.Context, client pd.Client) error {
-	BRVersion, err := semver.NewVersion(removeVAndHash(build.ReleaseVersion))
-	if err != nil {
-		return errors.Annotatef(berrors.ErrVersionMismatch, "%s: invalid version, please recompile using `git fetch origin --tags && make build`", err)
-	}
+func CheckClusterVersion(ctx context.Context, client pd.Client, checker VerChecker) error {
 	stores, err := client.GetAllStores(ctx, pd.WithExcludeTombstone())
 	if err != nil {
 		return errors.Trace(err)
@@ -99,44 +99,67 @@ func CheckClusterVersion(ctx context.Context, client pd.Client) error {
 		}
 
 		tikvVersionString := removeVAndHash(s.Version)
-		tikvVersion, err := semver.NewVersion(tikvVersionString)
-		if err != nil {
-			return errors.Annotatef(berrors.ErrVersionMismatch, "%s: TiKV node %s version %s is invalid", err, s.Address, tikvVersionString)
+		tikvVersion, getVersionErr := semver.NewVersion(tikvVersionString)
+		if getVersionErr != nil {
+			return errors.Annotatef(berrors.ErrVersionMismatch, "%s: TiKV node %s version %s is invalid", getVersionErr, s.Address, tikvVersionString)
 		}
-
-		if tikvVersion.Compare(*minTiKVVersion) < 0 {
-			return errors.Annotatef(berrors.ErrVersionMismatch, "TiKV node %s version %s don't support BR, please upgrade cluster to %s",
-				s.Address, tikvVersionString, build.ReleaseVersion)
+		if checkerErr := checker(s, tikvVersion); checkerErr != nil {
+			return checkerErr
 		}
+	}
+	return nil
+}
 
-		if tikvVersion.Major != BRVersion.Major {
-			return errors.Annotatef(berrors.ErrVersionMismatch, "TiKV node %s version %s and BR %s major version mismatch, please use the same version of BR",
-				s.Address, tikvVersionString, build.ReleaseVersion)
+// CheckVersionForBackup checks the version for backup and
+func CheckVersionForBackup(backupVersion *semver.Version) VerChecker {
+	return func(store *metapb.Store, ver *semver.Version) error {
+		if backupVersion.Major > ver.Major {
+			return errors.Annotatef(berrors.ErrVersionMismatch,
+				"backup with cluster version %s cannot be restored at cluster of version %s: major version mismatches",
+				backupVersion, ver)
 		}
+		return nil
+	}
+}
 
-		// 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.Annotatef(berrors.ErrVersionMismatch, "TiKV node %s version %s and BR %s version mismatch, please use the same version of BR",
-					s.Address, tikvVersionString, build.ReleaseVersion)
-			}
-		}
+// CheckVersionForBR checks whether version of the cluster and BR itself is compatible.
+func CheckVersionForBR(s *metapb.Store, tikvVersion *semver.Version) error {
+	BRVersion, err := semver.NewVersion(removeVAndHash(build.ReleaseVersion))
+	if err != nil {
+		return errors.Annotatef(berrors.ErrVersionMismatch, "%s: invalid version, please recompile using `git fetch origin --tags && make build`", err)
+	}
 
-		if tikvVersion.Major == 4 {
-			if tikvVersion.Compare(*incompatibleTiKVMajor4) < 0 && BRVersion.Compare(*incompatibleTiKVMajor4) >= 0 {
-				return errors.Annotatef(berrors.ErrVersionMismatch, "TiKV node %s version %s and BR %s version mismatch, please use the same version of BR",
-					s.Address, tikvVersionString, build.ReleaseVersion)
-			}
+	if tikvVersion.Compare(*minTiKVVersion) < 0 {
+		return errors.Annotatef(berrors.ErrVersionMismatch, "TiKV node %s version %s don't support BR, please upgrade cluster to %s",
+			s.Address, tikvVersion, build.ReleaseVersion)
+	}
+
+	if tikvVersion.Major != BRVersion.Major {
+		return errors.Annotatef(berrors.ErrVersionMismatch, "TiKV node %s version %s and BR %s major version mismatch, please use the same version of BR",
+			s.Address, tikvVersion, build.ReleaseVersion)
+	}
+
+	// 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.Annotatef(berrors.ErrVersionMismatch, "TiKV node %s version %s and BR %s version mismatch, please use the same version of BR",
+				s.Address, tikvVersion, build.ReleaseVersion)
 		}
+	}
 
-		// don't warn if we are the master build, which always have the version v4.0.0-beta.2-*
-		if build.GitBranch != "master" && tikvVersion.Compare(*BRVersion) > 0 {
-			log.Warn(fmt.Sprintf("BR version is outdated, please consider use version %s of BR", tikvVersionString))
-			break
+	if tikvVersion.Major == 4 {
+		if tikvVersion.Compare(*incompatibleTiKVMajor4) < 0 && BRVersion.Compare(*incompatibleTiKVMajor4) >= 0 {
+			return errors.Annotatef(berrors.ErrVersionMismatch, "TiKV node %s version %s and BR %s version mismatch, please use the same version of BR",
+				s.Address, tikvVersion, build.ReleaseVersion)
 		}
 	}
+
+	// don't warn if we are the master build, which always have the version v4.0.0-beta.2-*
+	if build.GitBranch != "master" && tikvVersion.Compare(*BRVersion) > 0 {
+		log.Warn(fmt.Sprintf("BR version is outdated, please consider use version %s of BR", tikvVersion))
+	}
 	return nil
 }
 
@@ -199,3 +222,20 @@ func CheckTiDBVersion(versionStr string, requiredMinVersion, requiredMaxVersion
 	}
 	return CheckVersion("TiDB", *version, requiredMinVersion, requiredMaxVersion)
 }
+
+// NormalizeBackupVersion normalizes the version string from backupmeta.
+func NormalizeBackupVersion(version string) *semver.Version {
+	// We need to unquote here because we get the version from PD HTTP API,
+	// which returns quoted string.
+	trimmedVerStr := strings.TrimSpace(version)
+	unquotedVerStr, err := strconv.Unquote(trimmedVerStr)
+	if err != nil {
+		unquotedVerStr = trimmedVerStr
+	}
+	normalizedVerStr := strings.TrimSpace(unquotedVerStr)
+	ver, err := semver.NewVersion(normalizedVerStr)
+	if err != nil {
+		log.Warn("cannot parse backup version", zap.String("version", normalizedVerStr), zap.Error(err))
+	}
+	return ver
+}

pkg/version/version_test.go

@@ -4,6 +4,7 @@ package version
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/coreos/go-semver/semver"
@@ -50,7 +51,7 @@ func (s *checkSuite) TestCheckClusterVersion(c *C) {
 		mock.getAllStores = func() []*metapb.Store {
 			return tiflash("v4.0.0-rc.1")
 		}
-		err := CheckClusterVersion(context.Background(), &mock)
+		err := CheckClusterVersion(context.Background(), &mock, CheckVersionForBR)
 		c.Assert(err, ErrorMatches, `incompatible.*version v4.0.0-rc.1, try update it to 4.0.0.*`)
 	}
 
@@ -59,7 +60,7 @@ func (s *checkSuite) TestCheckClusterVersion(c *C) {
 		mock.getAllStores = func() []*metapb.Store {
 			return tiflash("v3.1.0-beta.1")
 		}
-		err := CheckClusterVersion(context.Background(), &mock)
+		err := CheckClusterVersion(context.Background(), &mock, CheckVersionForBR)
 		c.Assert(err, ErrorMatches, `incompatible.*version v3.1.0-beta.1, try update it to 3.1.0.*`)
 	}
 
@@ -68,7 +69,7 @@ func (s *checkSuite) TestCheckClusterVersion(c *C) {
 		mock.getAllStores = func() []*metapb.Store {
 			return tiflash("v3.0.15")
 		}
-		err := CheckClusterVersion(context.Background(), &mock)
+		err := CheckClusterVersion(context.Background(), &mock, CheckVersionForBR)
 		c.Assert(err, ErrorMatches, `incompatible.*version v3.0.15, try update it to 3.1.0.*`)
 	}
 
@@ -77,7 +78,7 @@ func (s *checkSuite) TestCheckClusterVersion(c *C) {
 		mock.getAllStores = func() []*metapb.Store {
 			return []*metapb.Store{{Version: minTiKVVersion.String()}}
 		}
-		err := CheckClusterVersion(context.Background(), &mock)
+		err := CheckClusterVersion(context.Background(), &mock, CheckVersionForBR)
 		c.Assert(err, IsNil)
 	}
 
@@ -87,7 +88,7 @@ func (s *checkSuite) TestCheckClusterVersion(c *C) {
 			// TiKV is too lower to support BR
 			return []*metapb.Store{{Version: `v2.1.0`}}
 		}
-		err := CheckClusterVersion(context.Background(), &mock)
+		err := CheckClusterVersion(context.Background(), &mock, CheckVersionForBR)
 		c.Assert(err, ErrorMatches, ".*TiKV .* don't support BR, please upgrade cluster .*")
 	}
 
@@ -97,7 +98,7 @@ func (s *checkSuite) TestCheckClusterVersion(c *C) {
 			// TiKV v3.1.0-beta.2 is incompatible with BR v3.1.0
 			return []*metapb.Store{{Version: minTiKVVersion.String()}}
 		}
-		err := CheckClusterVersion(context.Background(), &mock)
+		err := CheckClusterVersion(context.Background(), &mock, CheckVersionForBR)
 		c.Assert(err, ErrorMatches, "TiKV .* mismatch, please .*")
 	}
 
@@ -107,7 +108,7 @@ func (s *checkSuite) TestCheckClusterVersion(c *C) {
 			// 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)
+		err := CheckClusterVersion(context.Background(), &mock, CheckVersionForBR)
 		c.Assert(err, ErrorMatches, "TiKV .* major version mismatch, please .*")
 	}
 
@@ -117,7 +118,7 @@ func (s *checkSuite) TestCheckClusterVersion(c *C) {
 			// 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)
+		err := CheckClusterVersion(context.Background(), &mock, CheckVersionForBR)
 		c.Assert(err, ErrorMatches, "TiKV .* mismatch, please .*")
 	}
 
@@ -127,17 +128,35 @@ func (s *checkSuite) TestCheckClusterVersion(c *C) {
 			// 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)
+		err := CheckClusterVersion(context.Background(), &mock, CheckVersionForBR)
 		c.Assert(err, IsNil)
 	}
 
+	{
+		// Even across many patch versions, backup should be usable.
+		mock.getAllStores = func() []*metapb.Store {
+			return []*metapb.Store{{Version: "v4.0.0-rc.1"}}
+		}
+		err := CheckClusterVersion(context.Background(), &mock, CheckVersionForBackup(semver.New("4.0.12")))
+		c.Assert(err, IsNil)
+	}
+
+	{
+		// Restore across major version isn't allowed.
+		mock.getAllStores = func() []*metapb.Store {
+			return []*metapb.Store{{Version: "v4.0.0-rc.1"}}
+		}
+		err := CheckClusterVersion(context.Background(), &mock, CheckVersionForBackup(semver.New("5.0.0-rc")))
+		c.Assert(err, Not(IsNil))
+	}
+
 	{
 		build.ReleaseVersion = "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)
+		err := CheckClusterVersion(context.Background(), &mock, CheckVersionForBR)
 		c.Assert(err, IsNil)
 	}
 }
@@ -229,3 +248,50 @@ func (s *checkSuite) TestCheckVersion(c *C) {
 	err = CheckVersion("TiNB", *semver.New("3.0.0-beta"), *semver.New("2.3.5"), *semver.New("3.0.0"))
 	c.Assert(err, ErrorMatches, "TiNB version too new.*")
 }
+
+type versionEqualsC struct{}
+
+func (v versionEqualsC) Info() *CheckerInfo {
+	return &CheckerInfo{
+		Name:   "VersionEquals",
+		Params: []string{"source", "target"},
+	}
+}
+
+func (v versionEqualsC) Check(params []interface{}, names []string) (result bool, error string) {
+	source := params[0].(*semver.Version)
+	target := params[1].(*semver.Version)
+
+	if source == nil || target == nil {
+		if target == source {
+			return true, ""
+		}
+		return false, fmt.Sprintf("one of version is nil but another is not (%s and %s)", params[0], params[1])
+	}
+
+	if source.Equal(*target) {
+		return true, ""
+	}
+	return false, fmt.Sprintf("version not equal (%s vs %s)", source, target)
+}
+
+var versionEquals versionEqualsC
+
+func (s *checkSuite) TestNormalizeBackupVersion(c *C) {
+	cases := []struct {
+		target string
+		source string
+	}{
+		{"4.0.0", `"4.0.0\n"`},
+		{"5.0.0-rc.x", `"5.0.0-rc.x\n"`},
+		{"5.0.0-rc.x", `5.0.0-rc.x`},
+		{"4.0.12", `"4.0.12"` + "\n"},
+		{"<error-version>", ""},
+	}
+
+	for _, testCase := range cases {
+		target, _ := semver.NewVersion(testCase.target)
+		source := NormalizeBackupVersion(testCase.source)
+		c.Assert(source, versionEquals, target)
+	}
+}