mirror of
https://github.com/grafana/grafana.git
synced 2025-02-25 18:55:37 -06:00
Fix evaluation of alert rules for datasources with custom headers (#44862)
* Fix evaluation of alert rules for datasources with custom headers * Fix unit tests * Fix integration tests * Evaluator fields should be package private
This commit is contained in:
parent
3cf31451ec
commit
9df43abbb5
@ -103,6 +103,7 @@ func (api *API) RegisterAPIEndpoints(m *metrics.API) {
|
||||
Cfg: api.Cfg,
|
||||
ExpressionService: api.ExpressionService,
|
||||
DatasourceCache: api.DatasourceCache,
|
||||
secretsService: api.SecretsService,
|
||||
log: logger,
|
||||
}), m)
|
||||
api.RegisterConfigurationApiEndpoints(NewForkedConfiguration(
|
||||
|
@ -15,6 +15,7 @@ import (
|
||||
"github.com/grafana/grafana/pkg/services/datasources"
|
||||
apimodels "github.com/grafana/grafana/pkg/services/ngalert/api/tooling/definitions"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/eval"
|
||||
"github.com/grafana/grafana/pkg/services/secrets"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/web"
|
||||
)
|
||||
@ -25,6 +26,7 @@ type TestingApiSrv struct {
|
||||
ExpressionService *expr.Service
|
||||
DatasourceCache datasources.CacheService
|
||||
log log.Logger
|
||||
secretsService secrets.Service
|
||||
}
|
||||
|
||||
func (srv TestingApiSrv) RouteTestRuleConfig(c *models.ReqContext, body apimodels.TestRulePayload) response.Response {
|
||||
@ -33,7 +35,7 @@ func (srv TestingApiSrv) RouteTestRuleConfig(c *models.ReqContext, body apimodel
|
||||
if body.Type() != apimodels.GrafanaBackend || body.GrafanaManagedCondition == nil {
|
||||
return ErrResp(http.StatusBadRequest, errors.New("unexpected payload"), "")
|
||||
}
|
||||
return conditionEval(c, *body.GrafanaManagedCondition, srv.DatasourceCache, srv.ExpressionService, srv.Cfg, srv.log)
|
||||
return conditionEval(c, *body.GrafanaManagedCondition, srv.DatasourceCache, srv.ExpressionService, srv.secretsService, srv.Cfg, srv.log)
|
||||
}
|
||||
|
||||
if body.Type() != apimodels.LoTexRulerBackend {
|
||||
@ -86,7 +88,7 @@ func (srv TestingApiSrv) RouteEvalQueries(c *models.ReqContext, cmd apimodels.Ev
|
||||
return ErrResp(http.StatusBadRequest, err, "invalid queries or expressions")
|
||||
}
|
||||
|
||||
evaluator := eval.Evaluator{Cfg: srv.Cfg, Log: srv.log, DataSourceCache: srv.DatasourceCache}
|
||||
evaluator := eval.NewEvaluator(srv.Cfg, srv.log, srv.DatasourceCache, srv.secretsService)
|
||||
evalResults, err := evaluator.QueriesAndExpressionsEval(c.SignedInUser.OrgId, cmd.Data, now, srv.ExpressionService)
|
||||
if err != nil {
|
||||
return ErrResp(http.StatusBadRequest, err, "Failed to evaluate queries and expressions")
|
||||
|
@ -22,6 +22,7 @@ import (
|
||||
apimodels "github.com/grafana/grafana/pkg/services/ngalert/api/tooling/definitions"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/eval"
|
||||
ngmodels "github.com/grafana/grafana/pkg/services/ngalert/models"
|
||||
"github.com/grafana/grafana/pkg/services/secrets"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/util"
|
||||
"github.com/grafana/grafana/pkg/web"
|
||||
@ -233,7 +234,7 @@ func validateQueriesAndExpressions(ctx context.Context, data []ngmodels.AlertQue
|
||||
return refIDs, nil
|
||||
}
|
||||
|
||||
func conditionEval(c *models.ReqContext, cmd ngmodels.EvalAlertConditionCommand, datasourceCache datasources.CacheService, expressionService *expr.Service, cfg *setting.Cfg, log log.Logger) response.Response {
|
||||
func conditionEval(c *models.ReqContext, cmd ngmodels.EvalAlertConditionCommand, datasourceCache datasources.CacheService, expressionService *expr.Service, secretsService secrets.Service, cfg *setting.Cfg, log log.Logger) response.Response {
|
||||
evalCond := ngmodels.Condition{
|
||||
Condition: cmd.Condition,
|
||||
OrgID: c.SignedInUser.OrgId,
|
||||
@ -248,7 +249,7 @@ func conditionEval(c *models.ReqContext, cmd ngmodels.EvalAlertConditionCommand,
|
||||
now = timeNow()
|
||||
}
|
||||
|
||||
evaluator := eval.Evaluator{Cfg: cfg, Log: log, DataSourceCache: datasourceCache}
|
||||
evaluator := eval.NewEvaluator(cfg, log, datasourceCache, secretsService)
|
||||
evalResults, err := evaluator.ConditionEval(&evalCond, now, expressionService)
|
||||
if err != nil {
|
||||
return ErrResp(http.StatusBadRequest, err, "Failed to evaluate conditions")
|
||||
|
@ -11,12 +11,13 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana/pkg/components/simplejson"
|
||||
"github.com/grafana/grafana/pkg/expr/classic"
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
m "github.com/grafana/grafana/pkg/models"
|
||||
"github.com/grafana/grafana/pkg/services/datasources"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/models"
|
||||
|
||||
"github.com/grafana/grafana/pkg/services/secrets"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
@ -25,9 +26,23 @@ import (
|
||||
)
|
||||
|
||||
type Evaluator struct {
|
||||
Cfg *setting.Cfg
|
||||
Log log.Logger
|
||||
DataSourceCache datasources.CacheService
|
||||
cfg *setting.Cfg
|
||||
log log.Logger
|
||||
dataSourceCache datasources.CacheService
|
||||
secretsService secrets.Service
|
||||
}
|
||||
|
||||
func NewEvaluator(
|
||||
cfg *setting.Cfg,
|
||||
log log.Logger,
|
||||
datasourceCache datasources.CacheService,
|
||||
secretsService secrets.Service) *Evaluator {
|
||||
return &Evaluator{
|
||||
cfg: cfg,
|
||||
log: log,
|
||||
dataSourceCache: datasourceCache,
|
||||
secretsService: secretsService,
|
||||
}
|
||||
}
|
||||
|
||||
// invalidEvalResultFormatError is an error for invalid format of the alert definition evaluation results.
|
||||
@ -124,7 +139,7 @@ type AlertExecCtx struct {
|
||||
}
|
||||
|
||||
// GetExprRequest validates the condition, gets the datasource information and creates an expr.Request from it.
|
||||
func GetExprRequest(ctx AlertExecCtx, data []models.AlertQuery, now time.Time, dsCacheService datasources.CacheService) (*expr.Request, error) {
|
||||
func GetExprRequest(ctx AlertExecCtx, data []models.AlertQuery, now time.Time, dsCacheService datasources.CacheService, secretsService secrets.Service) (*expr.Request, error) {
|
||||
req := &expr.Request{
|
||||
OrgId: ctx.OrgID,
|
||||
Headers: map[string]string{
|
||||
@ -167,6 +182,20 @@ func GetExprRequest(ctx AlertExecCtx, data []models.AlertQuery, now time.Time, d
|
||||
}
|
||||
datasources[q.DatasourceUID] = ds
|
||||
}
|
||||
|
||||
// If the datasource has been configured with custom HTTP headers
|
||||
// then we need to add these to the request
|
||||
decryptedData, err := secretsService.DecryptJsonData(ctx.Ctx, ds.SecureJsonData)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
customHeaders := getCustomHeaders(ds.JsonData, decryptedData)
|
||||
for k, v := range customHeaders {
|
||||
if _, ok := req.Headers[k]; !ok {
|
||||
req.Headers[k] = v
|
||||
}
|
||||
}
|
||||
|
||||
req.Queries = append(req.Queries, expr.Query{
|
||||
TimeRange: expr.TimeRange{
|
||||
From: q.RelativeTimeRange.ToTimeRange(now).From,
|
||||
@ -183,14 +212,40 @@ func GetExprRequest(ctx AlertExecCtx, data []models.AlertQuery, now time.Time, d
|
||||
return req, nil
|
||||
}
|
||||
|
||||
func getCustomHeaders(jsonData *simplejson.Json, decryptedValues map[string]string) map[string]string {
|
||||
headers := make(map[string]string)
|
||||
if jsonData == nil {
|
||||
return headers
|
||||
}
|
||||
|
||||
index := 1
|
||||
for {
|
||||
headerNameSuffix := fmt.Sprintf("httpHeaderName%d", index)
|
||||
headerValueSuffix := fmt.Sprintf("httpHeaderValue%d", index)
|
||||
|
||||
key := jsonData.Get(headerNameSuffix).MustString()
|
||||
if key == "" {
|
||||
// No (more) header values are available
|
||||
break
|
||||
}
|
||||
|
||||
if val, ok := decryptedValues[headerValueSuffix]; ok {
|
||||
headers[key] = val
|
||||
}
|
||||
index++
|
||||
}
|
||||
|
||||
return headers
|
||||
}
|
||||
|
||||
type NumberValueCapture struct {
|
||||
Var string // RefID
|
||||
Labels data.Labels
|
||||
Value *float64
|
||||
}
|
||||
|
||||
func executeCondition(ctx AlertExecCtx, c *models.Condition, now time.Time, exprService *expr.Service, dsCacheService datasources.CacheService) ExecutionResults {
|
||||
execResp, err := executeQueriesAndExpressions(ctx, c.Data, now, exprService, dsCacheService)
|
||||
func executeCondition(ctx AlertExecCtx, c *models.Condition, now time.Time, exprService *expr.Service, dsCacheService datasources.CacheService, secretsService secrets.Service) ExecutionResults {
|
||||
execResp, err := executeQueriesAndExpressions(ctx, c.Data, now, exprService, dsCacheService, secretsService)
|
||||
if err != nil {
|
||||
return ExecutionResults{Error: err}
|
||||
}
|
||||
@ -273,7 +328,7 @@ func executeCondition(ctx AlertExecCtx, c *models.Condition, now time.Time, expr
|
||||
return result
|
||||
}
|
||||
|
||||
func executeQueriesAndExpressions(ctx AlertExecCtx, data []models.AlertQuery, now time.Time, exprService *expr.Service, dsCacheService datasources.CacheService) (resp *backend.QueryDataResponse, err error) {
|
||||
func executeQueriesAndExpressions(ctx AlertExecCtx, data []models.AlertQuery, now time.Time, exprService *expr.Service, dsCacheService datasources.CacheService, secretsService secrets.Service) (resp *backend.QueryDataResponse, err error) {
|
||||
defer func() {
|
||||
if e := recover(); e != nil {
|
||||
ctx.Log.Error("alert rule panic", "error", e, "stack", string(debug.Stack()))
|
||||
@ -286,7 +341,7 @@ func executeQueriesAndExpressions(ctx AlertExecCtx, data []models.AlertQuery, no
|
||||
}
|
||||
}()
|
||||
|
||||
queryDataReq, err := GetExprRequest(ctx, data, now, dsCacheService)
|
||||
queryDataReq, err := GetExprRequest(ctx, data, now, dsCacheService, secretsService)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@ -522,12 +577,12 @@ func (evalResults Results) AsDataFrame() data.Frame {
|
||||
|
||||
// ConditionEval executes conditions and evaluates the result.
|
||||
func (e *Evaluator) ConditionEval(condition *models.Condition, now time.Time, expressionService *expr.Service) (Results, error) {
|
||||
alertCtx, cancelFn := context.WithTimeout(context.Background(), e.Cfg.UnifiedAlerting.EvaluationTimeout)
|
||||
alertCtx, cancelFn := context.WithTimeout(context.Background(), e.cfg.UnifiedAlerting.EvaluationTimeout)
|
||||
defer cancelFn()
|
||||
|
||||
alertExecCtx := AlertExecCtx{OrgID: condition.OrgID, Ctx: alertCtx, ExpressionsEnabled: e.Cfg.ExpressionsEnabled, Log: e.Log}
|
||||
alertExecCtx := AlertExecCtx{OrgID: condition.OrgID, Ctx: alertCtx, ExpressionsEnabled: e.cfg.ExpressionsEnabled, Log: e.log}
|
||||
|
||||
execResult := executeCondition(alertExecCtx, condition, now, expressionService, e.DataSourceCache)
|
||||
execResult := executeCondition(alertExecCtx, condition, now, expressionService, e.dataSourceCache, e.secretsService)
|
||||
|
||||
evalResults := evaluateExecutionResult(execResult, now)
|
||||
return evalResults, nil
|
||||
@ -535,12 +590,12 @@ func (e *Evaluator) ConditionEval(condition *models.Condition, now time.Time, ex
|
||||
|
||||
// QueriesAndExpressionsEval executes queries and expressions and returns the result.
|
||||
func (e *Evaluator) QueriesAndExpressionsEval(orgID int64, data []models.AlertQuery, now time.Time, expressionService *expr.Service) (*backend.QueryDataResponse, error) {
|
||||
alertCtx, cancelFn := context.WithTimeout(context.Background(), e.Cfg.UnifiedAlerting.EvaluationTimeout)
|
||||
alertCtx, cancelFn := context.WithTimeout(context.Background(), e.cfg.UnifiedAlerting.EvaluationTimeout)
|
||||
defer cancelFn()
|
||||
|
||||
alertExecCtx := AlertExecCtx{OrgID: orgID, Ctx: alertCtx, ExpressionsEnabled: e.Cfg.ExpressionsEnabled, Log: e.Log}
|
||||
alertExecCtx := AlertExecCtx{OrgID: orgID, Ctx: alertCtx, ExpressionsEnabled: e.cfg.ExpressionsEnabled, Log: e.log}
|
||||
|
||||
execResult, err := executeQueriesAndExpressions(alertExecCtx, data, now, expressionService, e.DataSourceCache)
|
||||
execResult, err := executeQueriesAndExpressions(alertExecCtx, data, now, expressionService, e.dataSourceCache, e.secretsService)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to execute conditions: %w", err)
|
||||
}
|
||||
|
@ -122,7 +122,7 @@ func (ng *AlertNG) init() error {
|
||||
BaseInterval: baseInterval,
|
||||
Logger: ng.Log,
|
||||
MaxAttempts: ng.Cfg.UnifiedAlerting.MaxAttempts,
|
||||
Evaluator: eval.Evaluator{Cfg: ng.Cfg, Log: ng.Log, DataSourceCache: ng.DataSourceCache},
|
||||
Evaluator: eval.NewEvaluator(ng.Cfg, ng.Log, ng.DataSourceCache, ng.SecretsService),
|
||||
InstanceStore: store,
|
||||
RuleStore: store,
|
||||
AdminConfigStore: store,
|
||||
|
@ -75,7 +75,7 @@ type schedule struct {
|
||||
|
||||
log log.Logger
|
||||
|
||||
evaluator eval.Evaluator
|
||||
evaluator *eval.Evaluator
|
||||
|
||||
ruleStore store.RuleStore
|
||||
instanceStore store.InstanceStore
|
||||
@ -108,7 +108,7 @@ type SchedulerCfg struct {
|
||||
EvalAppliedFunc func(models.AlertRuleKey, time.Time)
|
||||
MaxAttempts int64
|
||||
StopAppliedFunc func(models.AlertRuleKey)
|
||||
Evaluator eval.Evaluator
|
||||
Evaluator *eval.Evaluator
|
||||
RuleStore store.RuleStore
|
||||
OrgStore store.OrgStore
|
||||
InstanceStore store.InstanceStore
|
||||
|
@ -1021,7 +1021,7 @@ func setupScheduler(t *testing.T, rs store.RuleStore, is store.InstanceStore, ac
|
||||
C: mockedClock,
|
||||
BaseInterval: time.Second,
|
||||
MaxAttempts: 1,
|
||||
Evaluator: eval.Evaluator{Cfg: &setting.Cfg{ExpressionsEnabled: true}, Log: logger},
|
||||
Evaluator: eval.NewEvaluator(&setting.Cfg{ExpressionsEnabled: true}, logger, nil, secretsService),
|
||||
RuleStore: rs,
|
||||
InstanceStore: is,
|
||||
AdminConfigStore: acs,
|
||||
|
Loading…
Reference in New Issue
Block a user