add alert rule api

model/alerting/resource.go

@@ -5,11 +5,11 @@
 package alerting
 
 type Resource struct {
-	ID string `json:"id"`
+	ID string `json:"resource_id" elastic_mapping:"resource_id:{type:keyword}"`
-	Type string `json:"type"`
+	Type string `json:"type" elastic_mapping:"type:{type:keyword}"`
 	Objects []string `json:"objects" elastic_mapping:"objects:{type:keyword,copy_to:search_text}"`
 	Filter Filter `json:"filter,omitempty" elastic_mapping:"-"`
 	RawFilter map[string]interface{} `json:"raw_filter,omitempty"`
-	TimeField string `json:"time_field,omitempty"`
+	TimeField string `json:"time_field,omitempty" elastic_mapping:"id:{type:keyword}"`
 	Context Context `json:"context"`
 }

model/alerting/rule.go

@@ -5,12 +5,13 @@
 package alerting
 
 import (
-	"infini.sh/framework/core/orm"
 	"time"
 )
 
 type Rule struct {
-	orm.ORMObjectBase
+	ID      string    `json:"id,omitempty"      elastic_meta:"_id" elastic_mapping:"id: { type: keyword }"`
+	Created time.Time `json:"created,omitempty" elastic_mapping:"created: { type: date }"`
+	Updated time.Time `json:"updated,omitempty" elastic_mapping:"updated: { type: date }"`
 	//Name string `json:"name" elastic_mapping:"name:{type:keyword,copy_to:search_text}"`
 	Enabled bool `json:"enabled" elastic_mapping:"enabled:{type:keyword}"`
 	Resource Resource `json:"resource" elastic_mapping:"resource:{type:object}"`

plugin/api/alerting/api.go

@@ -13,7 +13,10 @@ type AlertAPI struct {
 
 func init() {
 	alert:=AlertAPI{}
-	api.HandleAPIMethod(api.GET, "/elasticsearch/:id/alerting/rule/:rule_id", alert.getRule)
+	api.HandleAPIMethod(api.GET, "/alerting/rule/:rule_id", alert.getRule)
-	api.HandleAPIMethod(api.POST, "/elasticsearch/:id/alerting/rule", alert.createRule)
+	api.HandleAPIMethod(api.POST, "/alerting/rule", alert.createRule)
+	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

@@ -5,15 +5,17 @@
 package alerting
 
 import (
-	log "github.com/cihub/seelog"
+	"fmt"
 	"infini.sh/console/model/alerting"
 	alerting2 "infini.sh/console/service/alerting"
-	"infini.sh/console/service/alerting/elasticsearch"
+	_ "infini.sh/console/service/alerting/elasticsearch"
 	httprouter "infini.sh/framework/core/api/router"
+	"infini.sh/framework/core/elastic"
 	"infini.sh/framework/core/orm"
 	"infini.sh/framework/core/task"
 	"infini.sh/framework/core/util"
 	"net/http"
+	log "src/github.com/cihub/seelog"
 	"time"
 )
 
@@ -28,6 +30,13 @@ func (alertAPI *AlertAPI) createRule(w http.ResponseWriter, req *http.Request, p
 	}
 	var ids []string
 	for _, rule := range rules {
+		exists, err := checkResourceExists(&rule)
+		if err != nil || !exists {
+			alertAPI.WriteJSON(w, util.MapStr{
+				"error": err.Error(),
+			}, http.StatusInternalServerError)
+			return
+		}
 		err = rule.Metrics.RefreshExpression()
 		if err != nil {
 			alertAPI.WriteJSON(w, util.MapStr{
@@ -71,86 +80,177 @@ func (alertAPI *AlertAPI) createRule(w http.ResponseWriter, req *http.Request, p
 	}, http.StatusOK)
 }
 func (alertAPI *AlertAPI) getRule(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
-	rule := alerting.Rule{
+	id := ps.MustGetParameter("rule_id")
-		ORMObjectBase: orm.ORMObjectBase{
-			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{
-						//{
-						//	"term": util.MapStr{
-						//		"metadata.labels.cluster_id": util.MapStr{
-						//			"value": "xxx",
-						//		},
-						//	},
-						//},
-					},
-				},
-			},
-		},
 
-		Metrics: alerting.Metric{
+	obj := alerting.Rule{}
-			PeriodInterval: "1m",
+	obj.ID = id
-			MaxPeriods:     15,
-			Items: []alerting.MetricItem{
-				{Name: "a", Field: "payload.elasticsearch.node_stats.fs.total.free_in_bytes", Statistic: "min", Group: []string{"metadata.labels.cluster_id", "metadata.labels.node_id"}},
-				{Name: "b", Field: "payload.elasticsearch.node_stats.fs.total.total_in_bytes", Statistic: "max",Group: []string{"metadata.labels.cluster_id", "metadata.labels.node_id"}},
-			},
-			Formula: "a/b*100",
-			//Expression: "min(fs.free_in_bytes)/max(fs.total_in_bytes)*100",
-		},
-		Conditions: alerting.Condition{
-			Operator: "any",
-			Items: []alerting.ConditionItem{
-				{MinimumPeriodMatch: 10, Operator: "lte", Values: []string{"76"}, Severity: "warning", Message: "磁盘可用率小于20%"},
-				{MinimumPeriodMatch: 1, Operator: "lte", Values: []string{"75"}, Severity: "error", Message: "磁盘可用率小于10%"},
-			},
-		},
 
-		Channels: alerting.RuleChannel{
+	exists, err := orm.Get(&obj)
-			Normal: []alerting.Channel{
+	if !exists || err != nil {
-				{Name: "钉钉", Type: alerting.ChannelWebhook, Webhook: &alerting.CustomWebhook{
+		alertAPI.WriteJSON(w, util.MapStr{
-					HeaderParams: map[string]string{
+			"_id":   id,
-						"Content-Type": "application/json",
+			"found": false,
-					},
+		}, http.StatusNotFound)
-					Body:   `{"msgtype": "text","text": {"content":"告警通知: {{ctx.message}}"}}`,
+		return
-					Method: http.MethodPost,
-					URL:    "https://oapi.dingtalk.com/robot/send?access_token=XXXXXX",
-				}},
-			},
-			Escalation: []alerting.Channel{
-				{Type: alerting.ChannelWebhook, Name: "微信", Webhook: &alerting.CustomWebhook{
-					HeaderParams: map[string]string{
-						"Content-Type": "application/json",
-					},
-					Body:   `{"msgtype": "text","text": {"content":"告警通知: {{ctx.message}}"}}`,
-					Method: http.MethodPost,
-					URL:    "https://oapi.weixin.com/robot/send?access_token=XXXXXX",
-				}},
-			},
-			ThrottlePeriod: "1h",
-			AcceptTimeRange: alerting.TimeRange{
-				Start: "8:00",
-				End: "21:00",
-			},
-			EscalationEnabled: true,
-			EscalationThrottlePeriod: "30m",
-		},
 	}
-	eng := &elasticsearch.Engine{}
-	result, err := eng.ExecuteQuery(&rule)
 	if err != nil {
+		alertAPI.WriteError(w, err.Error(), http.StatusInternalServerError)
 		log.Error(err)
+		return
 	}
-	alertAPI.WriteJSON(w, result, http.StatusOK)
+
+	alertAPI.WriteJSON(w, util.MapStr{
+		"found":   true,
+		"_id":     id,
+		"_source": obj,
+	}, 200)
 
 }
+
+func (alertAPI *AlertAPI) updateRule(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+	id := ps.MustGetParameter("rule_id")
+	obj := alerting.Rule{}
+
+	obj.ID = id
+	exists, err := orm.Get(&obj)
+	if !exists || err != nil {
+		alertAPI.WriteJSON(w, util.MapStr{
+			"_id":    id,
+			"result": "not_found",
+		}, http.StatusNotFound)
+		return
+	}
+
+	id = obj.ID
+	create := obj.Created
+	obj = alerting.Rule{}
+	err = alertAPI.DecodeJSON(req, &obj)
+	if err != nil {
+		alertAPI.WriteError(w, err.Error(), http.StatusInternalServerError)
+		log.Error(err)
+		return
+	}
+
+	//protect
+	obj.ID = id
+	obj.Created = create
+	obj.Updated = time.Now()
+	err = orm.Update(&obj)
+	if err != nil {
+		alertAPI.WriteError(w, err.Error(), http.StatusInternalServerError)
+		log.Error(err)
+		return
+	}
+
+	if obj.Enabled {
+		//update task
+		task.StopTask(id)
+		eng := alerting2.GetEngine(obj.Resource.Type)
+		ruleTask := task.ScheduleTask{
+			ID:          obj.ID,
+			Interval:    obj.Schedule.Interval,
+			Description: obj.Metrics.Expression,
+			Task:        eng.GenerateTask(&obj),
+		}
+		task.RegisterScheduleTask(ruleTask)
+		task.StartTask(ruleTask.ID)
+	}else{
+		task.DeleteTask(id)
+	}
+
+	alertAPI.WriteJSON(w, util.MapStr{
+		"_id":    obj.ID,
+		"result": "updated",
+	}, 200)
+}
+
+func (alertAPI *AlertAPI) deleteRule(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+	id := ps.MustGetParameter("rule_id")
+
+	obj := alerting.Rule{}
+	obj.ID = id
+
+	exists, err := orm.Get(&obj)
+	if !exists || err != nil {
+		alertAPI.WriteJSON(w, util.MapStr{
+			"_id":    id,
+			"result": "not_found",
+		}, http.StatusNotFound)
+		return
+	}
+
+	err = orm.Delete(&obj)
+	if err != nil {
+		alertAPI.WriteError(w, err.Error(), http.StatusInternalServerError)
+		log.Error(err)
+		return
+	}
+	task.DeleteTask(obj.ID)
+
+	alertAPI.WriteJSON(w, util.MapStr{
+		"_id":    obj.ID,
+		"result": "deleted",
+	}, 200)
+}
+
+func (alertAPI *AlertAPI) searchRule(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+	var (
+		keyword = alertAPI.GetParameterOrDefault(req, "keyword", "")
+		from = alertAPI.GetIntOrDefault(req, "from", 0)
+		size = alertAPI.GetIntOrDefault(req, "size", 20)
+	)
+
+	mustQuery := []util.MapStr{
+	}
+	if keyword != "" {
+		mustQuery = append(mustQuery, util.MapStr{
+			"match": util.MapStr{
+				"search_text": util.MapStr{
+					"query":                keyword,
+					"fuzziness":            "AUTO",
+					"max_expansions":       10,
+					"prefix_length":        2,
+					"fuzzy_transpositions": true,
+					"boost":                50,
+				},
+			},
+		})
+	}
+	queryDSL := util.MapStr{
+		"from": from,
+		"size": size,
+		"query": util.MapStr{
+			"bool": util.MapStr{
+				"must": mustQuery,
+			},
+		},
+	}
+	q := &orm.Query{
+		RawQuery: util.MustToJSONBytes(queryDSL),
+	}
+	err, searchResult := orm.Search(alerting.Rule{}, q)
+	if err != nil {
+		alertAPI.WriteError(w, err.Error(), http.StatusInternalServerError)
+		log.Error(err)
+		return
+	}
+	w.Write(searchResult.Raw)
+}
+
+func checkResourceExists(rule *alerting.Rule) (bool, error) {
+	if rule.Resource.ID == "" {
+		return false, fmt.Errorf("resource id can not be empty")
+	}
+	switch rule.Resource.Type {
+	case "elasticsearch":
+		obj := elastic.ElasticsearchConfig{}
+		obj.ID = rule.Resource.ID
+		ok, err := orm.Get(&obj)
+		if err != nil {
+			return false, err
+		}
+		return ok && obj.Name != "", nil
+	default:
+		return false, fmt.Errorf("unsupport resource type: %s", rule.Resource.Type)
+	}
+}