Alerting: Improve rule validation to check if rule uses backend datasources (#58986)

* validate if rule uses backend datasources

* add backend datasource to test

* fix tests

* another forgotten import

* remove unused var

pkg/services/ngalert/eval/eval.go

@@ -15,6 +15,7 @@ import (
 	"github.com/grafana/grafana/pkg/expr"
 	"github.com/grafana/grafana/pkg/expr/classic"
 	"github.com/grafana/grafana/pkg/infra/log"
+	"github.com/grafana/grafana/pkg/plugins"
 	"github.com/grafana/grafana/pkg/services/datasources"
 	"github.com/grafana/grafana/pkg/services/ngalert/models"
 	"github.com/grafana/grafana/pkg/setting"
@@ -83,16 +84,20 @@ type evaluatorImpl struct {
 	evaluationTimeout time.Duration
 	dataSourceCache   datasources.CacheService
 	expressionService *expr.Service
+	pluginsStore      plugins.Store
 }
 
 func NewEvaluatorFactory(
 	cfg setting.UnifiedAlertingSettings,
 	datasourceCache datasources.CacheService,
-	expressionService *expr.Service) EvaluatorFactory {
+	expressionService *expr.Service,
+	pluginsStore plugins.Store,
+) EvaluatorFactory {
 	return &evaluatorImpl{
 		evaluationTimeout: cfg.EvaluationTimeout,
 		dataSourceCache:   datasourceCache,
 		expressionService: expressionService,
+		pluginsStore:      pluginsStore,
 	}
 }
 
@@ -591,7 +596,23 @@ func (evalResults Results) AsDataFrame() data.Frame {
 }
 
 func (e *evaluatorImpl) Validate(ctx EvaluationContext, condition models.Condition) error {
-	_, err := e.Create(ctx, condition)
+	req, err := getExprRequest(ctx, condition.Data, e.dataSourceCache)
+	if err != nil {
+		return err
+	}
+	for _, query := range req.Queries {
+		if query.DataSource == nil || expr.IsDataSource(query.DataSource.Uid) {
+			continue
+		}
+		p, found := e.pluginsStore.Plugin(ctx.Ctx, query.DataSource.Type)
+		if !found { // technically this should fail earlier during datasource resolution phase.
+			return fmt.Errorf("datasource refID %s could not be found: %w", query.RefID, plugins.ErrPluginUnavailable)
+		}
+		if !p.Backend {
+			return fmt.Errorf("datasource refID %s is not a backend datasource", query.RefID)
+		}
+	}
+	_, err = e.create(condition, req)
 	return err
 }
 
@@ -606,6 +627,10 @@ func (e *evaluatorImpl) Create(ctx EvaluationContext, condition models.Condition
 	if err != nil {
 		return nil, err
 	}
+	return e.create(condition, req)
+}
+
+func (e *evaluatorImpl) create(condition models.Condition, req *expr.Request) (ConditionEvaluator, error) {
 	pipeline, err := e.expressionService.BuildPipeline(req)
 	if err != nil {
 		return nil, err

pkg/services/ngalert/eval/eval_test.go

@@ -12,11 +12,13 @@ import (
 	ptr "github.com/xorcare/pointer"
 
 	"github.com/grafana/grafana/pkg/expr"
+	"github.com/grafana/grafana/pkg/plugins"
 	"github.com/grafana/grafana/pkg/services/datasources"
 	fakes "github.com/grafana/grafana/pkg/services/datasources/fakes"
 	"github.com/grafana/grafana/pkg/services/ngalert/models"
 	"github.com/grafana/grafana/pkg/services/user"
 	"github.com/grafana/grafana/pkg/setting"
+	"github.com/grafana/grafana/pkg/util"
 )
 
 func TestEvaluateExecutionResult(t *testing.T) {
@@ -351,15 +353,20 @@ func TestEvaluateExecutionResultsNoData(t *testing.T) {
 }
 
 func TestValidate(t *testing.T) {
+	type services struct {
+		cache        *fakes.FakeCacheService
+		pluginsStore *plugins.FakePluginStore
+	}
+
 	testCases := []struct {
 		name      string
-		condition func(service *fakes.FakeCacheService) models.Condition
+		condition func(services services) models.Condition
 		error     bool
 	}{
 		{
 			name:  "fail if no expressions",
 			error: true,
-			condition: func(service *fakes.FakeCacheService) models.Condition {
+			condition: func(_ services) models.Condition {
 				return models.Condition{
 					Condition: "A",
 					Data:      []models.AlertQuery{},
@@ -369,17 +376,25 @@ func TestValidate(t *testing.T) {
 		{
 			name:  "fail if condition RefID does not exist",
 			error: true,
-			condition: func(service *fakes.FakeCacheService) models.Condition {
+			condition: func(services services) models.Condition {
-				ds := models.GenerateAlertQuery()
+				dsQuery := models.GenerateAlertQuery()
-				service.DataSources = append(service.DataSources, &datasources.DataSource{
+				ds := &datasources.DataSource{
-					Uid: ds.DatasourceUID,
+					Uid:  dsQuery.DatasourceUID,
+					Type: util.GenerateShortUID(),
+				}
+				services.cache.DataSources = append(services.cache.DataSources, ds)
+				services.pluginsStore.PluginList = append(services.pluginsStore.PluginList, plugins.PluginDTO{
+					JSONData: plugins.JSONData{
+						ID:      ds.Type,
+						Backend: true,
+					},
 				})
 
 				return models.Condition{
 					Condition: "C",
 					Data: []models.AlertQuery{
-						ds,
+						dsQuery,
-						models.CreateClassicConditionExpression("B", ds.RefID, "last", "gt", rand.Int()),
+						models.CreateClassicConditionExpression("B", dsQuery.RefID, "last", "gt", rand.Int()),
 					},
 				}
 			},
@@ -387,17 +402,24 @@ func TestValidate(t *testing.T) {
 		{
 			name:  "fail if condition RefID is empty",
 			error: true,
-			condition: func(service *fakes.FakeCacheService) models.Condition {
+			condition: func(services services) models.Condition {
-				ds := models.GenerateAlertQuery()
+				dsQuery := models.GenerateAlertQuery()
-				service.DataSources = append(service.DataSources, &datasources.DataSource{
+				ds := &datasources.DataSource{
-					Uid: ds.DatasourceUID,
+					Uid:  dsQuery.DatasourceUID,
+					Type: util.GenerateShortUID(),
+				}
+				services.cache.DataSources = append(services.cache.DataSources, ds)
+				services.pluginsStore.PluginList = append(services.pluginsStore.PluginList, plugins.PluginDTO{
+					JSONData: plugins.JSONData{
+						ID:      ds.Type,
+						Backend: true,
+					},
 				})
-
 				return models.Condition{
 					Condition: "",
 					Data: []models.AlertQuery{
-						ds,
+						dsQuery,
-						models.CreateClassicConditionExpression("B", ds.RefID, "last", "gt", rand.Int()),
+						models.CreateClassicConditionExpression("B", dsQuery.RefID, "last", "gt", rand.Int()),
 					},
 				}
 			},
@@ -405,13 +427,68 @@ func TestValidate(t *testing.T) {
 		{
 			name:  "fail if datasource with UID does not exists",
 			error: true,
-			condition: func(service *fakes.FakeCacheService) models.Condition {
+			condition: func(services services) models.Condition {
-				ds := models.GenerateAlertQuery()
+				dsQuery := models.GenerateAlertQuery()
 				// do not update the cache service
 				return models.Condition{
-					Condition: ds.RefID,
+					Condition: dsQuery.RefID,
 					Data: []models.AlertQuery{
-						ds,
+						dsQuery,
+					},
+				}
+			},
+		},
+		{
+			name:  "fail if datasource cannot be found in plugin store",
+			error: true,
+			condition: func(services services) models.Condition {
+				dsQuery := models.GenerateAlertQuery()
+				ds := &datasources.DataSource{
+					Uid:  dsQuery.DatasourceUID,
+					Type: util.GenerateShortUID(),
+				}
+				services.cache.DataSources = append(services.cache.DataSources, ds)
+				// do not update the plugin store
+				return models.Condition{
+					Condition: dsQuery.RefID,
+					Data: []models.AlertQuery{
+						dsQuery,
+					},
+				}
+			},
+		},
+		{
+			name:  "fail if datasource is not backend one",
+			error: true,
+			condition: func(services services) models.Condition {
+				dsQuery1 := models.GenerateAlertQuery()
+				dsQuery2 := models.GenerateAlertQuery()
+				ds1 := &datasources.DataSource{
+					Uid:  dsQuery1.DatasourceUID,
+					Type: util.GenerateShortUID(),
+				}
+				ds2 := &datasources.DataSource{
+					Uid:  dsQuery2.DatasourceUID,
+					Type: util.GenerateShortUID(),
+				}
+				services.cache.DataSources = append(services.cache.DataSources, ds1, ds2)
+				services.pluginsStore.PluginList = append(services.pluginsStore.PluginList, plugins.PluginDTO{
+					JSONData: plugins.JSONData{
+						ID:      ds1.Type,
+						Backend: false,
+					},
+				}, plugins.PluginDTO{
+					JSONData: plugins.JSONData{
+						ID:      ds2.Type,
+						Backend: true,
+					},
+				})
+				// do not update the plugin store
+				return models.Condition{
+					Condition: dsQuery1.RefID,
+					Data: []models.AlertQuery{
+						dsQuery1,
+						dsQuery2,
 					},
 				}
 			},
@@ -419,17 +496,25 @@ func TestValidate(t *testing.T) {
 		{
 			name:  "pass if datasource exists and condition is correct",
 			error: false,
-			condition: func(service *fakes.FakeCacheService) models.Condition {
+			condition: func(services services) models.Condition {
-				ds := models.GenerateAlertQuery()
+				dsQuery := models.GenerateAlertQuery()
-				service.DataSources = append(service.DataSources, &datasources.DataSource{
+				ds := &datasources.DataSource{
-					Uid: ds.DatasourceUID,
+					Uid:  dsQuery.DatasourceUID,
+					Type: util.GenerateShortUID(),
+				}
+				services.cache.DataSources = append(services.cache.DataSources, ds)
+				services.pluginsStore.PluginList = append(services.pluginsStore.PluginList, plugins.PluginDTO{
+					JSONData: plugins.JSONData{
+						ID:      ds.Type,
+						Backend: true,
+					},
 				})
 
 				return models.Condition{
 					Condition: "B",
 					Data: []models.AlertQuery{
-						ds,
+						dsQuery,
-						models.CreateClassicConditionExpression("B", ds.RefID, "last", "gt", rand.Int()),
+						models.CreateClassicConditionExpression("B", dsQuery.RefID, "last", "gt", rand.Int()),
 					},
 				}
 			},
@@ -441,12 +526,16 @@ func TestValidate(t *testing.T) {
 
 		t.Run(testCase.name, func(t *testing.T) {
 			cacheService := &fakes.FakeCacheService{}
-			condition := testCase.condition(cacheService)
+			store := &plugins.FakePluginStore{}
+			condition := testCase.condition(services{
+				cache:        cacheService,
+				pluginsStore: store,
+			})
 
-			evaluator := NewEvaluatorFactory(setting.UnifiedAlertingSettings{}, cacheService, expr.ProvideService(&setting.Cfg{ExpressionsEnabled: true}, nil, nil))
+			evaluator := NewEvaluatorFactory(setting.UnifiedAlertingSettings{}, cacheService, expr.ProvideService(&setting.Cfg{ExpressionsEnabled: true}, nil, nil), store)
 			evalCtx := Context(context.Background(), u)
 
-			_, err := evaluator.Create(evalCtx, condition)
+			err := evaluator.Validate(evalCtx, condition)
 			if testCase.error {
 				require.Error(t, err)
 			} else {

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/plugins"
 	"github.com/grafana/grafana/pkg/services/accesscontrol"
 	"github.com/grafana/grafana/pkg/services/annotations"
 	"github.com/grafana/grafana/pkg/services/dashboards"
@@ -62,6 +63,7 @@ func ProvideService(
 	bus bus.Bus,
 	accesscontrolService accesscontrol.Service,
 	annotationsRepo annotations.Repository,
+	pluginsStore plugins.Store,
 ) (*AlertNG, error) {
 	ng := &AlertNG{
 		Cfg:                  cfg,
@@ -85,6 +87,7 @@ func ProvideService(
 		bus:                  bus,
 		accesscontrolService: accesscontrolService,
 		annotationsRepo:      annotationsRepo,
+		pluginsStore:         pluginsStore,
 	}
 
 	if ng.IsDisabled() {
@@ -130,6 +133,7 @@ type AlertNG struct {
 	store                *store.DBstore
 
 	bus          bus.Bus
+	pluginsStore plugins.Store
 }
 
 func (ng *AlertNG) init() error {
@@ -182,7 +186,7 @@ func (ng *AlertNG) init() error {
 
 	ng.AlertsRouter = alertsRouter
 
-	evalFactory := eval.NewEvaluatorFactory(ng.Cfg.UnifiedAlerting, ng.DataSourceCache, ng.ExpressionService)
+	evalFactory := eval.NewEvaluatorFactory(ng.Cfg.UnifiedAlerting, ng.DataSourceCache, ng.ExpressionService, ng.pluginsStore)
 	schedCfg := schedule.SchedulerCfg{
 		MaxAttempts:          ng.Cfg.UnifiedAlerting.MaxAttempts,
 		C:                    clk,

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/plugins"
 	"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/metrics"
@@ -651,7 +652,7 @@ func setupScheduler(t *testing.T, rs *fakeRulesStore, is *state.FakeInstanceStor
 
 	var evaluator = evalMock
 	if evalMock == nil {
-		evaluator = eval.NewEvaluatorFactory(setting.UnifiedAlertingSettings{}, nil, expr.ProvideService(&setting.Cfg{ExpressionsEnabled: true}, nil, nil))
+		evaluator = eval.NewEvaluatorFactory(setting.UnifiedAlertingSettings{}, nil, expr.ProvideService(&setting.Cfg{ExpressionsEnabled: true}, nil, nil), &plugins.FakePluginStore{})
 	}
 
 	if registry == nil {

pkg/services/ngalert/tests/util.go

@@ -18,6 +18,7 @@ import (
 	"github.com/grafana/grafana/pkg/infra/db"
 	"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"
 	acmock "github.com/grafana/grafana/pkg/services/accesscontrol/mock"
 	"github.com/grafana/grafana/pkg/services/annotations/annotationstest"
@@ -99,7 +100,7 @@ func SetupTestEnv(tb testing.TB, baseInterval time.Duration) (*ngalert.AlertNG,
 
 	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(),
+		secretsService, nil, m, folderService, ac, &dashboards.FakeDashboardService{}, nil, bus, ac, annotationstest.NewFakeAnnotationsRepo(), &plugins.FakePluginStore{},
 	)
 	require.NoError(tb, err)
 	return ng, &store.DBstore{

pkg/services/quota/quotaimpl/quota_test.go

@@ -5,10 +5,15 @@ import (
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
+	"github.com/xorcare/pointer"
+
 	"github.com/grafana/grafana/pkg/api/routing"
 	"github.com/grafana/grafana/pkg/bus"
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/infra/tracing"
+	"github.com/grafana/grafana/pkg/plugins"
 	acmock "github.com/grafana/grafana/pkg/services/accesscontrol/mock"
 	"github.com/grafana/grafana/pkg/services/annotations/annotationstest"
 	"github.com/grafana/grafana/pkg/services/apikey"
@@ -38,9 +43,6 @@ import (
 	"github.com/grafana/grafana/pkg/services/user"
 	"github.com/grafana/grafana/pkg/services/user/userimpl"
 	"github.com/grafana/grafana/pkg/setting"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/stretchr/testify/require"
-	"github.com/xorcare/pointer"
 )
 
 func TestQuotaService(t *testing.T) {
@@ -476,7 +478,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(),
+		secretsService, nil, m, &foldertest.FakeService{}, &acmock.Mock{}, &dashboards.FakeDashboardService{}, nil, b, &acmock.Mock{}, annotationstest.NewFakeAnnotationsRepo(), &plugins.FakePluginStore{},
 	)
 	require.NoError(t, err)
 	_, err = storesrv.ProvideService(sqlStore, featuremgmt.WithFeatures(), sqlStore.Cfg, quotaService, storesrv.ProvideSystemUsersService())