refactor(dbm-services): 增大遍重载模拟执行任务的次数 #8834

Author: ymakedaq

dbm-services/mysql/db-simulation/app/service/simulation_task.go

@@ -172,8 +172,12 @@ func init() {
 	})
 	go func() {
 		// 等待一个周期的原因，避免重载执行正常的任务
-		time.Sleep(time.Duration(HeartbeatInterval) * time.Second)
-		reloadRunningTaskFromdb()
+		// 因为http svr 是graceful shutdown, 可能旧的服务会接受请求
+		for i := 0; i < 5; i++ {
+			logger.Info("the %d times reload running task", i)
+			time.Sleep(time.Duration(HeartbeatInterval) * time.Second)
+			reloadRunningTaskFromdb(i + HeartbeatInterval)
+		}
 		rdb.Close()
 	}()
 }
@@ -185,7 +189,8 @@ type ReloadParam struct {
 }
 
 // reloadRunningTaskFromdb 重载重启服务前运行的任务 加锁是避免多个服务同时启动，避免相同任务被重载
-func reloadRunningTaskFromdb() {
+// nolint
+func reloadRunningTaskFromdb(heartbeatInterval int) {
 	key := "simulation:reload:lock"
 	locker := redislock.New(rdb)
 	ctx := context.Background()
@@ -200,9 +205,8 @@ func reloadRunningTaskFromdb() {
 	}()
 	var tks []model.TbSimulationTask
 	if err := model.DB.Model(model.TbSimulationTask{}).Where(
-		//nolint
-		"phase not in (?) and create_time >  DATE_SUB(NOW(),INTERVAL 2 HOUR) and time_to_sec(timediff(now(),heartbeat_time)) > ? ",
-		[]string{model.PhaseDone, model.PhaseReloading}, HeartbeatInterval).Scan(&tks).Error; err != nil {
+		"phase not in (?) and create_time > DATE_SUB(NOW(),INTERVAL 2 HOUR) and time_to_sec(timediff(now(),heartbeat_time)) > ?",
+		[]string{model.PhaseDone, model.PhaseReloading}, heartbeatInterval).Scan(&tks).Error; err != nil {
 		logger.Error("get running task failed %s", err.Error())
 		return
 	}

dbm-services/mysql/db-simulation/app/syntax/syntax.go

@@ -142,7 +142,7 @@ func (tf *TmysqlParseFile) Do(dbtype string, versions []string) (result map[stri
 }
 
 func (tf *TmysqlParseFile) doSingleVersion(dbtype string, mysqlVersion string) (err error) {
-	errChan := make(chan error, len(tf.Param.FileNames))
+	errChan := make(chan error)
 	alreadExecutedSqlfileChan := make(chan string, len(tf.Param.FileNames))
 	signalChan := make(chan struct{})
 
@@ -257,7 +257,7 @@ func (t *TmysqlParse) delTempDir() {
 // Downloadfile download sqlfile
 func (tf *TmysqlParseFile) Downloadfile() (err error) {
 	wg := &sync.WaitGroup{}
-	errCh := make(chan error, 10)
+	errCh := make(chan error)
 	c := make(chan struct{}, 5)
 	for _, fileName := range tf.Param.FileNames {
 		wg.Add(1)
@@ -335,7 +335,7 @@ func (tf *TmysqlParseFile) Execute(alreadExecutedSqlfileCh chan string, version
 	var wg sync.WaitGroup
 	var errs []error
 	c := make(chan struct{}, 10) // Semaphore to limit concurrent goroutines
-	errChan := make(chan error, len(tf.Param.FileNames))
+	errChan := make(chan error)
 
 	// Iterate through all SQL files
 	for _, fileName := range tf.Param.FileNames {
@@ -383,7 +383,7 @@ func (t *TmysqlParse) AnalyzeParseResult(alreadExecutedSqlfileCh chan string, my
 	dbtype string) (err error) {
 	var errs []error
 	c := make(chan struct{}, 10)
-	errChan := make(chan error, 5)
+	errChan := make(chan error)
 	wg := &sync.WaitGroup{}
 
 	for sqlfile := range alreadExecutedSqlfileCh {

dbm-services/mysql/db-simulation/main.go

@@ -1,3 +1,4 @@
+// Package main TODO
 /*
  * TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-DB管理系统(BlueKing-BK-DBM) available.
  * Copyright (C) 2017-2023 THL A29 Limited, a Tencent company. All rights reserved.
@@ -7,7 +8,7 @@
  * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
  * specific language governing permissions and limitations under the License.
  */
-
+// main
 package main
 
 import (
@@ -26,7 +27,6 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/samber/lo"
 	"go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin"
-
 	_ "go.uber.org/automaxprocs"
 
 	"dbm-services/common/go-pubpkg/apm/metric"

dbm-services/mysql/db-simulation/pkg/util/spider.go

@@ -49,7 +49,7 @@ func ParseGetShardKeyForSpider(tableComment string) (string, error) {
 	if end-pos <= 0 {
 		return "", errors.New("parse error")
 	}
-
+	//nolint
 	len := uint(end - pos)
 	keyBuf := make([]byte, len)
 	copy(keyBuf, tableComment[pos:end])