Emit error when the rule synchronization fails

Signed-off-by: SungJin1212 <[email protected]>

Author: SungJin1212

CHANGELOG.md

@@ -18,6 +18,7 @@
 * [FEATURE] Distributor/Ingester: Implemented experimental feature to use gRPC stream connection for push requests. This can be enabled by setting `-distributor.use-stream-push=true`. #6580
 * [FEATURE] Compactor: Add support for percentage based sharding for compactors. #6738
 * [FEATURE] Querier: Allow choosing PromQL engine via header. #6777
+* [ENHANCEMENT] Ruler: Emit an error message when the rule synchronization fails. #6902
 * [ENHANCEMENT] Tenant Federation: Add a # of query result limit logic when the `-tenant-federation.regex-matcher-enabled` is enabled. #6845
 * [ENHANCEMENT] Query Frontend: Add a `cortex_slow_queries_total` metric to track # of slow queries per user. #6859
 * [ENHANCEMENT] Query Frontend: Change to return 400 when the tenant resolving fail. #6715

pkg/ruler/ruler.go

@@ -683,29 +683,40 @@ func (r *Ruler) run(ctx context.Context) error {
 		ringTickerChan = ringTicker.C
 	}
 
-	r.syncRules(ctx, rulerSyncReasonInitial)
+	syncRuleErrMsg := func(syncRulesErr error) {
+		level.Error(r.logger).Log("msg", "failed to sync rules", "err", syncRulesErr)
+	}
+
+	initialSyncErr := r.syncRules(ctx, rulerSyncReasonInitial)
+	if initialSyncErr != nil {
+		syncRuleErrMsg(initialSyncErr)
+	}
 	for {
+		var syncRulesErr error
 		select {
 		case <-ctx.Done():
 			return nil
 		case <-tick.C:
-			r.syncRules(ctx, rulerSyncReasonPeriodic)
+			syncRulesErr = r.syncRules(ctx, rulerSyncReasonPeriodic)
 		case <-ringTickerChan:
 			// We ignore the error because in case of error it will return an empty
 			// replication set which we use to compare with the previous state.
 			currRingState, _ := r.ring.GetAllHealthy(RingOp)
 
 			if ring.HasReplicationSetChanged(ringLastState, currRingState) {
 				ringLastState = currRingState
-				r.syncRules(ctx, rulerSyncReasonRingChange)
+				syncRulesErr = r.syncRules(ctx, rulerSyncReasonRingChange)
 			}
 		case err := <-r.subservicesWatcher.Chan():
 			return errors.Wrap(err, "ruler subservice failed")
 		}
+		if syncRulesErr != nil {
+			syncRuleErrMsg(syncRulesErr)
+		}
 	}
 }
 
-func (r *Ruler) syncRules(ctx context.Context, reason string) {
+func (r *Ruler) syncRules(ctx context.Context, reason string) error {
 	level.Info(r.logger).Log("msg", "syncing rules", "reason", reason)
 	r.rulerSync.WithLabelValues(reason).Inc()
 	timer := prometheus.NewTimer(nil)
@@ -717,19 +728,21 @@ func (r *Ruler) syncRules(ctx context.Context, reason string) {
 
 	loadedConfigs, backupConfigs, err := r.loadRuleGroups(ctx)
 	if err != nil {
-		return
+		return err
 	}
 
 	if ctx.Err() != nil {
 		level.Info(r.logger).Log("msg", "context is canceled. not syncing rules")
-		return
+		return err
 	}
 	// This will also delete local group files for users that are no longer in 'configs' map.
 	r.manager.SyncRuleGroups(ctx, loadedConfigs)
 
 	if r.cfg.RulesBackupEnabled() {
 		r.manager.BackUpRuleGroups(ctx, backupConfigs)
 	}
+
+	return nil
 }
 
 func (r *Ruler) loadRuleGroups(ctx context.Context) (map[string]rulespb.RuleGroupList, map[string]rulespb.RuleGroupList, error) {

pkg/ruler/ruler_test.go

@@ -1342,7 +1342,8 @@ func TestGetRules(t *testing.T) {
 
 			// Sync Rules
 			forEachRuler(func(_ string, r *Ruler) {
-				r.syncRules(context.Background(), rulerSyncReasonInitial)
+				err := r.syncRules(context.Background(), rulerSyncReasonInitial)
+				require.NoError(t, err)
 			})
 
 			if tc.sharding {
@@ -1572,7 +1573,8 @@ func TestGetRulesFromBackup(t *testing.T) {
 
 	// Sync Rules
 	forEachRuler(func(_ string, r *Ruler) {
-		r.syncRules(context.Background(), rulerSyncReasonInitial)
+		err := r.syncRules(context.Background(), rulerSyncReasonInitial)
+		require.NoError(t, err)
 	})
 
 	// update the State of the rulers in the ring based on tc.rulerStateMap
@@ -1788,7 +1790,8 @@ func getRulesHATest(replicationFactor int) func(t *testing.T) {
 
 		// Sync Rules
 		forEachRuler(func(_ string, r *Ruler) {
-			r.syncRules(context.Background(), rulerSyncReasonInitial)
+			err := r.syncRules(context.Background(), rulerSyncReasonInitial)
+			require.NoError(t, err)
 		})
 
 		// update the State of the rulers in the ring based on tc.rulerStateMap
@@ -1811,8 +1814,10 @@ func getRulesHATest(replicationFactor int) func(t *testing.T) {
 			t.Errorf("ruler %s was not terminated with error %s", "ruler1", err.Error())
 		}
 
-		rulerAddrMap["ruler2"].syncRules(context.Background(), rulerSyncReasonPeriodic)
-		rulerAddrMap["ruler3"].syncRules(context.Background(), rulerSyncReasonPeriodic)
+		err = rulerAddrMap["ruler2"].syncRules(context.Background(), rulerSyncReasonPeriodic)
+		require.NoError(t, err)
+		err = rulerAddrMap["ruler3"].syncRules(context.Background(), rulerSyncReasonPeriodic)
+		require.NoError(t, err)
 
 		requireGroupStateEqual := func(a *GroupStateDesc, b *GroupStateDesc) {
 			require.Equal(t, a.Group.Interval, b.Group.Interval)
@@ -2800,7 +2805,8 @@ func TestRecoverAlertsPostOutage(t *testing.T) {
 	evalFunc := func(ctx context.Context, g *promRules.Group, evalTimestamp time.Time) {}
 
 	r, _ := buildRulerWithIterFunc(t, rulerCfg, &querier.TestConfig{Cfg: querierConfig, Distributor: d, Stores: queryables}, store, nil, evalFunc)
-	r.syncRules(context.Background(), rulerSyncReasonInitial)
+	err := r.syncRules(context.Background(), rulerSyncReasonInitial)
+	require.NoError(t, err)
 
 	// assert initial state of rule group
 	ruleGroup := r.manager.GetRules("user1")[0]
@@ -3265,7 +3271,8 @@ func TestGetShardSizeForUser(t *testing.T) {
 
 			// Sync Rules
 			forEachRuler(func(_ string, r *Ruler) {
-				r.syncRules(context.Background(), rulerSyncReasonInitial)
+				err := r.syncRules(context.Background(), rulerSyncReasonInitial)
+				require.NoError(t, err)
 			})
 
 			result := testRuler.getShardSizeForUser(tc.userID)