Merge pull request #749 from WeBankPartners/dev

zgyzgyhero · web-flow · commit 114ffecbc1e3 · 2021-11-03T02:23:45.000+08:00
Dev
diff --git a/monitor-agent/archive_mysql_tool/funcs/db.go b/monitor-agent/archive_mysql_tool/funcs/db.go
@@ -50,6 +50,29 @@ func InitDbEngine(databaseName string) (err error) {
 	return err
 }
 
+func ResetDbEngine()  {
+	err := mysqlEngine.Close()
+	if err != nil {
+		log.Printf("close mysql engine fail,%s \n", err.Error())
+	}
+	time.Sleep(30*time.Second)
+	databaseName := Config().Mysql.DatabasePrefix + time.Now().Format("2006")
+	connectStr := fmt.Sprintf("%s:%s@%s(%s:%s)/%s?collation=utf8mb4_unicode_ci&allowNativePasswords=true",
+		Config().Mysql.User, Config().Mysql.Password, "tcp", Config().Mysql.Server, Config().Mysql.Port, databaseName)
+	mysqlEngine, err = xorm.NewEngine("mysql", connectStr)
+	if err != nil {
+		log.Printf("init mysql fail with connect: %s error: %v \n", connectStr, err)
+	} else {
+		mysqlEngine.SetMaxIdleConns(Config().Mysql.MaxIdle)
+		mysqlEngine.SetMaxOpenConns(Config().Mysql.MaxOpen)
+		mysqlEngine.SetConnMaxLifetime(time.Duration(Config().Mysql.Timeout) * time.Second)
+		mysqlEngine.Charset("utf8")
+		// 使用驼峰式映射
+		mysqlEngine.SetMapper(core.SnakeMapper{})
+	}
+	log.Println("Reset db engine done! ")
+}
+
 func InitMonitorDbEngine() (err error) {
 	connectStr := fmt.Sprintf("%s:%s@%s(%s:%s)/%s?collation=utf8mb4_unicode_ci&allowNativePasswords=true",
 		Config().Monitor.Mysql.User, Config().Monitor.Mysql.Password, "tcp", Config().Monitor.Mysql.Server, Config().Monitor.Mysql.Port, Config().Monitor.Mysql.DataBase)
@@ -91,7 +114,7 @@ func insertMysql(rows []*ArchiveTable, tableName string) error {
 	for sqlIndex, v := range sqlList {
 		var tmpErr error
 		for i := 0; i < 3; i++ {
-			log.Printf("start try %d to insert mysql,data num:%d \n", i+1, rowCountList[sqlIndex])
+			//log.Printf("start try %d to insert mysql,data num:%d \n", i+1, rowCountList[sqlIndex])
 			_, err := mysqlEngine.Exec(v)
 			if err != nil {
 				tmpErr = err
@@ -104,6 +127,7 @@ func insertMysql(rows []*ArchiveTable, tableName string) error {
 			}
 		}
 		if tmpErr != nil {
+			log.Printf("Exec sql error:%s \n", tmpErr.Error())
 			tmpErrorString := tmpErr.Error()
 			if len(tmpErrorString) > 200 {
 				tmpErrorString = tmpErrorString[:200]
diff --git a/monitor-agent/archive_mysql_tool/funcs/job.go b/monitor-agent/archive_mysql_tool/funcs/job.go
@@ -140,6 +140,11 @@ func consumeJob() {
 		endTime := time.Now()
 		useTime := float64(endTime.Sub(startTime).Nanoseconds()) / 1e6
 		log.Printf("done with consume job,use time: %.3f ms", useTime)
+		if int(endTime.Sub(startTime).Seconds()) >= jobTimeout {
+			log.Println("job timeout,try to reset db connection ")
+			ResetDbEngine()
+		}
+		gTransport.CloseIdleConnections()
 	}
 }
 
@@ -151,7 +156,7 @@ func checkJobStatus() {
 			log.Printf("archive job done \n")
 			break
 		}
-		time.Sleep(10 * time.Second)
+		time.Sleep(30 * time.Second)
 	}
 }
 
@@ -165,8 +170,9 @@ func archiveTimeoutAction(param ArchiveActionList) {
 	case <-timeoutChan:
 		log.Printf("done archive action,job length:%d \n", len(param))
 	case <-time.After(time.Duration(jobTimeout) * time.Second):
-		log.Printf("timeout archive action in 10min,job length:%d \n", len(param))
+		log.Printf("timeout archive action in %d s,job length:%d \n", jobTimeout, len(param))
 	}
+	close(timeoutChan)
 }
 
 func archiveAction(param ArchiveActionList) {
diff --git a/monitor-agent/archive_mysql_tool/funcs/prometheus.go b/monitor-agent/archive_mysql_tool/funcs/prometheus.go
@@ -67,13 +67,16 @@ func getPrometheusData(param *PrometheusQueryParam) error {
 		return fmt.Errorf("http request error, %v \n", err)
 	}
 	if resp.StatusCode/100 != 2 {
+		if resp.Body != nil {
+			resp.Body.Close()
+		}
 		return fmt.Errorf("http response status %d \n", resp.StatusCode)
 	}
 	bodyBytes,err := ioutil.ReadAll(resp.Body)
+	resp.Body.Close()
 	if err != nil {
 		return fmt.Errorf("http read body error, %v \n", err)
 	}
-	defer resp.Body.Close()
 	var result PrometheusResponse
 	err = json.Unmarshal(bodyBytes, &result)
 	if err != nil {
diff --git a/monitor-server/api/v1/agent/kubernetes.go b/monitor-server/api/v1/agent/kubernetes.go
@@ -12,6 +12,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"time"
 )
 
 func UpdateKubernetesCluster(c *gin.Context)  {
@@ -72,6 +73,8 @@ func UpdateKubernetesCluster(c *gin.Context)  {
 	if err != nil {
 		mid.ReturnHandleError(c, err.Error(), err)
 	}else{
+		time.Sleep(20*time.Second)
+		db.SyncPodToEndpoint()
 		mid.ReturnSuccess(c)
 	}
 }
@@ -184,6 +187,7 @@ func handleAddKubernetesCluster(input k8sClusterRequestInputObj) error  {
 	}else {
 		err = db.AddKubernetesCluster(m.KubernetesClusterParam{ClusterName: input.ClusterName, Ip: input.Ip, Port: input.Port, Token: input.Token})
 	}
+	db.SyncPodToEndpoint()
 	return err
 }
 
diff --git a/monitor-server/api/v1/alarm/alarm.go b/monitor-server/api/v1/alarm/alarm.go
@@ -545,6 +545,8 @@ func QueryEntityAlarm(c *gin.Context)  {
 	value := param.Criteria.Condition
 	if strings.Contains(value, "monitor-check") {
 		alarmObj := db.GetCheckProgressContent(value)
+		alarmObj.Id = value
+		alarmObj.DisplayName = alarmObj.Subject
 		result.Data = append(result.Data, &alarmObj)
 	}else {
 		if strings.Contains(value, "-") {
@@ -571,6 +573,8 @@ func QueryEntityAlarm(c *gin.Context)  {
 			mid.ReturnData(c, result)
 			return
 		}
+		alarmObj.Id = value
+		alarmObj.DisplayName = alarmObj.Subject
 		result.Data = append(result.Data, &alarmObj)
 	}
 	result.Status = "OK"
diff --git a/monitor-server/models/entity.go b/monitor-server/models/entity.go
@@ -76,6 +76,7 @@ type AlarmEntity struct {
 
 type AlarmEntityObj struct {
 	Id         string `json:"id"`
+	DisplayName string `json:"displayName"`
 	Status     string `json:"status"`
 	Subject    string `json:"subject"`
 	Content    string `json:"content"`
diff --git a/monitor-server/services/db/dashboard.go b/monitor-server/services/db/dashboard.go
@@ -140,7 +140,7 @@ func GetPromMetric(endpoint []string, metric string) (error, string) {
 			reg = strings.Replace(reg, "$guid", host.Guid, -1)
 		}
 		if strings.Contains(reg, "$pod") {
-			reg = strings.Replace(reg, "$pod", host.Name[len(host.ExportVersion)+1:], -1)
+			reg = strings.Replace(reg, "$pod", host.Name, -1)
 		}
 		if strings.Contains(reg, "$k8s_namespace") {
 			reg = strings.Replace(reg, "$k8s_namespace", host.ExportVersion, -1)
@@ -182,7 +182,7 @@ func ReplacePromQlKeyword(promQl, metric string, host m.EndpointTable) string {
 		promQl = strings.Replace(promQl, "$guid", host.Guid, -1)
 	}
 	if strings.Contains(promQl, "$pod") {
-		promQl = strings.Replace(promQl, "$pod", host.Name[len(host.ExportVersion)+1:], -1)
+		promQl = strings.Replace(promQl, "$pod", host.Name, -1)
 	}
 	if strings.Contains(promQl, "$k8s_namespace") {
 		promQl = strings.Replace(promQl, "$k8s_namespace", host.ExportVersion, -1)
diff --git a/monitor-server/services/db/kubernetes.go b/monitor-server/services/db/kubernetes.go
@@ -166,11 +166,11 @@ func StartCronSyncKubernetesPod(interval int)  {
 	t := time.NewTicker(time.Duration(interval*10)*time.Second).C
 	for {
 		<- t
-		go syncPodToEndpoint()
+		go SyncPodToEndpoint()
 	}
 }
 
-func syncPodToEndpoint() bool {
+func SyncPodToEndpoint() bool {
 	log.Logger.Info("start to sync kubernetes pod")
 	var kubernetesTables []*m.KubernetesClusterTable
 	result := false
@@ -190,7 +190,11 @@ func syncPodToEndpoint() bool {
 		x.SQL("select * from kubernetes_endpoint_rel where kubernete_id=?", v.Id).Find(&tmpKubernetesEndpointTables)
 		tmpApiServerIp := v.ApiServer[:strings.Index(v.ApiServer, ":")]
 		for _,vv := range series {
-			tmpEndpointGuid := fmt.Sprintf("%s_%s_pod", vv.Name, tmpApiServerIp)
+			tmpPodName := vv.Name
+			if strings.HasPrefix(tmpPodName, "pod=") {
+				tmpPodName = tmpPodName[4:]
+			}
+			tmpEndpointGuid := fmt.Sprintf("%s_%s_pod", tmpPodName, tmpApiServerIp)
 			existsFlag := false
 			for _,ke := range tmpKubernetesEndpointTables {
 				if ke.EndpointGuid == tmpEndpointGuid {
@@ -204,8 +208,8 @@ func syncPodToEndpoint() bool {
 				}else{
 					tmpGuidMap[tmpEndpointGuid] = 1
 				}
-				endpointTables = append(endpointTables, &m.EndpointTable{Guid:tmpEndpointGuid, Name:vv.Name, Ip:tmpApiServerIp, ExportType:"pod", Step:10, OsType:v.ClusterName})
-				kubernetesEndpointTables = append(kubernetesEndpointTables, &m.KubernetesEndpointRelTable{KuberneteId:v.Id, EndpointGuid:tmpEndpointGuid})
+				endpointTables = append(endpointTables, &m.EndpointTable{Guid:tmpEndpointGuid, Name:tmpPodName, Ip:tmpApiServerIp, ExportType:"pod", Step:10, OsType:v.ClusterName})
+				kubernetesEndpointTables = append(kubernetesEndpointTables, &m.KubernetesEndpointRelTable{KuberneteId:v.Id, EndpointGuid:tmpEndpointGuid, PodGuid: tmpPodName})
 			}
 		}
 	}
@@ -242,9 +246,9 @@ func syncPodToEndpoint() bool {
 	}
 	if len(kubernetesEndpointTables) > 0 {
 		result = true
-		keRelSql := "insert into kubernetes_endpoint_rel(kubernete_id,endpoint_guid) values "
+		keRelSql := "insert into kubernetes_endpoint_rel(kubernete_id,endpoint_guid,pod_guid) values "
 		for i,v := range kubernetesEndpointTables {
-			keRelSql += fmt.Sprintf("(%d,'%s')", v.KuberneteId, v.EndpointGuid)
+			keRelSql += fmt.Sprintf("(%d,'%s','%s')", v.KuberneteId, v.EndpointGuid, v.PodGuid)
 			if i < len(kubernetesEndpointTables)-1 {
 				keRelSql += ","
 			}

Original file line number	Diff line number	Diff line change
`@@ -140,6 +140,11 @@ func consumeJob() {`
`140`	`140`	`endTime := time.Now()`
`141`	`141`	`useTime := float64(endTime.Sub(startTime).Nanoseconds()) / 1e6`
`142`	`142`	`log.Printf("done with consume job,use time: %.3f ms", useTime)`
	`143`	`+ if int(endTime.Sub(startTime).Seconds()) >= jobTimeout {`
	`144`	`+ log.Println("job timeout,try to reset db connection ")`
	`145`	`+ ResetDbEngine()`
	`146`	`+ }`
	`147`	`+ gTransport.CloseIdleConnections()`
`143`	`148`	`}`
`144`	`149`	`}`
`145`	`150`
`@@ -151,7 +156,7 @@ func checkJobStatus() {`
`151`	`156`	`log.Printf("archive job done \n")`
`152`	`157`	`break`
`153`	`158`	`}`
`154`		`- time.Sleep(10 * time.Second)`
	`159`	`+ time.Sleep(30 * time.Second)`
`155`	`160`	`}`
`156`	`161`	`}`
`157`	`162`
`@@ -165,8 +170,9 @@ func archiveTimeoutAction(param ArchiveActionList) {`
`165`	`170`	`case <-timeoutChan:`
`166`	`171`	`log.Printf("done archive action,job length:%d \n", len(param))`
`167`	`172`	`case <-time.After(time.Duration(jobTimeout) * time.Second):`
`168`		`- log.Printf("timeout archive action in 10min,job length:%d \n", len(param))`
	`173`	`+ log.Printf("timeout archive action in %d s,job length:%d \n", jobTimeout, len(param))`
`169`	`174`	`}`
	`175`	`+ close(timeoutChan)`
`170`	`176`	`}`
`171`	`177`
`172`	`178`	`func archiveAction(param ArchiveActionList) {`
Original file line number	Diff line number	Diff line change
`@@ -67,13 +67,16 @@ func getPrometheusData(param *PrometheusQueryParam) error {`
`67`	`67`	`return fmt.Errorf("http request error, %v \n", err)`
`68`	`68`	`}`
`69`	`69`	`if resp.StatusCode/100 != 2 {`
	`70`	`+ if resp.Body != nil {`
	`71`	`+ resp.Body.Close()`
	`72`	`+ }`
`70`	`73`	`return fmt.Errorf("http response status %d \n", resp.StatusCode)`
`71`	`74`	`}`
`72`	`75`	`bodyBytes,err := ioutil.ReadAll(resp.Body)`
	`76`	`+ resp.Body.Close()`
`73`	`77`	`if err != nil {`
`74`	`78`	`return fmt.Errorf("http read body error, %v \n", err)`
`75`	`79`	`}`
`76`		`- defer resp.Body.Close()`
`77`	`80`	`var result PrometheusResponse`
`78`	`81`	`err = json.Unmarshal(bodyBytes, &result)`
`79`	`82`	`if err != nil {`
Original file line number	Diff line number	Diff line change
`@@ -12,6 +12,7 @@ import (`
`12`	`12`	`"net/http"`
`13`	`13`	`"strconv"`
`14`	`14`	`"strings"`
	`15`	`+ "time"`
`15`	`16`	`)`
`16`	`17`
`17`	`18`	`func UpdateKubernetesCluster(c *gin.Context) {`
`@@ -72,6 +73,8 @@ func UpdateKubernetesCluster(c *gin.Context) {`
`72`	`73`	`if err != nil {`
`73`	`74`	`mid.ReturnHandleError(c, err.Error(), err)`
`74`	`75`	`}else{`
	`76`	`+ time.Sleep(20*time.Second)`
	`77`	`+ db.SyncPodToEndpoint()`
`75`	`78`	`mid.ReturnSuccess(c)`
`76`	`79`	`}`
`77`	`80`	`}`
`@@ -184,6 +187,7 @@ func handleAddKubernetesCluster(input k8sClusterRequestInputObj) error {`
`184`	`187`	`}else {`
`185`	`188`	`err = db.AddKubernetesCluster(m.KubernetesClusterParam{ClusterName: input.ClusterName, Ip: input.Ip, Port: input.Port, Token: input.Token})`
`186`	`189`	`}`
	`190`	`+ db.SyncPodToEndpoint()`
`187`	`191`	`return err`
`188`	`192`	`}`
`189`	`193`
Original file line number	Diff line number	Diff line change
`@@ -140,7 +140,7 @@ func GetPromMetric(endpoint []string, metric string) (error, string) {`
`140`	`140`	`reg = strings.Replace(reg, "$guid", host.Guid, -1)`
`141`	`141`	`}`
`142`	`142`	`if strings.Contains(reg, "$pod") {`
`143`		`- reg = strings.Replace(reg, "$pod", host.Name[len(host.ExportVersion)+1:], -1)`
	`143`	`+ reg = strings.Replace(reg, "$pod", host.Name, -1)`
`144`	`144`	`}`
`145`	`145`	`if strings.Contains(reg, "$k8s_namespace") {`
`146`	`146`	`reg = strings.Replace(reg, "$k8s_namespace", host.ExportVersion, -1)`
`@@ -182,7 +182,7 @@ func ReplacePromQlKeyword(promQl, metric string, host m.EndpointTable) string {`
`182`	`182`	`promQl = strings.Replace(promQl, "$guid", host.Guid, -1)`
`183`	`183`	`}`
`184`	`184`	`if strings.Contains(promQl, "$pod") {`
`185`		`- promQl = strings.Replace(promQl, "$pod", host.Name[len(host.ExportVersion)+1:], -1)`
	`185`	`+ promQl = strings.Replace(promQl, "$pod", host.Name, -1)`
`186`	`186`	`}`
`187`	`187`	`if strings.Contains(promQl, "$k8s_namespace") {`
`188`	`188`	`promQl = strings.Replace(promQl, "$k8s_namespace", host.ExportVersion, -1)`