Alerting: Add compressed periodic save for alert instances (#111803)

What is this feature?

This PR implements compressed periodic save for alert state storage, providing a more efficient alternative to regular periodic saves by grouping alert instances by rule UID and storing them using protobuf and snappy compression. When enabled via the state_compressed_periodic_save_enabled configuration option, the system groups alert instances by their alert rule, compresses each group using protobuf serialization and snappy compression, and processes all rules within a single database transaction at specified intervals instead of syncing after every alert evaluation cycle.

Why do we need this feature?

During discussions in PR #111357, we identified the need for a compressed approach to periodic alert state storage that could further reduce database load beyond the jitter mechanism. While the jitter feature distributes database operations over time, this compressed periodic save approach reduces the frequency of database operations by batching alert state updates at explicitly declared intervals rather than syncing after every alert evaluation cycle.
This approach provides several key benefits:

- Reduced Database Frequency: Instead of frequent sync operations tied to alert evaluation cycles, updates occur only at configured intervals
- Storage Efficiency: Rule-based grouping with protobuf and snappy compression significantly reduces storage requirements

The compressed periodic save complements the existing jitter mechanism by providing an alternative strategy focused on reducing overall database interaction frequency while maintaining data integrity through compression and batching.

Who is this feature for?

- Platform/Infrastructure teams managing large-scale Grafana deployments with high alert cardinality
- Organizations looking to optimize storage costs and database performance for alerting workloads
- Production environments with 1000+ alert rules where database write frequency is a concern

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.
 

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)

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

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

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

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
 	}

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 {

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