From 934ba1aaa1fa4c1b9f1dcc0d6753c67a23c4f466 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jean-Philippe=20Qu=C3=A9m=C3=A9ner?= Date: Fri, 16 Jun 2023 19:55:49 +0200 Subject: [PATCH] Alerting: Rewrite range to instant queries if possible (#69976) --- .../feature-toggles/index.md | 1 + .../src/types/featureToggles.gen.ts | 1 + pkg/services/featuremgmt/registry.go | 7 + pkg/services/featuremgmt/toggles_gen.csv | 1 + pkg/services/featuremgmt/toggles_gen.go | 4 + pkg/services/ngalert/store/alert_rule.go | 14 +- pkg/services/ngalert/store/alert_rule_test.go | 8 +- .../ngalert/store/loki_range_to_instant.go | 63 ++++++ .../store/loki_range_to_instant_test.go | 182 ++++++++++++++++++ 9 files changed, 277 insertions(+), 4 deletions(-) create mode 100644 pkg/services/ngalert/store/loki_range_to_instant.go create mode 100644 pkg/services/ngalert/store/loki_range_to_instant_test.go diff --git a/docs/sources/setup-grafana/configure-grafana/feature-toggles/index.md b/docs/sources/setup-grafana/configure-grafana/feature-toggles/index.md index d61fabe78a8..35adb9b4ee7 100644 --- a/docs/sources/setup-grafana/configure-grafana/feature-toggles/index.md +++ b/docs/sources/setup-grafana/configure-grafana/feature-toggles/index.md @@ -114,6 +114,7 @@ Experimental features might be changed or removed without prior notice. | `cloudWatchLogsMonacoEditor` | Enables the Monaco editor for CloudWatch Logs queries | | `exploreScrollableLogsContainer` | Improves the scrolling behavior of logs in Explore | | `recordedQueriesMulti` | Enables writing multiple items from a single query within Recorded Queries | +| `alertingLokiRangeToInstant` | Rewrites eligible loki range queries to instant queries | ## Development feature toggles diff --git a/packages/grafana-data/src/types/featureToggles.gen.ts b/packages/grafana-data/src/types/featureToggles.gen.ts index 656f35f3260..63cdb3ba10c 100644 --- a/packages/grafana-data/src/types/featureToggles.gen.ts +++ b/packages/grafana-data/src/types/featureToggles.gen.ts @@ -101,4 +101,5 @@ export interface FeatureToggles { cloudWatchLogsMonacoEditor?: boolean; exploreScrollableLogsContainer?: boolean; recordedQueriesMulti?: boolean; + alertingLokiRangeToInstant?: boolean; } diff --git a/pkg/services/featuremgmt/registry.go b/pkg/services/featuremgmt/registry.go index 87881d457fa..2a453d80d46 100644 --- a/pkg/services/featuremgmt/registry.go +++ b/pkg/services/featuremgmt/registry.go @@ -564,5 +564,12 @@ var ( Stage: FeatureStageExperimental, Owner: grafanaObservabilityMetricsSquad, }, + { + Name: "alertingLokiRangeToInstant", + Description: "Rewrites eligible loki range queries to instant queries", + Stage: FeatureStageExperimental, + FrontendOnly: false, + Owner: grafanaAlertingSquad, + }, } ) diff --git a/pkg/services/featuremgmt/toggles_gen.csv b/pkg/services/featuremgmt/toggles_gen.csv index 402eff8fd73..fc563b88b95 100644 --- a/pkg/services/featuremgmt/toggles_gen.csv +++ b/pkg/services/featuremgmt/toggles_gen.csv @@ -82,3 +82,4 @@ sqlDatasourceDatabaseSelection,preview,@grafana/grafana-bi-squad,false,false,fal cloudWatchLogsMonacoEditor,experimental,@grafana/aws-plugins,false,false,false,true exploreScrollableLogsContainer,experimental,@grafana/observability-logs,false,false,false,true recordedQueriesMulti,experimental,@grafana/observability-metrics,false,false,false,false +alertingLokiRangeToInstant,experimental,@grafana/alerting-squad,false,false,false,false diff --git a/pkg/services/featuremgmt/toggles_gen.go b/pkg/services/featuremgmt/toggles_gen.go index 108242217dd..f783217a187 100644 --- a/pkg/services/featuremgmt/toggles_gen.go +++ b/pkg/services/featuremgmt/toggles_gen.go @@ -338,4 +338,8 @@ const ( // FlagRecordedQueriesMulti // Enables writing multiple items from a single query within Recorded Queries FlagRecordedQueriesMulti = "recordedQueriesMulti" + + // FlagAlertingLokiRangeToInstant + // Rewrites eligible loki range queries to instant queries + FlagAlertingLokiRangeToInstant = "alertingLokiRangeToInstant" ) diff --git a/pkg/services/ngalert/store/alert_rule.go b/pkg/services/ngalert/store/alert_rule.go index 78f468a1ca6..4e8fe18481c 100644 --- a/pkg/services/ngalert/store/alert_rule.go +++ b/pkg/services/ngalert/store/alert_rule.go @@ -10,6 +10,7 @@ import ( "github.com/grafana/grafana/pkg/infra/db" "github.com/grafana/grafana/pkg/services/dashboards" + "github.com/grafana/grafana/pkg/services/featuremgmt" "github.com/grafana/grafana/pkg/services/folder" ngmodels "github.com/grafana/grafana/pkg/services/ngalert/models" "github.com/grafana/grafana/pkg/services/search/model" @@ -539,7 +540,7 @@ func (st DBstore) GetAlertRulesForScheduling(ctx context.Context, query *ngmodel st.Logger.Error("unable to close rows session", "error", err) } }() - + lokiRangeToInstantEnabled := st.FeatureToggles.IsEnabled(featuremgmt.FlagAlertingLokiRangeToInstant) // Deserialize each rule separately in case any of them contain invalid JSON. for rows.Next() { rule := new(ngmodels.AlertRule) @@ -548,6 +549,17 @@ func (st DBstore) GetAlertRulesForScheduling(ctx context.Context, query *ngmodel st.Logger.Error("Invalid rule found in DB store, ignoring it", "func", "GetAlertRulesForScheduling", "error", err) continue } + // This was added to mitigate the high load that could be created by loki range queries. + // In previous versions of Grafana, Loki datasources would default to range queries + // instead of instant queries, sometimes creating unnecessary load. This is only + // done for Grafana Cloud. + if lokiRangeToInstantEnabled && canBeInstant(rule) { + if err := migrateToInstant(rule); err != nil { + st.Logger.Error("Could not migrate rule from range to instant query", "rule", rule.UID, "err", err) + } else { + st.Logger.Info("Migrated rule from range to instant query", "rule", rule.UID) + } + } rules = append(rules, rule) } diff --git a/pkg/services/ngalert/store/alert_rule_test.go b/pkg/services/ngalert/store/alert_rule_test.go index 493d267cce6..7323097b4e9 100644 --- a/pkg/services/ngalert/store/alert_rule_test.go +++ b/pkg/services/ngalert/store/alert_rule_test.go @@ -12,6 +12,7 @@ import ( "github.com/grafana/grafana/pkg/infra/log" "github.com/grafana/grafana/pkg/infra/log/logtest" "github.com/grafana/grafana/pkg/infra/tracing" + "github.com/grafana/grafana/pkg/services/featuremgmt" "github.com/grafana/grafana/pkg/services/folder" "github.com/grafana/grafana/pkg/services/folder/folderimpl" "github.com/grafana/grafana/pkg/services/ngalert/testutil" @@ -324,9 +325,10 @@ func TestIntegration_GetAlertRulesForScheduling(t *testing.T) { sqlStore := db.InitTestDB(t) store := &DBstore{ - SQLStore: sqlStore, - Cfg: cfg.UnifiedAlerting, - FolderService: setupFolderService(t, sqlStore, cfg), + SQLStore: sqlStore, + Cfg: cfg.UnifiedAlerting, + FolderService: setupFolderService(t, sqlStore, cfg), + FeatureToggles: featuremgmt.WithFeatures(), } generator := models.AlertRuleGen(withIntervalMatching(store.Cfg.BaseInterval), models.WithUniqueID(), models.WithUniqueOrgID()) diff --git a/pkg/services/ngalert/store/loki_range_to_instant.go b/pkg/services/ngalert/store/loki_range_to_instant.go new file mode 100644 index 00000000000..4524ed066bc --- /dev/null +++ b/pkg/services/ngalert/store/loki_range_to_instant.go @@ -0,0 +1,63 @@ +package store + +import ( + "encoding/json" + + "github.com/grafana/grafana/pkg/services/ngalert/models" +) + +const ( + grafanaCloudLogs = "grafanacloud-logs" + grafanaCloudUsageInsights = "grafanacloud-usage-insights" + grafanaCloudStateHistory = "grafanacloud-loki-alert-state-history" +) + +func canBeInstant(r *models.AlertRule) bool { + if len(r.Data) < 2 { + return false + } + // First query part should be range query. + if r.Data[0].QueryType != "range" { + return false + } + // First query part should go to cloud logs or insights. + if r.Data[0].DatasourceUID != grafanaCloudLogs && + r.Data[0].DatasourceUID != grafanaCloudUsageInsights && + r.Data[0].DatasourceUID != grafanaCloudStateHistory { + return false + } + // Second query part should be and expression, '-100' is the legacy way to define it. + if r.Data[1].DatasourceUID != "__expr__" && r.Data[1].DatasourceUID != "-100" { + return false + } + exprRaw := make(map[string]interface{}) + if err := json.Unmarshal(r.Data[1].Model, &exprRaw); err != nil { + return false + } + // Second query part should be "last()" + if val, ok := exprRaw["reducer"].(string); !ok || val != "last" { + return false + } + // Second query part should use first query part as expression. + if ref, ok := exprRaw["expression"].(string); !ok || ref != r.Data[0].RefID { + return false + } + return true +} + +// migrateToInstant will move a range-query to an instant query. This should only +// be used for loki. +func migrateToInstant(r *models.AlertRule) error { + modelRaw := make(map[string]interface{}) + if err := json.Unmarshal(r.Data[0].Model, &modelRaw); err != nil { + return err + } + modelRaw["queryType"] = "instant" + model, err := json.Marshal(modelRaw) + if err != nil { + return err + } + r.Data[0].Model = model + r.Data[0].QueryType = "instant" + return nil +} diff --git a/pkg/services/ngalert/store/loki_range_to_instant_test.go b/pkg/services/ngalert/store/loki_range_to_instant_test.go new file mode 100644 index 00000000000..46aa34133de --- /dev/null +++ b/pkg/services/ngalert/store/loki_range_to_instant_test.go @@ -0,0 +1,182 @@ +package store + +import ( + "encoding/json" + "testing" + + "github.com/grafana/grafana/pkg/services/ngalert/models" + "github.com/stretchr/testify/require" +) + +func TestCanBeInstant(t *testing.T) { + tcs := []struct { + name string + expected bool + rule *models.AlertRule + }{ + { + name: "valid rule that can be migrated from range to instant", + expected: true, + rule: createMigrateableLokiRule(t), + }, + { + name: "invalid rule where the data array is too short to be migrateable", + expected: false, + rule: createMigrateableLokiRule(t, func(r *models.AlertRule) { + r.Data = []models.AlertQuery{r.Data[0]} + }), + }, + { + name: "invalid rule that is not a range query", + expected: false, + rule: createMigrateableLokiRule(t, func(r *models.AlertRule) { + r.Data[0].QueryType = "something-else" + }), + }, + { + name: "invalid rule that does not use a cloud datasource", + expected: false, + rule: createMigrateableLokiRule(t, func(r *models.AlertRule) { + r.Data[0].DatasourceUID = "something-else" + }), + }, + { + name: "invalid rule that has no aggregation as second item", + expected: false, + rule: createMigrateableLokiRule(t, func(r *models.AlertRule) { + r.Data[1].DatasourceUID = "something-else" + }), + }, + { + name: "invalid rule that has not last() as aggregation", + expected: false, + rule: createMigrateableLokiRule(t, func(r *models.AlertRule) { + raw := make(map[string]interface{}) + err := json.Unmarshal(r.Data[1].Model, &raw) + require.NoError(t, err) + raw["reducer"] = "avg" + r.Data[1].Model, err = json.Marshal(raw) + require.NoError(t, err) + }), + }, + { + name: "invalid rule that has not last() pointing to range query", + expected: false, + rule: createMigrateableLokiRule(t, func(r *models.AlertRule) { + raw := make(map[string]interface{}) + err := json.Unmarshal(r.Data[1].Model, &raw) + require.NoError(t, err) + raw["expression"] = "C" + r.Data[1].Model, err = json.Marshal(raw) + require.NoError(t, err) + }), + }, + } + for _, tc := range tcs { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.expected, canBeInstant(tc.rule)) + }) + } +} + +func TestMigrateLokiQueryToInstant(t *testing.T) { + original := createMigrateableLokiRule(t) + mirgrated := createMigrateableLokiRule(t, func(r *models.AlertRule) { + r.Data[0].QueryType = "instant" + r.Data[0].Model = []byte(`{ + "datasource": { + "type": "loki", + "uid": "grafanacloud-logs" + }, + "editorMode": "code", + "expr": "1", + "hide": false, + "intervalMs": 1000, + "maxDataPoints": 43200, + "queryType": "instant", + "refId": "A" + }`) + }) + + require.True(t, canBeInstant(original)) + require.NoError(t, migrateToInstant(original)) + + require.Equal(t, mirgrated.Data[0].QueryType, original.Data[0].QueryType) + + originalModel := make(map[string]interface{}) + require.NoError(t, json.Unmarshal(original.Data[0].Model, &originalModel)) + migratedModel := make(map[string]interface{}) + require.NoError(t, json.Unmarshal(mirgrated.Data[0].Model, &migratedModel)) + + require.Equal(t, migratedModel, originalModel) + + require.False(t, canBeInstant(original)) +} + +func createMigrateableLokiRule(t *testing.T, muts ...func(*models.AlertRule)) *models.AlertRule { + t.Helper() + r := &models.AlertRule{ + Data: []models.AlertQuery{ + { + RefID: "A", + QueryType: "range", + DatasourceUID: grafanaCloudLogs, + Model: []byte(`{ + "datasource": { + "type": "loki", + "uid": "grafanacloud-logs" + }, + "editorMode": "code", + "expr": "1", + "hide": false, + "intervalMs": 1000, + "maxDataPoints": 43200, + "queryType": "range", + "refId": "A" + }`), + }, + { + RefID: "B", + DatasourceUID: "__expr__", + Model: []byte(`{ + "conditions": [ + { + "evaluator": { + "params": [], + "type": "gt" + }, + "operator": { + "type": "and" + }, + "query": { + "params": [ + "B" + ] + }, + "reducer": { + "params": [], + "type": "last" + }, + "type": "query" + } + ], + "datasource": { + "type": "__expr__", + "uid": "__expr__" + }, + "expression": "A", + "hide": false, + "intervalMs": 1000, + "maxDataPoints": 43200, + "reducer": "last", + "refId": "B", + "type": "reduce" + }`), + }, + }, + } + for _, m := range muts { + m(r) + } + return r +}