Refactoring filterRuleGroups, filterBackupRuleGroups and removed ruler address from LivenessCheckRequest gRPC call

Signed-off-by: Anand Rajagopal <[email protected]>

Author: rajagopalanand

pkg/ruler/ruler.go

@@ -512,7 +512,7 @@ func tokenForGroup(g *rulespb.RuleGroupDesc) uint32 {
 	return ringHasher.Sum32()
 }
 
-func (r *Ruler) instanceOwnsRuleGroup(rr ring.ReadRing, g *rulespb.RuleGroupDesc, disabledRuleGroups validation.DisabledRuleGroups, instanceAddr string, forBackup bool) (bool, error) {
+func (r *Ruler) instanceOwnsRuleGroup(rr ring.ReadRing, g *rulespb.RuleGroupDesc, disabledRuleGroups validation.DisabledRuleGroups, forBackup bool) (bool, error) {
 
 	hash := tokenForGroup(g)
 
@@ -521,6 +521,7 @@ func (r *Ruler) instanceOwnsRuleGroup(rr ring.ReadRing, g *rulespb.RuleGroupDesc
 		return false, errors.Wrap(err, "error reading ring to verify rule group ownership")
 	}
 
+	instanceAddr := r.lifecycler.GetInstanceAddr()
 	if forBackup {
 		// Only the second up to the last replica are used as backup
 		for i := 1; i < len(rlrs.Instances); i++ {
@@ -536,7 +537,7 @@ func (r *Ruler) instanceOwnsRuleGroup(rr ring.ReadRing, g *rulespb.RuleGroupDesc
 				level.Debug(r.Logger()).Log("msg", "primary taking ownership", "user", g.User, "group", g.Name, "namespace", g.Namespace, "ruler", instanceAddr)
 				return ownsRuleGroupOrDisable(g, disabledRuleGroups)
 			}
-			if ruler.Addr == instanceAddr && r.nonPrimaryInstanceOwnsRuleGroup(g, rlrs.GetAddresses()[:i], instanceAddr) {
+			if ruler.Addr == instanceAddr && r.nonPrimaryInstanceOwnsRuleGroup(g, rlrs.GetAddresses()[:i]) {
 				level.Info(r.Logger()).Log("msg", "non-primary ruler taking ownership", "user", g.User, "group", g.Name, "namespace", g.Namespace, "ruler", instanceAddr)
 				return ownsRuleGroupOrDisable(g, disabledRuleGroups)
 			}
@@ -569,7 +570,7 @@ func (r *Ruler) LivenessCheck(_ context.Context, request *LivenessCheckRequest)
 
 // This function performs a liveness check against the provided replicas. If any one of the replicas responds with a state = Running, then
 // this Ruler should not take ownership of the rule group. Otherwise, this Ruler must take ownership of the rule group to avoid missing evaluations
-func (r *Ruler) nonPrimaryInstanceOwnsRuleGroup(g *rulespb.RuleGroupDesc, replicas []string, selfAddress string) bool {
+func (r *Ruler) nonPrimaryInstanceOwnsRuleGroup(g *rulespb.RuleGroupDesc, replicas []string) bool {
 	userID := g.User
 
 	jobs := concurrency.CreateJobsFromStrings(replicas)
@@ -586,14 +587,12 @@ func (r *Ruler) nonPrimaryInstanceOwnsRuleGroup(g *rulespb.RuleGroupDesc, replic
 		rulerClient, err := r.GetClientFor(addr)
 		if err != nil {
 			errorChan <- err
-			level.Debug(r.Logger()).Log("msg", "unable to get client for ruler", "ruler addr", addr)
+			level.Error(r.Logger()).Log("msg", "unable to get client for ruler", "ruler addr", addr)
 			return nil
 		}
-		level.Debug(r.Logger()).Log("msg", "performing liveness check against", "addr", addr, "for", g.Name, "instance addr", selfAddress)
+		level.Debug(r.Logger()).Log("msg", "performing liveness check against", "addr", addr, "for", g.Name)
 
-		resp, err := rulerClient.LivenessCheck(ctx, &LivenessCheckRequest{
-			RulerAddress: selfAddress,
-		})
+		resp, err := rulerClient.LivenessCheck(ctx, &LivenessCheckRequest{})
 		if err != nil {
 			errorChan <- err
 			level.Debug(r.Logger()).Log("msg", "liveness check failed", "addr", addr, "for", g.Name, "err", err.Error())
@@ -804,12 +803,12 @@ func (r *Ruler) listRulesShardingDefault(ctx context.Context) (map[string]rulesp
 	backedUpConfigs := make(map[string]rulespb.RuleGroupList)
 	for userID, groups := range configs {
 		ruleGroupCounts[userID] = len(groups)
-		owned := r.filterRuleGroups(userID, groups, r.limits.DisabledRuleGroups(userID), r.ring, r.lifecycler.GetInstanceAddr(), r.logger, r.ringCheckErrors)
+		owned := r.filterRuleGroups(userID, groups, r.ring)
 		if len(owned) > 0 {
 			ownedConfigs[userID] = owned
 		}
 		if r.cfg.RulesBackupEnabled() {
-			backup := r.filterBackupRuleGroups(userID, groups, owned, r.limits.DisabledRuleGroups(userID), r.ring, r.lifecycler.GetInstanceAddr(), r.logger, r.ringCheckErrors)
+			backup := r.filterBackupRuleGroups(userID, groups, owned, r.ring)
 			if len(backup) > 0 {
 				backedUpConfigs[userID] = backup
 			}
@@ -876,10 +875,10 @@ func (r *Ruler) listRulesShuffleSharding(ctx context.Context) (map[string]rulesp
 				ruleGroupCounts[userID] = len(groups)
 				gLock.Unlock()
 
-				filterOwned := r.filterRuleGroups(userID, groups, r.limits.DisabledRuleGroups(userID), userRings[userID], r.lifecycler.GetInstanceAddr(), r.logger, r.ringCheckErrors)
+				filterOwned := r.filterRuleGroups(userID, groups, userRings[userID])
 				var filterBackup []*rulespb.RuleGroupDesc
 				if r.cfg.RulesBackupEnabled() {
-					filterBackup = r.filterBackupRuleGroups(userID, groups, filterOwned, r.limits.DisabledRuleGroups(userID), userRings[userID], r.lifecycler.GetInstanceAddr(), r.logger, r.ringCheckErrors)
+					filterBackup = r.filterBackupRuleGroups(userID, groups, filterOwned, userRings[userID])
 				}
 				if len(filterOwned) == 0 && len(filterBackup) == 0 {
 					continue
@@ -906,28 +905,29 @@ func (r *Ruler) listRulesShuffleSharding(ctx context.Context) (map[string]rulesp
 // This function only uses User, Namespace, and Name fields of individual RuleGroups.
 //
 // This method must not use r.ring, but only ring passed as parameter.
-func (r *Ruler) filterRuleGroups(userID string, ruleGroups []*rulespb.RuleGroupDesc, disabledRuleGroups validation.DisabledRuleGroups, ring ring.ReadRing, instanceAddr string, log log.Logger, ringCheckErrors prometheus.Counter) []*rulespb.RuleGroupDesc {
+func (r *Ruler) filterRuleGroups(userID string, ruleGroups []*rulespb.RuleGroupDesc, ring ring.ReadRing) []*rulespb.RuleGroupDesc {
 	// Prune the rule group to only contain rules that this ruler is responsible for, based on ring.
 	var result []*rulespb.RuleGroupDesc
+
 	for _, g := range ruleGroups {
-		owned, err := r.instanceOwnsRuleGroup(ring, g, disabledRuleGroups, instanceAddr, false)
+		owned, err := r.instanceOwnsRuleGroup(ring, g, r.limits.DisabledRuleGroups(userID), false)
 		if err != nil {
 			switch e := err.(type) {
 			case *DisabledRuleGroupErr:
-				level.Info(log).Log("msg", e.Message)
+				level.Info(r.logger).Log("msg", e.Message)
 				continue
 			default:
-				ringCheckErrors.Inc()
-				level.Error(log).Log("msg", "failed to check if the ruler replica owns the rule group", "user", userID, "namespace", g.Namespace, "group", g.Name, "err", err)
+				r.ringCheckErrors.Inc()
+				level.Error(r.logger).Log("msg", "failed to check if the ruler replica owns the rule group", "user", userID, "namespace", g.Namespace, "group", g.Name, "err", err)
 				continue
 			}
 		}
 
 		if owned {
-			level.Debug(log).Log("msg", "rule group owned", "user", g.User, "namespace", g.Namespace, "name", g.Name)
+			level.Debug(r.logger).Log("msg", "rule group owned", "user", g.User, "namespace", g.Namespace, "name", g.Name)
 			result = append(result, g)
 		} else {
-			level.Debug(log).Log("msg", "rule group not owned, ignoring", "user", g.User, "namespace", g.Namespace, "name", g.Name)
+			level.Debug(r.logger).Log("msg", "rule group not owned, ignoring", "user", g.User, "namespace", g.Namespace, "name", g.Name)
 		}
 	}
 
@@ -938,7 +938,7 @@ func (r *Ruler) filterRuleGroups(userID string, ruleGroups []*rulespb.RuleGroupD
 // This function only uses User, Namespace, and Name fields of individual RuleGroups.
 //
 // This method must not use r.ring, but only ring passed as parameter
-func (r *Ruler) filterBackupRuleGroups(userID string, ruleGroups []*rulespb.RuleGroupDesc, owned []*rulespb.RuleGroupDesc, disabledRuleGroups validation.DisabledRuleGroups, ring ring.ReadRing, instanceAddr string, log log.Logger, ringCheckErrors prometheus.Counter) []*rulespb.RuleGroupDesc {
+func (r *Ruler) filterBackupRuleGroups(userID string, ruleGroups []*rulespb.RuleGroupDesc, owned []*rulespb.RuleGroupDesc, ring ring.ReadRing) []*rulespb.RuleGroupDesc {
 	var result []*rulespb.RuleGroupDesc
 	ownedMap := map[uint32]struct{}{}
 	for _, g := range owned {
@@ -951,24 +951,24 @@ func (r *Ruler) filterBackupRuleGroups(userID string, ruleGroups []*rulespb.Rule
 		if _, OK := ownedMap[hash]; OK {
 			continue
 		}
-		backup, err := r.instanceOwnsRuleGroup(ring, g, disabledRuleGroups, instanceAddr, true)
+		backup, err := r.instanceOwnsRuleGroup(ring, g, r.limits.DisabledRuleGroups(userID), true)
 		if err != nil {
 			switch e := err.(type) {
 			case *DisabledRuleGroupErr:
-				level.Info(log).Log("msg", e.Message)
+				level.Info(r.logger).Log("msg", e.Message)
 				continue
 			default:
-				ringCheckErrors.Inc()
-				level.Error(log).Log("msg", "failed to check if the ruler replica backs up the rule group", "user", userID, "namespace", g.Namespace, "group", g.Name, "err", err)
+				r.ringCheckErrors.Inc()
+				level.Error(r.logger).Log("msg", "failed to check if the ruler replica backs up the rule group", "user", userID, "namespace", g.Namespace, "group", g.Name, "err", err)
 				continue
 			}
 		}
 
 		if backup {
-			level.Debug(log).Log("msg", "rule group backed up", "user", g.User, "namespace", g.Namespace, "name", g.Name)
+			level.Debug(r.logger).Log("msg", "rule group backed up", "user", g.User, "namespace", g.Namespace, "name", g.Name)
 			result = append(result, g)
 		} else {
-			level.Debug(log).Log("msg", "rule group not backed up, ignoring", "user", g.User, "namespace", g.Namespace, "name", g.Name)
+			level.Debug(r.logger).Log("msg", "rule group not backed up, ignoring", "user", g.User, "namespace", g.Namespace, "name", g.Name)
 		}
 	}