From 48f1db63ff5efc45039e64498dae8eacb3537ca3 Mon Sep 17 00:00:00 2001 From: Yuri Tseretyan Date: Fri, 6 Jan 2023 21:21:43 -0500 Subject: [PATCH] Alerting: Add support for tracing to alerting scheduler (#61057) --- pkg/services/ngalert/ngalert.go | 5 ++ pkg/services/ngalert/schedule/schedule.go | 47 ++++++++++++++++++- .../ngalert/schedule/schedule_unit_test.go | 5 ++ pkg/services/ngalert/tests/util.go | 5 +- pkg/services/quota/quotaimpl/quota_test.go | 3 +- 5 files changed, 60 insertions(+), 5 deletions(-) diff --git a/pkg/services/ngalert/ngalert.go b/pkg/services/ngalert/ngalert.go index 8baab25f734..f715455f7ef 100644 --- a/pkg/services/ngalert/ngalert.go +++ b/pkg/services/ngalert/ngalert.go @@ -15,6 +15,7 @@ import ( "github.com/grafana/grafana/pkg/infra/db" "github.com/grafana/grafana/pkg/infra/kvstore" "github.com/grafana/grafana/pkg/infra/log" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/plugins" "github.com/grafana/grafana/pkg/services/accesscontrol" "github.com/grafana/grafana/pkg/services/annotations" @@ -64,6 +65,7 @@ func ProvideService( accesscontrolService accesscontrol.Service, annotationsRepo annotations.Repository, pluginsStore plugins.Store, + tracer tracing.Tracer, ) (*AlertNG, error) { ng := &AlertNG{ Cfg: cfg, @@ -88,6 +90,7 @@ func ProvideService( accesscontrolService: accesscontrolService, annotationsRepo: annotationsRepo, pluginsStore: pluginsStore, + tracer: tracer, } if ng.IsDisabled() { @@ -134,6 +137,7 @@ type AlertNG struct { bus bus.Bus pluginsStore plugins.Store + tracer tracing.Tracer } func (ng *AlertNG) init() error { @@ -198,6 +202,7 @@ func (ng *AlertNG) init() error { RuleStore: store, Metrics: ng.Metrics.GetSchedulerMetrics(), AlertSender: alertsRouter, + Tracer: ng.tracer, } history, err := configureHistorianBackend(ng.Cfg.UnifiedAlerting.StateHistory, ng.annotationsRepo, ng.dashboardService) diff --git a/pkg/services/ngalert/schedule/schedule.go b/pkg/services/ngalert/schedule/schedule.go index 503885fea5d..f05b650c90c 100644 --- a/pkg/services/ngalert/schedule/schedule.go +++ b/pkg/services/ngalert/schedule/schedule.go @@ -7,9 +7,12 @@ import ( "net/url" "time" + "github.com/hashicorp/go-multierror" prometheusModel "github.com/prometheus/common/model" + "go.opentelemetry.io/otel/attribute" "github.com/grafana/grafana/pkg/infra/log" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/services/datasources" "github.com/grafana/grafana/pkg/services/ngalert/api/tooling/definitions" "github.com/grafana/grafana/pkg/services/ngalert/eval" @@ -93,6 +96,8 @@ type schedule struct { // current tick depends on its evaluation interval and when it was // last evaluated. schedulableAlertRules alertRulesRegistry + + tracer tracing.Tracer } // SchedulerCfg is the scheduler configuration. @@ -107,6 +112,7 @@ type SchedulerCfg struct { RuleStore RulesStore Metrics *metrics.Scheduler AlertSender AlertsSender + Tracer tracing.Tracer } // NewScheduler returns a new schedule. @@ -126,6 +132,7 @@ func NewScheduler(cfg SchedulerCfg, stateManager *state.Manager) *schedule { minRuleInterval: cfg.MinRuleInterval, schedulableAlertRules: alertRulesRegistry{rules: make(map[ngmodels.AlertRuleKey]*ngmodels.AlertRule)}, alertsSender: cfg.AlertSender, + tracer: cfg.Tracer, } return &sch @@ -323,7 +330,7 @@ func (sch *schedule) ruleRoutine(grafanaCtx context.Context, key ngmodels.AlertR } } - evaluate := func(ctx context.Context, attempt int64, e *evaluation) { + evaluate := func(ctx context.Context, attempt int64, e *evaluation, span tracing.Span) { logger := logger.New("version", e.rule.Version, "attempt", attempt, "now", e.scheduledAt) start := sch.clock.Now() @@ -363,8 +370,28 @@ func (sch *schedule) ruleRoutine(grafanaCtx context.Context, key ngmodels.AlertR if results == nil { results = append(results, eval.NewResultFromError(err, e.scheduledAt, dur)) } + if err == nil { + for _, result := range results { + if result.Error != nil { + err = multierror.Append(err, result.Error) + } + } + } + span.RecordError(err) + span.AddEvents( + []string{"error", "message"}, + []tracing.EventValue{ + {Str: fmt.Sprintf("%v", err)}, + {Str: "rule evaluation failed"}, + }) } else { logger.Debug("Alert rule evaluated", "results", results, "duration", dur) + span.AddEvents( + []string{"message", "results"}, + []tracing.EventValue{ + {Str: "rule evaluated"}, + {Num: int64(len(results))}, + }) } if ctx.Err() != nil { // check if the context is not cancelled. The evaluation can be a long-running task. logger.Debug("Skip updating the state because the context has been cancelled") @@ -372,6 +399,13 @@ func (sch *schedule) ruleRoutine(grafanaCtx context.Context, key ngmodels.AlertR } processedStates := sch.stateManager.ProcessEvalResults(ctx, e.scheduledAt, e.rule, results, sch.getRuleExtraLabels(e)) alerts := FromStateTransitionToPostableAlerts(processedStates, sch.stateManager, sch.appURL) + span.AddEvents( + []string{"message", "state_transitions", "alerts_to_send"}, + []tracing.EventValue{ + {Str: "results processed"}, + {Num: int64(len(processedStates))}, + {Num: int64(len(alerts.PostableAlerts))}, + }) if len(alerts.PostableAlerts) > 0 { sch.alertsSender.Send(key, alerts) } @@ -434,7 +468,16 @@ func (sch *schedule) ruleRoutine(grafanaCtx context.Context, key ngmodels.AlertR } currentRuleVersion = newVersion } - evaluate(grafanaCtx, attempt, ctx) + tracingCtx, span := sch.tracer.Start(grafanaCtx, "alert rule execution") + defer span.End() + + span.SetAttributes("rule_uid", ctx.rule.UID, attribute.String("rule_uid", ctx.rule.UID)) + span.SetAttributes("org_id", ctx.rule.OrgID, attribute.Int64("org_id", ctx.rule.OrgID)) + span.SetAttributes("rule_version", ctx.rule.Version, attribute.Int64("rule_version", ctx.rule.Version)) + utcTick := ctx.scheduledAt.UTC().Format(time.RFC3339Nano) + span.SetAttributes("tick", utcTick, attribute.String("tick", utcTick)) + + evaluate(tracingCtx, attempt, ctx, span) return nil }) if err != nil { diff --git a/pkg/services/ngalert/schedule/schedule_unit_test.go b/pkg/services/ngalert/schedule/schedule_unit_test.go index b595e831f3e..78086d21028 100644 --- a/pkg/services/ngalert/schedule/schedule_unit_test.go +++ b/pkg/services/ngalert/schedule/schedule_unit_test.go @@ -21,6 +21,7 @@ import ( "golang.org/x/sync/errgroup" "github.com/grafana/grafana/pkg/expr" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/plugins" "github.com/grafana/grafana/pkg/services/ngalert/api/tooling/definitions" "github.com/grafana/grafana/pkg/services/ngalert/eval" @@ -37,6 +38,7 @@ type evalAppliedInfo struct { } func TestProcessTicks(t *testing.T) { + testTracer := tracing.InitializeTracerForTest() testMetrics := metrics.NewNGAlert(prometheus.NewPedanticRegistry()) ctx := context.Background() dispatcherGroup, ctx := errgroup.WithContext(ctx) @@ -67,6 +69,7 @@ func TestProcessTicks(t *testing.T) { RuleStore: ruleStore, Metrics: testMetrics.GetSchedulerMetrics(), AlertSender: notifier, + Tracer: testTracer, } st := state.NewManager(testMetrics.GetStateMetrics(), nil, nil, &state.NoopImageService{}, mockedClock, &state.FakeHistorian{}) @@ -639,6 +642,7 @@ func TestSchedule_DeleteAlertRule(t *testing.T) { func setupScheduler(t *testing.T, rs *fakeRulesStore, is *state.FakeInstanceStore, registry *prometheus.Registry, senderMock *AlertsSenderMock, evalMock eval.EvaluatorFactory) *schedule { t.Helper() + testTracer := tracing.InitializeTracerForTest() mockedClock := clock.NewMock() @@ -684,6 +688,7 @@ func setupScheduler(t *testing.T, rs *fakeRulesStore, is *state.FakeInstanceStor RuleStore: rs, Metrics: m.GetSchedulerMetrics(), AlertSender: senderMock, + Tracer: testTracer, } st := state.NewManager(m.GetStateMetrics(), nil, is, &state.NoopImageService{}, mockedClock, &state.FakeHistorian{}) diff --git a/pkg/services/ngalert/tests/util.go b/pkg/services/ngalert/tests/util.go index b03c54a751f..c7df12d3774 100644 --- a/pkg/services/ngalert/tests/util.go +++ b/pkg/services/ngalert/tests/util.go @@ -95,12 +95,13 @@ func SetupTestEnv(tb testing.TB, baseInterval time.Duration) (*ngalert.AlertNG, features, folderPermissions, dashboardPermissions, ac, ) - bus := bus.ProvideBus(tracing.InitializeTracerForTest()) + tracer := tracing.InitializeTracerForTest() + bus := bus.ProvideBus(tracer) folderService := folderimpl.ProvideService(ac, bus, cfg, dashboardService, dashboardStore, nil, features, folderPermissions, nil) ng, err := ngalert.ProvideService( cfg, &FakeFeatures{}, nil, nil, routing.NewRouteRegister(), sqlStore, nil, nil, nil, quotatest.New(false, nil), - secretsService, nil, m, folderService, ac, &dashboards.FakeDashboardService{}, nil, bus, ac, annotationstest.NewFakeAnnotationsRepo(), &plugins.FakePluginStore{}, + secretsService, nil, m, folderService, ac, &dashboards.FakeDashboardService{}, nil, bus, ac, annotationstest.NewFakeAnnotationsRepo(), &plugins.FakePluginStore{}, tracer, ) require.NoError(tb, err) return ng, &store.DBstore{ diff --git a/pkg/services/quota/quotaimpl/quota_test.go b/pkg/services/quota/quotaimpl/quota_test.go index f1943c51432..3bacfb87d96 100644 --- a/pkg/services/quota/quotaimpl/quota_test.go +++ b/pkg/services/quota/quotaimpl/quota_test.go @@ -465,6 +465,7 @@ func getQuotaBySrvTargetScope(t *testing.T, quotaService quota.Service, srv quot } func setupEnv(t *testing.T, sqlStore *sqlstore.SQLStore, b bus.Bus, quotaService quota.Service) { + tracer := tracing.InitializeTracerForTest() _, err := apikeyimpl.ProvideService(sqlStore, sqlStore.Cfg, quotaService) require.NoError(t, err) _, err = authimpl.ProvideUserAuthTokenService(sqlStore, nil, nil, featuremgmt.WithFeatures(), quotaService, sqlStore.Cfg) @@ -478,7 +479,7 @@ func setupEnv(t *testing.T, sqlStore *sqlstore.SQLStore, b bus.Bus, quotaService m := metrics.NewNGAlert(prometheus.NewRegistry()) _, err = ngalert.ProvideService( sqlStore.Cfg, &ngalerttests.FakeFeatures{}, nil, nil, routing.NewRouteRegister(), sqlStore, nil, nil, nil, quotaService, - secretsService, nil, m, &foldertest.FakeService{}, &acmock.Mock{}, &dashboards.FakeDashboardService{}, nil, b, &acmock.Mock{}, annotationstest.NewFakeAnnotationsRepo(), &plugins.FakePluginStore{}, + secretsService, nil, m, &foldertest.FakeService{}, &acmock.Mock{}, &dashboards.FakeDashboardService{}, nil, b, &acmock.Mock{}, annotationstest.NewFakeAnnotationsRepo(), &plugins.FakePluginStore{}, tracer, ) require.NoError(t, err) _, err = storesrv.ProvideService(sqlStore, featuremgmt.WithFeatures(), sqlStore.Cfg, quotaService, storesrv.ProvideSystemUsersService())