# 修复io.Reader导致空body问题

go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a // indirect
 	github.com/robfig/cron v0.0.0-20180505203441-b41be1df6967
 	github.com/timest/env v0.0.0-20180717050204-5fce78d35255
+	golang.org/x/text v0.3.0
 	gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc // indirect
 	gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df
 )

main.go

@@ -3,8 +3,8 @@ package main
 import (
 	"flag"
 	"git.quantgroup.cn/DevOps/enoch/service"
-	"git.quantgroup.cn/DevOps/enoch/service/alarm"
 	"git.quantgroup.cn/DevOps/enoch/service/conf"
+	"git.quantgroup.cn/DevOps/enoch/service/job"
 	"git.quantgroup.cn/DevOps/enoch/service/log"
 	"log"
 	"net/http"
@@ -31,16 +31,16 @@ func main() {
 	//data.RedisPoolInit()
 
 	//初始化alarm
-	alarm.Load()
 
 	//初始化kafka 消费者
-	go service.AgentClusterConsumer(conf.HealthTopic(), service.HealthMessageHandler{})
+	//go service.AgentClusterConsumer(conf.HealthTopic(), service.HealthMessageHandler{})
 	//go service.AgentClusterConsumer(conf.BraveTopic(), service.BraveMessageHandler{})
 	intPort, _ := strconv.Atoi(port)
 
 	if quartz {
 		log.Println("启动定时任务")
 		//job.AutoEmailPerformInfo()
+		job.AutoAlarm()
 	}
 
 	go func() {
@@ -49,10 +49,10 @@ func main() {
 
 	http.HandleFunc("/duration", service.DurationInterface)
 	http.HandleFunc("/counter", service.CounterInterface)
-
 	err := http.ListenAndServe(":"+port, nil)
 	if err != nil {
 		logger.Error.Fatalln("服务启动失败", err)
 	}
 
+
 }

service/alarm/alarm_engine.go

-package alarm
-
-import (
-	"git.quantgroup.cn/DevOps/enoch/service/util"
-	"net/http"
-	"strings"
-)
-
-func Sender(target []string, triggerStrategy string)  {
-
-}
-
-/**
-	加载发送策略
- */
-func LoadSendStrategy()  {
-
-}
-
-func senderMail(info string, receiver string)  {
-	util.SendEmail("接口耗时情况", info, receiver)
-}
-
-func senderDingDing(info string, receiver string)  {
-
-	http.Post(receiver, "application/json", )
-}
-
-/**
-判断是否符合报警的时间间隔
- */
-func isExpired()  {
-
-}
-
-/**
-	持久化报警
- */
-func persistence()  {
-
-}
-
-

service/alarm/load_data.go → service/alarm/load.go

@@ -24,7 +24,6 @@ func Load() {
 	}
 
 	sql := BuildSql(strategies)
-	logger.Info.Println(sql)
 	result := QueryMonitor(sql)
 	DealResult(result, strategies)
 

service/alarm/notice_engine.go

+package alarm
+
+import (
+	"encoding/json"
+	"git.quantgroup.cn/DevOps/enoch/service/log"
+	"git.quantgroup.cn/DevOps/enoch/service/util"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+)
+
+var SenderWatcher sync.Map
+
+func Sender(target []string, title string, info string, notice Notice) {
+	uniqueKey := buildUniqueKey(target)
+	if !isExpired(uniqueKey, notice.Interval) { //小于报警周期
+		return
+	}
+	info = uniqueKey + info
+	switch notice.Sender {
+	case DING_DING:
+		senderDingDing(title, info, notice.Receiver)
+		break
+	case MAIL:
+		senderMail(title, info, notice.Receiver)
+	default:
+		logger.Error.Println("策略配置错误，未匹配到对应的Sender")
+
+	}
+}
+
+func senderMail(title string, info string, receiver []string) {
+	for _, r := range receiver {
+		util.SendEmail(title, info, r)
+	}
+}
+
+func senderDingDing(title string, info string, receiver [] string) {
+	bodyStr := buildDingDingMsg(title, info)
+
+	for _, r := range receiver {
+		data := strings.NewReader(string(bodyStr))
+		_, err := http.Post(r, "application/json;charset=utf-8", data)
+		if err != nil {
+			logger.Error.Println(err)
+		}
+	}
+}
+
+type dinDingMsg struct {
+	MsgType string `json:"msgtype"`
+	Link    link   `json:"link"`
+}
+type link struct {
+	Title      string `json:"title"`
+	Text       string `json:"text"`
+	MessageUrl string `json:"messageUrl"`
+}
+
+func buildDingDingMsg(title string, info string) []byte {
+	msg := dinDingMsg{
+		MsgType: "link",
+		Link: link{
+			Title:      title,
+			Text:       info,
+			MessageUrl: "https://open-doc.dingtalk.com/docs/doc.htm?spm=a219a.7629140.0.0.Rqyvqo&treeId=257&articleId=105735&docType=1",
+		},
+	}
+
+	msgStr, err := json.Marshal(msg)
+
+	if nil != err {
+		logger.Error.Println("无法序列化ding ding msg", err)
+	}
+	logger.Info.Println("buildDingDingMsg", msgStr)
+	return msgStr
+
+}
+
+/**
+判断是否符合报警的时间间隔
+ */
+func isExpired(key string, interval int64) bool {
+	now := time.Now().Unix()
+	lastTime, hasValue := SenderWatcher.LoadOrStore(key, now)
+	logger.Info.Println("---------时间间隔：old:", lastTime, ",new:", now)
+	if hasValue { // 存在旧值，判断是否过期
+		if now-lastTime.(int64) >= interval { //过期
+			return true
+		}
+		//不过期，重新赋周期内最小值
+		SenderWatcher.Store(key, lastTime)
+	}
+	return true
+}
+
+func buildUniqueKey(target []string) string {
+	length := len(target)
+	str := strings.Builder{}
+	for i := length - 1; i >= 0; i-- {
+		str.WriteString(target[i])
+		str.WriteString(":")
+	}
+	return str.String()
+}
+
+/**
+	持久化报警
+ */
+func persistence() {
+
+}

service/alarm/operator.go

@@ -3,16 +3,9 @@ package alarm
 import (
 	"git.quantgroup.cn/DevOps/enoch/service/log"
 	"strconv"
+	"strings"
 )
 
-type Operator interface {
-	Less(alter string, real string)
-	Grater(alter string, real string)
-	Equal(alter string, real string)
-	Between(floor string, ceil string, real string)
-	ComparedWithSame(old string, current string, alter string)
-}
-
 type Compare struct {
 }
 
@@ -47,7 +40,7 @@ func (Compare) Between(floor string, ceil string, real string) bool {
 func (Compare) ComparedWithSame(alter string, old string, current string) bool {
 	cf := parseToFloat(current)
 	of := parseToFloat(old)
-	return (cf - of)/of > parseToFloat(alter)
+	return (cf-of)/of > parseToFloat(alter)
 }
 
 func parseToFloat(value string) float64 {
@@ -57,3 +50,58 @@ func parseToFloat(value string) float64 {
 	}
 	return rs
 }
+
+type MsgBuilder struct {
+}
+
+func (MsgBuilder) Less(alter string, real string) string {
+	str := strings.Builder{}
+	str.WriteString("的值为：")
+	str.WriteString(real)
+	str.WriteString("已经低于阈值：")
+	str.WriteString(alter)
+	return str.String()
+}
+
+/**
+	real > alter
+ */
+func (MsgBuilder) Greater(alter string, real string) string {
+	str := strings.Builder{}
+	str.WriteString("的值为：")
+	str.WriteString(real)
+	str.WriteString("已经高于阈值：")
+	str.WriteString(alter)
+	return str.String()
+}
+
+/**
+	real = alter
+ */
+func (MsgBuilder) Equal(alter string, real string) string {
+	str := strings.Builder{}
+	str.WriteString("的值为：")
+	str.WriteString(real)
+	str.WriteString("已经达到阈值：")
+	str.WriteString(alter)
+	return str.String()
+}
+
+func (MsgBuilder) Between(floor string, ceil string, real string) string {
+	str := strings.Builder{}
+	str.WriteString("的值为：")
+	str.WriteString(real)
+	str.WriteString("已经介于阈值：")
+	str.WriteString(ceil)
+	str.WriteString("~")
+	str.WriteString(floor)
+	str.WriteString("之间")
+	return str.String()
+}
+
+/**
+	同比超过alter
+ */
+func (MsgBuilder) ComparedWithSame(alter string, old string, current string) string {
+	return "同比超过" + alter
+}

service/alarm/query_engine.go

@@ -7,57 +7,22 @@ import (
 	"strings"
 )
 
-func BuildSql(strategies []Strategy) string {
-	var sqlBuff strings.Builder
-	for _, strategy := range strategies {
-		sqlBuff.WriteString(strategy.Sql)
-	}
-	return sqlBuff.String()
-}
-
 var operators = reflect.ValueOf(Compare{})
-
-func needIgnore(ignoreTag map[string]bool, tags []string) bool {
-	for _, k := range tags {
-		if ignoreTag[k] {
-			return true
-		}
-	}
-	return false
-}
-
-func getAlterValue(personalAlterValue map[string][]string, uniqueTag string, tags []string, alterValue []string) []string {
-	value := personalAlterValue[uniqueTag]
-	if nil != value {
-		return value
-	}
-	tagLen := len(tags)
-	for i := 1; i < tagLen; i++ {
-		if nil != personalAlterValue[tags[i]] {
-			return personalAlterValue[tags[i]]
-		}
-	}
-	return alterValue
-}
-
-func getTagValues(keys []string, tag map[string]string) []string {
-	values := make([]string, len(keys))
-	for i, k := range keys {
-		values[i] = tag[k]
-	}
-	return values
-}
+var msgBuilder = reflect.ValueOf(MsgBuilder{})
 
 func DealResult(res []client.Result, strategies []Strategy) {
 	var resIndex = 0
 	for _, strategy := range strategies {
 		logger.Info.Println("-------", strategy.Name, resIndex, "---------")
 		operator := strategy.Operator
-		method := operators.MethodByName(operator)
+		operatorMethod := operators.MethodByName(operator)
+		buildMsgMethod := msgBuilder.MethodByName(operator)
 		alterValue := strategy.AlterValue
 		alterValueLen := len(alterValue)
 		ignoreTag := strategy.IgnoreTag
 		tags := strategy.Tag
+		notice := strategy.Notice
+		noDataAlter := strategy.NoDataAlter
 		personalAlterValue := strategy.PersonalAlterValue
 		if strategy.SqlLen == 1 { //单sql
 			result := res[resIndex]
@@ -68,8 +33,9 @@ func DealResult(res []client.Result, strategies []Strategy) {
 					continue
 				}
 				value := series.Values[0][1]
-				if nil == value { //todo 空值报警
-					logger.Warning.Println(strategy.Name, ":", uniqueTag, ":", "空值")
+				if nil == value && noDataAlter { //空值报警
+
+					Sender(tagValues, strategy.Name, "no data", notice)
 				}
 
 				currentAlterValue := getAlterValue(personalAlterValue, uniqueTag, tagValues, alterValue)
@@ -78,12 +44,17 @@ func DealResult(res []client.Result, strategies []Strategy) {
 					params[j] = reflect.ValueOf(arg)
 				}
 				params[alterValueLen] = reflect.ValueOf(reflect.ValueOf(value).String())
-				logger.Info.Println(uniqueTag, params)
-				rs := method.Call(params)
-				logger.Info.Println(uniqueTag, ": real:", value, ": check", rs[0])
+				logger.Info.Println(uniqueTag, ":", params)
+				rs := operatorMethod.Call(params)
+				logger.Info.Println(uniqueTag, ":", params, rs[0])
+				if rs[0].Bool() { //触发报警策略
+					Sender(tagValues, strategy.Name, buildMsgMethod.Call(params)[0].String(), notice) // 报警
+				}
+
 			}
 		} else {
 			params := make(map[string][]reflect.Value)
+			tagValueMap := make(map[string][]string)
 			for i := 0; i < strategy.SqlLen; i++ {
 				resIndex += i
 				for _, series := range res[resIndex].Series {
@@ -92,41 +63,41 @@ func DealResult(res []client.Result, strategies []Strategy) {
 					}
 					uniqueTag := series.Tags[tags[0]]
 					tagValues := getTagValues(tags, series.Tags)
+					tagValueMap[uniqueTag] = tagValues
 					value := series.Values[0][1]
 
-					if nil == value { //todo 空值报警
-						params[uniqueTag] = nil
-						logger.Warning.Println(strategy.Name, ":", uniqueTag, ":", "空值")
-					}
-
 					if nil == params[uniqueTag] {
 						currentParams := make([]reflect.Value, strategy.SqlLen+alterValueLen)
-
 						currentAlterValue := getAlterValue(personalAlterValue, uniqueTag, tagValues, alterValue)
 						for k, arg := range currentAlterValue {
 							currentParams[k] = reflect.ValueOf(arg)
 						}
-
 						params[uniqueTag] = currentParams
 					}
-					params[uniqueTag][alterValueLen+i] = reflect.ValueOf(reflect.ValueOf(value).String())
+					valueStr := reflect.ValueOf(value).String()
+
+					params[uniqueTag][alterValueLen+i] = reflect.ValueOf(valueStr)
 
 				}
 			}
+
 			for k, v := range params {
-				if nil == v {
+				if nil == v || hasNilValue(v) {
 					continue
 				}
-				logger.Info.Println(k, v)
-				rs := method.Call(v)
-				//todo 结果报警
-				logger.Info.Println(k, ": real:", v, ": check", rs[0])
+
+				logger.Info.Println(k, ":", v)
+				rs := operatorMethod.Call(v)
+				logger.Info.Println(k, ":", v, rs[0])
+				//结果报警
+				if rs[0].Bool() { //触发报警策略
+					Sender(tagValueMap[k], strategy.Name, buildMsgMethod.Call(v)[0].String(), notice) // 报警
+				}
 			}
 		}
 		//循环结果集
 		resIndex += 1
 	}
-	logger.Info.Fatal("结束")
 }
 
 func CheckArray(strategies []Strategy) bool {
@@ -164,3 +135,49 @@ func Check(strategy Strategy) bool {
 
 	return true
 }
+
+func BuildSql(strategies []Strategy) string {
+	var sqlBuff strings.Builder
+	for _, strategy := range strategies {
+		sqlBuff.WriteString(strategy.Sql)
+	}
+	return sqlBuff.String()
+}
+
+func hasNilValue(params []reflect.Value) bool {
+	for _, c := range params {
+		return c.IsValid()
+	}
+	return false
+}
+
+func needIgnore(ignoreTag map[string]bool, tags []string) bool {
+	for _, k := range tags {
+		if ignoreTag[k] {
+			return true
+		}
+	}
+	return false
+}
+
+func getAlterValue(personalAlterValue map[string][]string, uniqueTag string, tags []string, alterValue []string) []string {
+	value := personalAlterValue[uniqueTag]
+	if nil != value {
+		return value
+	}
+	tagLen := len(tags)
+	for i := 1; i < tagLen; i++ {
+		if nil != personalAlterValue[tags[i]] {
+			return personalAlterValue[tags[i]]
+		}
+	}
+	return alterValue
+}
+
+func getTagValues(keys []string, tag map[string]string) []string {
+	values := make([]string, len(keys))
+	for i, k := range keys {
+		values[i] = tag[k]
+	}
+	return values
+}

service/alarm/strategy.go

 package alarm
 
 type StrategyType int8
+type NoticeSender int8
 
 const (
 	SERVICE StrategyType = 1 //监控服务的策略
 	MACHINE StrategyType = 2 //监控机器的策略
 	COMMON  StrategyType = 3 //通用策略
+)
 
+const (
+	DING_DING NoticeSender = 1 //钉钉
+	MAIL NoticeSender = 2 //邮箱
 )
 
+
 type Strategy struct {
-	Name               string              `json:"name"`
-	Sql                string              `json:"sql"`
-	SqlLen             int                 `json:"sql_len"`
-	Type               StrategyType        `json:"type"`
-	Operator           string              `json:"operator"`
-	AlterValue         []string            `json:"alter_value"`
-	Tag                []string            `json:"tag"`
-	IgnoreTag          map[string]bool     `json:"ignore_tag"`
-	PersonalAlterValue map[string][]string `json:"personal_alter_value"`
+	Name               string              `json:"name"` //策略名称
+	Sql                string              `json:"sql"`//sql 每条sql用 ; 分割
+	SqlLen             int                 `json:"sql_len"` //包含的sql条数，
+	Type               StrategyType        `json:"type"` //策略类型
+	Operator           string              `json:"operator"` //结果比较方式
+	AlterValue         []string            `json:"alter_value"` //触发点
+	Tag                []string            `json:"tag"` //sql中需要展示的tag名称,将最细粒度的tag名称放在array[0], sys_name>host
+	IgnoreTag          map[string]bool     `json:"ignore_tag"` //需要忽略的服务、主机
+	PersonalAlterValue map[string][]string `json:"personal_alter_value"` //针对服务、主机个性化触发点配置
+	NoDataAlter bool `json:"no_data_alter"` //空值报警
+	Notice Notice `json:"notice"` //通知方式
+}
+
+type Notice struct {
+	 Sender NoticeSender `json:"sender"` //发送方式 dingDing(1)/mail(2)
+	 Interval int64 `json:"interval"` //相同的(同服务/主机,同策略)报警发送间隔 单位 秒
+	 Receiver []string //接收者，dingDing 则为webHookUrl, mail 则为邮箱
 }
 

service/conf/dev/tsconfig.json

@@ -8,7 +8,9 @@
     "alter_value": [
       "0.8"
     ],
-    "tag": ["sys_name"],
+    "tag": [
+      "sys_name"
+    ],
     "ignore_tag": {
       "msg": true
     },
@@ -19,6 +21,14 @@
       "xyqb": [
         "1"
       ]
+    },
+    "notice": {
+      "sender": 1,
+      "interval": 120,
+      "receiver": [
+        "https://oapi.dingtalk.com/robot/send?access_token=9ffab8e4ae5f94e0fbf84aa91c9cb474d9e3d5bd0bb3c2daffe4cdfe0c2cbbc7",
+        "https://oapi.dingtalk.com/robot/send?access_token=50cab90234d56e4151ef58132714482ce20f2dce791b2a492439fdfa5fc092ca"
+      ]
     }
   },
   {
@@ -30,7 +40,9 @@
     "alter_value": [
       "0.1"
     ],
-    "tag": ["sys_name"],
+    "tag": [
+      "sys_name"
+    ],
     "ignore_tag": {
       "msg": true
     },
@@ -41,6 +53,13 @@
       "xyqb": [
         "1"
       ]
+    },
+    "notice": {
+      "sender": 2,
+      "interval": 120,
+      "receiver": [
+        "zhongxin.men@quantgroup.cn"
+      ]
     }
   },
   {
@@ -52,7 +71,10 @@
     "alter_value": [
       "0.9"
     ],
-    "tag": ["host", "sys_name"],
+    "tag": [
+      "host",
+      "sys_name"
+    ],
     "personal_alter_value": {
       "feature": [
         "0.7"
@@ -60,6 +82,14 @@
       "xyqb": [
         "1"
       ]
+    },
+    "notice": {
+      "sender": 1,
+      "interval": 360,
+      "receiver": [
+        "https://oapi.dingtalk.com/robot/send?access_token=9ffab8e4ae5f94e0fbf84aa91c9cb474d9e3d5bd0bb3c2daffe4cdfe0c2cbbc7",
+        "https://oapi.dingtalk.com/robot/send?access_token=50cab90234d56e4151ef58132714482ce20f2dce791b2a492439fdfa5fc092ca"
+      ]
     }
   },
   {
@@ -71,7 +101,10 @@
     "alter_value": [
       "0.2"
     ],
-    "tag": ["host", "sys_name"],
+    "tag": [
+      "host",
+      "sys_name"
+    ],
     "personal_alter_value": {
       "feature": [
         "0.7"
@@ -79,6 +112,13 @@
       "xyqb": [
         "1"
       ]
+    },
+    "notice": {
+      "sender": 2,
+      "interval": 120,
+      "receiver": [
+        "zhongxin.men@quantgroup.cn"
+      ]
     }
   },
   {
@@ -90,7 +130,10 @@
     "alter_value": [
       "0.2"
     ],
-    "tag": ["host", "sys_name"],
+    "tag": [
+      "host",
+      "sys_name"
+    ],
     "personal_alter_value": {
       "feature": [
         "0.7"
@@ -98,6 +141,13 @@
       "xyqb": [
         "1"
       ]
+    },
+    "notice": {
+      "sender": 2,
+      "interval": 120,
+      "receiver": [
+        "zhongxin.men@quantgroup.cn"
+      ]
     }
   }
 ]

service/job/alarm.go

+package job
+
+import (
+	"fmt"
+	"git.quantgroup.cn/DevOps/enoch/service/alarm"
+	"git.quantgroup.cn/DevOps/enoch/service/log"
+	"github.com/robfig/cron"
+	"time"
+)
+
+/**
+	报警定时任务，每分钟执行一次
+ */
+func AutoAlarm() {
+	c := cron.New()
+
+	err := c.AddFunc("@every 1m", func() {
+		logger.Info.Println("开始执行定时任务", time.Now().Minute())
+		alarm.Load()
+	})
+	if err != nil {
+		fmt.Print("err")
+	}
+
+	c.Start()
+}