Merge branch 'master' into kennytm/disk-quota

lightning/backend/local.go

@@ -1168,7 +1168,34 @@ loopWrite:
 			for errCnt < maxRetryTimes {
 				log.L().Debug("ingest meta", zap.Reflect("meta", meta))
 				var resp *sst.IngestResponse
-				resp, err = local.Ingest(ctx, meta, region)
+				failpoint.Inject("FailIngestMeta", func(val failpoint.Value) {
+					// only inject the error once
+					switch val.(string) {
+					case "notleader":
+						resp = &sst.IngestResponse{
+							Error: &errorpb.Error{
+								NotLeader: &errorpb.NotLeader{
+									RegionId: region.Region.Id,
+									Leader:   region.Leader,
+								},
+							},
+						}
+					case "epochnotmatch":
+						resp = &sst.IngestResponse{
+							Error: &errorpb.Error{
+								EpochNotMatch: &errorpb.EpochNotMatch{
+									CurrentRegions: []*metapb.Region{region.Region},
+								},
+							},
+						}
+					}
+					if resp != nil {
+						err = nil
+					}
+				})
+				if resp == nil {
+					resp, err = local.Ingest(ctx, meta, region)
+				}
 				if err != nil {
 					if errors.Cause(err) == context.Canceled {
 						return nil, err
@@ -1178,16 +1205,7 @@ loopWrite:
 					errCnt++
 					continue
 				}
-				failpoint.Inject("FailIngestMeta", func(val failpoint.Value) {
-					switch val.(string) {
-					case "notleader":
-						resp.Error.NotLeader = &errorpb.NotLeader{
-							RegionId: region.Region.Id, Leader: region.Leader}
-					case "epochnotmatch":
-						resp.Error.EpochNotMatch = &errorpb.EpochNotMatch{
-							CurrentRegions: []*metapb.Region{region.Region}}
-					}
-				})
+
 				var retryTy retryType
 				var newRegion *split.RegionInfo
 				retryTy, newRegion, err = local.isIngestRetryable(ctx, resp, region, meta)

lightning/backend/local_test.go

@@ -17,7 +17,6 @@ import (
 	"bytes"
 	"context"
 	"encoding/binary"
-	"fmt"
 	"math"
 	"math/rand"
 	"os"
@@ -379,16 +378,13 @@ func testLocalWriter(c *C, needSort bool, partitialSort bool) {
 }
 
 func (s *localSuite) TestLocalWriterWithSort(c *C) {
-	fmt.Println("TestLocalWriterWithSort")
 	testLocalWriter(c, false, false)
 }
 
 func (s *localSuite) TestLocalWriterWithIngest(c *C) {
-	fmt.Println("TestLocalWriterWithIngest")
 	testLocalWriter(c, true, false)
 }
 
 func (s *localSuite) TestLocalWriterWithIngestUnsort(c *C) {
-	fmt.Println("TestLocalWriterWithIngestUnsort")
 	testLocalWriter(c, true, true)
 }

lightning/backend/tidb.go

@@ -188,13 +188,14 @@ func (enc *tidbEncoder) appendSQL(sb *strings.Builder, datum *types.Datum, col *
 		value := strconv.AppendFloat(buffer[:0], datum.GetFloat64(), 'g', -1, 64)
 		sb.Write(value)
 	case types.KindString:
-		if enc.mode.HasStrictMode() {
-			d, err := table.CastValue(enc.se, *datum, col.ToInfo(), false, false)
-			if err != nil {
-				return errors.Trace(err)
-			}
-			datum = &d
-		}
+		// See: https://github.com/pingcap/tidb-lightning/issues/550
+		//if enc.mode.HasStrictMode() {
+		//	d, err := table.CastValue(enc.se, *datum, col.ToInfo(), false, false)
+		//	if err != nil {
+		//		return errors.Trace(err)
+		//	}
+		//	datum = &d
+		//}
 
 		enc.appendSQLBytes(sb, datum.GetBytes())
 	case types.KindBytes:

lightning/backend/tidb_test.go

@@ -183,7 +183,8 @@ func (s *mysqlSuite) TestWriteRowsErrorOnDup(c *C) {
 	c.Assert(err, IsNil)
 }
 
-func (s *mysqlSuite) TestStrictMode(c *C) {
+// TODO: temporarily disable this test before we fix strict mode
+func (s *mysqlSuite) testStrictMode(c *C) {
 	ft := *types.NewFieldType(mysql.TypeVarchar)
 	ft.Charset = charset.CharsetUTF8MB4
 	col0 := &model.ColumnInfo{ID: 1, Name: model.NewCIStr("s0"), State: model.StatePublic, Offset: 0, FieldType: ft}

lightning/config/config.go

@@ -306,6 +306,8 @@ type Security struct {
 	CAPath   string `toml:"ca-path" json:"ca-path"`
 	CertPath string `toml:"cert-path" json:"cert-path"`
 	KeyPath  string `toml:"key-path" json:"key-path"`
+	// RedactInfoLog indicates that whether enabling redact log
+	RedactInfoLog bool `toml:"redact-info-log" json:"redact-info-log"`
 }
 
 // RegistersMySQL registers (or deregisters) the TLS config with name "cluster"

lightning/config/global.go

@@ -144,7 +144,6 @@ func LoadGlobalConfig(args []string, extraFlags func(*flag.FlagSet)) (*GlobalCon
 
 	logLevel := flagext.ChoiceVar(fs, "L", "", `log level: info, debug, warn, error, fatal (default info)`, "", "info", "debug", "warn", "warning", "error", "fatal")
 	logFilePath := fs.String("log-file", "", "log file path")
-	redactLog := fs.Bool("redact-log", false, "whether to redact sensitive info in log")
 	tidbHost := fs.String("tidb-host", "", "TiDB server host")
 	tidbPort := fs.Int("tidb-port", 0, "TiDB server port (default 4000)")
 	tidbUser := fs.String("tidb-user", "", "TiDB user name to connect")
@@ -163,6 +162,7 @@ func LoadGlobalConfig(args []string, extraFlags func(*flag.FlagSet)) (*GlobalCon
 	tlsCAPath := fs.String("ca", "", "CA certificate path for TLS connection")
 	tlsCertPath := fs.String("cert", "", "certificate path for TLS connection")
 	tlsKeyPath := fs.String("key", "", "private key path for TLS connection")
+	redactInfoLog := fs.Bool("redact-info-log", false, "whether to redact sensitive info in log")
 
 	statusAddr := fs.String("status-addr", "", "the Lightning server address")
 	serverMode := fs.Bool("server-mode", false, "start Lightning in server mode, wait for multiple tasks instead of starting immediately")
@@ -199,9 +199,6 @@ func LoadGlobalConfig(args []string, extraFlags func(*flag.FlagSet)) (*GlobalCon
 	if *logFilePath != "" {
 		cfg.App.Config.File = *logFilePath
 	}
-	if *redactLog {
-		cfg.App.Config.RedactLog = *redactLog
-	}
 	// "-" is a special config for log to stdout
 	if cfg.App.Config.File == "-" {
 		cfg.App.Config.File = ""
@@ -271,6 +268,9 @@ func LoadGlobalConfig(args []string, extraFlags func(*flag.FlagSet)) (*GlobalCon
 	if *tlsKeyPath != "" {
 		cfg.Security.KeyPath = *tlsKeyPath
 	}
+	if *redactInfoLog {
+		cfg.Security.RedactInfoLog = *redactInfoLog
+	}
 	if len(filter) > 0 {
 		cfg.Mydumper.Filter = filter
 	}

lightning/lightning.go

@@ -81,6 +81,8 @@ func New(globalCfg *config.GlobalConfig) *Lightning {
 		log.L().Fatal("failed to load TLS certificates", zap.Error(err))
 	}
 
+	log.InitRedact(globalCfg.Security.RedactInfoLog)
+
 	ctx, shutdown := context.WithCancel(context.Background())
 	return &Lightning{
 		globalCfg: globalCfg,

lightning/log/log.go

@@ -44,8 +44,6 @@ type Config struct {
 	FileMaxDays int `toml:"max-days" json:"max-days"`
 	// Maximum number of old log files to retain.
 	FileMaxBackups int `toml:"max-backups" json:"max-backups"`
-	// Redact sensitive logs during the whole process
-	RedactLog bool `toml:"redact-log" json:"redact-log"`
 }
 
 func (cfg *Config) Adjust() {
@@ -100,8 +98,6 @@ func InitLogger(cfg *Config, tidbLoglevel string) error {
 	appLogger = Logger{logger.WithOptions(zap.AddStacktrace(zap.DPanicLevel))}
 	appLevel = props.Level
 
-	InitRedact(cfg.RedactLog)
-
 	return nil
 }
 

tests/local_backend/run.sh

@@ -23,7 +23,7 @@ ENGINE_COUNT=6
 rm -f "$TEST_DIR/lightning-local.log"
 rm -f "/tmp/tidb_lightning_checkpoint_local_backend_test.pb"
 run_sql 'DROP DATABASE IF EXISTS cpeng;'
-export GO_FAILPOINTS='github.com/pingcap/tidb-lightning/lightning/backend/local/FailIngestMeta=return("notleader")'
+export GO_FAILPOINTS='github.com/pingcap/tidb-lightning/lightning/backend/FailIngestMeta=1*return("notleader")'
 
 run_lightning --backend local --enable-checkpoint=1 --log-file "$TEST_DIR/lightning-local.log" --config "tests/$TEST_NAME/config.toml"
 
@@ -40,7 +40,7 @@ check_contains 'sum(c): 46'
 run_sql 'DROP DATABASE cpeng;'
 rm -f "/tmp/tidb_lightning_checkpoint_local_backend_test.pb"
 
-export GO_FAILPOINTS='github.com/pingcap/tidb-lightning/lightning/backend/local/FailIngestMeta=return("epochnotmatch")'
+export GO_FAILPOINTS='github.com/pingcap/tidb-lightning/lightning/backend/FailIngestMeta=2*return("epochnotmatch")'
 
 run_lightning --backend local --enable-checkpoint=1 --log-file "$TEST_DIR/lightning-local.log" --config "tests/$TEST_NAME/config.toml"
 

tidb-lightning.toml

@@ -41,8 +41,6 @@ file = "tidb-lightning.log"
 max-size = 128 # MB
 max-days = 28
 max-backups = 14
-# If set to true, lightning will redact sensitive infomation in log.
-redact-log = false
 
 [security]
 # specifies certificates and keys for TLS connections within the cluster.
@@ -52,6 +50,8 @@ redact-log = false
 # cert-path = "/path/to/lightning.pem"
 # private key of this service.
 # key-path = "/path/to/lightning.key"
+# If set to true, lightning will redact sensitive infomation in log.
+# redact-info-log = false
 
 [checkpoint]
 # Whether to enable checkpoints.