diff --git a/docs/sources/alerting/set-up/performance-limitations/index.md b/docs/sources/alerting/set-up/performance-limitations/index.md index 82e8ccd41f1..b18f3b6ec68 100644 --- a/docs/sources/alerting/set-up/performance-limitations/index.md +++ b/docs/sources/alerting/set-up/performance-limitations/index.md @@ -68,7 +68,21 @@ You can change this behavior by disabling the `alertingSaveStateCompressed` feat You can also reduce database load by writing states periodically instead of after every evaluation. -To save state periodically: +There are two approaches for periodic state saving: + +#### Compressed periodic saves + +You can combine compressed alert state storage with periodic saves by enabling both `alertingSaveStateCompressed` and `alertingSaveStatePeriodic` feature toggles together. + +This approach groups all alert instances by rule UID and compresses them together for efficient storage. + +When both feature toggles are enabled, Grafana will save compressed alert states at the interval specified by `state_periodic_save_interval`. Note that in compressed mode, the `state_periodic_save_batch_size` setting is ignored as the system groups instances by rule UID rather than by batch size. + +#### Batch-based periodic saves + +Alternatively, you can use batch-based periodic saves without compression: + +This approach processes individual alert instances in batches of a specified size. 1. Enable the `alertingSaveStatePeriodic` feature toggle. 1. Disable the `alertingSaveStateCompressed` feature toggle. @@ -77,7 +91,7 @@ By default, it saves the states every 5 minutes to the database and on each shut can also be configured using the `state_periodic_save_interval` configuration flag. During this process, Grafana deletes all existing alert instances from the database and then writes the entire current set of instances back in batches in a single transaction. Configure the size of each batch using the `state_periodic_save_batch_size` configuration option. -#### Jitter for periodic saves +##### Jitter for batch-based periodic saves To further distribute database load, you can enable jitter for periodic state saves by setting `state_periodic_save_jitter_enabled = true`. When jitter is enabled, instead of saving all batches simultaneously, Grafana spreads the batch writes across a calculated time window of 85% of the save interval. diff --git a/pkg/services/ngalert/ngalert.go b/pkg/services/ngalert/ngalert.go index ebb7e1061c2..6de3f7b1ba0 100644 --- a/pkg/services/ngalert/ngalert.go +++ b/pkg/services/ngalert/ngalert.go @@ -504,13 +504,6 @@ func initInstanceStore(sqlStore db.DB, logger log.Logger, featureToggles feature if featureToggles.IsEnabledGlobally(featuremgmt.FlagAlertingSaveStateCompressed) { logger.Info("Using protobuf-based alert instance store") instanceStore = protoInstanceStore - // If FlagAlertingSaveStateCompressed is enabled, ProtoInstanceDBStore is used, - // which functions differently from InstanceDBStore. FlagAlertingSaveStatePeriodic is - // not applicable to ProtoInstanceDBStore, so a warning is logged if it is set. - //nolint:staticcheck // not yet migrated to OpenFeature - if featureToggles.IsEnabledGlobally(featuremgmt.FlagAlertingSaveStatePeriodic) { - logger.Warn("alertingSaveStatePeriodic is not used when alertingSaveStateCompressed feature flag enabled") - } } else { logger.Info("Using simple database alert instance store") instanceStore = simpleInstanceStore @@ -525,7 +518,15 @@ func initStatePersister(uaCfg setting.UnifiedAlertingSettings, cfg state.Manager //nolint:staticcheck // not yet migrated to OpenFeature if featureToggles.IsEnabledGlobally(featuremgmt.FlagAlertingSaveStateCompressed) { logger.Info("Using rule state persister") - statePersister = state.NewSyncRuleStatePersisiter(logger, cfg) + + if featureToggles.IsEnabledGlobally(featuremgmt.FlagAlertingSaveStatePeriodic) { + logger.Info("Compressed storage with periodic save enabled") + ticker := clock.New().Ticker(cfg.StatePeriodicSaveInterval) + statePersister = state.NewSyncRuleStatePersisiter(logger, ticker, cfg) + } else { + logger.Info("Compressed storage FullSync disabled") + statePersister = state.NewSyncRuleStatePersisiter(logger, nil, cfg) + } } else if featureToggles.IsEnabledGlobally(featuremgmt.FlagAlertingSaveStatePeriodic) { logger.Info("Using periodic state persister") ticker := clock.New().Ticker(uaCfg.StatePeriodicSaveInterval) diff --git a/pkg/services/ngalert/ngalert_test.go b/pkg/services/ngalert/ngalert_test.go index c9adb2603cc..ef81499cd77 100644 --- a/pkg/services/ngalert/ngalert_test.go +++ b/pkg/services/ngalert/ngalert_test.go @@ -436,7 +436,9 @@ func TestInitStatePersister(t *testing.T) { ua := setting.UnifiedAlertingSettings{ StatePeriodicSaveInterval: 1 * time.Minute, } - cfg := state.ManagerCfg{} + cfg := state.ManagerCfg{ + StatePeriodicSaveInterval: 1 * time.Minute, + } tests := []struct { name string diff --git a/pkg/services/ngalert/state/persist.go b/pkg/services/ngalert/state/persist.go index 7c7442b986e..1ad5beed0ca 100644 --- a/pkg/services/ngalert/state/persist.go +++ b/pkg/services/ngalert/state/persist.go @@ -8,6 +8,10 @@ import ( history_model "github.com/grafana/grafana/pkg/services/ngalert/state/historian/model" ) +type AlertInstancesProvider interface { + GetAlertInstances() []models.AlertInstance +} + // InstanceStore represents the ability to fetch and write alert instances. type InstanceStore interface { InstanceReader diff --git a/pkg/services/ngalert/state/persister_async.go b/pkg/services/ngalert/state/persister_async.go index 0de1dd5cdde..f76eed5f06e 100644 --- a/pkg/services/ngalert/state/persister_async.go +++ b/pkg/services/ngalert/state/persister_async.go @@ -12,10 +12,6 @@ import ( "github.com/grafana/grafana/pkg/services/ngalert/models" ) -type AlertInstancesProvider interface { - GetAlertInstances() []models.AlertInstance -} - type AsyncStatePersister struct { log log.Logger batchSize int diff --git a/pkg/services/ngalert/state/persister_sync_rule.go b/pkg/services/ngalert/state/persister_sync_rule.go index da2494d3647..a237d2983f2 100644 --- a/pkg/services/ngalert/state/persister_sync_rule.go +++ b/pkg/services/ngalert/state/persister_sync_rule.go @@ -4,6 +4,7 @@ import ( "context" "time" + "github.com/benbjohnson/clock" "go.opentelemetry.io/otel/trace" "github.com/grafana/grafana/pkg/infra/log" @@ -11,22 +12,63 @@ import ( ) type SyncRuleStatePersister struct { - log log.Logger - store InstanceStore + log log.Logger + store InstanceStore + ticker *clock.Ticker } -func NewSyncRuleStatePersisiter(log log.Logger, cfg ManagerCfg) StatePersister { +func NewSyncRuleStatePersisiter(log log.Logger, ticker *clock.Ticker, cfg ManagerCfg) StatePersister { return &SyncRuleStatePersister{ - log: log, - store: cfg.InstanceStore, + log: log, + store: cfg.InstanceStore, + ticker: ticker, } } -func (a *SyncRuleStatePersister) Async(_ context.Context, _ AlertInstancesProvider) { - a.log.Debug("Async: No-Op") +func (a *SyncRuleStatePersister) Async(ctx context.Context, instancesProvider AlertInstancesProvider) { + if a.ticker == nil { + return + } + + for { + select { + case <-a.ticker.C: + if err := a.fullSync(ctx, instancesProvider); err != nil { + a.log.Error("Failed to do a full compressed state sync to database", "err", err) + } + case <-ctx.Done(): + a.log.Info("Scheduler is shutting down, doing a final state sync.") + if err := a.fullSync(context.Background(), instancesProvider); err != nil { + a.log.Error("Failed to do a full compressed state sync to database", "err", err) + } + a.ticker.Stop() + a.log.Info("Compressed state async worker is shut down.") + return + } + } +} + +func (a *SyncRuleStatePersister) fullSync(ctx context.Context, instancesProvider AlertInstancesProvider) error { + startTime := time.Now() + a.log.Debug("Full compressed state sync start") + instances := instancesProvider.GetAlertInstances() + + // batchSize is set to 0 because compressed storage groups instances by ruleUID, not by batch size + err := a.store.FullSync(ctx, instances, 0, nil) + if err != nil { + a.log.Error("Full compressed state sync failed", "duration", time.Since(startTime), "instances", len(instances)) + return err + } + a.log.Debug("Full compressed state sync done", "duration", time.Since(startTime), "instances", len(instances)) + return nil } func (a *SyncRuleStatePersister) Sync(ctx context.Context, span trace.Span, ruleKey models.AlertRuleKeyWithGroup, states StateTransitions) { + if a.ticker != nil { + a.log.Debug("Skip immediate save, using periodic save instead") + return + } + if a.store == nil || len(states) == 0 { return } diff --git a/pkg/services/ngalert/store/proto_instance_database.go b/pkg/services/ngalert/store/proto_instance_database.go index eaa58253ac4..a9450d0851f 100644 --- a/pkg/services/ngalert/store/proto_instance_database.go +++ b/pkg/services/ngalert/store/proto_instance_database.go @@ -9,6 +9,7 @@ import ( "time" "github.com/golang/snappy" + "github.com/grafana/grafana/pkg/services/sqlstore" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" @@ -98,28 +99,13 @@ func (st ProtoInstanceDBStore) SaveAlertInstancesForRule(ctx context.Context, ke logger := st.Logger.FromContext(ctx) logger.Debug("SaveAlertInstancesForRule called", "rule_uid", key.UID, "org_id", key.OrgID, "instances", len(instances)) - alert_instances_proto := make([]*pb.AlertInstance, len(instances)) - - for i, instance := range instances { - alert_instances_proto[i] = alertInstanceModelToProto(instance) - } - - compressedAlertInstances, err := compressAlertInstances(alert_instances_proto) + compressedAlertInstances, err := convertAndCompressAlertInstances(instances) if err != nil { return fmt.Errorf("failed to compress alert instances: %w", err) } - return st.SQLStore.WithTransactionalDbSession(ctx, func(sess *db.Session) error { - params := []any{key.OrgID, key.UID, compressedAlertInstances, time.Now()} - - upsertSQL := st.SQLStore.GetDialect().UpsertSQL( - "alert_rule_state", - []string{"org_id", "rule_uid"}, - []string{"org_id", "rule_uid", "data", "updated_at"}, - ) - _, err = sess.SQL(upsertSQL, params...).Query() - - return err + return st.SQLStore.WithTransactionalDbSession(ctx, func(sess *sqlstore.DBSession) error { + return st.upsertCompressedAlertInstances(sess, key.OrgID, key.UID, compressedAlertInstances, time.Now()) }) } @@ -134,9 +120,68 @@ func (st ProtoInstanceDBStore) DeleteAlertInstancesByRule(ctx context.Context, k } func (st ProtoInstanceDBStore) FullSync(ctx context.Context, instances []models.AlertInstance, batchSize int, jitterFunc func(int) time.Duration) error { + if len(instances) == 0 { + return nil + } + logger := st.Logger.FromContext(ctx) - logger.Error("FullSync called and not implemented") - return errors.New("fullsync is not implemented for proto instance database store") + logger.Debug("FullSync called", "total_instances", len(instances)) + + ruleGroups := make(map[models.AlertRuleKeyWithGroup][]models.AlertInstance) + for _, instance := range instances { + ruleKey := models.AlertRuleKeyWithGroup{ + AlertRuleKey: models.AlertRuleKey{ + OrgID: instance.RuleOrgID, + UID: instance.RuleUID, + }, + RuleGroup: "", + } + ruleGroups[ruleKey] = append(ruleGroups[ruleKey], instance) + } + + type preparedRule struct { + ruleKey models.AlertRuleKeyWithGroup + compressedData []byte + } + preparedRules := make([]preparedRule, 0, len(ruleGroups)) + + for ruleKey, ruleInstances := range ruleGroups { + // Convert and compress instances + compressedAlertInstances, err := convertAndCompressAlertInstances(ruleInstances) + if err != nil { + logger.Error("Failed to compress instances for rule", "rule_uid", ruleKey.UID, "error", err) + continue + } + + preparedRules = append(preparedRules, preparedRule{ + ruleKey: ruleKey, + compressedData: compressedAlertInstances, + }) + + logger.Debug("Prepared rule for sync", "rule_uid", ruleKey.UID, "org_id", ruleKey.OrgID, "instances", len(ruleInstances)) + } + + return st.SQLStore.WithTransactionalDbSession(ctx, func(sess *sqlstore.DBSession) error { + syncTimestamp := time.Now() + logger.Debug("Starting FullSync transaction", "rules_count", len(preparedRules), "timestamp", syncTimestamp) + + // First we delete all records from the table + if _, err := sess.Exec("DELETE FROM alert_rule_state"); err != nil { + return fmt.Errorf("failed to delete alert_rule_state: %w", err) + } + + for i, prepared := range preparedRules { + logger.Debug("Executing UPSERT for rule", "rule_uid", prepared.ruleKey.UID, "org_id", prepared.ruleKey.OrgID, "rule_index", i+1, "total_rules", len(preparedRules)) + + // Execute UPSERT with pre-compressed data using helper method + if err := st.upsertCompressedAlertInstances(sess, prepared.ruleKey.OrgID, prepared.ruleKey.UID, prepared.compressedData, syncTimestamp); err != nil { + return fmt.Errorf("failed to save instances for rule %s: %w", prepared.ruleKey.UID, err) + } + } + + logger.Debug("FullSync transaction completed successfully", "rules_synced", len(preparedRules)) + return nil + }) } func alertInstanceModelToProto(modelInstance models.AlertInstance) *pb.AlertInstance { @@ -155,6 +200,30 @@ func alertInstanceModelToProto(modelInstance models.AlertInstance) *pb.AlertInst } } +// convertAndCompressAlertInstances converts model instances to protobuf and compresses them +func convertAndCompressAlertInstances(instances []models.AlertInstance) ([]byte, error) { + alertInstancesProto := make([]*pb.AlertInstance, len(instances)) + for i, instance := range instances { + alertInstancesProto[i] = alertInstanceModelToProto(instance) + } + + return compressAlertInstances(alertInstancesProto) +} + +// upsertCompressedAlertInstances performs upsert operation for compressed alert instances +func (st ProtoInstanceDBStore) upsertCompressedAlertInstances(sess *sqlstore.DBSession, orgID int64, ruleUID string, compressedData []byte, timestamp time.Time) error { + upsertSQL := st.SQLStore.GetDialect().UpsertSQL( + "alert_rule_state", + []string{"org_id", "rule_uid"}, + []string{"org_id", "rule_uid", "data", "updated_at"}, + ) + + params := []any{orgID, ruleUID, compressedData, timestamp} + _, err := sess.SQL(upsertSQL, params...).Query() + + return err +} + func compressAlertInstances(instances []*pb.AlertInstance) ([]byte, error) { mProto, err := proto.Marshal(&pb.AlertInstances{Instances: instances}) if err != nil { diff --git a/pkg/services/ngalert/store/proto_instance_database_test.go b/pkg/services/ngalert/store/proto_instance_database_test.go index 04428a0efae..c98923995dd 100644 --- a/pkg/services/ngalert/store/proto_instance_database_test.go +++ b/pkg/services/ngalert/store/proto_instance_database_test.go @@ -174,6 +174,166 @@ func TestCompressAndDecompressAlertInstances(t *testing.T) { require.EqualExportedValues(t, alertInstances[1], decompressedInstances[1]) } +func TestConvertAndCompressAlertInstances(t *testing.T) { + now := time.Now() + + modelInstances := []models.AlertInstance{ + { + AlertInstanceKey: models.AlertInstanceKey{ + RuleUID: "rule-uid-1", + RuleOrgID: 1, + LabelsHash: "hash-1", + }, + Labels: map[string]string{"label-1": "value-1"}, + CurrentState: models.InstanceStateFiring, + CurrentStateSince: now, + CurrentStateEnd: now.Add(time.Hour), + CurrentReason: "reason-1", + LastEvalTime: now.Add(-time.Minute), + LastSentAt: &now, + FiredAt: &now, + ResolvedAt: nil, + ResultFingerprint: "fingerprint-1", + }, + { + AlertInstanceKey: models.AlertInstanceKey{ + RuleUID: "rule-uid-1", + RuleOrgID: 1, + LabelsHash: "hash-2", + }, + Labels: map[string]string{"label-2": "value-2"}, + CurrentState: models.InstanceStateNormal, + CurrentStateSince: now, + CurrentStateEnd: now.Add(time.Hour), + CurrentReason: "reason-2", + LastEvalTime: now.Add(-time.Minute), + LastSentAt: nil, + FiredAt: nil, + ResolvedAt: &now, + ResultFingerprint: "fingerprint-2", + }, + } + + compressedData, err := convertAndCompressAlertInstances(modelInstances) + require.NoError(t, err) + require.NotEmpty(t, compressedData) + + // Verify we can decompress and get back the same data + decompressedInstances, err := decompressAlertInstances(compressedData) + require.NoError(t, err) + require.Len(t, decompressedInstances, 2) + + // Convert back to model to compare + for i, protoInstance := range decompressedInstances { + modelInstance := alertInstanceProtoToModel("rule-uid-1", 1, protoInstance) + require.Equal(t, modelInstances[i].Labels, modelInstance.Labels) + require.Equal(t, modelInstances[i].CurrentState, modelInstance.CurrentState) + require.Equal(t, modelInstances[i].LabelsHash, modelInstance.LabelsHash) + require.Equal(t, modelInstances[i].ResultFingerprint, modelInstance.ResultFingerprint) + } +} + +func TestConvertAndCompressAlertInstances_EmptyInput(t *testing.T) { + emptyInstances := []models.AlertInstance{} + + compressedData, err := convertAndCompressAlertInstances(emptyInstances) + require.NoError(t, err) + + decompressedInstances, err := decompressAlertInstances(compressedData) + require.NoError(t, err) + require.Empty(t, decompressedInstances) +} + +func TestFullSyncGroupingLogic(t *testing.T) { + now := time.Now() + + // Test instances from multiple rules to verify grouping logic + instances := []models.AlertInstance{ + { + AlertInstanceKey: models.AlertInstanceKey{ + RuleUID: "rule-1", + RuleOrgID: 1, + LabelsHash: "hash-1-1", + }, + Labels: models.InstanceLabels{"rule1": "instance1"}, + CurrentState: models.InstanceStateFiring, + CurrentStateSince: now, + CurrentStateEnd: now.Add(time.Hour), + CurrentReason: "test reason 1", + LastEvalTime: now.Add(-time.Minute), + ResultFingerprint: "fingerprint-1-1", + }, + { + AlertInstanceKey: models.AlertInstanceKey{ + RuleUID: "rule-1", + RuleOrgID: 1, + LabelsHash: "hash-1-2", + }, + Labels: models.InstanceLabels{"rule1": "instance2"}, + CurrentState: models.InstanceStateNormal, + CurrentStateSince: now, + CurrentStateEnd: now.Add(time.Hour), + CurrentReason: "test reason 2", + LastEvalTime: now.Add(-time.Minute), + ResultFingerprint: "fingerprint-1-2", + }, + { + AlertInstanceKey: models.AlertInstanceKey{ + RuleUID: "rule-2", + RuleOrgID: 1, + LabelsHash: "hash-2-1", + }, + Labels: models.InstanceLabels{"rule2": "instance1"}, + CurrentState: models.InstanceStatePending, + CurrentStateSince: now, + CurrentStateEnd: now.Add(time.Hour), + CurrentReason: "test reason 3", + LastEvalTime: now.Add(-time.Minute), + ResultFingerprint: "fingerprint-2-1", + }, + } + + // Test the grouping logic that FullSync uses internally + ruleGroups := make(map[models.AlertRuleKeyWithGroup][]models.AlertInstance) + for _, instance := range instances { + ruleKey := models.AlertRuleKeyWithGroup{ + AlertRuleKey: models.AlertRuleKey{ + OrgID: instance.RuleOrgID, + UID: instance.RuleUID, + }, + RuleGroup: "", + } + ruleGroups[ruleKey] = append(ruleGroups[ruleKey], instance) + } + + // Verify grouping worked correctly + require.Len(t, ruleGroups, 2, "Should have 2 rule groups") + + rule1Key := models.AlertRuleKeyWithGroup{ + AlertRuleKey: models.AlertRuleKey{OrgID: 1, UID: "rule-1"}, + RuleGroup: "", + } + rule2Key := models.AlertRuleKeyWithGroup{ + AlertRuleKey: models.AlertRuleKey{OrgID: 1, UID: "rule-2"}, + RuleGroup: "", + } + + require.Len(t, ruleGroups[rule1Key], 2, "Rule 1 should have 2 instances") + require.Len(t, ruleGroups[rule2Key], 1, "Rule 2 should have 1 instance") + + // Test compression for each group + for ruleKey, ruleInstances := range ruleGroups { + compressedData, err := convertAndCompressAlertInstances(ruleInstances) + require.NoError(t, err, "Compression should succeed for rule %s", ruleKey.UID) + require.NotEmpty(t, compressedData, "Compressed data should not be empty for rule %s", ruleKey.UID) + + // Verify decompression works + decompressedInstances, err := decompressAlertInstances(compressedData) + require.NoError(t, err, "Decompression should succeed for rule %s", ruleKey.UID) + require.Len(t, decompressedInstances, len(ruleInstances), "Should have same number of instances after decompression for rule %s", ruleKey.UID) + } +} + func toProtoTimestampPtr(tm *time.Time) *timestamppb.Timestamp { if tm == nil { return nil