添加consumer destroy 功能

service/agent_msg_process.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"git.quantgroup.cn/DevOps/enoch/service/end_points"
+	"git.quantgroup.cn/DevOps/enoch/service/log"
 	"github.com/influxdata/influxdb/client/v2"
 	"math/big"
 	"net"
@@ -26,6 +27,18 @@ func (HealthMessageHandler) MsgProcess(msg string) {
 	buildMsg(chunkMsg)
 }
 
+func (HealthMessageHandler) Destroy()  {
+	if len(metricsPointSlice) > 0 {
+		logger.Info.Println("metricsMessageHandler 提交本地缓存数据：", len(metricsPointSlice))
+		batchWrite(metricsPointSlice)
+	}
+
+	if len(healthPointSlice) > 0 {
+		logger.Info.Println("HealthMessageHandler 提交本地缓存数据：", len(healthPointSlice))
+		batchWrite(healthPointSlice)
+	}
+}
+
 func buildHealthInfluxMsg(appName string, ip string, timestamp time.Time, submitLimit int, db map[string]end_points.DBDetail) {
 	tags := make(map[string]string, )
 	tags["sys_name"] = appName

service/kafka_agent_sarama.go

@@ -49,6 +49,7 @@ func AgentClusterConsumer(kafkaConf conf.KafkaConf, messageHandle MessageHandler
 				consumer.MarkOffset(msg, "") // mark message as processed
 			}
 		case <-signals:
+			messageHandle.Destroy()
 			return
 		}
 	}

service/messageHandler.go

@@ -2,4 +2,5 @@ package service
 
 type MessageHandler interface {
 	MsgProcess(msg string)
+	Destroy()
 }

service/msg_process.go

@@ -30,6 +30,13 @@ func (BraveMessageHandler) MsgProcess(msg string) {
 	//msgRedisProcess(traceMsg)
 	msgInfluxProcess(traceMsg)
 }
+func (BraveMessageHandler) Destroy() {
+	//系统关系时，提交所有数据
+	if len(pointSlice) > 0 {
+		logger.Info.Println("braveMessageHandler 提交本地缓存数据：", len(pointSlice))
+		batchWrite(pointSlice)
+	}
+}
 
 var batchSize = 5000
 var pointSlice = make([]*client.Point, 0, batchSize)
@@ -97,8 +104,6 @@ func batchWrite(pointArray []*client.Point) {
 	}
 }
 
-
-
 func httpWrite(pointArray []*client.Point) error {
 	c := data.NewClient()
 	defer func() { _ = c.Close() }()
@@ -130,7 +135,7 @@ func ReSubmit(data []string) error {
 		if len(pointSlice) > 1000 {
 			err := httpWrite(pointSlice)
 			if err != nil {
-				return  err
+				return err
 			}
 			logger.Info.Println("缓存重新提交：1000")
 			pointSlice = make([]*client.Point, 0)
@@ -140,7 +145,7 @@ func ReSubmit(data []string) error {
 		err := httpWrite(pointSlice)
 		if err != nil {
 			logger.Info.Println(pointSlice)
-			return  err
+			return err
 		}
 		logger.Info.Println("缓存重新提交：", len(pointSlice))
 	}
@@ -200,5 +205,4 @@ func Counter(day string, fun func(sysName string, durations map[string]string))
 		}
 		fun(string(redisKey.([]uint8)), reply2)
 	}
-
 }