Alerting: Add recording rule target datasource support to Prometheus conversion API (#101799)

What is this feature?

Adds target datasource UID to the recording rules so that they write to the same datasource used for alerting rule queries after the import.

Why do we need this feature?

Target datasourse support was added in #101678, and under a feature flag grafanaManagedRecordingRulesDatasources (#101778).

This PR makes the importing process:
    Check if the import contains recording rules
    Verify both recording rules and the grafanaManagedRecordingRulesDatasources feature flag are enabled
    If either check fails, return an error
    If both checks pass, create recording rules with the provided datasource UID set as both the query and target datasource

pkg/services/ngalert/api/api.go

@@ -188,7 +188,14 @@ func (api *API) RegisterAPIEndpoints(m *metrics.API) {
 
 	if api.FeatureManager.IsEnabledGlobally(featuremgmt.FlagAlertingConversionAPI) {
 		api.RegisterConvertPrometheusApiEndpoints(NewConvertPrometheusApi(
-			NewConvertPrometheusSrv(&api.Cfg.UnifiedAlerting, logger, api.RuleStore, api.DatasourceCache, api.AlertRules),
+			NewConvertPrometheusSrv(
+				&api.Cfg.UnifiedAlerting,
+				logger,
+				api.RuleStore,
+				api.DatasourceCache,
+				api.AlertRules,
+				api.FeatureManager,
+			),
 		), m)
 	}
 }

pkg/services/ngalert/api/api_convert_prometheus.go

@@ -18,6 +18,7 @@ import (
 	contextmodel "github.com/grafana/grafana/pkg/services/contexthandler/model"
 	"github.com/grafana/grafana/pkg/services/dashboards"
 	"github.com/grafana/grafana/pkg/services/datasources"
+	"github.com/grafana/grafana/pkg/services/featuremgmt"
 	"github.com/grafana/grafana/pkg/services/folder"
 	apimodels "github.com/grafana/grafana/pkg/services/ngalert/api/tooling/definitions"
 	"github.com/grafana/grafana/pkg/services/ngalert/models"
@@ -47,7 +48,17 @@ var (
 	).Errorf("missing datasource UID header")
 
 	errInvalidHeaderValueMsg  = "Invalid value for header {{.Public.Header}}: must be 'true' or 'false'"
-	errInvalidHeaderValueBase = errutil.ValidationFailed("aleting.invalidHeaderValue").MustTemplate(errInvalidHeaderValueMsg, errutil.WithPublic(errInvalidHeaderValueMsg))
+	errInvalidHeaderValueBase = errutil.ValidationFailed("alerting.invalidHeaderValue").MustTemplate(errInvalidHeaderValueMsg, errutil.WithPublic(errInvalidHeaderValueMsg))
+
+	errRecordingRulesNotEnabled = errutil.ValidationFailed(
+		"alerting.recordingRulesNotEnabled",
+		errutil.WithPublicMessage("Cannot import recording rules: Feature not enabled."),
+	).Errorf("recording rules not enabled")
+
+	errRecordingRulesDatasourcesNotEnabled = errutil.ValidationFailed(
+		"alerting.recordingRulesDatasourcesNotEnabled",
+		errutil.WithPublicMessage("Cannot import recording rules: Configuration of target datasources not enabled."),
+	).Errorf("recording rules target datasources configuration not enabled")
 )
 
 func errInvalidHeaderValue(header string) error {
@@ -98,15 +109,24 @@ type ConvertPrometheusSrv struct {
 	ruleStore        RuleStore
 	datasourceCache  datasources.CacheService
 	alertRuleService *provisioning.AlertRuleService
+	featureToggles   featuremgmt.FeatureToggles
 }
 
-func NewConvertPrometheusSrv(cfg *setting.UnifiedAlertingSettings, logger log.Logger, ruleStore RuleStore, datasourceCache datasources.CacheService, alertRuleService *provisioning.AlertRuleService) *ConvertPrometheusSrv {
+func NewConvertPrometheusSrv(
+	cfg *setting.UnifiedAlertingSettings,
+	logger log.Logger,
+	ruleStore RuleStore,
+	datasourceCache datasources.CacheService,
+	alertRuleService *provisioning.AlertRuleService,
+	featureToggles featuremgmt.FeatureToggles,
+) *ConvertPrometheusSrv {
 	return &ConvertPrometheusSrv{
 		cfg:              cfg,
 		logger:           logger,
 		ruleStore:        ruleStore,
 		datasourceCache:  datasourceCache,
 		alertRuleService: alertRuleService,
+		featureToggles:   featureToggles,
 	}
 }
 
@@ -301,6 +321,20 @@ func (srv *ConvertPrometheusSrv) RouteConvertPrometheusPostRuleGroup(c *contextm
 	workingFolderUID := getWorkingFolderUID(c)
 	logger = logger.New("folder_title", namespaceTitle, "group", promGroup.Name, "working_folder_uid", workingFolderUID)
 
+	// If we're importing recording rules, we can only import them if the feature is enabled,
+	// and the feature flag that enables configuring target datasources per-rule is also enabled.
+	if promGroupHasRecordingRules(promGroup) {
+		if !srv.cfg.RecordingRules.Enabled {
+			logger.Error("Cannot import recording rules", "error", errRecordingRulesNotEnabled)
+			return errorToResponse(errRecordingRulesNotEnabled)
+		}
+
+		if !srv.featureToggles.IsEnabledGlobally(featuremgmt.FlagGrafanaManagedRecordingRulesDatasources) {
+			logger.Error("Cannot import recording rules", "error", errRecordingRulesDatasourcesNotEnabled)
+			return errorToResponse(errRecordingRulesDatasourcesNotEnabled)
+		}
+	}
+
 	logger.Info("Converting Prometheus rule group", "rules", len(promGroup.Rules))
 
 	ns, errResp := srv.getOrCreateNamespace(c, namespaceTitle, logger, workingFolderUID)
@@ -494,3 +528,12 @@ func namespaceErrorResponse(err error) response.Response {
 
 	return toNamespaceErrorResponse(err)
 }
+
+func promGroupHasRecordingRules(promGroup apimodels.PrometheusRuleGroup) bool {
+	for _, rule := range promGroup.Rules {
+		if rule.Record != "" {
+			return true
+		}
+	}
+	return false
+}

pkg/services/ngalert/api/api_convert_prometheus_test.go

@@ -17,6 +17,7 @@ import (
 	contextmodel "github.com/grafana/grafana/pkg/services/contexthandler/model"
 	"github.com/grafana/grafana/pkg/services/datasources"
 	dsfakes "github.com/grafana/grafana/pkg/services/datasources/fakes"
+	"github.com/grafana/grafana/pkg/services/featuremgmt"
 	"github.com/grafana/grafana/pkg/services/folder"
 	"github.com/grafana/grafana/pkg/services/folder/foldertest"
 	acfakes "github.com/grafana/grafana/pkg/services/ngalert/accesscontrol/fakes"
@@ -48,6 +49,13 @@ func TestRouteConvertPrometheusPostRuleGroup(t *testing.T) {
 					"severity": "critical",
 				},
 			},
+			{
+				Record: "recorded-metric",
+				Expr:   "vector(1)",
+				Labels: map[string]string{
+					"severity": "warning",
+				},
+			},
 		},
 	}
 
@@ -104,21 +112,39 @@ func TestRouteConvertPrometheusPostRuleGroup(t *testing.T) {
 		response := srv.RouteConvertPrometheusPostRuleGroup(rc, fldr.Title, simpleGroup)
 		require.Equal(t, http.StatusAccepted, response.Status())
 
-		// Get the updated rule
+		// Get the rules
 		remaining, err := ruleStore.ListAlertRules(context.Background(), &models.ListAlertRulesQuery{
 			OrgID: 1,
 		})
 		require.NoError(t, err)
-		require.Len(t, remaining, 1)
+		require.Len(t, remaining, 2)
-
+
-		require.Equal(t, simpleGroup.Name, remaining[0].RuleGroup)
+		// Create a map of rule titles to their expected definitions
-		require.Equal(t, fmt.Sprintf("[%s] %s", simpleGroup.Name, simpleGroup.Rules[0].Alert), remaining[0].Title)
+		expectedRules := make(map[string]string)
-		promRuleYAML, err := yaml.Marshal(simpleGroup.Rules[0])
+		for _, rule := range simpleGroup.Rules {
+			if rule.Alert != "" {
+				title := fmt.Sprintf("[%s] %s", simpleGroup.Name, rule.Alert)
+				promRuleYAML, err := yaml.Marshal(rule)
 				require.NoError(t, err)
+				expectedRules[title] = string(promRuleYAML)
+			} else if rule.Record != "" {
+				title := fmt.Sprintf("[%s] %s", simpleGroup.Name, rule.Record)
+				promRuleYAML, err := yaml.Marshal(rule)
+				require.NoError(t, err)
+				expectedRules[title] = string(promRuleYAML)
+			}
+		}
 
-		promDefinition, err := remaining[0].PrometheusRuleDefinition()
+		// Verify each rule matches its expected definition
+		for _, r := range remaining {
+			require.Equal(t, simpleGroup.Name, r.RuleGroup)
+			expectedDef, exists := expectedRules[r.Title]
+			require.True(t, exists, "unexpected rule title: %s", r.Title)
+
+			promDefinition, err := r.PrometheusRuleDefinition()
 			require.NoError(t, err)
-		require.Equal(t, string(promRuleYAML), promDefinition)
+			require.Equal(t, expectedDef, promDefinition)
+		}
 	})
 
 	t.Run("should fail to replace a provisioned rule group", func(t *testing.T) {
@@ -263,6 +289,58 @@ func TestRouteConvertPrometheusPostRuleGroup(t *testing.T) {
 		response := srv.RouteConvertPrometheusPostRuleGroup(rc, "test", simpleGroup)
 		require.Equal(t, http.StatusAccepted, response.Status())
 	})
+
+	t.Run("with disabled recording rules", func(t *testing.T) {
+		testCases := []struct {
+			name                   string
+			recordingRules         bool
+			recordingRulesTargetDS bool
+			expectedStatus         int
+		}{
+			{
+				name:                   "when recording rules are enabled",
+				recordingRules:         true,
+				recordingRulesTargetDS: true,
+				expectedStatus:         http.StatusAccepted,
+			},
+			{
+				name:                   "when recording rules are disabled",
+				recordingRules:         false,
+				recordingRulesTargetDS: true,
+				expectedStatus:         http.StatusBadRequest,
+			},
+			{
+				name:                   "when target datasources for recording rules are disabled",
+				recordingRules:         true,
+				recordingRulesTargetDS: false,
+				expectedStatus:         http.StatusBadRequest,
+			},
+			{
+				name:                   "when both recording rules and target datasources are disabled",
+				recordingRules:         false,
+				recordingRulesTargetDS: false,
+				expectedStatus:         http.StatusBadRequest,
+			},
+		}
+
+		for _, tc := range testCases {
+			t.Run(tc.name, func(t *testing.T) {
+				var features featuremgmt.FeatureToggles
+				if tc.recordingRulesTargetDS {
+					features = featuremgmt.WithFeatures(featuremgmt.FlagGrafanaManagedRecordingRulesDatasources)
+				} else {
+					features = featuremgmt.WithFeatures()
+				}
+
+				srv, _, _, _ := createConvertPrometheusSrv(t, withFeatureToggles(features))
+				srv.cfg.RecordingRules.Enabled = tc.recordingRules
+				rc := createRequestCtx()
+
+				response := srv.RouteConvertPrometheusPostRuleGroup(rc, "test", simpleGroup)
+				require.Equal(t, tc.expectedStatus, response.Status())
+			})
+		}
+	})
 }
 
 func TestRouteConvertPrometheusGetRuleGroup(t *testing.T) {
@@ -783,6 +861,7 @@ type convertPrometheusSrvOptions struct {
 	provenanceStore              provisioning.ProvisioningStore
 	fakeAccessControlRuleService *acfakes.FakeRuleService
 	quotaChecker                 *provisioning.MockQuotaChecker
+	featureToggles               featuremgmt.FeatureToggles
 }
 
 type convertPrometheusSrvOptionsFunc func(*convertPrometheusSrvOptions)
@@ -805,6 +884,12 @@ func withQuotaChecker(checker *provisioning.MockQuotaChecker) convertPrometheusS
 	}
 }
 
+func withFeatureToggles(toggles featuremgmt.FeatureToggles) convertPrometheusSrvOptionsFunc {
+	return func(opts *convertPrometheusSrvOptions) {
+		opts.featureToggles = toggles
+	}
+}
+
 func createConvertPrometheusSrv(t *testing.T, opts ...convertPrometheusSrvOptionsFunc) (*ConvertPrometheusSrv, datasources.CacheService, *fakes.RuleStore, *foldertest.FakeService) {
 	t.Helper()
 
@@ -816,6 +901,7 @@ func createConvertPrometheusSrv(t *testing.T, opts ...convertPrometheusSrvOption
 		provenanceStore:              fakes.NewFakeProvisioningStore(),
 		fakeAccessControlRuleService: &acfakes.FakeRuleService{},
 		quotaChecker:                 quotas,
+		featureToggles:               featuremgmt.WithFeatures(featuremgmt.FlagGrafanaManagedRecordingRulesDatasources),
 	}
 
 	for _, opt := range opts {
@@ -851,9 +937,12 @@ func createConvertPrometheusSrv(t *testing.T, opts ...convertPrometheusSrvOption
 
 	cfg := &setting.UnifiedAlertingSettings{
 		DefaultRuleEvaluationInterval: 1 * time.Minute,
+		RecordingRules: setting.RecordingRuleSettings{
+			Enabled: true,
+		},
 	}
 
-	srv := NewConvertPrometheusSrv(cfg, log.NewNopLogger(), ruleStore, dsCache, alertRuleService)
+	srv := NewConvertPrometheusSrv(cfg, log.NewNopLogger(), ruleStore, dsCache, alertRuleService, options.featureToggles)
 
 	return srv, dsCache, ruleStore, folderService
 }

pkg/services/ngalert/prom/convert.go

@@ -193,6 +193,7 @@ func (p *Converter) convertRule(orgID int64, namespaceUID string, promGroup Prom
 		record = &models.Record{
 			From:                queryRefID,
 			Metric:              rule.Record,
+			TargetDatasourceUID: p.cfg.DatasourceUID,
 		}
 
 		isPaused = p.cfg.RecordingRules.IsPaused

pkg/services/ngalert/prom/convert_test.go

@@ -198,6 +198,7 @@ func TestPrometheusRulesToGrafana(t *testing.T) {
 					require.NotNil(t, grafanaRule.Record)
 					require.Equal(t, grafanaRule.Record.From, queryRefID)
 					require.Equal(t, promRule.Record, grafanaRule.Record.Metric)
+					require.Equal(t, tc.config.DatasourceUID, grafanaRule.Record.TargetDatasourceUID)
 				} else {
 					require.Equal(t, fmt.Sprintf("[%s] %s", tc.promGroup.Name, promRule.Alert), grafanaRule.Title)
 				}

pkg/tests/api/alerting/api_convert_prometheus_test.go

@@ -108,7 +108,8 @@ func TestIntegrationConvertPrometheusEndpoints(t *testing.T) {
 			EnableUnifiedAlerting: true,
 			DisableAnonymous:      true,
 			AppModeProduction:     true,
-			EnableFeatureToggles:  []string{"alertingConversionAPI"},
+			EnableFeatureToggles:  []string{"alertingConversionAPI", "grafanaManagedRecordingRulesDatasources", "grafanaManagedRecordingRules"},
+			EnableRecordingRules:  true,
 		})
 
 		grafanaListedAddr, env := testinfra.StartGrafanaEnv(t, dir, gpath)
@@ -219,7 +220,8 @@ func TestIntegrationConvertPrometheusEndpoints_UpdateRule(t *testing.T) {
 			EnableUnifiedAlerting: true,
 			DisableAnonymous:      true,
 			AppModeProduction:     true,
-			EnableFeatureToggles:  []string{"alertingConversionAPI"},
+			EnableFeatureToggles:  []string{"alertingConversionAPI", "grafanaManagedRecordingRulesDatasources", "grafanaManagedRecordingRules"},
+			EnableRecordingRules:  true,
 		})
 
 		grafanaListedAddr, env := testinfra.StartGrafanaEnv(t, dir, gpath)
@@ -305,7 +307,8 @@ func TestIntegrationConvertPrometheusEndpoints_Conflict(t *testing.T) {
 			EnableUnifiedAlerting: true,
 			DisableAnonymous:      true,
 			AppModeProduction:     true,
-			EnableFeatureToggles:  []string{"alertingConversionAPI"},
+			EnableFeatureToggles:  []string{"alertingConversionAPI", "grafanaManagedRecordingRulesDatasources", "grafanaManagedRecordingRules"},
+			EnableRecordingRules:  true,
 		})
 
 		grafanaListedAddr, env := testinfra.StartGrafanaEnv(t, dir, gpath)
@@ -392,7 +395,8 @@ func TestIntegrationConvertPrometheusEndpoints_CreatePausedRules(t *testing.T) {
 			EnableUnifiedAlerting: true,
 			DisableAnonymous:      true,
 			AppModeProduction:     true,
-			EnableFeatureToggles:  []string{"alertingConversionAPI"},
+			EnableFeatureToggles:  []string{"alertingConversionAPI", "grafanaManagedRecordingRulesDatasources", "grafanaManagedRecordingRules"},
+			EnableRecordingRules:  true,
 		})
 
 		grafanaListedAddr, env := testinfra.StartGrafanaEnv(t, dir, path)
@@ -507,7 +511,8 @@ func TestIntegrationConvertPrometheusEndpoints_FolderUIDHeader(t *testing.T) {
 			EnableUnifiedAlerting: true,
 			DisableAnonymous:      true,
 			AppModeProduction:     true,
-			EnableFeatureToggles:  []string{"alertingConversionAPI"},
+			EnableFeatureToggles:  []string{"alertingConversionAPI", "grafanaManagedRecordingRulesDatasources", "grafanaManagedRecordingRules"},
+			EnableRecordingRules:  true,
 		})
 
 		grafanaListedAddr, env := testinfra.StartGrafanaEnv(t, dir, path)
@@ -608,7 +613,8 @@ func TestIntegrationConvertPrometheusEndpoints_Delete(t *testing.T) {
 			EnableUnifiedAlerting: true,
 			DisableAnonymous:      true,
 			AppModeProduction:     true,
-			EnableFeatureToggles:  []string{"alertingConversionAPI"},
+			EnableFeatureToggles:  []string{"alertingConversionAPI", "grafanaManagedRecordingRulesDatasources", "grafanaManagedRecordingRules"},
+			EnableRecordingRules:  true,
 		})
 
 		grafanaListedAddr, env := testinfra.StartGrafanaEnv(t, dir, gpath)

pkg/tests/testinfra/testinfra.go

@@ -295,6 +295,13 @@ func CreateGrafDir(t *testing.T, opts GrafanaOpts) (string, string) {
 	_, err = alertingSect.NewKey("max_attempts", "3")
 	require.NoError(t, err)
 
+	if opts.EnableRecordingRules {
+		recordingRulesSect, err := cfg.NewSection("recording_rules")
+		require.NoError(t, err)
+		_, err = recordingRulesSect.NewKey("enabled", "true")
+		require.NoError(t, err)
+	}
+
 	if opts.LicensePath != "" {
 		section, err := cfg.NewSection("enterprise")
 		require.NoError(t, err)
@@ -537,6 +544,7 @@ type GrafanaOpts struct {
 	UnifiedStorageConfig                  map[string]setting.UnifiedStorageConfig
 	GrafanaComSSOAPIToken                 string
 	LicensePath                           string
+	EnableRecordingRules                  bool
 
 	// When "unified-grpc" is selected it will also start the grpc server
 	APIServerStorageType options.StorageType