From e3a4bde622ce3784c416618712ffebd9c8fb0293 Mon Sep 17 00:00:00 2001 From: Yuriy Tseretyan Date: Wed, 2 Nov 2022 10:13:39 -0400 Subject: [PATCH] Alerting: Condition evaluator with cached pipeline (#57479) * create rule evaluator * load header from the context * init one factory * update scheduler --- pkg/services/ngalert/api/api.go | 9 +- pkg/services/ngalert/api/api_testing.go | 40 +++-- pkg/services/ngalert/api/api_testing_test.go | 65 +++---- pkg/services/ngalert/eval/context.go | 27 +-- pkg/services/ngalert/eval/eval.go | 159 +++++++++-------- .../eval/eval_mocks/ConditionEvaluator.go | 137 +++++++++++++++ .../ngalert/eval/eval_mocks/FakeFactory.go | 32 ++++ pkg/services/ngalert/eval/eval_test.go | 4 +- pkg/services/ngalert/eval/evaluator_mock.go | 160 ------------------ pkg/services/ngalert/ngalert.go | 15 +- pkg/services/ngalert/schedule/schedule.go | 43 +++-- .../ngalert/schedule/schedule_unit_test.go | 18 +- .../api/alerting/api_alertmanager_test.go | 2 +- 13 files changed, 371 insertions(+), 340 deletions(-) create mode 100644 pkg/services/ngalert/eval/eval_mocks/ConditionEvaluator.go create mode 100644 pkg/services/ngalert/eval/eval_mocks/FakeFactory.go delete mode 100644 pkg/services/ngalert/eval/evaluator_mock.go diff --git a/pkg/services/ngalert/api/api.go b/pkg/services/ngalert/api/api.go index aea2a7a5c56..bf0c3953c8f 100644 --- a/pkg/services/ngalert/api/api.go +++ b/pkg/services/ngalert/api/api.go @@ -6,7 +6,6 @@ import ( "time" "github.com/grafana/grafana/pkg/api/routing" - "github.com/grafana/grafana/pkg/expr" "github.com/grafana/grafana/pkg/infra/log" "github.com/grafana/grafana/pkg/services/accesscontrol" "github.com/grafana/grafana/pkg/services/datasourceproxy" @@ -64,7 +63,6 @@ type API struct { DatasourceCache datasources.CacheService DatasourceService datasources.DataSourceService RouteRegister routing.RouteRegister - ExpressionService *expr.Service QuotaService quota.Service Schedule schedule.ScheduleService TransactionManager provisioning.TransactionManager @@ -82,6 +80,7 @@ type API struct { MuteTimings *provisioning.MuteTimingService AlertRules *provisioning.AlertRuleService AlertsRouter *sender.AlertsRouter + EvaluatorFactory eval.EvaluatorFactory } // RegisterAPIEndpoints registers API handlers @@ -92,8 +91,6 @@ func (api *API) RegisterAPIEndpoints(m *metrics.API) { ac: api.AccessControl, } - evaluator := eval.NewEvaluator(api.Cfg, api.DatasourceCache, api.ExpressionService) - // Register endpoints for proxying to Alertmanager-compatible backends. api.RegisterAlertmanagerApiEndpoints(NewForkingAM( api.DatasourceCache, @@ -111,7 +108,7 @@ func (api *API) RegisterAPIEndpoints(m *metrics.API) { api.DatasourceCache, NewLotexRuler(proxy, logger), &RulerSrv{ - conditionValidator: evaluator, + conditionValidator: api.EvaluatorFactory, QuotaService: api.QuotaService, scheduleService: api.Schedule, store: api.RuleStore, @@ -128,7 +125,7 @@ func (api *API) RegisterAPIEndpoints(m *metrics.API) { DatasourceCache: api.DatasourceCache, log: logger, accessControl: api.AccessControl, - evaluator: evaluator, + evaluator: api.EvaluatorFactory, }), m) api.RegisterConfigurationApiEndpoints(NewConfiguration( &ConfigSrv{ diff --git a/pkg/services/ngalert/api/api_testing.go b/pkg/services/ngalert/api/api_testing.go index 5d61b4e39ab..77ad87403f6 100644 --- a/pkg/services/ngalert/api/api_testing.go +++ b/pkg/services/ngalert/api/api_testing.go @@ -24,7 +24,7 @@ type TestingApiSrv struct { DatasourceCache datasources.CacheService log log.Logger accessControl accesscontrol.AccessControl - evaluator eval.Evaluator + evaluator eval.EvaluatorFactory } func (srv TestingApiSrv) RouteTestGrafanaRuleConfig(c *models.ReqContext, body apimodels.TestRulePayload) response.Response { @@ -44,16 +44,20 @@ func (srv TestingApiSrv) RouteTestGrafanaRuleConfig(c *models.ReqContext, body a } ctx := eval.Context(c.Req.Context(), c.SignedInUser) - if err := srv.evaluator.Validate(ctx, evalCond); err != nil { + conditionEval, err := srv.evaluator.Create(ctx, evalCond) + if err != nil { return ErrResp(http.StatusBadRequest, err, "invalid condition") } - ctx = ctx.When(body.GrafanaManagedCondition.Now) - if ctx.At.IsZero() { - ctx = ctx.When(timeNow()) + now := body.GrafanaManagedCondition.Now + if now.IsZero() { + now = timeNow() } - evalResults := srv.evaluator.ConditionEval(ctx, evalCond) + evalResults, err := conditionEval.Evaluate(c.Req.Context(), now) + if err != nil { + return ErrResp(500, err, "Failed to evaluate the rule") + } frame := evalResults.AsDataFrame() return response.JSONStreaming(http.StatusOK, util.DynMap{ @@ -106,14 +110,28 @@ func (srv TestingApiSrv) RouteEvalQueries(c *models.ReqContext, cmd apimodels.Ev return ErrResp(http.StatusUnauthorized, fmt.Errorf("%w to query one or many data sources used by the rule", ErrAuthorization), "") } - ctx := eval.Context(c.Req.Context(), c.SignedInUser).When(cmd.Now) - if ctx.At.IsZero() { - ctx = ctx.When(timeNow()) + cond := ngmodels.Condition{ + Condition: "", + Data: cmd.Data, + } + if len(cmd.Data) > 0 { + cond.Condition = cmd.Data[0].RefID + } + evaluator, err := srv.evaluator.Create(eval.Context(c.Req.Context(), c.SignedInUser), cond) + + if err != nil { + return ErrResp(http.StatusBadRequest, err, "Failed to build evaluator for queries and expressions") } - evalResults, err := srv.evaluator.QueriesAndExpressionsEval(ctx, cmd.Data) + now := cmd.Now + if now.IsZero() { + now = timeNow() + } + + evalResults, err := evaluator.EvaluateRaw(c.Req.Context(), now) + if err != nil { - return ErrResp(http.StatusBadRequest, err, "Failed to evaluate queries and expressions") + return ErrResp(http.StatusInternalServerError, err, "Failed to evaluate queries and expressions") } return response.JSONStreaming(http.StatusOK, evalResults) diff --git a/pkg/services/ngalert/api/api_testing_test.go b/pkg/services/ngalert/api/api_testing_test.go index f802bb561ef..e76f5ca7749 100644 --- a/pkg/services/ngalert/api/api_testing_test.go +++ b/pkg/services/ngalert/api/api_testing_test.go @@ -16,6 +16,7 @@ import ( fakes "github.com/grafana/grafana/pkg/services/datasources/fakes" "github.com/grafana/grafana/pkg/services/ngalert/api/tooling/definitions" "github.com/grafana/grafana/pkg/services/ngalert/eval" + "github.com/grafana/grafana/pkg/services/ngalert/eval/eval_mocks" "github.com/grafana/grafana/pkg/services/ngalert/models" "github.com/grafana/grafana/pkg/services/user" "github.com/grafana/grafana/pkg/web" @@ -58,6 +59,8 @@ func TestRouteTestGrafanaRuleConfig(t *testing.T) { data1 := models.GenerateAlertQuery() data2 := models.GenerateAlertQuery() + currentTime := time.Now() + ac := acMock.New().WithPermissions([]accesscontrol.Permission{ {Action: datasources.ActionQuery, Scope: datasources.ScopeProvider.GetResourceScopeUID(data1.DatasourceUID)}, {Action: datasources.ActionQuery, Scope: datasources.ScopeProvider.GetResourceScopeUID(data2.DatasourceUID)}, @@ -68,26 +71,26 @@ func TestRouteTestGrafanaRuleConfig(t *testing.T) { {Uid: data2.DatasourceUID}, }} - evaluator := &eval.FakeEvaluator{} var result []eval.Result - evaluator.EXPECT().Validate(mock.Anything, mock.Anything).Return(nil) - evaluator.EXPECT().ConditionEval(mock.Anything, mock.Anything).Return(result) + evaluator := &eval_mocks.ConditionEvaluatorMock{} + evaluator.EXPECT().Evaluate(mock.Anything, mock.Anything).Return(result, nil) - srv := createTestingApiSrv(ds, ac, evaluator) + evalFactory := eval_mocks.NewEvaluatorFactory(evaluator) + + srv := createTestingApiSrv(ds, ac, evalFactory) response := srv.RouteTestGrafanaRuleConfig(rc, definitions.TestRulePayload{ Expr: "", GrafanaManagedCondition: &definitions.EvalAlertConditionCommand{ Condition: data1.RefID, Data: []models.AlertQuery{data1, data2}, - Now: time.Time{}, + Now: currentTime, }, }) require.Equal(t, http.StatusOK, response.Status()) - evaluator.AssertCalled(t, "ConditionEval", mock.Anything, mock.Anything, mock.Anything, mock.Anything) - evaluator.AssertCalled(t, "Validate", mock.Anything, mock.Anything, mock.Anything) + evaluator.AssertCalled(t, "Evaluate", mock.Anything, currentTime) }) }) @@ -109,26 +112,25 @@ func TestRouteTestGrafanaRuleConfig(t *testing.T) { ds := &fakes.FakeCacheService{DataSources: []*datasources.DataSource{ {Uid: data1.DatasourceUID}, }} + currentTime := time.Now() - evaluator := &eval.FakeEvaluator{} + evaluator := &eval_mocks.ConditionEvaluatorMock{} var result []eval.Result - evaluator.EXPECT().Validate(mock.Anything, mock.Anything).Return(nil) - evaluator.EXPECT().ConditionEval(mock.Anything, mock.Anything).Return(result) + evaluator.EXPECT().Evaluate(mock.Anything, mock.Anything).Return(result, nil) - srv := createTestingApiSrv(ds, ac, evaluator) + srv := createTestingApiSrv(ds, ac, eval_mocks.NewEvaluatorFactory(evaluator)) response := srv.RouteTestGrafanaRuleConfig(rc, definitions.TestRulePayload{ Expr: "", GrafanaManagedCondition: &definitions.EvalAlertConditionCommand{ Condition: data1.RefID, Data: []models.AlertQuery{data1}, - Now: time.Time{}, + Now: currentTime, }, }) require.Equal(t, http.StatusUnauthorized, response.Status()) - evaluator.AssertNotCalled(t, "ConditionEval", mock.Anything, mock.Anything, mock.Anything, mock.Anything) - evaluator.AssertNotCalled(t, "Validate", mock.Anything, mock.Anything, mock.Anything) + evaluator.AssertNotCalled(t, "Evaluate", mock.Anything, currentTime) rc.IsSignedIn = true @@ -137,14 +139,13 @@ func TestRouteTestGrafanaRuleConfig(t *testing.T) { GrafanaManagedCondition: &definitions.EvalAlertConditionCommand{ Condition: data1.RefID, Data: []models.AlertQuery{data1}, - Now: time.Time{}, + Now: currentTime, }, }) require.Equal(t, http.StatusOK, response.Status()) - evaluator.AssertCalled(t, "ConditionEval", mock.Anything, mock.Anything, mock.Anything, mock.Anything) - evaluator.AssertCalled(t, "Validate", mock.Anything, mock.Anything, mock.Anything) + evaluator.AssertCalled(t, "Evaluate", mock.Anything, currentTime) }) }) } @@ -184,6 +185,8 @@ func TestRouteEvalQueries(t *testing.T) { data1 := models.GenerateAlertQuery() data2 := models.GenerateAlertQuery() + currentTime := time.Now() + ac := acMock.New().WithPermissions([]accesscontrol.Permission{ {Action: datasources.ActionQuery, Scope: datasources.ScopeProvider.GetResourceScopeUID(data1.DatasourceUID)}, {Action: datasources.ActionQuery, Scope: datasources.ScopeProvider.GetResourceScopeUID(data2.DatasourceUID)}, @@ -194,7 +197,7 @@ func TestRouteEvalQueries(t *testing.T) { {Uid: data2.DatasourceUID}, }} - evaluator := &eval.FakeEvaluator{} + evaluator := &eval_mocks.ConditionEvaluatorMock{} result := &backend.QueryDataResponse{ Responses: map[string]backend.DataResponse{ "test": { @@ -203,18 +206,18 @@ func TestRouteEvalQueries(t *testing.T) { }, }, } - evaluator.EXPECT().QueriesAndExpressionsEval(mock.Anything, mock.Anything).Return(result, nil) + evaluator.EXPECT().EvaluateRaw(mock.Anything, mock.Anything).Return(result, nil) - srv := createTestingApiSrv(ds, ac, evaluator) + srv := createTestingApiSrv(ds, ac, eval_mocks.NewEvaluatorFactory(evaluator)) response := srv.RouteEvalQueries(rc, definitions.EvalQueriesPayload{ Data: []models.AlertQuery{data1, data2}, - Now: time.Time{}, + Now: currentTime, }) require.Equal(t, http.StatusOK, response.Status()) - evaluator.AssertCalled(t, "QueriesAndExpressionsEval", mock.Anything, mock.Anything, mock.Anything, mock.Anything) + evaluator.AssertCalled(t, "EvaluateRaw", mock.Anything, currentTime) }) }) @@ -237,7 +240,9 @@ func TestRouteEvalQueries(t *testing.T) { {Uid: data1.DatasourceUID}, }} - evaluator := &eval.FakeEvaluator{} + currentTime := time.Now() + + evaluator := &eval_mocks.ConditionEvaluatorMock{} result := &backend.QueryDataResponse{ Responses: map[string]backend.DataResponse{ "test": { @@ -246,33 +251,33 @@ func TestRouteEvalQueries(t *testing.T) { }, }, } - evaluator.EXPECT().QueriesAndExpressionsEval(mock.Anything, mock.Anything).Return(result, nil) + evaluator.EXPECT().EvaluateRaw(mock.Anything, mock.Anything).Return(result, nil) - srv := createTestingApiSrv(ds, ac, evaluator) + srv := createTestingApiSrv(ds, ac, eval_mocks.NewEvaluatorFactory(evaluator)) response := srv.RouteEvalQueries(rc, definitions.EvalQueriesPayload{ Data: []models.AlertQuery{data1}, - Now: time.Time{}, + Now: currentTime, }) require.Equal(t, http.StatusUnauthorized, response.Status()) - evaluator.AssertNotCalled(t, "QueriesAndExpressionsEval", mock.Anything, mock.Anything, mock.Anything, mock.Anything) + evaluator.AssertNotCalled(t, "EvaluateRaw", mock.Anything, mock.Anything) rc.IsSignedIn = true response = srv.RouteEvalQueries(rc, definitions.EvalQueriesPayload{ Data: []models.AlertQuery{data1}, - Now: time.Time{}, + Now: currentTime, }) require.Equal(t, http.StatusOK, response.Status()) - evaluator.AssertCalled(t, "QueriesAndExpressionsEval", mock.Anything, mock.Anything, mock.Anything, mock.Anything) + evaluator.AssertCalled(t, "EvaluateRaw", mock.Anything, currentTime) }) }) } -func createTestingApiSrv(ds *fakes.FakeCacheService, ac *acMock.Mock, evaluator *eval.FakeEvaluator) *TestingApiSrv { +func createTestingApiSrv(ds *fakes.FakeCacheService, ac *acMock.Mock, evaluator eval.EvaluatorFactory) *TestingApiSrv { if ac == nil { ac = acMock.New().WithDisabled() } diff --git a/pkg/services/ngalert/eval/context.go b/pkg/services/ngalert/eval/context.go index 7172b0fd34d..f84a0e607f6 100644 --- a/pkg/services/ngalert/eval/context.go +++ b/pkg/services/ngalert/eval/context.go @@ -2,42 +2,19 @@ package eval import ( "context" - "time" - "github.com/grafana/grafana/pkg/services/ngalert/models" "github.com/grafana/grafana/pkg/services/user" ) // EvaluationContext represents the context in which a condition is evaluated. type EvaluationContext struct { - Ctx context.Context - User *user.SignedInUser - At time.Time - RuleUID string + Ctx context.Context + User *user.SignedInUser } func Context(ctx context.Context, user *user.SignedInUser) EvaluationContext { return EvaluationContext{ Ctx: ctx, User: user, - At: time.Now(), } } - -func (c EvaluationContext) When(t time.Time) EvaluationContext { - c.At = t - return c -} - -func (c EvaluationContext) WithRule(r *models.AlertRule) EvaluationContext { - if r != nil { - c.RuleUID = r.UID - } - return c -} - -func (c EvaluationContext) WithTimeout(timeout time.Duration) (EvaluationContext, context.CancelFunc) { - timeoutCtx, cancel := context.WithTimeout(c.Ctx, timeout) - c.Ctx = timeoutCtx - return c, cancel -} diff --git a/pkg/services/ngalert/eval/eval.go b/pkg/services/ngalert/eval/eval.go index 0578b5f047f..c7725c0a396 100644 --- a/pkg/services/ngalert/eval/eval.go +++ b/pkg/services/ngalert/eval/eval.go @@ -3,6 +3,7 @@ package eval import ( + "context" "errors" "fmt" "runtime/debug" @@ -24,28 +25,72 @@ import ( var logger = log.New("ngalert.eval") -//go:generate mockery --name Evaluator --structname FakeEvaluator --inpackage --filename evaluator_mock.go --with-expecter -type Evaluator interface { - // ConditionEval executes conditions and evaluates the result. - ConditionEval(ctx EvaluationContext, condition models.Condition) Results - // QueriesAndExpressionsEval executes queries and expressions and returns the result. - QueriesAndExpressionsEval(ctx EvaluationContext, data []models.AlertQuery) (*backend.QueryDataResponse, error) +type EvaluatorFactory interface { // Validate validates that the condition is correct. Returns nil if the condition is correct. Otherwise, error that describes the failure Validate(ctx EvaluationContext, condition models.Condition) error + // BuildRuleEvaluator build an evaluator pipeline ready to evaluate a rule's query + Create(ctx EvaluationContext, condition models.Condition) (ConditionEvaluator, error) +} + +//go:generate mockery --name ConditionEvaluator --structname ConditionEvaluatorMock --with-expecter --output eval_mocks --outpkg eval_mocks +type ConditionEvaluator interface { + // EvaluateRaw evaluates the condition and returns raw backend response backend.QueryDataResponse + EvaluateRaw(ctx context.Context, now time.Time) (resp *backend.QueryDataResponse, err error) + // Evaluate evaluates the condition and converts the response to Results + Evaluate(ctx context.Context, now time.Time) (Results, error) +} + +type conditionEvaluator struct { + pipeline expr.DataPipeline + expressionService *expr.Service + condition models.Condition + evalTimeout time.Duration +} + +func (r *conditionEvaluator) EvaluateRaw(ctx context.Context, now time.Time) (resp *backend.QueryDataResponse, err error) { + defer func() { + if e := recover(); e != nil { + logger.FromContext(ctx).Error("alert rule panic", "error", e, "stack", string(debug.Stack())) + panicErr := fmt.Errorf("alert rule panic; please check the logs for the full stack") + if err != nil { + err = fmt.Errorf("queries and expressions execution failed: %w; %v", err, panicErr.Error()) + } else { + err = panicErr + } + } + }() + + execCtx := ctx + if r.evalTimeout <= 0 { + timeoutCtx, cancel := context.WithTimeout(ctx, r.evalTimeout) + defer cancel() + execCtx = timeoutCtx + } + return r.expressionService.ExecutePipeline(execCtx, now, r.pipeline) +} + +// Evaluate evaluates the condition and converts the response to Results +func (r *conditionEvaluator) Evaluate(ctx context.Context, now time.Time) (Results, error) { + response, err := r.EvaluateRaw(ctx, now) + if err != nil { + return nil, err + } + execResults := queryDataResponseToExecutionResults(r.condition, response) + return evaluateExecutionResult(execResults, now), nil } type evaluatorImpl struct { - cfg *setting.Cfg + evaluationTimeout time.Duration dataSourceCache datasources.CacheService expressionService *expr.Service } -func NewEvaluator( - cfg *setting.Cfg, +func NewEvaluatorFactory( + cfg setting.UnifiedAlertingSettings, datasourceCache datasources.CacheService, - expressionService *expr.Service) Evaluator { + expressionService *expr.Service) EvaluatorFactory { return &evaluatorImpl{ - cfg: cfg, + evaluationTimeout: cfg.EvaluationTimeout, dataSourceCache: datasourceCache, expressionService: expressionService, } @@ -122,6 +167,15 @@ type Result struct { EvaluationString string } +func NewResultFromError(err error, evaluatedAt time.Time, duration time.Duration) Result { + return Result{ + State: Error, + Error: err, + EvaluatedAt: evaluatedAt, + EvaluationDuration: duration, + } +} + // State is an enum of the evaluation State for an alert instance. type State int @@ -169,8 +223,9 @@ func buildDatasourceHeaders(ctx EvaluationContext) map[string]string { "X-Cache-Skip": "true", } - if ctx.RuleUID != "" { - headers["X-Rule-Uid"] = ctx.RuleUID + key, ok := models.RuleKeyFromContext(ctx.Ctx) + if ok { + headers["X-Rule-Uid"] = key.UID } return headers @@ -312,27 +367,6 @@ func queryDataResponseToExecutionResults(c models.Condition, execResp *backend.Q return result } -func executeQueriesAndExpressions(ctx EvaluationContext, data []models.AlertQuery, exprService *expr.Service, dsCacheService datasources.CacheService) (resp *backend.QueryDataResponse, err error) { - defer func() { - if e := recover(); e != nil { - logger.FromContext(ctx.Ctx).Error("alert rule panic", "error", e, "stack", string(debug.Stack())) - panicErr := fmt.Errorf("alert rule panic; please check the logs for the full stack") - if err != nil { - err = fmt.Errorf("queries and expressions execution failed: %w; %v", err, panicErr.Error()) - } else { - err = panicErr - } - } - }() - - queryDataReq, err := getExprRequest(ctx, data, dsCacheService) - if err != nil { - return nil, err - } - - return exprService.TransformData(ctx.Ctx, ctx.At, queryDataReq) -} - // datasourceUIDsToRefIDs returns a sorted slice of Ref IDs for each Datasource UID. // // If refIDsToDatasourceUIDs is nil then this function also returns nil. Likewise, @@ -399,12 +433,7 @@ func evaluateExecutionResult(execResults ExecutionResults, ts time.Time) Results evalResults := make([]Result, 0) appendErrRes := func(e error) { - evalResults = append(evalResults, Result{ - State: Error, - Error: e, - EvaluatedAt: ts, - EvaluationDuration: time.Since(ts), - }) + evalResults = append(evalResults, NewResultFromError(e, ts, time.Since(ts))) } appendNoData := func(labels data.Labels) { @@ -560,55 +589,37 @@ func (evalResults Results) AsDataFrame() data.Frame { return *frame } -// ConditionEval executes conditions and evaluates the result. -func (e *evaluatorImpl) ConditionEval(ctx EvaluationContext, condition models.Condition) Results { - execResp, err := e.QueriesAndExpressionsEval(ctx, condition.Data) - var execResults ExecutionResults - if err != nil { - execResults = ExecutionResults{Error: err} - } else { - execResults = queryDataResponseToExecutionResults(condition, execResp) - } - return evaluateExecutionResult(execResults, ctx.At) -} - -// QueriesAndExpressionsEval executes queries and expressions and returns the result. -func (e *evaluatorImpl) QueriesAndExpressionsEval(ctx EvaluationContext, data []models.AlertQuery) (*backend.QueryDataResponse, error) { - timeoutCtx, cancelFn := ctx.WithTimeout(e.cfg.UnifiedAlerting.EvaluationTimeout) - defer cancelFn() - - execResult, err := executeQueriesAndExpressions(timeoutCtx, data, e.expressionService, e.dataSourceCache) - if err != nil { - return nil, fmt.Errorf("failed to execute conditions: %w", err) - } - - return execResult, nil -} - func (e *evaluatorImpl) Validate(ctx EvaluationContext, condition models.Condition) error { + _, err := e.Create(ctx, condition) + return err +} + +func (e *evaluatorImpl) Create(ctx EvaluationContext, condition models.Condition) (ConditionEvaluator, error) { if len(condition.Data) == 0 { - return errors.New("expression list is empty. must be at least 1 expression") + return nil, errors.New("expression list is empty. must be at least 1 expression") } if len(condition.Condition) == 0 { - return errors.New("condition must not be empty") + return nil, errors.New("condition must not be empty") } - - ctx.At = time.Now() - req, err := getExprRequest(ctx, condition.Data, e.dataSourceCache) if err != nil { - return err + return nil, err } pipeline, err := e.expressionService.BuildPipeline(req) if err != nil { - return err + return nil, err } conditions := make([]string, 0, len(pipeline)) for _, node := range pipeline { if node.RefID() == condition.Condition { - return nil + return &conditionEvaluator{ + pipeline: pipeline, + expressionService: e.expressionService, + condition: condition, + evalTimeout: e.evaluationTimeout, + }, nil } conditions = append(conditions, node.RefID()) } - return fmt.Errorf("condition %s does not exist, must be one of %v", condition.Condition, conditions) + return nil, fmt.Errorf("condition %s does not exist, must be one of %v", condition.Condition, conditions) } diff --git a/pkg/services/ngalert/eval/eval_mocks/ConditionEvaluator.go b/pkg/services/ngalert/eval/eval_mocks/ConditionEvaluator.go new file mode 100644 index 00000000000..3d82a79a5e4 --- /dev/null +++ b/pkg/services/ngalert/eval/eval_mocks/ConditionEvaluator.go @@ -0,0 +1,137 @@ +// Code generated by mockery v2.14.0. DO NOT EDIT. + +package eval_mocks + +import ( + context "context" + + backend "github.com/grafana/grafana-plugin-sdk-go/backend" + + eval "github.com/grafana/grafana/pkg/services/ngalert/eval" + + mock "github.com/stretchr/testify/mock" + + time "time" +) + +// ConditionEvaluatorMock is an autogenerated mock type for the ConditionEvaluator type +type ConditionEvaluatorMock struct { + mock.Mock +} + +type ConditionEvaluatorMock_Expecter struct { + mock *mock.Mock +} + +func (_m *ConditionEvaluatorMock) EXPECT() *ConditionEvaluatorMock_Expecter { + return &ConditionEvaluatorMock_Expecter{mock: &_m.Mock} +} + +// Evaluate provides a mock function with given fields: ctx, now +func (_m *ConditionEvaluatorMock) Evaluate(ctx context.Context, now time.Time) (eval.Results, error) { + ret := _m.Called(ctx, now) + + var r0 eval.Results + if rf, ok := ret.Get(0).(func(context.Context, time.Time) eval.Results); ok { + r0 = rf(ctx, now) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(eval.Results) + } + } + + var r1 error + if rf, ok := ret.Get(1).(func(context.Context, time.Time) error); ok { + r1 = rf(ctx, now) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// ConditionEvaluatorMock_Evaluate_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Evaluate' +type ConditionEvaluatorMock_Evaluate_Call struct { + *mock.Call +} + +// Evaluate is a helper method to define mock.On call +// - ctx context.Context +// - now time.Time +func (_e *ConditionEvaluatorMock_Expecter) Evaluate(ctx interface{}, now interface{}) *ConditionEvaluatorMock_Evaluate_Call { + return &ConditionEvaluatorMock_Evaluate_Call{Call: _e.mock.On("Evaluate", ctx, now)} +} + +func (_c *ConditionEvaluatorMock_Evaluate_Call) Run(run func(ctx context.Context, now time.Time)) *ConditionEvaluatorMock_Evaluate_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context), args[1].(time.Time)) + }) + return _c +} + +func (_c *ConditionEvaluatorMock_Evaluate_Call) Return(_a0 eval.Results, _a1 error) *ConditionEvaluatorMock_Evaluate_Call { + _c.Call.Return(_a0, _a1) + return _c +} + +// EvaluateRaw provides a mock function with given fields: ctx, now +func (_m *ConditionEvaluatorMock) EvaluateRaw(ctx context.Context, now time.Time) (*backend.QueryDataResponse, error) { + ret := _m.Called(ctx, now) + + var r0 *backend.QueryDataResponse + if rf, ok := ret.Get(0).(func(context.Context, time.Time) *backend.QueryDataResponse); ok { + r0 = rf(ctx, now) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*backend.QueryDataResponse) + } + } + + var r1 error + if rf, ok := ret.Get(1).(func(context.Context, time.Time) error); ok { + r1 = rf(ctx, now) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// ConditionEvaluatorMock_EvaluateRaw_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'EvaluateRaw' +type ConditionEvaluatorMock_EvaluateRaw_Call struct { + *mock.Call +} + +// EvaluateRaw is a helper method to define mock.On call +// - ctx context.Context +// - now time.Time +func (_e *ConditionEvaluatorMock_Expecter) EvaluateRaw(ctx interface{}, now interface{}) *ConditionEvaluatorMock_EvaluateRaw_Call { + return &ConditionEvaluatorMock_EvaluateRaw_Call{Call: _e.mock.On("EvaluateRaw", ctx, now)} +} + +func (_c *ConditionEvaluatorMock_EvaluateRaw_Call) Run(run func(ctx context.Context, now time.Time)) *ConditionEvaluatorMock_EvaluateRaw_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context), args[1].(time.Time)) + }) + return _c +} + +func (_c *ConditionEvaluatorMock_EvaluateRaw_Call) Return(resp *backend.QueryDataResponse, err error) *ConditionEvaluatorMock_EvaluateRaw_Call { + _c.Call.Return(resp, err) + return _c +} + +type mockConstructorTestingTNewConditionEvaluatorMock interface { + mock.TestingT + Cleanup(func()) +} + +// NewConditionEvaluatorMock creates a new instance of ConditionEvaluatorMock. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. +func NewConditionEvaluatorMock(t mockConstructorTestingTNewConditionEvaluatorMock) *ConditionEvaluatorMock { + mock := &ConditionEvaluatorMock{} + mock.Mock.Test(t) + + t.Cleanup(func() { mock.AssertExpectations(t) }) + + return mock +} diff --git a/pkg/services/ngalert/eval/eval_mocks/FakeFactory.go b/pkg/services/ngalert/eval/eval_mocks/FakeFactory.go new file mode 100644 index 00000000000..8101a583bea --- /dev/null +++ b/pkg/services/ngalert/eval/eval_mocks/FakeFactory.go @@ -0,0 +1,32 @@ +package eval_mocks + +import ( + "errors" + + "github.com/grafana/grafana/pkg/services/ngalert/eval" + "github.com/grafana/grafana/pkg/services/ngalert/models" +) + +type fakeEvaluatorFactory struct { + err error + evaluator eval.ConditionEvaluator +} + +func NewEvaluatorFactory(evaluator eval.ConditionEvaluator) eval.EvaluatorFactory { + return &fakeEvaluatorFactory{evaluator: evaluator} +} + +func NewFailingEvaluatorFactory(err error) eval.EvaluatorFactory { + if err == nil { + err = errors.New("test") + } + return &fakeEvaluatorFactory{err: err} +} + +func (f fakeEvaluatorFactory) Validate(ctx eval.EvaluationContext, condition models.Condition) error { + return f.err +} + +func (f fakeEvaluatorFactory) Create(ctx eval.EvaluationContext, condition models.Condition) (eval.ConditionEvaluator, error) { + return f.evaluator, f.err +} diff --git a/pkg/services/ngalert/eval/eval_test.go b/pkg/services/ngalert/eval/eval_test.go index e38e9c3bffd..1b77efc04c3 100644 --- a/pkg/services/ngalert/eval/eval_test.go +++ b/pkg/services/ngalert/eval/eval_test.go @@ -443,10 +443,10 @@ func TestValidate(t *testing.T) { cacheService := &fakes.FakeCacheService{} condition := testCase.condition(cacheService) - evaluator := NewEvaluator(&setting.Cfg{ExpressionsEnabled: true}, cacheService, expr.ProvideService(&setting.Cfg{ExpressionsEnabled: true}, nil, nil)) + evaluator := NewEvaluatorFactory(setting.UnifiedAlertingSettings{}, cacheService, expr.ProvideService(&setting.Cfg{ExpressionsEnabled: true}, nil, nil)) evalCtx := Context(context.Background(), u) - err := evaluator.Validate(evalCtx, condition) + _, err := evaluator.Create(evalCtx, condition) if testCase.error { require.Error(t, err) } else { diff --git a/pkg/services/ngalert/eval/evaluator_mock.go b/pkg/services/ngalert/eval/evaluator_mock.go deleted file mode 100644 index b26ef309b6d..00000000000 --- a/pkg/services/ngalert/eval/evaluator_mock.go +++ /dev/null @@ -1,160 +0,0 @@ -// Code generated by mockery v2.12.0. DO NOT EDIT. - -package eval - -import ( - backend "github.com/grafana/grafana-plugin-sdk-go/backend" - mock "github.com/stretchr/testify/mock" - - models "github.com/grafana/grafana/pkg/services/ngalert/models" - - testing "testing" -) - -// FakeEvaluator is an autogenerated mock type for the Evaluator type -type FakeEvaluator struct { - mock.Mock -} - -type FakeEvaluator_Expecter struct { - mock *mock.Mock -} - -func (_m *FakeEvaluator) EXPECT() *FakeEvaluator_Expecter { - return &FakeEvaluator_Expecter{mock: &_m.Mock} -} - -// ConditionEval provides a mock function with given fields: ctx, condition -func (_m *FakeEvaluator) ConditionEval(ctx EvaluationContext, condition models.Condition) Results { - ret := _m.Called(ctx, condition) - - var r0 Results - if rf, ok := ret.Get(0).(func(EvaluationContext, models.Condition) Results); ok { - r0 = rf(ctx, condition) - } else { - if ret.Get(0) != nil { - r0 = ret.Get(0).(Results) - } - } - - return r0 -} - -// FakeEvaluator_ConditionEval_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ConditionEval' -type FakeEvaluator_ConditionEval_Call struct { - *mock.Call -} - -// ConditionEval is a helper method to define mock.On call -// - ctx EvaluationContext -// - condition models.Condition -func (_e *FakeEvaluator_Expecter) ConditionEval(ctx interface{}, condition interface{}) *FakeEvaluator_ConditionEval_Call { - return &FakeEvaluator_ConditionEval_Call{Call: _e.mock.On("ConditionEval", ctx, condition)} -} - -func (_c *FakeEvaluator_ConditionEval_Call) Run(run func(ctx EvaluationContext, condition models.Condition)) *FakeEvaluator_ConditionEval_Call { - _c.Call.Run(func(args mock.Arguments) { - run(args[0].(EvaluationContext), args[1].(models.Condition)) - }) - return _c -} - -func (_c *FakeEvaluator_ConditionEval_Call) Return(_a0 Results) *FakeEvaluator_ConditionEval_Call { - _c.Call.Return(_a0) - return _c -} - -// QueriesAndExpressionsEval provides a mock function with given fields: ctx, data -func (_m *FakeEvaluator) QueriesAndExpressionsEval(ctx EvaluationContext, data []models.AlertQuery) (*backend.QueryDataResponse, error) { - ret := _m.Called(ctx, data) - - var r0 *backend.QueryDataResponse - if rf, ok := ret.Get(0).(func(EvaluationContext, []models.AlertQuery) *backend.QueryDataResponse); ok { - r0 = rf(ctx, data) - } else { - if ret.Get(0) != nil { - r0 = ret.Get(0).(*backend.QueryDataResponse) - } - } - - var r1 error - if rf, ok := ret.Get(1).(func(EvaluationContext, []models.AlertQuery) error); ok { - r1 = rf(ctx, data) - } else { - r1 = ret.Error(1) - } - - return r0, r1 -} - -// FakeEvaluator_QueriesAndExpressionsEval_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'QueriesAndExpressionsEval' -type FakeEvaluator_QueriesAndExpressionsEval_Call struct { - *mock.Call -} - -// QueriesAndExpressionsEval is a helper method to define mock.On call -// - ctx EvaluationContext -// - data []models.AlertQuery -func (_e *FakeEvaluator_Expecter) QueriesAndExpressionsEval(ctx interface{}, data interface{}) *FakeEvaluator_QueriesAndExpressionsEval_Call { - return &FakeEvaluator_QueriesAndExpressionsEval_Call{Call: _e.mock.On("QueriesAndExpressionsEval", ctx, data)} -} - -func (_c *FakeEvaluator_QueriesAndExpressionsEval_Call) Run(run func(ctx EvaluationContext, data []models.AlertQuery)) *FakeEvaluator_QueriesAndExpressionsEval_Call { - _c.Call.Run(func(args mock.Arguments) { - run(args[0].(EvaluationContext), args[1].([]models.AlertQuery)) - }) - return _c -} - -func (_c *FakeEvaluator_QueriesAndExpressionsEval_Call) Return(_a0 *backend.QueryDataResponse, _a1 error) *FakeEvaluator_QueriesAndExpressionsEval_Call { - _c.Call.Return(_a0, _a1) - return _c -} - -// Validate provides a mock function with given fields: ctx, condition -func (_m *FakeEvaluator) Validate(ctx EvaluationContext, condition models.Condition) error { - ret := _m.Called(ctx, condition) - - var r0 error - if rf, ok := ret.Get(0).(func(EvaluationContext, models.Condition) error); ok { - r0 = rf(ctx, condition) - } else { - r0 = ret.Error(0) - } - - return r0 -} - -// FakeEvaluator_Validate_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Validate' -type FakeEvaluator_Validate_Call struct { - *mock.Call -} - -// Validate is a helper method to define mock.On call -// - ctx EvaluationContext -// - condition models.Condition -func (_e *FakeEvaluator_Expecter) Validate(ctx interface{}, condition interface{}) *FakeEvaluator_Validate_Call { - return &FakeEvaluator_Validate_Call{Call: _e.mock.On("Validate", ctx, condition)} -} - -func (_c *FakeEvaluator_Validate_Call) Run(run func(ctx EvaluationContext, condition models.Condition)) *FakeEvaluator_Validate_Call { - _c.Call.Run(func(args mock.Arguments) { - run(args[0].(EvaluationContext), args[1].(models.Condition)) - }) - return _c -} - -func (_c *FakeEvaluator_Validate_Call) Return(_a0 error) *FakeEvaluator_Validate_Call { - _c.Call.Return(_a0) - return _c -} - -// NewFakeEvaluator creates a new instance of FakeEvaluator. It also registers the testing.TB interface on the mock and a cleanup function to assert the mocks expectations. -func NewFakeEvaluator(t testing.TB) *FakeEvaluator { - mock := &FakeEvaluator{} - mock.Mock.Test(t) - - t.Cleanup(func() { mock.AssertExpectations(t) }) - - return mock -} diff --git a/pkg/services/ngalert/ngalert.go b/pkg/services/ngalert/ngalert.go index 9e38ba31e06..d44b1dc46b6 100644 --- a/pkg/services/ngalert/ngalert.go +++ b/pkg/services/ngalert/ngalert.go @@ -180,13 +180,14 @@ func (ng *AlertNG) init() error { ng.AlertsRouter = alertsRouter + evalFactory := eval.NewEvaluatorFactory(ng.Cfg.UnifiedAlerting, ng.DataSourceCache, ng.ExpressionService) schedCfg := schedule.SchedulerCfg{ - Cfg: ng.Cfg.UnifiedAlerting, - C: clk, - Evaluator: eval.NewEvaluator(ng.Cfg, ng.DataSourceCache, ng.ExpressionService), - RuleStore: store, - Metrics: ng.Metrics.GetSchedulerMetrics(), - AlertSender: alertsRouter, + Cfg: ng.Cfg.UnifiedAlerting, + C: clk, + EvaluatorFactory: evalFactory, + RuleStore: store, + Metrics: ng.Metrics.GetSchedulerMetrics(), + AlertSender: alertsRouter, } historian := historian.NewAnnotationHistorian(ng.annotationsRepo, ng.dashboardService) @@ -215,7 +216,6 @@ func (ng *AlertNG) init() error { DatasourceCache: ng.DataSourceCache, DatasourceService: ng.DataSourceService, RouteRegister: ng.RouteRegister, - ExpressionService: ng.ExpressionService, Schedule: ng.schedule, DataProxy: ng.DataProxy, QuotaService: ng.QuotaService, @@ -233,6 +233,7 @@ func (ng *AlertNG) init() error { MuteTimings: muteTimingService, AlertRules: alertRuleService, AlertsRouter: alertsRouter, + EvaluatorFactory: evalFactory, } api.RegisterAPIEndpoints(ng.Metrics.GetAPIMetrics()) diff --git a/pkg/services/ngalert/schedule/schedule.go b/pkg/services/ngalert/schedule/schedule.go index 8efb9608f5c..23b01d13b6c 100644 --- a/pkg/services/ngalert/schedule/schedule.go +++ b/pkg/services/ngalert/schedule/schedule.go @@ -78,7 +78,7 @@ type schedule struct { log log.Logger - evaluator eval.Evaluator + evaluatorFactory eval.EvaluatorFactory ruleStore RulesStore @@ -101,14 +101,14 @@ type schedule struct { // SchedulerCfg is the scheduler configuration. type SchedulerCfg struct { - Cfg setting.UnifiedAlertingSettings - C clock.Clock - EvalAppliedFunc func(ngmodels.AlertRuleKey, time.Time) - StopAppliedFunc func(ngmodels.AlertRuleKey) - Evaluator eval.Evaluator - RuleStore RulesStore - Metrics *metrics.Scheduler - AlertSender AlertsSender + Cfg setting.UnifiedAlertingSettings + C clock.Clock + EvalAppliedFunc func(ngmodels.AlertRuleKey, time.Time) + StopAppliedFunc func(ngmodels.AlertRuleKey) + EvaluatorFactory eval.EvaluatorFactory + RuleStore RulesStore + Metrics *metrics.Scheduler + AlertSender AlertsSender } // NewScheduler returns a new schedule. @@ -121,7 +121,7 @@ func NewScheduler(cfg SchedulerCfg, appURL *url.URL, stateManager *state.Manager log: log.New("ngalert.scheduler"), evalAppliedFunc: cfg.EvalAppliedFunc, stopAppliedFunc: cfg.StopAppliedFunc, - evaluator: cfg.Evaluator, + evaluatorFactory: cfg.EvaluatorFactory, ruleStore: cfg.RuleStore, metrics: cfg.Metrics, appURL: appURL, @@ -340,15 +340,28 @@ func (sch *schedule) ruleRoutine(grafanaCtx context.Context, key ngmodels.AlertR }, }, } - evalCtx := eval.Context(ctx, schedulerUser).When(e.scheduledAt).WithRule(e.rule) + evalCtx := eval.Context(ctx, schedulerUser) + ruleEval, err := sch.evaluatorFactory.Create(evalCtx, e.rule.GetEvalCondition()) + var results eval.Results + var dur time.Duration + if err == nil { + results, err = ruleEval.Evaluate(ctx, e.scheduledAt) + if err != nil { + logger.Error("Failed to evaluate rule", "error", err, "duration", dur) + } + } else { + logger.Error("Failed to build rule evaluator", "error", err) + } + dur = sch.clock.Now().Sub(start) - results := sch.evaluator.ConditionEval(evalCtx, e.rule.GetEvalCondition()) - dur := sch.clock.Now().Sub(start) evalTotal.Inc() evalDuration.Observe(dur.Seconds()) - if results.HasErrors() { + + if err != nil || results.HasErrors() { evalTotalFailures.Inc() - logger.Error("Failed to evaluate alert rule", "results", results, "duration", dur) + if results == nil { + results = append(results, eval.NewResultFromError(err, e.scheduledAt, dur)) + } } else { logger.Debug("Alert rule evaluated", "results", results, "duration", dur) } diff --git a/pkg/services/ngalert/schedule/schedule_unit_test.go b/pkg/services/ngalert/schedule/schedule_unit_test.go index 70f6051ba93..e30e53e2490 100644 --- a/pkg/services/ngalert/schedule/schedule_unit_test.go +++ b/pkg/services/ngalert/schedule/schedule_unit_test.go @@ -477,7 +477,7 @@ func TestSchedule_DeleteAlertRule(t *testing.T) { }) } -func setupScheduler(t *testing.T, rs *fakeRulesStore, is *state.FakeInstanceStore, registry *prometheus.Registry, senderMock *AlertsSenderMock, evalMock *eval.FakeEvaluator) *schedule { +func setupScheduler(t *testing.T, rs *fakeRulesStore, is *state.FakeInstanceStore, registry *prometheus.Registry, senderMock *AlertsSenderMock, evalMock eval.EvaluatorFactory) *schedule { t.Helper() mockedClock := clock.NewMock() @@ -490,9 +490,9 @@ func setupScheduler(t *testing.T, rs *fakeRulesStore, is *state.FakeInstanceStor is = &state.FakeInstanceStore{} } - var evaluator eval.Evaluator = evalMock + var evaluator = evalMock if evalMock == nil { - evaluator = eval.NewEvaluator(&setting.Cfg{ExpressionsEnabled: true}, nil, expr.ProvideService(&setting.Cfg{ExpressionsEnabled: true}, nil, nil)) + evaluator = eval.NewEvaluatorFactory(setting.UnifiedAlertingSettings{}, nil, expr.ProvideService(&setting.Cfg{ExpressionsEnabled: true}, nil, nil)) } if registry == nil { @@ -516,12 +516,12 @@ func setupScheduler(t *testing.T, rs *fakeRulesStore, is *state.FakeInstanceStor } schedCfg := SchedulerCfg{ - Cfg: cfg, - C: mockedClock, - Evaluator: evaluator, - RuleStore: rs, - Metrics: m.GetSchedulerMetrics(), - AlertSender: senderMock, + Cfg: cfg, + C: mockedClock, + EvaluatorFactory: evaluator, + RuleStore: rs, + Metrics: m.GetSchedulerMetrics(), + AlertSender: senderMock, } stateRs := state.FakeRuleReader{} diff --git a/pkg/tests/api/alerting/api_alertmanager_test.go b/pkg/tests/api/alerting/api_alertmanager_test.go index cf01a88c32b..9f79c9fe17c 100644 --- a/pkg/tests/api/alerting/api_alertmanager_test.go +++ b/pkg/tests/api/alerting/api_alertmanager_test.go @@ -2483,7 +2483,7 @@ func TestEval(t *testing.T) { if setting.IsEnterprise { return "user is not authorized to query one or many data sources used by the rule" } - return "Failed to evaluate queries and expressions: failed to execute conditions: failed to build query 'A': data source not found" + return "Failed to build evaluator for queries and expressions: failed to build query 'A': data source not found" }, }, }