refactor: es query support concurrency

alert/astats/stats.go

@@ -17,6 +17,7 @@ type Stats struct {
 	CounterRuleEval             *prometheus.CounterVec
 	CounterQueryDataErrorTotal  *prometheus.CounterVec
 	CounterQueryDataTotal       *prometheus.CounterVec
+	CounterVarFillingQuery      *prometheus.CounterVec
 	CounterRecordEval           *prometheus.CounterVec
 	CounterRecordEvalErrorTotal *prometheus.CounterVec
 	CounterMuteTotal            *prometheus.CounterVec
@@ -54,7 +55,7 @@ func NewSyncStats() *Stats {
 		Subsystem: subsystem,
 		Name:      "query_data_total",
 		Help:      "Number of rule eval query data.",
-	}, []string{"datasource"})
+	}, []string{"datasource", "rule_id"})
 
 	CounterRecordEval := prometheus.NewCounterVec(prometheus.CounterOpts{
 		Namespace: namespace,
@@ -135,6 +136,13 @@ func NewSyncStats() *Stats {
 		Help:      "The size of notify record queue.",
 	})
 
+	CounterVarFillingQuery := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "var_filling_query_total",
+		Help:      "Number of var filling query.",
+	}, []string{"rule_id", "datasource_id", "ref", "typ"})
+
 	prometheus.MustRegister(
 		CounterAlertsTotal,
 		GaugeAlertQueueSize,
@@ -151,6 +159,7 @@ func NewSyncStats() *Stats {
 		CounterSubEventTotal,
 		GaugeQuerySeriesCount,
 		GaugeNotifyRecordQueueSize,
+		CounterVarFillingQuery,
 	)
 
 	return &Stats{
@@ -169,5 +178,6 @@ func NewSyncStats() *Stats {
 		CounterSubEventTotal:        CounterSubEventTotal,
 		GaugeQuerySeriesCount:       GaugeQuerySeriesCount,
 		GaugeNotifyRecordQueueSize:  GaugeNotifyRecordQueueSize,
+		CounterVarFillingQuery:      CounterVarFillingQuery,
 	}
 }

alert/dispatch/dispatch.go

@@ -732,10 +732,10 @@ func getSendTarget(customParams map[string]string, sendtos []string) string {
 
 	values := make([]string, 0)
 	for _, value := range customParams {
-		if len(value) <= 4 {
+		runes := []rune(value)
+		if len(runes) <= 4 {
 			values = append(values, value)
 		} else {
-			runes := []rune(value)
 			maskedValue := string(runes[:len(runes)-4]) + "****"
 			values = append(values, maskedValue)
 		}

alert/eval/eval.go

@@ -13,6 +13,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/ccfos/nightingale/v6/alert/astats"
 	"github.com/ccfos/nightingale/v6/alert/common"
 	"github.com/ccfos/nightingale/v6/alert/process"
 	"github.com/ccfos/nightingale/v6/dscache"
@@ -99,7 +100,7 @@ func NewAlertRuleWorker(rule *models.AlertRule, datasourceId int64, Processor *p
 		rule.CronPattern = fmt.Sprintf("@every %ds", interval)
 	}
 
-	arw.Scheduler = cron.New(cron.WithSeconds())
+	arw.Scheduler = cron.New(cron.WithSeconds(), cron.WithChain(cron.SkipIfStillRunning(cron.DefaultLogger)))
 
 	entryID, err := arw.Scheduler.AddFunc(rule.CronPattern, func() {
 		arw.Eval()
@@ -281,9 +282,21 @@ func (arw *AlertRuleWorker) GetPromAnomalyPoint(ruleConfig string) ([]models.Ano
 			if hasLabelLossAggregator(query) || notExactMatch(query) {
 				// 若有聚合函数或非精确匹配则需要先填充变量然后查询，这个方式效率较低
 				anomalyPoints = arw.VarFillingBeforeQuery(query, readerClient)
+				arw.Processor.Stats.CounterVarFillingQuery.WithLabelValues(
+					fmt.Sprintf("%v", arw.Rule.Id),
+					fmt.Sprintf("%v", arw.Processor.DatasourceId()),
+					fmt.Sprintf("%v", i),
+					"BeforeQuery",
+				).Inc()
 			} else {
 				// 先查询再过滤变量，效率较高，但无法处理有聚合函数的情况
 				anomalyPoints = arw.VarFillingAfterQuery(query, readerClient)
+				arw.Processor.Stats.CounterVarFillingQuery.WithLabelValues(
+					fmt.Sprintf("%v", arw.Rule.Id),
+					fmt.Sprintf("%v", arw.Processor.DatasourceId()),
+					fmt.Sprintf("%v", i),
+					"AfterQuery",
+				).Inc()
 			}
 			lst = append(lst, anomalyPoints...)
 		} else {
@@ -302,7 +315,7 @@ func (arw *AlertRuleWorker) GetPromAnomalyPoint(ruleConfig string) ([]models.Ano
 			}
 
 			var warnings promsdk.Warnings
-			arw.Processor.Stats.CounterQueryDataTotal.WithLabelValues(fmt.Sprintf("%d", arw.DatasourceId)).Inc()
+			arw.Processor.Stats.CounterQueryDataTotal.WithLabelValues(fmt.Sprintf("%d", arw.DatasourceId), fmt.Sprintf("%d", arw.Rule.Id)).Inc()
 			value, warnings, err := readerClient.Query(context.Background(), promql, time.Now())
 			if err != nil {
 				logger.Errorf("rule_eval:%s promql:%s, error:%v", arw.Key(), promql, err)
@@ -413,6 +426,7 @@ func (arw *AlertRuleWorker) VarFillingAfterQuery(query models.PromQuery, readerC
 				realQuery = strings.Replace(realQuery, fmt.Sprintf("$%s", key), val, -1)
 			}
 			// 得到满足值变量的所有结果
+			arw.Processor.Stats.CounterQueryDataTotal.WithLabelValues(fmt.Sprintf("%d", arw.DatasourceId), fmt.Sprintf("%d", arw.Rule.Id)).Inc()
 			value, _, err := readerClient.Query(context.Background(), curQuery, time.Now())
 			if err != nil {
 				logger.Errorf("rule_eval:%s, promql:%s, error:%v", arw.Key(), curQuery, err)
@@ -574,7 +588,7 @@ func (arw *AlertRuleWorker) getParamPermutation(paramVal map[string]models.Param
 				logger.Errorf("query:%s fail to unmarshalling into string slice, error:%v", paramQuery.Query, err)
 			}
 			if len(query) == 0 {
-				paramsKeyAllLabel, err := getParamKeyAllLabel(varToLabel[paramKey], originPromql, readerClient)
+				paramsKeyAllLabel, err := getParamKeyAllLabel(varToLabel[paramKey], originPromql, readerClient, arw.DatasourceId, arw.Rule.Id, arw.Processor.Stats)
 				if err != nil {
 					logger.Errorf("rule_eval:%s, fail to getParamKeyAllLabel, error:%v query:%s", arw.Key(), err, paramQuery.Query)
 				}
@@ -605,7 +619,7 @@ func (arw *AlertRuleWorker) getParamPermutation(paramVal map[string]models.Param
 	return res, nil
 }
 
-func getParamKeyAllLabel(paramKey string, promql string, client promsdk.API) ([]string, error) {
+func getParamKeyAllLabel(paramKey string, promql string, client promsdk.API, dsId int64, rid int64, stats *astats.Stats) ([]string, error) {
 	labels, metricName, err := promql2.GetLabelsAndMetricNameWithReplace(promql, "$")
 	if err != nil {
 		return nil, fmt.Errorf("promql:%s, get labels error:%v", promql, err)
@@ -619,6 +633,7 @@ func getParamKeyAllLabel(paramKey string, promql string, client promsdk.API) ([]
 	}
 	pr := metricName + "{" + strings.Join(labelstrs, ",") + "}"
 
+	stats.CounterQueryDataTotal.WithLabelValues(fmt.Sprintf("%d", dsId), fmt.Sprintf("%d", rid)).Inc()
 	value, _, err := client.Query(context.Background(), pr, time.Now())
 	if err != nil {
 		return nil, fmt.Errorf("promql: %s query error: %v", pr, err)
@@ -1276,6 +1291,7 @@ func (arw *AlertRuleWorker) VarFillingBeforeQuery(query models.PromQuery, reader
 						<-semaphore
 						wg.Done()
 					}()
+					arw.Processor.Stats.CounterQueryDataTotal.WithLabelValues(fmt.Sprintf("%d", arw.DatasourceId), fmt.Sprintf("%d", arw.Rule.Id)).Inc()
 					value, _, err := readerClient.Query(context.Background(), promql, time.Now())
 					if err != nil {
 						logger.Errorf("rule_eval:%s, promql:%s, error:%v", arw.Key(), promql, err)
@@ -1456,7 +1472,7 @@ func (arw *AlertRuleWorker) GetAnomalyPoint(rule *models.AlertRule, dsId int64)
 
 			ctx := context.WithValue(context.Background(), "delay", int64(rule.Delay))
 			series, err := plug.QueryData(ctx, query)
-			arw.Processor.Stats.CounterQueryDataTotal.WithLabelValues(fmt.Sprintf("%d", arw.DatasourceId)).Inc()
+			arw.Processor.Stats.CounterQueryDataTotal.WithLabelValues(fmt.Sprintf("%d", arw.DatasourceId), fmt.Sprintf("%d", rule.Id)).Inc()
 			if err != nil {
 				logger.Warningf("rule_eval rid:%d query data error: %v", rule.Id, err)
 				arw.Processor.Stats.CounterRuleEvalErrorTotal.WithLabelValues(fmt.Sprintf("%v", arw.Processor.DatasourceId()), GET_CLIENT, arw.Processor.BusiGroupCache.GetNameByBusiGroupId(arw.Rule.GroupId), fmt.Sprintf("%v", arw.Rule.Id)).Inc()

alert/record/prom_rule.go

@@ -39,7 +39,7 @@ func NewRecordRuleContext(rule *models.RecordingRule, datasourceId int64, promCl
 		rule.CronPattern = fmt.Sprintf("@every %ds", rule.PromEvalInterval)
 	}
 
-	rrc.scheduler = cron.New(cron.WithSeconds())
+	rrc.scheduler = cron.New(cron.WithSeconds(), cron.WithChain(cron.SkipIfStillRunning(cron.DefaultLogger)))
 	_, err := rrc.scheduler.AddFunc(rule.CronPattern, func() {
 		rrc.Eval()
 	})

center/router/router_query.go

@@ -3,6 +3,7 @@ package router
 import (
 	"fmt"
 	"sort"
+	"sync"
 
 	"github.com/ccfos/nightingale/v6/dscache"
 	"github.com/ccfos/nightingale/v6/models"
@@ -38,71 +39,116 @@ type LogResp struct {
 	List  []interface{} `json:"list"`
 }
 
-func (rt *Router) QueryLogBatch(c *gin.Context) {
-	var f QueryFrom
-	ginx.BindJSON(c, &f)
-
+func QueryLogBatchConcurrently(anonymousAccess bool, ctx *gin.Context, f QueryFrom) (LogResp, error) {
 	var resp LogResp
-	var errMsg string
+	var mu sync.Mutex
+	var wg sync.WaitGroup
+	var errs []error
+
 	for _, q := range f.Queries {
-		if !rt.Center.AnonymousAccess.PromQuerier && !CheckDsPerm(c, q.Did, q.DsCate, q) {
-			ginx.Bomb(200, "no permission")
+		if !anonymousAccess && !CheckDsPerm(ctx, q.Did, q.DsCate, q) {
+			return LogResp{}, fmt.Errorf("no permission")
 		}
 
 		plug, exists := dscache.DsCache.Get(q.DsCate, q.Did)
 		if !exists {
 			logger.Warningf("cluster:%d not exists query:%+v", q.Did, q)
-			ginx.Bomb(200, "cluster not exists")
+			return LogResp{}, fmt.Errorf("cluster not exists")
 		}
 
-		data, total, err := plug.QueryLog(c.Request.Context(), q.Query)
-		if err != nil {
-			errMsg += fmt.Sprintf("query data error: %v query:%v\n ", err, q)
-			logger.Warningf("query data error: %v query:%v", err, q)
-			continue
-		}
+		wg.Add(1)
+		go func(query Query) {
+			defer wg.Done()
 
-		m := make(map[string]interface{})
-		m["ref"] = q.Ref
-		m["ds_id"] = q.Did
-		m["ds_cate"] = q.DsCate
-		m["data"] = data
-		resp.List = append(resp.List, m)
-		resp.Total += total
+			data, total, err := plug.QueryLog(ctx.Request.Context(), query.Query)
+			mu.Lock()
+			defer mu.Unlock()
+			if err != nil {
+				errMsg := fmt.Sprintf("query data error: %v query:%v\n ", err, query)
+				logger.Warningf(errMsg)
+				errs = append(errs, err)
+				return
+			}
+
+			m := make(map[string]interface{})
+			m["ref"] = query.Ref
+			m["ds_id"] = query.Did
+			m["ds_cate"] = query.DsCate
+			m["data"] = data
+
+			resp.List = append(resp.List, m)
+			resp.Total += total
+		}(q)
 	}
 
-	if errMsg != "" || len(resp.List) == 0 {
-		ginx.Bomb(200, errMsg)
+	wg.Wait()
+
+	if len(errs) > 0 {
+		return LogResp{}, errs[0]
+	}
+
+	if len(resp.List) == 0 {
+		return LogResp{}, fmt.Errorf("no data")
+	}
+
+	return resp, nil
+}
+
+func (rt *Router) QueryLogBatch(c *gin.Context) {
+	var f QueryFrom
+	ginx.BindJSON(c, &f)
+
+	resp, err := QueryLogBatchConcurrently(rt.Center.AnonymousAccess.PromQuerier, c, f)
+	if err != nil {
+		ginx.Bomb(200, "err:%v", err)
 	}
 
 	ginx.NewRender(c).Data(resp, nil)
 }
 
-func (rt *Router) QueryData(c *gin.Context) {
-	var f models.QueryParam
-	ginx.BindJSON(c, &f)
-
+func QueryDataConcurrently(anonymousAccess bool, ctx *gin.Context, f models.QueryParam) ([]models.DataResp, error) {
 	var resp []models.DataResp
-	var err error
+	var mu sync.Mutex
+	var wg sync.WaitGroup
+	var errs []error
+
 	for _, q := range f.Querys {
-		if !rt.Center.AnonymousAccess.PromQuerier && !CheckDsPerm(c, f.DatasourceId, f.Cate, q) {
-			ginx.Bomb(403, "no permission")
+		if !anonymousAccess && !CheckDsPerm(ctx, f.DatasourceId, f.Cate, q) {
+			return nil, fmt.Errorf("no permission")
 		}
 
 		plug, exists := dscache.DsCache.Get(f.Cate, f.DatasourceId)
 		if !exists {
 			logger.Warningf("cluster:%d not exists", f.DatasourceId)
-			ginx.Bomb(200, "cluster not exists")
+			return nil, fmt.Errorf("cluster not exists")
 		}
-		var datas []models.DataResp
-		datas, err = plug.QueryData(c.Request.Context(), q)
-		if err != nil {
-			logger.Warningf("query data error: req:%+v err:%v", q, err)
-			ginx.Bomb(200, "err:%v", err)
-		}
-		logger.Debugf("query data: req:%+v resp:%+v", q, datas)
-		resp = append(resp, datas...)
+
+		wg.Add(1)
+		go func(query interface{}) {
+			defer wg.Done()
+
+			datas, err := plug.QueryData(ctx.Request.Context(), query)
+			if err != nil {
+				logger.Warningf("query data error: req:%+v err:%v", query, err)
+				mu.Lock()
+				errs = append(errs, err)
+				mu.Unlock()
+				return
+			}
+
+			logger.Debugf("query data: req:%+v resp:%+v", query, datas)
+			mu.Lock()
+			resp = append(resp, datas...)
+			mu.Unlock()
+		}(q)
 	}
+
+	wg.Wait()
+
+	if len(errs) > 0 {
+		return nil, errs[0]
+	}
+
 	// 面向API的统一处理
 	// 按照 .Metric 排序
 	// 确保仪表盘中相同图例的曲线颜色相同
@@ -115,41 +161,80 @@ func (rt *Router) QueryData(c *gin.Context) {
 		})
 	}
 
-	ginx.NewRender(c).Data(resp, err)
+	return resp, nil
+}
+
+func (rt *Router) QueryData(c *gin.Context) {
+	var f models.QueryParam
+	ginx.BindJSON(c, &f)
+
+	resp, err := QueryDataConcurrently(rt.Center.AnonymousAccess.PromQuerier, c, f)
+	if err != nil {
+		ginx.Bomb(200, "err:%v", err)
+	}
+
+	ginx.NewRender(c).Data(resp, nil)
+}
+
+// QueryLogConcurrently 并发查询日志
+func QueryLogConcurrently(anonymousAccess bool, ctx *gin.Context, f models.QueryParam) (LogResp, error) {
+	var resp LogResp
+	var mu sync.Mutex
+	var wg sync.WaitGroup
+	var errs []error
+
+	for _, q := range f.Querys {
+		if !anonymousAccess && !CheckDsPerm(ctx, f.DatasourceId, f.Cate, q) {
+			return LogResp{}, fmt.Errorf("no permission")
+		}
+
+		plug, exists := dscache.DsCache.Get(f.Cate, f.DatasourceId)
+		if !exists {
+			logger.Warningf("cluster:%d not exists query:%+v", f.DatasourceId, f)
+			return LogResp{}, fmt.Errorf("cluster not exists")
+		}
+
+		wg.Add(1)
+		go func(query interface{}) {
+			defer wg.Done()
+
+			data, total, err := plug.QueryLog(ctx.Request.Context(), query)
+			logger.Debugf("query log: req:%+v resp:%+v", query, data)
+			if err != nil {
+				errMsg := fmt.Sprintf("query data error: %v query:%v\n ", err, query)
+				logger.Warningf(errMsg)
+				mu.Lock()
+				errs = append(errs, err)
+				mu.Unlock()
+				return
+			}
+
+			mu.Lock()
+			resp.List = append(resp.List, data...)
+			resp.Total += total
+			mu.Unlock()
+		}(q)
+	}
+
+	wg.Wait()
+
+	if len(errs) > 0 {
+		return LogResp{}, errs[0]
+	}
+
+	if len(resp.List) == 0 {
+		return LogResp{}, fmt.Errorf("no data")
+	}
+
+	return resp, nil
 }
 
 func (rt *Router) QueryLogV2(c *gin.Context) {
 	var f models.QueryParam
 	ginx.BindJSON(c, &f)
 
-	var resp LogResp
-	var errMsg string
-	for _, q := range f.Querys {
-		if !rt.Center.AnonymousAccess.PromQuerier && !CheckDsPerm(c, f.DatasourceId, f.Cate, q) {
-			ginx.Bomb(200, "no permission")
-		}
-
-		plug, exists := dscache.DsCache.Get(f.Cate, f.DatasourceId)
-		if !exists {
-			logger.Warningf("cluster:%d not exists query:%+v", f.DatasourceId, f)
-			ginx.Bomb(200, "cluster not exists")
-		}
-
-		data, total, err := plug.QueryLog(c.Request.Context(), q)
-		if err != nil {
-			errMsg += fmt.Sprintf("query data error: %v query:%v\n ", err, q)
-			logger.Warningf("query data error: %v query:%v", err, q)
-			continue
-		}
-		resp.List = append(resp.List, data...)
-		resp.Total += total
-	}
-
-	if errMsg != "" || len(resp.List) == 0 {
-		ginx.Bomb(200, errMsg)
-	}
-
-	ginx.NewRender(c).Data(resp, nil)
+	resp, err := QueryLogConcurrently(rt.Center.AnonymousAccess.PromQuerier, c, f)
+	ginx.NewRender(c).Data(resp, err)
 }
 
 func (rt *Router) QueryLog(c *gin.Context) {

datasource/commons/eslike/eslike.go

@@ -481,7 +481,7 @@ func QueryData(ctx context.Context, queryParam interface{}, cliTimeout int64, ve
 
 	source, _ := queryString.Source()
 	b, _ := json.Marshal(source)
-	logger.Debugf("query_data q:%+v tsAggr:%+v query_string:%s", param, tsAggr, string(b))
+	logger.Debugf("query_data q:%+v indexArr:%+v tsAggr:%+v query_string:%s", param, indexArr, tsAggr, string(b))
 
 	searchSource := elastic.NewSearchSource().
 		Query(queryString).

memsto/notify_channel_cache.go

@@ -121,7 +121,6 @@ func (ncc *NotifyChannelCacheType) SyncNotifyChannels() {
 	err := ncc.syncNotifyChannels()
 	if err != nil {
 		fmt.Println("failed to sync notify channels:", err)
-		exit(1)
 	}
 
 	go ncc.loopSyncNotifyChannels()

pkg/prom/reader.go

@@ -701,6 +701,7 @@ func (h *httpAPI) Query(ctx context.Context, query string, ts time.Time) (model.
 	var warnings Warnings
 	var value model.Value
 	var statusCode int
+  
 	for i := 0; i < 1; i++ {
 		value, warnings, statusCode, err = h.query(ctx, query, ts)
 		if err == nil {