Method run

pkg/ruler/ruler.go:690–741

(ctx context.Context)

}

func (r *Ruler) run(ctx context.Context) error {
	level.Info(r.logger).Log("msg", "ruler up and running")

	tick := time.NewTicker(r.cfg.PollInterval)
	defer tick.Stop()

	var ringTickerChan <-chan time.Time
	var ringLastState ring.ReplicationSet

	if r.cfg.EnableSharding {
		ringLastState, _ = r.ring.GetAllHealthy(RingOp)
		ringTicker := time.NewTicker(util.DurationWithJitter(r.cfg.RingCheckPeriod, 0.2))
		defer ringTicker.Stop()
		ringTickerChan = ringTicker.C
	}

	if r.userIndexUpdater != nil {
		go r.userIndexUpdateLoop(ctx)
	}

	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:
			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
				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) userIndexUpdateLoop(ctx context.Context) {
	// Hardcode ID to check which ruler owns updating user index.