fix: 调整超时重试时长 #6619

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

@@ -3,6 +3,7 @@ package main
 import (
 	"dbm-services/mysql/db-partition/monitor"
 	"dbm-services/mysql/db-partition/service"
+	"dbm-services/mysql/db-partition/util"
 	"net/http"
 	"os"
 
@@ -33,6 +34,10 @@ func main() {
 		}
 	}
 
+	util.DbmetaClient = util.NewClientByHosts(viper.GetString("db_meta_service"))
+	util.DrsClient = util.NewClientByHosts(viper.GetString("db_remote_service"))
+	util.TicketClient = util.NewClientByHosts(viper.GetString("dbm_ticket_service"))
+
 	// 获取监控配置，多次尝试，获取监控配置失败
 	monitor.InitMonitor()
 

dbm-services/mysql/db-partition/model/init_redis.go

@@ -13,13 +13,15 @@ var rdb *redis.Client
 
 // InitClient 初始化连接
 func InitClient() (err error) {
+	// 初始化redis客户端
 	rdb = redis.NewClient(&redis.Options{
 		Addr:     fmt.Sprintf("%s:%d", viper.GetString("redis.host"), viper.GetInt("redis.port")),
 		Password: viper.GetString("redis.password"),
 		DB:       0,
 	})
 	slog.Info("redis info", "host", viper.GetString("redis.host"),
 		"port", viper.GetInt("redis.port"))
+	// 检查连通性
 	_, err = rdb.Ping().Result()
 	if err != nil {
 		slog.Error("redis db", "ping err", err)
@@ -33,15 +35,3 @@ func Lock(key string) (bool, error) {
 	slog.Info("msg", "key", key)
 	return rdb.SetNX(key, `{"lock":1}`, 30*time.Hour).Result()
 }
-
-/*
-UnLock 解锁
-func UnLock(key string) int64 {
-	nums, err := rdb.Del(key).Result()
-	if err != nil {
-		log.Println(err.Error())
-		return 0
-	}
-	return nums
-}
-*/

dbm-services/mysql/db-partition/service/db_meta_service.go

@@ -34,15 +34,9 @@ const Fail string = "failed"
 
 const Success string = "succeeded"
 
-// ExecuteAsynchronous TODO
-const ExecuteAsynchronous string = "UNKNOWN"
-
 // BackendMaster TODO
 const BackendMaster string = "backend_master"
 
-// RemoteMaster TODO
-const RemoteMaster string = "remote_master"
-
 // Orphan TODO
 const Orphan string = "orphan"
 
@@ -71,8 +65,7 @@ func CreateDbmTicket(config Ticket) (int, error) {
 	}
 
 	var resp Data
-	c := util.NewClientByHosts(viper.GetString("dbm_ticket_service"))
-	result, err := c.Do(http.MethodPost, "tickets/", config)
+	result, err := util.TicketClient.Do(http.MethodPost, "tickets/", config)
 	if err != nil {
 		slog.Error("msg", err)
 		return ticketId, err
@@ -164,9 +157,8 @@ type DownloadPartitionPara struct {
 
 // DownloadDbactor 下载dbactor
 func DownloadDbactor(bkCloudId int, ips []string) error {
-	c := util.NewClientByHosts(viper.GetString("db_meta_service"))
 	url := "/apis/v1/flow/scene/download_dbactor"
-	_, err := c.Do(http.MethodPost, url, DownloadPara{TicketType: "download_dbactor",
+	_, err := util.DbmetaClient.Do(http.MethodPost, url, DownloadPara{TicketType: "download_dbactor",
 		BkBizId:   viper.GetInt64("dba.bk_biz_id"),
 		BkCloudId: bkCloudId, DbType: "mysql", Ips: ips, CreatedBy: "admin"})
 	if err != nil {
@@ -178,9 +170,8 @@ func DownloadDbactor(bkCloudId int, ips []string) error {
 
 func DownloadFiles(files []Info) error {
 	path := "mysql/partition"
-	c := util.NewClientByHosts(viper.GetString("db_meta_service"))
 	url := "/apis/v1/flow/scene/download_file"
-	_, err := c.Do(http.MethodPost, url, DownloadPartitionPara{TicketType: "download_file",
+	_, err := util.DbmetaClient.Do(http.MethodPost, url, DownloadPartitionPara{TicketType: "download_file",
 		BkBizId: viper.GetInt64("dba.bk_biz_id"),
 		Files:   files, CreatedBy: "admin", Path: path})
 	if err != nil {
@@ -192,10 +183,9 @@ func DownloadFiles(files []Info) error {
 
 // GetCluster 根据域名获取集群信息
 func GetCluster(dns Domain, ClusterType string) (Instance, error) {
-	c := util.NewClientByHosts(viper.GetString("db_meta_service"))
 	var resp Instance
 	url := fmt.Sprintf("/apis/proxypass/dbmeta/priv_manager/mysql/%s/cluster_instances/", ClusterType)
-	result, err := c.Do(http.MethodPost, url, dns)
+	result, err := util.DbmetaClient.Do(http.MethodPost, url, dns)
 	if err != nil {
 		slog.Error("msg", url, err)
 		return resp, errno.DomainNotExists.Add(fmt.Sprintf(" %s: %s", dns.EntryName, err.Error()))
@@ -210,10 +200,9 @@ func GetCluster(dns Domain, ClusterType string) (Instance, error) {
 
 // GetAllClustersInfo 获取业务下所有集群信息
 func GetAllClustersInfo(id BkBizId) ([]Cluster, error) {
-	c := util.NewClientByHosts(viper.GetString("db_meta_service"))
 	var resp []Cluster
 	url := "/apis/proxypass/dbmeta/priv_manager/biz_clusters/"
-	result, err := c.Do(http.MethodPost, url, id)
+	result, err := util.DbmetaClient.Do(http.MethodPost, url, id)
 	if err != nil {
 		slog.Error("msg", url, err)
 		return resp, err
@@ -226,10 +215,9 @@ func GetAllClustersInfo(id BkBizId) ([]Cluster, error) {
 }
 
 func ListBizs() ([]Biz, error) {
-	c := util.NewClientByHosts(viper.GetString("db_meta_service"))
 	var resp []Biz
 	url := "/apis/cmdb/list_bizs/"
-	result, err := c.Do(http.MethodGet, url, nil)
+	result, err := util.DbmetaClient.Do(http.MethodGet, url, nil)
 	if err != nil {
 		slog.Error("msg", url, err)
 		return resp, err

dbm-services/mysql/db-partition/service/db_remote_service.go

@@ -8,8 +8,6 @@ import (
 	"strings"
 
 	"dbm-services/mysql/db-partition/util"
-
-	"github.com/spf13/viper"
 )
 
 // OneAddressExecuteSqlBasic OneAddressExecuteSql 通过db-remote-service服务连接mysql实例执行sql语句
@@ -18,14 +16,13 @@ func OneAddressExecuteSqlBasic(vtype string, queryRequest QueryRequest) (oneAddr
 	var errMsg []string
 	var result oneAddressResult
 	var temp []oneAddressResult
-	c := util.NewClientByHosts(viper.GetString("db_remote_service"))
 	var url string
 	if vtype == "mysql" {
 		url = "mysql/rpc/"
 	} else if vtype == "proxy" {
 		url = "proxy-admin/rpc/"
 	}
-	apiResp, err := c.Do(http.MethodPost, url, queryRequest)
+	apiResp, err := util.DrsClient.Do(http.MethodPost, url, queryRequest)
 	if err != nil {
 		slog.Error("drs err", err)
 		return result, err
@@ -83,17 +80,6 @@ type QueryRequest struct {
 	BkCloudId    int `form:"bk_cloud_id" json:"bk_cloud_id" url:"bk_cloud_id"`       // mysql服务所在的云域
 }
 
-// queryResponse db-remote-service服务/mysql/rpc接口返回的结构
-type queryResponse struct {
-	Code      int               `json:"code"`
-	Data      queryResponseData `json:"data"`
-	Msg       string            `json:"message"`
-	RequestId string            `json:"request_id"`
-}
-
-// queryResponseData 在多个ip:port执行sql返回的结果
-type queryResponseData []oneAddressResult
-
 // oneAddressResult 在一个ip:port执行sql返回的结果
 type oneAddressResult struct {
 	Address    string      `json:"address"`

dbm-services/mysql/db-partition/util/client.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"io/ioutil"
 	"log/slog"
-	"math/rand"
 	"net/http"
 	"net/http/httputil"
 	"strconv"
@@ -23,6 +22,12 @@ const (
 	statusSuccess int = 0
 )
 
+var (
+	DrsClient    *Client
+	DbmetaClient *Client
+	TicketClient *Client
+)
+
 // APIServerResponse TODO
 type APIServerResponse struct {
 	Code    int             `json:"code"`
@@ -77,13 +82,9 @@ func (c *Client) DoNew(method, url string, params interface{}, headers map[strin
 	var err error
 	for retryIdx := 0; retryIdx < 5; retryIdx++ {
 		response, err = c.doNewInner(method, url, params, headers)
-		if err == nil {
-			break
-		}
-		if strings.Contains(err.Error(), "cse.flowcontrol.Consumer.qps.limit") {
+		if err != nil {
 			slog.Error(fmt.Sprintf("DoNew failed, retryIdx:%d", retryIdx), err)
-			wait := retryIdx*retryIdx*1000 + rand.Intn(1000)
-			time.Sleep(time.Duration(wait) * time.Millisecond)
+			time.Sleep(time.Second)
 			continue
 		}
 		break
@@ -151,9 +152,12 @@ func (c *Client) doNewInner(method, url string, params interface{}, headers map[
 			break
 		}
 
-		wait := i*i*1000 + rand.Intn(1000)
-		time.Sleep(time.Duration(wait) * time.Millisecond)
-		slog.Warn(fmt.Sprintf("client.Do result with %s, wait %d milliSeconds and retry, url: %s", resp.Status, wait,
+		// 关闭前一个响应体，防止内存泄漏
+		if resp.Body != nil {
+			resp.Body.Close()
+		}
+		time.Sleep(time.Second)
+		slog.Warn(fmt.Sprintf("client.Do result with %s, wait 1 second and retry, url: %s", resp.Status,
 			req.URL.String()))
 		resp, err = c.client.Do(req)
 		if err != nil {

dbm-services/mysql/db-priv/util/client.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"io/ioutil"
 	"log/slog"
-	"math/rand"
 	"net/http"
 	"net/http/httputil"
 	"strings"
@@ -73,8 +72,7 @@ func (c *Client) DoNew(method, url string, params interface{}, headers map[strin
 		response, err = c.doNewInner(method, url, params, headers)
 		if err != nil {
 			slog.Error(fmt.Sprintf("DoNew failed, retryIdx:%d", retryIdx), err)
-			wait := retryIdx*retryIdx*1000 + rand.Intn(1000)
-			time.Sleep(time.Duration(wait) * time.Millisecond)
+			time.Sleep(time.Second)
 			continue
 		}
 		break
@@ -146,10 +144,8 @@ func (c *Client) doNewInner(method, url string, params interface{}, headers map[
 		if resp.Body != nil {
 			resp.Body.Close()
 		}
-
-		wait := i*i*1000 + rand.Intn(1000)
-		time.Sleep(time.Duration(wait) * time.Millisecond)
-		slog.Warn(fmt.Sprintf("client.Do result with %s, wait %d milliSeconds and retry, url: %s", resp.Status, wait,
+		time.Sleep(time.Second)
+		slog.Warn(fmt.Sprintf("client.Do result with %s, wait 1 second and retry, url: %s", resp.Status,
 			req.URL.String()))
 		resp, err = c.client.Do(req)
 		if err != nil {

dbm-services/sqlserver/db-tools/dbactuator/go.mod

@@ -57,7 +57,7 @@ require (
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/subosito/gotenv v1.4.2 // indirect
-	golang.org/x/crypto v0.9.0 // indirect
+	golang.org/x/crypto v0.9.0
 	golang.org/x/net v0.10.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect

dbm-services/sqlserver/db-tools/dbactuator/go.sum

@@ -418,6 +418,8 @@ golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.10.0 h1:SqMFp9UcQJZa+pmYuAKjd9xq1f0j5rLcDIk0mj4qAsA=
 golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.8.0 h1:n5xxQn2i3PC0yLAbjTpNT85q/Kgzcr2gIoX9OrJUols=
+golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=