5s延时问题监控(初版)

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/valyala/fasthttp v1.6.0
 	github.com/vrg0/go-common v0.0.0-20191213082238-e4e6080702f1
 	go.uber.org/zap v1.13.0
+	golang.org/x/sys v0.0.0-20200219091948-cb0a6d8edb6c // indirect
 	gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc // indirect
 	gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df
 )

pkg/global/global.go

@@ -48,6 +48,8 @@ var (
 	ConsulDc                       = ""
 	ConsulAddress                  = ""
 	ReportFormDir                  = ""
+	LogPath                        = ""
+	LogLevel                       = zapcore.DebugLevel
 	HttpPort                       = 9091
 )
 
@@ -92,9 +94,9 @@ func init() {
 	}
 
 	//初始化日志
-	logPath := Config.GetOrDefault(NamespaceApplication, "log.path", "/dev/stdout")
-	logLevel := getLoggerLevel(Config.GetOrDefault(NamespaceApplication, "log.level", "info"))
-	Logger = logger.New(logPath, logLevel)
+	LogPath = Config.GetOrDefault(NamespaceApplication, "log.path", "/dev/stdout")
+	LogLevel = getLoggerLevel(Config.GetOrDefault(NamespaceApplication, "log.level", "info"))
+	Logger = logger.New(LogPath, LogLevel)
 
 	//初始化kafka
 	kafkaLogger := Logger.GetStandardLogger()

pkg/points/delayed5s.go

+package points
+
+import (
+	"git.quantgroup.cn/DevOps/enoch/pkg/global"
+	"github.com/vrg0/go-common/logger"
+	"go.uber.org/zap/zapcore"
+	"sync"
+	"time"
+)
+
+type Node struct {
+	CreateTime time.Time
+	Points     TraceMsg
+}
+
+type Delayed5s struct {
+	Server map[string]Node
+	Client map[string]Node
+	lock   *sync.Mutex
+	save   *logger.Logger
+}
+
+func NewDelayed5s() *Delayed5s {
+	savePath := global.LogPath
+	if savePath != "/dev/stdout" {
+		savePath += "_delayed5s"
+	}
+
+	return &Delayed5s{
+		Server: make(map[string]Node),
+		Client: make(map[string]Node),
+		lock:   new(sync.Mutex),
+		save:   logger.New(savePath, zapcore.DebugLevel),
+	}
+}
+
+func (d *Delayed5s) Run() {
+	d.save.Info("5s延时监控程序启动")
+	go func() {
+		for {
+			time.Sleep(time.Second * 30)
+
+			now := time.Now()
+			d.lock.Lock()
+
+			//1、删除过期的client
+			for id, node := range d.Client {
+				if now.Unix() > node.CreateTime.Add(time.Minute * 10).Unix() {
+					delete(d.Client, id)
+				}
+			}
+
+			//2、删除过期的server
+			for id, node := range d.Server {
+				if now.Unix() > node.CreateTime.Add(time.Minute * 10).Unix() {
+					delete(d.Server, id)
+				}
+			}
+
+			//3、记录&删除延时请求
+			for id, clientNode := range d.Client {
+				serverNode, ok := d.Server[id]
+				if !ok {
+					continue
+				}
+
+				if clientNode.Points.Duration-serverNode.Points.Duration > int(1*time.Second) {
+					d.save.Info(clientNode.Points, serverNode.Points)
+					delete(d.Client, id)
+					delete(d.Server, id)
+				}
+			}
+
+			d.lock.Unlock()
+		}
+	}()
+}
+
+func (d *Delayed5s) AddNode(point TraceMsg) {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+
+	switch point.Kind {
+	case "SERVER":
+		d.Server[point.Id] = Node{time.Now(), point}
+	case "CLIENT":
+		d.Client[point.Id] = Node{time.Now(), point}
+	default:
+		//位置类型，不处理
+	}
+}

pkg/points/trace_msg.go

@@ -71,8 +71,17 @@ func TraceBaseInfoToPoint(data []byte) ([]*client.Point, error) {
 	return rtn, nil
 }
 
+var d5sObj = NewDelayed5s()
+
+func init() {
+	d5sObj.Run()
+}
+
 //失败返回nil
 func buildTraceInfluxMsg(traceInfo TraceMsg) *client.Point {
+	//临时：统计5s延时问题
+	d5sObj.AddNode(traceInfo)
+
 	if traceInfo.Kind != "SERVER" {
 		return nil
 	}