add alert test api

model/alerting/metric.go

@@ -44,6 +44,7 @@ type MetricItem struct {
 	Field string `json:"field"`
 	Statistic string `json:"statistic"`
 	Group []string `json:"group"` //bucket group
+	Limit int `json:"limit"`
 }
 
 type QueryResult struct {

plugin/api/alerting/api.go

@@ -18,6 +18,7 @@ type AlertAPI struct {
 func (alert *AlertAPI) Init() {
 	api.HandleAPIMethod(api.GET, "/alerting/rule/:rule_id", alert.getRule)
 	api.HandleAPIMethod(api.POST, "/alerting/rule", alert.createRule)
+	api.HandleAPIMethod(api.POST, "/alerting/rule/test", alert.sendTestMessage)
 	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)

plugin/api/alerting/rule.go

@@ -398,6 +398,29 @@ func (alertAPI *AlertAPI) enableRule(w http.ResponseWriter, req *http.Request, p
 		"_id": id,
 	}, http.StatusOK)
 }
+
+func (alertAPI *AlertAPI) sendTestMessage(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+	rule :=  alerting.Rule{}
+	err := alertAPI.DecodeJSON(req, &rule)
+	if err != nil {
+		alertAPI.WriteJSON(w, util.MapStr{
+			"error": err.Error(),
+		}, http.StatusInternalServerError)
+		return
+	}
+	eng := alerting2.GetEngine(rule.Resource.Type)
+	actionResults, err :=  eng.Test(&rule)
+	if err != nil {
+		alertAPI.WriteJSON(w, util.MapStr{
+			"error": err.Error(),
+		}, http.StatusInternalServerError)
+		return
+	}
+	alertAPI.WriteJSON(w, util.MapStr{
+		"action_results": actionResults,
+	}, http.StatusOK)
+
+}
 func checkResourceExists(rule *alerting.Rule) (bool, error) {
 	if rule.Resource.ID == "" {
 		return false, fmt.Errorf("resource id can not be empty")

service/alerting/elasticsearch/engine.go

@@ -67,6 +67,11 @@ func (engine *Engine) GenerateQuery(rule *alerting.Rule) (interface{}, error) {
 	}
 	var rootAggs util.MapStr
 	groups := rule.Metrics.Items[0].Group
+	limit := rule.Metrics.Items[0].Limit
+	//top group 10
+	if limit <= 0 {
+		limit = 10
+	}
 	if grpLength := len(groups); grpLength > 0 {
 		var lastGroupAgg util.MapStr
 
@@ -74,7 +79,7 @@ func (engine *Engine) GenerateQuery(rule *alerting.Rule) (interface{}, error) {
 			groupAgg := util.MapStr{
 				"terms": util.MapStr{
 					"field": groups[i],
-					"size": 500,
+					"size": limit,
 				},
 			}
 			groupID := util.GetUUID()
@@ -634,11 +639,31 @@ func (engine *Engine) Do(rule *alerting.Rule) error {
 	return nil
 }
 
+func (engine *Engine) Test(rule *alerting.Rule) ([]alerting.ActionExecutionResult, error) {
+	checkResults, err := engine.CheckCondition(rule)
+	if err != nil {
+		return nil, fmt.Errorf("check condition error:%w", err)
+	}
+	conditionResults := checkResults.ResultItems
+	var actionResults []alerting.ActionExecutionResult
+	if len(rule.Channels.Normal) > 0 {
+		actionResults = performChannels(rule.Channels.Normal, conditionResults)
+	}else if len(rule.Channels.Escalation) > 0{
+		actionResults = performChannels(rule.Channels.Escalation, conditionResults)
+	}else{
+		return nil, fmt.Errorf("no useable channel")
+	}
+	return actionResults, nil
+}
+
 func performChannels(channels []alerting.Channel, conditionResults []alerting.ConditionResultItem) []alerting.ActionExecutionResult {
 	var message string
 	for _, conditionResult := range conditionResults {
 		message += fmt.Sprintf("severity: %s\t message:%s\t groups:%v\t timestamp: %v;", conditionResult.ConditionItem.Severity, conditionResult.ConditionItem.Message, conditionResult.GroupValues, time.Now())
 	}
+	if message == ""{
+		message = "normal"
+	}
 	ctx := util.MapStr{
 		"ctx": util.MapStr{
 			"message": message,

service/alerting/engine.go

@@ -16,6 +16,7 @@ type Engine interface {
 	ExecuteQuery(rule *alerting.Rule)(*alerting.QueryResult, error)
 	CheckCondition(rule *alerting.Rule)(*alerting.ConditionResult, error)
 	GenerateTask(rule *alerting.Rule) func(ctx context.Context)
+	Test(rule *alerting.Rule) ([]alerting.ActionExecutionResult, error)
 }
 
 var (