Monitor worker alert (#22729)

* fix(monitor): GetMetricMeasurement 接口只查询过去30m的数据

避免数据过多导致 victoria-metrics 报错

* fix(monitor): wildchar of alert group by field

pkg/apis/monitor/metric.go

@@ -14,7 +14,11 @@
 
 package monitor
 
-import "yunion.io/x/onecloud/pkg/apis"
+import (
+	"yunion.io/x/pkg/util/sets"
+
+	"yunion.io/x/onecloud/pkg/apis"
+)
 
 const EXT_PREFIX = "ext"
 
@@ -59,6 +63,19 @@ const (
 )
 
 var (
+	MetricCloudResTypes = sets.NewString(
+		METRIC_RES_TYPE_HOST,
+		METRIC_RES_TYPE_AGENT,
+		METRIC_RES_TYPE_GUEST,
+		METRIC_RES_TYPE_CONTAINER,
+		METRIC_RES_TYPE_OSS,
+		METRIC_RES_TYPE_RDS,
+		METRIC_RES_TYPE_REDIS,
+		METRIC_RES_TYPE_TENANT,
+		METRIC_RES_TYPE_DOMAIN,
+		METRIC_RES_TYPE_STORAGE,
+		METRIC_RES_TYPE_CLOUDACCOUNT)
+
 	MetricResType = []string{METRIC_RES_TYPE_GUEST, METRIC_RES_TYPE_HOST, METRIC_RES_TYPE_REDIS, METRIC_RES_TYPE_OSS,
 		METRIC_RES_TYPE_RDS, METRIC_RES_TYPE_CLOUDACCOUNT}
 	MetricUnit = []string{METRIC_UNIT_PERCENT, METRIC_UNIT_BPS, METRIC_UNIT_MBPS, METRIC_UNIT_BYTEPS, "count/s",

pkg/cloudcommon/notifyclient/notify_internal.go

@@ -182,13 +182,18 @@ func lang(ctx context.Context, contactType npk.TNotifyChannel, reIds []string, c
 	return langMap, nil
 }
 
-func isEmptyRecipients(recipientId []string) bool {
+func isEmptyRecipients(recipientId, robots []string) bool {
 	var recvs []string
 	for _, c := range recipientId {
 		if len(c) > 0 {
 			recvs = append(recvs, c)
 		}
 	}
+	for _, robot := range robots {
+		if len(robot) > 0 {
+			recvs = append(recvs, robot)
+		}
+	}
 	return len(recvs) == 0
 }
 
@@ -216,7 +221,7 @@ func genMsgViaLang(ctx context.Context, p sNotifyParams) ([]npk.SNotifyMessage,
 		reIds = p.recipientId
 	}
 
-	if isEmptyRecipients(p.recipientId) {
+	if isEmptyRecipients(p.recipientId, p.robots) {
 		return nil, errors.Wrap(errors.ErrEmpty, "empty receipients")
 	}
 
@@ -414,10 +419,10 @@ func (t *notifyTask) Run() {
 }
 
 func intelliNotify(ctx context.Context, p sNotifyParams) {
-	if isEmptyRecipients(p.recipientId) {
+	if isEmptyRecipients(p.recipientId, p.robots) {
 		return
 	}
-	log.Infof("recipientId: %v, contacts: %v, event %s priority %s", p.recipientId, p.contacts, p.event, p.priority)
+	log.Infof("recipientId: %v, robots: %v, contacts: %v, event %s priority %s", p.recipientId, p.robots, p.contacts, p.event, p.priority)
 	msgs, err := genMsgViaLang(ctx, p)
 	if err != nil {
 		log.Errorf("unable send notification: %v", err)

pkg/monitor/alerting/conditions/query.go

@@ -103,6 +103,14 @@ func (c *QueryCondition) GenerateFormatCond(meta *monitor.QueryResultMeta, metri
 		Evaluator:    c.Evaluator,
 	}
 }
+
+func (c *QueryCondition) IsCloudResource() bool {
+	if c.ResType == "" {
+		return false
+	}
+	return monitor.MetricCloudResTypes.Has(c.ResType)
+}
+
 func (c FormatCond) String() string {
 	if c.QueryMeta != nil {
 		return fmt.Sprintf("%s(%q) %s", c.Reducer, c.QueryMeta.RawQuery, c.Evaluator.String())
@@ -172,7 +180,7 @@ func (c *QueryCondition) Eval(context *alerting.EvalContext) (*alerting.Conditio
 	}
 
 	for _, series := range seriesList {
-		if len(c.ResType) != 0 {
+		if c.IsCloudResource() {
 			isLatestOfSerie, resource := c.serieIsLatestResource(nil, series)
 			if !isLatestOfSerie {
 				continue
@@ -542,7 +550,11 @@ func (c *QueryCondition) checkGroupByField() {
 	}
 	for i, group := range c.Query.Model.GroupBy {
 		if group.Params[0] == "*" {
-			c.Query.Model.GroupBy[i].Params = []string{monitor.GetMeasurementTagIdKeyByResType(metricMeasurement.ResType)}
+			tagId := monitor.GetMeasurementTagIdKeyByResType(metricMeasurement.ResType)
+			if tagId == "" {
+				tagId = "*"
+			}
+			c.Query.Model.GroupBy[i].Params = []string{tagId}
 		}
 	}
 }

pkg/monitor/alerting/eval_context.go

@@ -17,6 +17,7 @@ package alerting
 import (
 	"context"
 	"fmt"
+	"sort"
 	"strings"
 	"time"
 
@@ -269,14 +270,30 @@ func (c *EvalContext) GetEvalMatches() []monitor.EvalMatch {
 	return ret
 }
 
+func (c *EvalContext) getTagsDesc(tags map[string]string) string {
+	strs := make([]string, 0)
+	for k, v := range tags {
+		if v == "" {
+			continue
+		}
+		strs = append(strs, k+"="+v)
+	}
+	sort.Strings(strs)
+	ret := strings.Join(strs, ",")
+	return "{" + ret + "}"
+}
+
 func (c *EvalContext) GetResourceNameOfMatches(matches []*monitor.EvalMatch) string {
 	names := strings.Builder{}
+	names.WriteString("\n")
 	for i, match := range matches {
-		if name, ok := match.Tags["name"]; ok {
-			names.WriteString(fmt.Sprintf("%s.%s(%s)", name, match.Metric, match.ValueStr))
-			if i < len(matches)-1 {
-				names.WriteString(", ")
-			}
+		if name, ok := match.Tags["name"]; ok && name != "" {
+			names.WriteString(fmt.Sprintf("- %s.%s: %s", name, match.Metric, match.ValueStr))
+		} else {
+			names.WriteString(fmt.Sprintf("- %s%s: %s", match.Metric, c.getTagsDesc(match.Tags), match.ValueStr))
+		}
+		if i < len(matches)-1 {
+			names.WriteString("\n")
 		}
 	}
 	return names.String()

pkg/monitor/models/datasource.go

@@ -474,6 +474,10 @@ func (m *SDataSourceManager) GetMetricMeasurement(userCred mcclient.TokenCredent
 	output.TagValue = make(map[string][]string, 0)
 
 	output.FieldKey = []string{field}
+	// 只查询过去 30m 的指标
+	if timeF.To == "now" {
+		timeF.From = "30m"
+	}
 	if err := getTagValues(userCred, output, timeF, tagFilter, true); err != nil {
 		return jsonutils.JSONNull, errors.Wrap(err, "getTagValues error")
 	}

pkg/monitor/models/unifiedmonitor.go

@@ -382,6 +382,11 @@ func (self *SUnifiedMonitorManager) ValidateInputQuery(query *monitor.AlertQuery
 	}
 	if input.Interval == "" {
 		input.Interval = "5m"
+		if input.To == "now" {
+			if input.From == "10m" {
+				input.Interval = "1m"
+			}
+		}
 	}
 
 	if query.From == "" {

pkg/monitor/tsdb/driver/victoriametrics/vm.go

@@ -346,7 +346,19 @@ func (vm *vmAdapter) FillSelect(query *monitor.AlertQuery, isAlert bool) *monito
 
 func (vm *vmAdapter) FillGroupBy(query *monitor.AlertQuery, inputQuery *monitor.MetricQueryInput, tagId string, isAlert bool) *monitor.AlertQuery {
 	if isAlert {
-		query = influxdb.FillGroupByWithWildChar(query, inputQuery, tagId)
+		query = FillGroupByWithWildChar(query, inputQuery, tagId)
 	}
 	return query
 }
+
+func FillGroupByWithWildChar(query *monitor.AlertQuery, inputQuery *monitor.MetricQueryInput, tagId string) *monitor.AlertQuery {
+	if len(tagId) == 0 {
+		tagId = "*"
+	}
+	query.Model.GroupBy = append(query.Model.GroupBy,
+		monitor.MetricQueryPart{
+			Type:   "field",
+			Params: []string{tagId},
+		})
+	return query
+}