Merge branch 'master' into fix-sequence-alter-version

.golangci.yml

@@ -3,7 +3,6 @@ run:
 linters:
   disable-all: true
   enable:
-    - misspell
     - ineffassign
     - typecheck
     - varcheck
@@ -20,7 +19,6 @@ linters:
     - bodyclose
     - exportloopref
     - rowserrcheck
-    - unconvert
     - makezero
     - durationcheck
     - prealloc

DEPS.bzl

@@ -173,8 +173,8 @@ def go_deps():
         name = "com_github_aws_aws_sdk_go",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/aws/aws-sdk-go",
-        sum = "h1:r0puXncSaAfRt7Btml2swUo74Kao+vKhO3VLjwDjK54=",
-        version = "v1.35.3",
+        sum = "h1:hAwyfe7eZa7sM+S5mIJZFiNFwJMia9Whz6CYblioLoU=",
+        version = "v1.36.30",
     )
     go_repository(
         name = "com_github_aymerick_raymond",
@@ -1354,15 +1354,15 @@ def go_deps():
         name = "com_github_google_uuid",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/google/uuid",
-        sum = "h1:EVhdT+1Kseyi1/pUmXKaFxYsDNy9RQYkMWRH68J/W7Y=",
-        version = "v1.1.2",
+        sum = "h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=",
+        version = "v1.3.0",
     )
     go_repository(
         name = "com_github_googleapis_gax_go_v2",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/googleapis/gax-go/v2",
-        sum = "h1:dp3bWCh+PPO1zjRRiCSczJav13sBvG4UhNyVTa1KqdU=",
-        version = "v2.1.1",
+        sum = "h1:s7jOdKSaksJVOxE0Y/S32otcfiP+UQ0cL8/GTKaONwE=",
+        version = "v2.2.0",
     )
     go_repository(
         name = "com_github_gopherjs_gopherjs",
@@ -2347,8 +2347,8 @@ def go_deps():
         name = "com_github_nxadm_tail",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/nxadm/tail",
-        sum = "h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=",
-        version = "v1.4.8",
+        sum = "h1:DQuhQpB1tVlglWS2hLQ5OV6B5r8aGxSrPc5Qo6uTN78=",
+        version = "v1.4.4",
     )
     go_repository(
         name = "com_github_oklog_ulid",
@@ -2531,8 +2531,8 @@ def go_deps():
         name = "com_github_pingcap_tipb",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/pingcap/tipb",
-        sum = "h1:hE1dQdnvxWCHhD0snX67paV9y6inq8TxVFbsKqjaTQk=",
-        version = "v0.0.0-20220714100504-7d3474676bc9",
+        sum = "h1:FBaTXU8C3xgt/drM58VHxojHo/QoG1oPsgWTGvaSpO4=",
+        version = "v0.0.0-20220718022156-3e2483c20a9e",
     )
     go_repository(
         name = "com_github_pkg_browser",
@@ -3012,8 +3012,8 @@ def go_deps():
         name = "com_github_tikv_client_go_v2",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/tikv/client-go/v2",
-        sum = "h1:nbcwXbkilywhMoAseLPzg/VHdFNhMEWy6JeqL/Gmq7A=",
-        version = "v2.0.1-0.20220713085647-57c12f7c64f6",
+        sum = "h1:OSxo1R2y6iyAAxbUPL7h1HC/17CNuwNkW2lF1JySJ7k=",
+        version = "v2.0.1-0.20220718080214-86d51ba7eb02",
     )
     go_repository(
         name = "com_github_tikv_pd_client",
@@ -3328,8 +3328,8 @@ def go_deps():
         name = "com_google_cloud_go_compute",
         build_file_proto_mode = "disable_global",
         importpath = "cloud.google.com/go/compute",
-        sum = "h1:EKki8sSdvDU0OO9mAXGwPXOTOgPz2l08R0/IutDH11I=",
-        version = "v1.2.0",
+        sum = "h1:b1zWmYuuHz7gO9kDcM/EpHGr06UgsYNRpNJzI2kFiLM=",
+        version = "v1.5.0",
     )
     go_repository(
         name = "com_google_cloud_go_datastore",
@@ -3722,8 +3722,8 @@ def go_deps():
         name = "org_golang_google_api",
         build_file_proto_mode = "disable_global",
         importpath = "google.golang.org/api",
-        sum = "h1:yHW5s2SFyDapr/43kYtIQmoaaFVW4baLMLwqV4auj2A=",
-        version = "v0.69.0",
+        sum = "h1:ExR2D+5TYIrMphWgs5JCgwRhEDlPDXXrLwHHMgPHTXE=",
+        version = "v0.74.0",
     )
     go_repository(
         name = "org_golang_google_appengine",
@@ -3736,15 +3736,15 @@ def go_deps():
         name = "org_golang_google_genproto",
         build_file_proto_mode = "disable_global",
         importpath = "google.golang.org/genproto",
-        sum = "h1:divpuJZKgX3Qt7MFDE5v62yu0yQcQbTCD9VJp9leX58=",
-        version = "v0.0.0-20220216160803-4663080d8bc8",
+        sum = "h1:0m9wktIpOxGw+SSKmydXWB3Z3GTfcPP6+q75HCQa6HI=",
+        version = "v0.0.0-20220324131243-acbaeb5b85eb",
     )
     go_repository(
         name = "org_golang_google_grpc",
         build_file_proto_mode = "disable_global",
         importpath = "google.golang.org/grpc",
-        sum = "h1:weqSxi/TMs1SqFRMHCtBgXRs8k3X39QIDEZ0pRcttUg=",
-        version = "v1.44.0",
+        sum = "h1:NEpgUqV3Z+ZjkqMsxMg11IaDrXY4RY6CQukSGK0uI1M=",
+        version = "v1.45.0",
     )
     go_repository(
         name = "org_golang_google_grpc_cmd_protoc_gen_go_grpc",
@@ -3764,8 +3764,8 @@ def go_deps():
         name = "org_golang_x_crypto",
         build_file_proto_mode = "disable_global",
         importpath = "golang.org/x/crypto",
-        sum = "h1:f+lwQ+GtmgoY+A2YaQxlSOnDjXcQ7ZRLWOHbC6HtRqE=",
-        version = "v0.0.0-20220214200702-86341886e292",
+        sum = "h1:7I4JAnoQBe7ZtJcBaYHi5UtiO8tQHbUSXxL+pnGRANg=",
+        version = "v0.0.0-20210921155107-089bfa567519",
     )
     go_repository(
         name = "org_golang_x_exp",
@@ -3814,15 +3814,15 @@ def go_deps():
         name = "org_golang_x_net",
         build_file_proto_mode = "disable_global",
         importpath = "golang.org/x/net",
-        sum = "h1:O7DYs+zxREGLKzKoMQrtrEacpb0ZVXA5rIwylE2Xchk=",
-        version = "v0.0.0-20220127200216-cd36cc0744dd",
+        sum = "h1:bRb386wvrE+oBNdF1d/Xh9mQrfQ4ecYhW5qJ5GvTGT4=",
+        version = "v0.0.0-20220412020605-290c469a71a5",
     )
     go_repository(
         name = "org_golang_x_oauth2",
         build_file_proto_mode = "disable_global",
         importpath = "golang.org/x/oauth2",
-        sum = "h1:RerP+noqYHUQ8CMRcPlC2nvTa4dcBIjegkuWdcUDuqg=",
-        version = "v0.0.0-20211104180415-d3ed0bb246c8",
+        sum = "h1:OSnWWcOd/CtWQC2cYSBgbTSJv3ciqd8r54ySIW2y3RE=",
+        version = "v0.0.0-20220411215720-9780585627b5",
     )
     go_repository(
         name = "org_golang_x_sync",

Makefile

@@ -31,7 +31,7 @@ dev: checklist check explaintest gogenerate br_unit_test test_part_parser_dev ut
 # Install the check tools.
 check-setup:tools/bin/revive tools/bin/goword
 
-check: fmt check-parallel lint tidy testSuite check-static vet errdoc
+check: check-parallel lint tidy testSuite check-static errdoc
 
 fmt:
 	@echo "gofmt (simplify)"

br/cmd/br/BUILD.bazel

@@ -13,6 +13,7 @@ go_library(
     importpath = "github.com/pingcap/tidb/br/cmd/br",
     visibility = ["//visibility:private"],
     deps = [
+        "//br/pkg/conn",
         "//br/pkg/errors",
         "//br/pkg/gluetidb",
         "//br/pkg/gluetikv",

br/cmd/br/debug.go

@@ -16,6 +16,7 @@ import (
 	backuppb "github.com/pingcap/kvproto/pkg/brpb"
 	"github.com/pingcap/kvproto/pkg/import_sstpb"
 	"github.com/pingcap/log"
+	"github.com/pingcap/tidb/br/pkg/conn"
 	berrors "github.com/pingcap/tidb/br/pkg/errors"
 	"github.com/pingcap/tidb/br/pkg/logutil"
 	"github.com/pingcap/tidb/br/pkg/metautil"
@@ -53,6 +54,7 @@ func NewDebugCommand() *cobra.Command {
 	meta.AddCommand(decodeBackupMetaCommand())
 	meta.AddCommand(encodeBackupMetaCommand())
 	meta.AddCommand(setPDConfigCommand())
+	meta.AddCommand(searchStreamBackupCommand())
 	meta.Hidden = true
 
 	return meta
@@ -383,7 +385,7 @@ func setPDConfigCommand() *cobra.Command {
 				return errors.Trace(err)
 			}
 
-			mgr, err := task.NewMgr(ctx, tidbGlue, cfg.PD, cfg.TLS, task.GetKeepalive(&cfg), cfg.CheckRequirements, false)
+			mgr, err := task.NewMgr(ctx, tidbGlue, cfg.PD, cfg.TLS, task.GetKeepalive(&cfg), cfg.CheckRequirements, false, conn.NormalVersionChecker)
 			if err != nil {
 				return errors.Trace(err)
 			}
@@ -398,3 +400,71 @@ func setPDConfigCommand() *cobra.Command {
 	}
 	return pdConfigCmd
 }
+
+func searchStreamBackupCommand() *cobra.Command {
+	searchBackupCMD := &cobra.Command{
+		Use:   "search-log-backup",
+		Short: "search log backup by key",
+		Args:  cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			ctx, cancel := context.WithCancel(GetDefaultContext())
+			defer cancel()
+
+			searchKey, err := cmd.Flags().GetString("search-key")
+			if err != nil {
+				return errors.Trace(err)
+			}
+			if searchKey == "" {
+				return errors.New("key param can't be empty")
+			}
+			keyBytes, err := hex.DecodeString(searchKey)
+			if err != nil {
+				return errors.Trace(err)
+			}
+
+			startTs, err := cmd.Flags().GetUint64("start-ts")
+			if err != nil {
+				return errors.Trace(err)
+			}
+			endTs, err := cmd.Flags().GetUint64("end-ts")
+			if err != nil {
+				return errors.Trace(err)
+			}
+
+			var cfg task.Config
+			if err = cfg.ParseFromFlags(cmd.Flags()); err != nil {
+				return errors.Trace(err)
+			}
+			_, s, err := task.GetStorage(ctx, cfg.Storage, &cfg)
+			if err != nil {
+				return errors.Trace(err)
+			}
+			comparator := restore.NewStartWithComparator()
+			bs := restore.NewStreamBackupSearch(s, comparator, keyBytes)
+			bs.SetStartTS(startTs)
+			bs.SetEndTs(endTs)
+
+			kvs, err := bs.Search(ctx)
+			if err != nil {
+				return errors.Trace(err)
+			}
+
+			kvsBytes, err := json.MarshalIndent(kvs, "", "  ")
+			if err != nil {
+				return errors.Trace(err)
+			}
+
+			cmd.Println("search result")
+			cmd.Println(string(kvsBytes))
+
+			return nil
+		},
+	}
+
+	flags := searchBackupCMD.Flags()
+	flags.String("search-key", "", "hex encoded key")
+	flags.Uint64("start-ts", 0, "search from start TSO, default is no start TSO limit")
+	flags.Uint64("end-ts", 0, "search to end TSO, default is no end TSO limit")
+
+	return searchBackupCMD
+}

br/pkg/conn/conn.go

@@ -47,6 +47,15 @@ const (
 	DefaultMergeRegionKeyCount uint64 = 960000
 )
 
+type VersionCheckerType int
+
+const (
+	// default version checker
+	NormalVersionChecker VersionCheckerType = iota
+	// version checker for PiTR
+	StreamVersionChecker
+)
+
 // Mgr manages connections to a TiDB cluster.
 type Mgr struct {
 	*pdutil.PdController
@@ -177,6 +186,7 @@ func NewMgr(
 	storeBehavior StoreBehavior,
 	checkRequirements bool,
 	needDomain bool,
+	versionCheckerType VersionCheckerType,
 ) (*Mgr, error) {
 	if span := opentracing.SpanFromContext(ctx); span != nil && span.Tracer() != nil {
 		span1 := span.Tracer().StartSpan("conn.NewMgr", opentracing.ChildOf(span.Context()))
@@ -192,7 +202,16 @@ func NewMgr(
 		return nil, errors.Trace(err)
 	}
 	if checkRequirements {
-		err = version.CheckClusterVersion(ctx, controller.GetPDClient(), version.CheckVersionForBR)
+		var checker version.VerChecker
+		switch versionCheckerType {
+		case NormalVersionChecker:
+			checker = version.CheckVersionForBR
+		case StreamVersionChecker:
+			checker = version.CheckVersionForBRPiTR
+		default:
+			return nil, errors.Errorf("unknown command type, comman code is %d", versionCheckerType)
+		}
+		err = version.CheckClusterVersion(ctx, controller.GetPDClient(), checker)
 		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.")

br/pkg/glue/BUILD.bazel

@@ -13,6 +13,7 @@ go_library(
         "//domain",
         "//kv",
         "//parser/model",
+        "//sessionctx",
         "@com_github_fatih_color//:color",
         "@com_github_pingcap_log//:log",
         "@com_github_tikv_pd_client//:client",

br/pkg/glue/console_glue.go

@@ -23,12 +23,40 @@ type ConsoleOperations struct {
 	ConsoleGlue
 }
 
-// StartTask prints a task start information, and mark as finished when the returned function called.
-func (ops ConsoleOperations) StartTask(message string) func() {
+// An extra field appending to the task.
+// return type is a {key: string, value: string} tuple.
+type ExtraField func() [2]string
+
+// NOTE:
+// Perhaps we'd better move these modifiers and terminal function to another package
+// like `glue/termutil?`
+
+// WithTimeCost adds the task information of time costing for `ShowTask`.
+func WithTimeCost() ExtraField {
 	start := time.Now()
+	return func() [2]string {
+		return [2]string{"take", time.Since(start).String()}
+	}
+}
+
+// WithConstExtraField adds an extra field with constant values.
+func WithConstExtraField(key string, value interface{}) ExtraField {
+	return func() [2]string {
+		return [2]string{key, fmt.Sprint(value)}
+	}
+}
+
+// ShowTask prints a task start information, and mark as finished when the returned function called.
+// This is for TUI presenting.
+func (ops ConsoleOperations) ShowTask(message string, extraFields ...ExtraField) func() {
 	ops.Print(message)
 	return func() {
-		ops.Printf("%s; take = %s\n", color.HiGreenString("DONE"), time.Since(start))
+		fields := make([]string, 0, len(extraFields))
+		for _, fieldFunc := range extraFields {
+			field := fieldFunc()
+			fields = append(fields, fmt.Sprintf("%s = %s", field[0], color.New(color.Bold).Sprint(field[1])))
+		}
+		ops.Printf("%s; %s\n", color.HiGreenString("DONE"), strings.Join(fields, ", "))
 	}
 }
 

br/pkg/logutil/logging.go

@@ -282,18 +282,27 @@ func (kr StringifyKeys) String() string {
 		if i > 0 {
 			sb.WriteString(", ")
 		}
-		sb.WriteString("[")
-		sb.WriteString(redact.Key(rng.StartKey))
-		sb.WriteString(", ")
-		var endKey string
-		if len(rng.EndKey) == 0 {
-			endKey = "inf"
-		} else {
-			endKey = redact.Key(rng.EndKey)
-		}
-		sb.WriteString(redact.String(endKey))
-		sb.WriteString(")")
+		sb.WriteString(StringifyRange(rng).String())
 	}
 	sb.WriteString("}")
 	return sb.String()
 }
+
+// StringifyRange is the wrapper for displaying a key range.
+type StringifyRange kv.KeyRange
+
+func (rng StringifyRange) String() string {
+	sb := new(strings.Builder)
+	sb.WriteString("[")
+	sb.WriteString(redact.Key(rng.StartKey))
+	sb.WriteString(", ")
+	var endKey string
+	if len(rng.EndKey) == 0 {
+		endKey = "inf"
+	} else {
+		endKey = redact.Key(rng.EndKey)
+	}
+	sb.WriteString(redact.String(endKey))
+	sb.WriteString(")")
+	return sb.String()
+}