update alert api

plugin/api/alerting/api.go

@@ -18,5 +18,14 @@ func init() {
 	api.HandleAPIMethod(api.DELETE, "/alerting/rule/:rule_id", alert.deleteRule)
 	api.HandleAPIMethod(api.PUT, "/alerting/rule/:rule_id", alert.updateRule)
 	api.HandleAPIMethod(api.GET, "/alerting/rule/_search", alert.searchRule)
+
+	api.HandleAPIMethod(api.GET, "/alerting/channel/:channel_id", alert.getChannel)
+	api.HandleAPIMethod(api.POST, "/alerting/channel", alert.createChannel)
+	api.HandleAPIMethod(api.DELETE, "/alerting/channel/:channel_id", alert.deleteChannel)
+	api.HandleAPIMethod(api.PUT, "/alerting/channel/:channel_id", alert.updateChannel)
+	api.HandleAPIMethod(api.GET, "/alerting/channel/_search", alert.searchChannel)
+
+	//just for test
+	//api.HandleAPIMethod(api.GET, "/alerting/rule/test", alert.testRule)
 }
 

plugin/api/alerting/channel.go

@@ -0,0 +1,169 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * web: https://infinilabs.com
+ * mail: hello#infini.ltd */
+
+package alerting
+
+import (
+	"fmt"
+	"infini.sh/console/model/alerting"
+	httprouter "infini.sh/framework/core/api/router"
+	"infini.sh/framework/core/orm"
+	"infini.sh/framework/core/util"
+	"net/http"
+	log "src/github.com/cihub/seelog"
+	"strconv"
+	"strings"
+)
+
+func (h *AlertAPI) createChannel(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+	var obj = &alerting.Channel{}
+	err := h.DecodeJSON(req, obj)
+	if err != nil {
+		h.WriteError(w, err.Error(), http.StatusInternalServerError)
+		log.Error(err)
+		return
+	}
+
+	err = orm.Create(obj)
+	if err != nil {
+		h.WriteError(w, err.Error(), http.StatusInternalServerError)
+		log.Error(err)
+		return
+	}
+
+	h.WriteJSON(w, util.MapStr{
+		"_id":    obj.ID,
+		"result": "created",
+	}, 200)
+
+}
+
+func (h *AlertAPI) getChannel(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+	id := ps.MustGetParameter("channel_id")
+
+	obj := alerting.Channel{}
+	obj.ID = id
+
+	exists, err := orm.Get(&obj)
+	if !exists || err != nil {
+		h.WriteJSON(w, util.MapStr{
+			"_id":   id,
+			"found": false,
+		}, http.StatusNotFound)
+		return
+	}
+	if err != nil {
+		h.WriteError(w, err.Error(), http.StatusInternalServerError)
+		log.Error(err)
+		return
+	}
+
+	h.WriteJSON(w, util.MapStr{
+		"found":   true,
+		"_id":     id,
+		"_source": obj,
+	}, 200)
+}
+
+func (h *AlertAPI) updateChannel(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+	id := ps.MustGetParameter("channel_id")
+	obj := alerting.Channel{}
+
+	obj.ID = id
+	exists, err := orm.Get(&obj)
+	if !exists || err != nil {
+		h.WriteJSON(w, util.MapStr{
+			"_id":    id,
+			"result": "not_found",
+		}, http.StatusNotFound)
+		return
+	}
+
+	id = obj.ID
+	create := obj.Created
+	obj = alerting.Channel{}
+	err = h.DecodeJSON(req, &obj)
+	if err != nil {
+		h.WriteError(w, err.Error(), http.StatusInternalServerError)
+		log.Error(err)
+		return
+	}
+
+	//protect
+	obj.ID = id
+	obj.Created = create
+	err = orm.Update(&obj)
+	if err != nil {
+		h.WriteError(w, err.Error(), http.StatusInternalServerError)
+		log.Error(err)
+		return
+	}
+
+	h.WriteJSON(w, util.MapStr{
+		"_id":    obj.ID,
+		"result": "updated",
+	}, 200)
+}
+
+func (h *AlertAPI) deleteChannel(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+	id := ps.MustGetParameter("channel_id")
+
+	obj := alerting.Channel{}
+	obj.ID = id
+
+	exists, err := orm.Get(&obj)
+	if !exists || err != nil {
+		h.WriteJSON(w, util.MapStr{
+			"_id":    id,
+			"result": "not_found",
+		}, http.StatusNotFound)
+		return
+	}
+
+	err = orm.Delete(&obj)
+	if err != nil {
+		h.WriteError(w, err.Error(), http.StatusInternalServerError)
+		log.Error(err)
+		return
+	}
+
+	h.WriteJSON(w, util.MapStr{
+		"_id":    obj.ID,
+		"result": "deleted",
+	}, 200)
+}
+
+func (h *AlertAPI) searchChannel(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+
+	var (
+		keyword        = h.GetParameterOrDefault(req, "keyword", "")
+		queryDSL    = `{"query":{"bool":{"must":[%s]}}, "size": %d, "from": %d}`
+		strSize     = h.GetParameterOrDefault(req, "size", "20")
+		strFrom     = h.GetParameterOrDefault(req, "from", "0")
+		mustBuilder = &strings.Builder{}
+	)
+	if keyword != "" {
+		mustBuilder.WriteString(fmt.Sprintf(`{"query_string":{"default_field":"*","query": "%s"}}`, keyword))
+	}
+	size, _ := strconv.Atoi(strSize)
+	if size <= 0 {
+		size = 20
+	}
+	from, _ := strconv.Atoi(strFrom)
+	if from < 0 {
+		from = 0
+	}
+
+	q := orm.Query{}
+	queryDSL = fmt.Sprintf(queryDSL, mustBuilder.String(), size, from)
+	q.RawQuery = []byte(queryDSL)
+
+	err, res := orm.Search(&alerting.Channel{}, &q)
+	if err != nil {
+		log.Error(err)
+		h.WriteError(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	h.Write(w, res.Raw)
+}

plugin/api/alerting/rule.go

@@ -234,6 +234,7 @@ func (alertAPI *AlertAPI) searchRule(w http.ResponseWriter, req *http.Request, p
 		log.Error(err)
 		return
 	}
+
 	w.Write(searchResult.Raw)
 }
 
@@ -254,3 +255,63 @@ func checkResourceExists(rule *alerting.Rule) (bool, error) {
 		return false, fmt.Errorf("unsupport resource type: %s", rule.Resource.Type)
 	}
 }
+
+//func (alertAPI *AlertAPI) testRule(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+//	rule := alerting.Rule{
+//		ID: util.GetUUID(),
+//		Created: time.Now(),
+//		Updated: time.Now(),
+//		Enabled: true,
+//		Resource: alerting.Resource{
+//			ID: "c8i18llath2blrusdjng",
+//			Type: "elasticsearch",
+//			Objects: []string{".infini_metrics*"},
+//			TimeField: "timestamp",
+//			RawFilter: map[string]interface{}{
+//				"bool": util.MapStr{
+//					"must": []util.MapStr{},
+//				},
+//			},
+//		},
+//
+//		Metrics: alerting.Metric{
+//			PeriodInterval: "1m",
+//			MaxPeriods:     15,
+//			Items: []alerting.MetricItem{
+//				{Name: "a", Field: "payload.elasticsearch.node_stats.os.cpu.percent", Statistic: "p99", Group: []string{"metadata.labels.cluster_id", "metadata.labels.node_id"}},
+//			},
+//		},
+//		Conditions: alerting.Condition{
+//			Operator: "any",
+//			Items: []alerting.ConditionItem{
+//				{MinimumPeriodMatch: 5, Operator: "gte", Values: []string{"90"}, Severity: "error", Message: "cpu使用率大于90%"},
+//			},
+//		},
+//
+//		Channels: alerting.RuleChannel{
+//			Normal: []alerting.Channel{
+//				{Name: "钉钉", Type: alerting.ChannelWebhook, Webhook: &alerting.CustomWebhook{
+//					HeaderParams: map[string]string{
+//						"Content-Type": "application/json",
+//					},
+//					Body:   `{"msgtype": "text","text": {"content":"告警通知: {{ctx.message}}"}}`,
+//					Method: http.MethodPost,
+//					URL:    "https://oapi.dingtalk.com/robot/send?access_token=XXXXXX",
+//				}},
+//			},
+//			ThrottlePeriod: "1h",
+//			AcceptTimeRange: alerting.TimeRange{
+//				Start: "8:00",
+//				End: "21:00",
+//			},
+//			EscalationEnabled: true,
+//			EscalationThrottlePeriod: "30m",
+//		},
+//	}
+//	eng := alerting2.GetEngine(rule.Resource.Type)
+//	data, err := eng.ExecuteQuery(&rule)
+//	if err != nil {
+//		log.Error(err)
+//	}
+//	alertAPI.WriteJSON(w, data, http.StatusOK)
+//}

service/alerting/elasticsearch/engine.go

@@ -19,6 +19,7 @@ import (
 	"infini.sh/framework/core/util"
 	"io"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -38,11 +39,7 @@ func (engine *Engine) GenerateQuery(rule *alerting.Rule) (interface{}, error) {
 	}
 	basicAggs := util.MapStr{}
 	for _, metricItem  := range rule.Metrics.Items {
-		basicAggs[metricItem.Name] = util.MapStr{
+		basicAggs[metricItem.Name] = engine.generateAgg(&metricItem)
-			metricItem.Statistic: util.MapStr{
-				"field": metricItem.Field,
-			},
-		}
 	}
 	timeAggs := util.MapStr{
 		"date_histogram": util.MapStr{
@@ -92,6 +89,37 @@ func (engine *Engine) GenerateQuery(rule *alerting.Rule) (interface{}, error) {
 		"aggs": rootAggs,
 	}, nil
 }
+func (engine *Engine) generateAgg(metricItem *alerting.MetricItem) interface{}{
+	var (
+		aggType = "value_count"
+		field = metricItem.Field
+	)
+	if field == "" || field == "*" {
+		field = "_id"
+	}
+	var percent = 0.0
+	switch metricItem.Statistic {
+	case "max", "min", "sum", "avg":
+		aggType = metricItem.Statistic
+	case "count", "value_count":
+		aggType = "value_count"
+	case "medium":
+		aggType = "median_absolute_deviation"
+	case "p99", "p95","p90","p80","p50":
+		aggType = "percentiles"
+		percentStr := strings.TrimPrefix(metricItem.Statistic, "p")
+		percent, _ = strconv.ParseFloat(percentStr, 32)
+	}
+	aggValue := util.MapStr{
+		"field": field,
+	}
+	if aggType == "percentiles" {
+		aggValue["percents"] = []interface{}{percent}
+	}
+	return util.MapStr{
+		aggType: aggValue,
+	}
+}
 
 func (engine *Engine) GenerateRawFilter(rule *alerting.Rule) (map[string]interface{}, error) {
 	query := map[string]interface{}{}
@@ -141,9 +169,15 @@ func (engine *Engine) GenerateRawFilter(rule *alerting.Rule) (map[string]interfa
 			}
 		}
 	}else{
-		query["bool"] = util.MapStr{
+		must := []interface{}{
-			"must": []interface{}{
 			timeQuery,
+		}
+		if _, ok := query["match_all"]; !ok {
+			must = append(must, query)
+		}
+		query = util.MapStr{
+			"bool":  util.MapStr{
+				"must": must,
 			},
 		}
 	}
@@ -465,7 +499,27 @@ func collectMetricData(agg interface{}, groupValues string, metricData *[]alerti
 								continue
 							}
 							if vm, ok := v.(map[string]interface{}); ok {
-								md.Data[k] = append(md.Data[k], alerting.TimeMetricData{bkM["key"], vm["value"]})
+								if metricVal, ok := vm["value"]; ok {
+									md.Data[k] = append(md.Data[k], alerting.TimeMetricData{bkM["key"], metricVal})
+								}else{
+									//percentiles agg type
+									switch  vm["values"].(type) {
+									case []interface{}:
+										for _, val := range vm["values"].([]interface{}) {
+											if valM, ok := val.(map[string]interface{}); ok {
+												md.Data[k] = append(md.Data[k], alerting.TimeMetricData{bkM["key"], valM["value"]})
+											}
+											break
+										}
+									case map[string]interface{}:
+										for _, val := range vm["values"].(map[string]interface{}) {
+												md.Data[k] = append(md.Data[k], alerting.TimeMetricData{bkM["key"], val})
+											break
+										}
+									}
+
+								}
+
 							}
 
 						}

service/alerting/elasticsearch/engine_test.go

@@ -7,7 +7,6 @@ package elasticsearch
 import (
 	"fmt"
 	"infini.sh/console/model/alerting"
-	"infini.sh/framework/core/orm"
 	"infini.sh/framework/core/util"
 	"net/http"
 	"sort"
@@ -17,11 +16,9 @@ import (
 
 func TestEngine( t *testing.T)  {
 	rule := alerting.Rule{
-		ORMObjectBase: orm.ORMObjectBase{
 		ID: util.GetUUID(),
 		Created: time.Now(),
 		Updated: time.Now(),
-		},
 		Enabled: true,
 		Resource: alerting.Resource{
 			ID: "c8i18llath2blrusdjng",
@@ -117,3 +114,73 @@ func TestEngine( t *testing.T)  {
 
 	//fmt.Println(util.MustToJSON(filter))
 }
+
+func TestGenerateAgg(t *testing.T)  {
+	eng := &Engine{}
+	agg := eng.generateAgg(&alerting.MetricItem{
+		Name: "a",
+		Field: "cpu.percent",
+		Statistic: "p99",
+	})
+	fmt.Println(util.MustToJSON(agg))
+}
+
+func TestGeneratePercentilesAggQuery(t *testing.T) {
+	rule := alerting.Rule{
+		ID: util.GetUUID(),
+		Created: time.Now(),
+		Updated: time.Now(),
+		Enabled: true,
+		Resource: alerting.Resource{
+			ID: "c8i18llath2blrusdjng",
+			Type: "elasticsearch",
+			Objects: []string{".infini_metrics*"},
+			TimeField: "timestamp",
+			RawFilter: map[string]interface{}{
+				"match_all": util.MapStr{
+
+				},
+			},
+		},
+
+		Metrics: alerting.Metric{
+			PeriodInterval: "1m",
+			MaxPeriods:     15,
+			Items: []alerting.MetricItem{
+				{Name: "a", Field: "payload.elasticsearch.node_stats.os.cpu.percent", Statistic: "p99", Group: []string{"metadata.labels.cluster_id", "metadata.labels.node_id"}},
+			},
+		},
+		Conditions: alerting.Condition{
+			Operator: "any",
+			Items: []alerting.ConditionItem{
+				{MinimumPeriodMatch: 5, Operator: "gte", Values: []string{"90"}, Severity: "error", Message: "cpu使用率大于90%"},
+			},
+		},
+
+		Channels: alerting.RuleChannel{
+			Normal: []alerting.Channel{
+				{Name: "钉钉", Type: alerting.ChannelWebhook, Webhook: &alerting.CustomWebhook{
+					HeaderParams: map[string]string{
+						"Content-Type": "application/json",
+					},
+					Body:   `{"msgtype": "text","text": {"content":"告警通知: {{ctx.message}}"}}`,
+					Method: http.MethodPost,
+					URL:    "https://oapi.dingtalk.com/robot/send?access_token=XXXXXX",
+				}},
+			},
+			ThrottlePeriod: "1h",
+			AcceptTimeRange: alerting.TimeRange{
+				Start: "8:00",
+				End: "21:00",
+			},
+			EscalationEnabled: true,
+			EscalationThrottlePeriod: "30m",
+		},
+	}
+	eng := &Engine{}
+	q, err := eng.GenerateQuery(&rule)
+	if err != nil {
+		t.Fatal(err)
+	}
+	fmt.Println(util.MustToJSON(q))
+}