lightning: support Re/ReregisterMySQL by different tls name (#30463)

br/pkg/lightning/config/config.go

@@ -552,27 +552,37 @@ type Security struct {
 	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"`
+
+	// TLSConfigName is used to set tls config for lightning in DM, so we don't expose this field to user
+	// DM may running many lightning instances at same time, so we need to set different tls config name for each lightning
+	TLSConfigName string `toml:"-" json:"-"`
 }
 
-// RegistersMySQL registers (or deregisters) the TLS config with name "cluster"
+// RegisterMySQL registers the TLS config with name "cluster" or security.TLSConfigName
 // for use in `sql.Open()`. This method is goroutine-safe.
 func (sec *Security) RegisterMySQL() error {
 	if sec == nil {
 		return nil
 	}
 	tlsConfig, err := common.ToTLSConfig(sec.CAPath, sec.CertPath, sec.KeyPath)
-	switch {
-	case err != nil:
+	if err != nil {
 		return errors.Trace(err)
-	case tlsConfig != nil:
+	}
+	if tlsConfig != nil {
 		// error happens only when the key coincides with the built-in names.
-		_ = gomysql.RegisterTLSConfig("cluster", tlsConfig)
-	default:
-		gomysql.DeregisterTLSConfig("cluster")
+		_ = gomysql.RegisterTLSConfig(sec.TLSConfigName, tlsConfig)
 	}
 	return nil
 }
 
+// DeregisterMySQL deregisters the TLS config with security.TLSConfigName
+func (sec *Security) DeregisterMySQL() {
+	if sec == nil || len(sec.CAPath) == 0 {
+		return
+	}
+	gomysql.DeregisterTLSConfig(sec.TLSConfigName)
+}
+
 // A duration which can be deserialized from a TOML string.
 // Implemented as https://github.com/BurntSushi/toml#using-the-encodingtextunmarshaler-interface
 type Duration struct {
@@ -1124,7 +1134,10 @@ func (cfg *Config) CheckAndAdjustSecurity() error {
 	switch cfg.TiDB.TLS {
 	case "":
 		if len(cfg.TiDB.Security.CAPath) > 0 {
-			cfg.TiDB.TLS = "cluster"
+			if cfg.TiDB.Security.TLSConfigName == "" {
+				cfg.TiDB.Security.TLSConfigName = "cluster" // adjust this the default value
+			}
+			cfg.TiDB.TLS = cfg.TiDB.Security.TLSConfigName
 		} else {
 			cfg.TiDB.TLS = "false"
 		}

br/pkg/lightning/config/config_test.go

@@ -323,6 +323,13 @@ func (s *configTestSuite) TestAdjustSecuritySection(c *C) {
 		c.Assert(cfg.TiDB.Security.CAPath, Equals, tc.expectedCA, comment)
 		c.Assert(cfg.TiDB.TLS, Equals, tc.expectedTLS, comment)
 	}
+	// test different tls config name
+	cfg := config.NewConfig()
+	assignMinimalLegalValue(cfg)
+	cfg.Security.CAPath = "/path/to/ca.pem"
+	cfg.Security.TLSConfigName = "tidb-tls"
+	c.Assert(cfg.Adjust(context.Background()), IsNil)
+	c.Assert(cfg.TiDB.Security.TLSConfigName, Equals, cfg.TiDB.TLS)
 }
 
 func (s *configTestSuite) TestInvalidCSV(c *C) {

br/pkg/lightning/lightning.go

@@ -274,10 +274,7 @@ func (l *Lightning) run(taskCtx context.Context, taskCfg *config.Config, g glue.
 		if taskCfg.TiDB.Security == nil {
 			return
 		}
-		taskCfg.TiDB.Security.CAPath = ""
-		if err := taskCfg.TiDB.Security.RegisterMySQL(); err != nil {
-			log.L().Warn("failed to deregister TLS config", log.ShortError(err))
-		}
+		taskCfg.TiDB.Security.DeregisterMySQL()
 	}()
 
 	// initiation of default glue should be after RegisterMySQL, which is ready to be called after taskCfg.Adjust

br/pkg/lightning/restore/restore.go

@@ -787,6 +787,7 @@ func (rc *Controller) restoreSchema(ctx context.Context) error {
 		os.Exit(0)
 	})
 
+	rc.checkpointsWg.Add(1) // checkpointsWg will be done in `rc.listenCheckpointUpdates`
 	go rc.listenCheckpointUpdates()
 
 	sysVars := ObtainImportantVariables(ctx, rc.tidbGlue.GetSQLExecutor(), !rc.isTiDBBackend())
@@ -993,7 +994,7 @@ func (rc *Controller) saveStatusCheckpoint(ctx context.Context, tableName string
 
 // listenCheckpointUpdates will combine several checkpoints together to reduce database load.
 func (rc *Controller) listenCheckpointUpdates() {
-	rc.checkpointsWg.Add(1)
+	defer rc.checkpointsWg.Done()
 
 	var lock sync.Mutex
 	coalesed := make(map[string]*checkpoints.TableCheckpointDiff)
@@ -1082,7 +1083,6 @@ func (rc *Controller) listenCheckpointUpdates() {
 			}
 		})
 	}
-	rc.checkpointsWg.Done()
 }
 
 // buildRunPeriodicActionAndCancelFunc build the runPeriodicAction func and a cancel func

br/pkg/lightning/restore/restore_test.go

@@ -1118,6 +1118,7 @@ func (s *tableRestoreSuite) TestSaveStatusCheckpoint(c *C) {
 		saveCpCh:      saveCpCh,
 		checkpointsDB: checkpoints.NewNullCheckpointsDB(),
 	}
+	rc.checkpointsWg.Add(1)
 	go rc.listenCheckpointUpdates()
 
 	start := time.Now()