Merge master to Release 4.0

README.md

@@ -56,7 +56,7 @@ go-ycsb load mysql -p workload=core \
 mysql -uroot -htidb -P4000 -E -e "SELECT COUNT(*) FROM test.usertable"
 
 # Build BR and backup!
-make release && \
+make build && \
 bin/br backup full --pd pd0:2379 --storage "local:///data/backup/full" \
     --log-file "/logs/br_backup.log"
 
@@ -69,6 +69,20 @@ bin/br restore full --pd pd0:2379 --storage "local:///data/backup/full" \
 
 # How many rows do we get again? Expected to be 100000 rows.
 mysql -uroot -htidb -P4000 -E -e "SELECT COUNT(*) FROM test.usertable"
+
+# Test S3 compatible storage (MinIO).
+# Create a bucket to save backup by mc (a MinIO Client).
+mc config host add minio $S3_ENDPOINT $MINIO_ACCESS_KEY $MINIO_SECRET_KEY && \
+mc mb minio/mybucket
+
+# Backup to S3 compatible storage.
+bin/br backup full --pd pd0:2379 --storage "s3://mybucket/full" \
+    --s3.endpoint="$S3_ENDPOINT"
+
+# Drop database and restore!
+mysql -uroot -htidb -P4000 -E -e "DROP DATABASE test; SHOW DATABASES;" && \
+bin/br restore full --pd pd0:2379 --storage "s3://mybucket/full" \
+    --s3.endpoint="$S3_ENDPOINT"
 ```
 
 ## Contributing

cmd/cmd.go

@@ -6,8 +6,11 @@ import (
 	"context"
 	"net/http"
 	"net/http/pprof"
+	"os"
+	"path/filepath"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/util/logutil"
@@ -16,15 +19,17 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/pingcap/br/pkg/gluetidb"
+	"github.com/pingcap/br/pkg/summary"
 	"github.com/pingcap/br/pkg/task"
 	"github.com/pingcap/br/pkg/utils"
 )
 
 var (
-	initOnce       = sync.Once{}
-	defaultContext context.Context
-	hasLogFile     uint64
-	tidbGlue       = gluetidb.Glue{}
+	initOnce        = sync.Once{}
+	defaultContext  context.Context
+	hasLogFile      uint64
+	tidbGlue        = gluetidb.Glue{}
+	envLogToTermKey = "BR_LOG_TO_TERM"
 )
 
 const (
@@ -41,6 +46,10 @@ const (
 	flagVersionShort = "V"
 )
 
+func timestampLogFileName() string {
+	return filepath.Join(os.TempDir(), "br.log."+time.Now().Format(time.RFC3339))
+}
+
 // AddFlags adds flags to the given cmd.
 func AddFlags(cmd *cobra.Command) {
 	cmd.Version = utils.BRInfo()
@@ -49,8 +58,8 @@ func AddFlags(cmd *cobra.Command) {
 
 	cmd.PersistentFlags().StringP(FlagLogLevel, "L", "info",
 		"Set the log level")
-	cmd.PersistentFlags().String(FlagLogFile, "",
-		"Set the log file path. If not set, logs will output to stdout")
+	cmd.PersistentFlags().String(FlagLogFile, timestampLogFileName(),
+		"Set the log file path. If not set, logs will output to temp file")
 	cmd.PersistentFlags().String(FlagStatusAddr, "",
 		"Set the HTTP listening address for the status report service. Set to empty string to disable")
 	task.DefineCommonFlags(cmd.PersistentFlags())
@@ -73,8 +82,15 @@ func Init(cmd *cobra.Command) (err error) {
 		if err != nil {
 			return
 		}
+		_, outputLogToTerm := os.LookupEnv(envLogToTermKey)
+		if outputLogToTerm {
+			// Log to term if env `BR_LOG_TO_TERM` is set.
+			conf.File.Filename = ""
+		}
 		if len(conf.File.Filename) != 0 {
 			atomic.StoreUint64(&hasLogFile, 1)
+			summary.InitCollector(true)
+			cmd.Printf("Detial BR log in %s\n", conf.File.Filename)
 		}
 		lg, p, e := log.InitLogger(conf)
 		if e != nil {

docker-compose.yaml

@@ -1,6 +1,6 @@
 ---
 # Source: tidb-docker-compose/templates/docker-compose.yml
-version: '2.1'
+version: '3.2'
 
 services:
   control:
@@ -10,11 +10,13 @@ services:
       dockerfile: ./docker/Dockerfile
     volumes:
       - ./docker/data:/data
-      - ./docker/logs:/logs
+      - ./docker/logs:/tmp
     command: -c "/usr/bin/tail -f /dev/null"
     depends_on:
       - "tidb"
     restart: on-failure
+    env_file:
+      - ./docker/minio.env
 
   pd0:
     image: pingcap/pd:latest
@@ -64,6 +66,8 @@ services:
     #     soft: 1000000
     #     hard: 1000000
     restart: on-failure
+    env_file:
+      - ./docker/minio.env
 
   tikv1:
     image: pingcap/tikv:latest
@@ -87,6 +91,8 @@ services:
     #     soft: 1000000
     #     hard: 1000000
     restart: on-failure
+    env_file:
+      - ./docker/minio.env
 
   tikv2:
     image: pingcap/tikv:latest
@@ -110,6 +116,8 @@ services:
     #     soft: 1000000
     #     hard: 1000000
     restart: on-failure
+    env_file:
+      - ./docker/minio.env
 
   tikv3:
     image: pingcap/tikv:latest
@@ -133,6 +141,8 @@ services:
     #     soft: 1000000
     #     hard: 1000000
     restart: on-failure
+    env_file:
+      - ./docker/minio.env
 
   tikv4:
     image: pingcap/tikv:latest
@@ -156,6 +166,8 @@ services:
     #     soft: 1000000
     #     hard: 1000000
     restart: on-failure
+    env_file:
+      - ./docker/minio.env
 
   tidb:
     image: pingcap/tidb:latest
@@ -185,6 +197,16 @@ services:
     #     hard: 1000000
     restart: on-failure
 
+  minio:
+    image: minio/minio
+    ports:
+      - 24927:24927
+    volumes:
+      - ./docker/data/s3:/data/s3
+    command: server --address=:24927 /data/s3
+    env_file:
+      - ./docker/minio.env
+
   tidb-vision:
     image: pingcap/tidb-vision:latest
     environment:

docker/Dockerfile

@@ -1,11 +1,13 @@
-FROM golang:1.13.8-buster as builder
-
 # For loading data to TiDB
+FROM golang:1.13.8-buster as go-ycsb-builder
 WORKDIR /go/src/github.com/pingcap/
 RUN git clone https://github.com/pingcap/go-ycsb.git && \
     cd go-ycsb && \
     make
 
+# For operating minio S3 compatible storage
+FROM minio/mc as mc-builder
+
 FROM golang:1.13.8-buster
 
 RUN apt-get update && apt-get install -y --no-install-recommends \
@@ -19,6 +21,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
 WORKDIR /go/src/github.com/pingcap/br
 COPY . .
 
-COPY --from=builder /go/src/github.com/pingcap/go-ycsb/bin/go-ycsb /go/bin/go-ycsb
+COPY --from=go-ycsb-builder /go/src/github.com/pingcap/go-ycsb/bin/go-ycsb /go/bin/go-ycsb
+COPY --from=mc-builder /usr/bin/mc /usr/bin/mc
 
 ENTRYPOINT ["/bin/bash"]

docker/minio.env

@@ -0,0 +1,6 @@
+MINIO_ACCESS_KEY=brs3accesskey
+MINIO_SECRET_KEY=brs3secretkey
+MINIO_BROWSER=off
+AWS_ACCESS_KEY_ID=brs3accesskey
+AWS_SECRET_ACCESS_KEY=brs3secretkey
+S3_ENDPOINT=http://minio:24927

go.sum

@@ -363,7 +363,6 @@ github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989/go.mod h1:O17Xtb
 github.com/pingcap/kvproto v0.0.0-20191211054548-3c6b38ea5107/go.mod h1:WWLmULLO7l8IOcQG+t+ItJ3fEcrL5FxF0Wu+HrMy26w=
 github.com/pingcap/kvproto v0.0.0-20200214064158-62d31900d88e/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
 github.com/pingcap/kvproto v0.0.0-20200221034943-a2aa1d1e20a8/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
-github.com/pingcap/kvproto v0.0.0-20200228095611-2cf9a243b8d5 h1:knEvP4R5v5b2T107/Q6VzB0C8/6T7NXB/V7Vl1FtQsg=
 github.com/pingcap/kvproto v0.0.0-20200228095611-2cf9a243b8d5/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
 github.com/pingcap/kvproto v0.0.0-20200317112120-78042b285b75 h1:DB3NTM0ilba/6sW+vccdEnP10bVvrVunDwWvRa0hSKc=
 github.com/pingcap/kvproto v0.0.0-20200317112120-78042b285b75/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=

pkg/backup/client.go

@@ -32,6 +32,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/pingcap/br/pkg/conn"
+	"github.com/pingcap/br/pkg/glue"
 	"github.com/pingcap/br/pkg/rtree"
 	"github.com/pingcap/br/pkg/storage"
 	"github.com/pingcap/br/pkg/summary"
@@ -309,7 +310,7 @@ func (bc *Client) BackupRanges(
 	ctx context.Context,
 	ranges []rtree.Range,
 	req kvproto.BackupRequest,
-	updateCh chan<- struct{},
+	updateCh glue.Progress,
 ) error {
 	start := time.Now()
 	defer func() {
@@ -374,7 +375,7 @@ func (bc *Client) BackupRange(
 	ctx context.Context,
 	startKey, endKey []byte,
 	req kvproto.BackupRequest,
-	updateCh chan<- struct{},
+	updateCh glue.Progress,
 ) (err error) {
 	start := time.Now()
 	defer func() {
@@ -486,7 +487,7 @@ func (bc *Client) fineGrainedBackup(
 	rateLimit uint64,
 	concurrency uint32,
 	rangeTree rtree.RangeTree,
-	updateCh chan<- struct{},
+	updateCh glue.Progress,
 ) error {
 	bo := tikv.NewBackoffer(ctx, backupFineGrainedMaxBackoff)
 	for {
@@ -561,7 +562,7 @@ func (bc *Client) fineGrainedBackup(
 				rangeTree.Put(resp.StartKey, resp.EndKey, resp.Files)
 
 				// Update progress
-				updateCh <- struct{}{}
+				updateCh.Inc()
 			}
 		}
 

pkg/backup/push.go

@@ -12,6 +12,7 @@ import (
 	"github.com/pingcap/log"
 	"go.uber.org/zap"
 
+	"github.com/pingcap/br/pkg/glue"
 	"github.com/pingcap/br/pkg/rtree"
 )
 
@@ -38,7 +39,7 @@ func newPushDown(ctx context.Context, mgr ClientMgr, cap int) *pushDown {
 func (push *pushDown) pushBackup(
 	req backup.BackupRequest,
 	stores []*metapb.Store,
-	updateCh chan<- struct{},
+	updateCh glue.Progress,
 ) (rtree.RangeTree, error) {
 	// Push down backup tasks to all tikv instances.
 	res := rtree.NewRangeTree()
@@ -90,7 +91,7 @@ func (push *pushDown) pushBackup(
 					resp.GetStartKey(), resp.GetEndKey(), resp.GetFiles())
 
 				// Update progress
-				updateCh <- struct{}{}
+				updateCh.Inc()
 			} else {
 				errPb := resp.GetError()
 				switch v := errPb.Detail.(type) {

pkg/backup/schema.go

@@ -18,6 +18,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/pingcap/br/pkg/checksum"
+	"github.com/pingcap/br/pkg/glue"
 	"github.com/pingcap/br/pkg/summary"
 	"github.com/pingcap/br/pkg/utils"
 )
@@ -67,7 +68,7 @@ func (pending *Schemas) Start(
 	store kv.Storage,
 	backupTS uint64,
 	concurrency uint,
-	updateCh chan<- struct{},
+	updateCh glue.Progress,
 ) {
 	workerPool := utils.NewWorkerPool(concurrency, "Schemas")
 	go func() {
@@ -82,7 +83,7 @@ func (pending *Schemas) Start(
 
 				if pending.skipChecksum {
 					pending.backupSchemaCh <- schema
-					updateCh <- struct{}{}
+					updateCh.Inc()
 					return
 				}
 
@@ -110,7 +111,7 @@ func (pending *Schemas) Start(
 					zap.Duration("take", time.Since(start)))
 				pending.backupSchemaCh <- schema
 
-				updateCh <- struct{}{}
+				updateCh.Inc()
 			})
 		}
 		pending.wg.Wait()

pkg/backup/schema_test.go

@@ -5,6 +5,7 @@ package backup
 import (
 	"context"
 	"math"
+	"sync/atomic"
 
 	. "github.com/pingcap/check"
 	"github.com/pingcap/tidb-tools/pkg/filter"
@@ -30,6 +31,24 @@ func (s *testBackupSchemaSuite) TearDownSuite(c *C) {
 	testleak.AfterTest(c)()
 }
 
+type simpleProgress struct {
+	counter int64
+}
+
+func (sp *simpleProgress) Inc() {
+	atomic.AddInt64(&sp.counter, 1)
+}
+
+func (sp *simpleProgress) Close() {}
+
+func (sp *simpleProgress) reset() {
+	atomic.StoreInt64(&sp.counter, 0)
+}
+
+func (sp *simpleProgress) get() int64 {
+	return atomic.LoadInt64(&sp.counter)
+}
+
 func (s *testBackupSchemaSuite) TestBuildBackupRangeAndSchema(c *C) {
 	c.Assert(s.mock.Start(), IsNil)
 	defer s.mock.Stop()
@@ -73,10 +92,10 @@ func (s *testBackupSchemaSuite) TestBuildBackupRangeAndSchema(c *C) {
 		s.mock.Domain, s.mock.Storage, testFilter, math.MaxUint64)
 	c.Assert(err, IsNil)
 	c.Assert(backupSchemas.Len(), Equals, 1)
-	updateCh := make(chan struct{}, 2)
+	updateCh := new(simpleProgress)
 	backupSchemas.Start(context.Background(), s.mock.Storage, math.MaxUint64, 1, updateCh)
 	schemas, err := backupSchemas.finishTableChecksum()
-	<-updateCh
+	c.Assert(updateCh.get(), Equals, int64(1))
 	c.Assert(err, IsNil)
 	c.Assert(len(schemas), Equals, 1)
 	// Cluster returns a dummy checksum (all fields are 1).
@@ -93,10 +112,10 @@ func (s *testBackupSchemaSuite) TestBuildBackupRangeAndSchema(c *C) {
 		s.mock.Domain, s.mock.Storage, noFilter, math.MaxUint64)
 	c.Assert(err, IsNil)
 	c.Assert(backupSchemas.Len(), Equals, 2)
+	updateCh.reset()
 	backupSchemas.Start(context.Background(), s.mock.Storage, math.MaxUint64, 2, updateCh)
 	schemas, err = backupSchemas.finishTableChecksum()
-	<-updateCh
-	<-updateCh
+	c.Assert(updateCh.get(), Equals, int64(2))
 	c.Assert(err, IsNil)
 	c.Assert(len(schemas), Equals, 2)
 	// Cluster returns a dummy checksum (all fields are 1).