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
(g *rulespb.RuleGroupDesc, replicas []string)
| 620 | // This function performs a liveness check against the provided replicas. If any one of the replicas responds with a state = Running, then |
| 621 | // this Ruler should not take ownership of the rule group. Otherwise, this Ruler must take ownership of the rule group to avoid missing evaluations |
| 622 | func (r *Ruler) nonPrimaryInstanceOwnsRuleGroup(g *rulespb.RuleGroupDesc, replicas []string) bool { |
| 623 | userID := g.User |
| 624 | |
| 625 | jobs := concurrency.CreateJobsFromStrings(replicas) |
| 626 | |
| 627 | errorChan := make(chan error, len(jobs)) |
| 628 | responseChan := make(chan *LivenessCheckResponse, len(jobs)) |
| 629 | |
| 630 | ctx := user.InjectOrgID(context.Background(), userID) |
| 631 | ctx, cancel := context.WithTimeout(ctx, r.cfg.LivenessCheckTimeout) |
| 632 | defer cancel() |
| 633 | |
| 634 | err := concurrency.ForEach(ctx, jobs, len(jobs), func(ctx context.Context, job any) error { |
| 635 | addr := job.(string) |
| 636 | rulerClient, err := r.GetClientFor(addr) |
| 637 | if err != nil { |
| 638 | errorChan <- err |
| 639 | level.Error(r.Logger()).Log("msg", "unable to get client for ruler", "ruler addr", addr) |
| 640 | return nil |
| 641 | } |
| 642 | level.Debug(r.Logger()).Log("msg", "performing liveness check against", "addr", addr, "for", g.Name) |
| 643 | |
| 644 | resp, err := rulerClient.LivenessCheck(ctx, &LivenessCheckRequest{}) |
| 645 | if err != nil { |
| 646 | errorChan <- err |
| 647 | level.Debug(r.Logger()).Log("msg", "liveness check failed", "addr", addr, "for", g.Name, "err", err.Error()) |
| 648 | return nil |
| 649 | } |
| 650 | level.Debug(r.Logger()).Log("msg", "liveness check succeeded ", "addr", addr, "for", g.Name, "ruler state", services.State(resp.GetState())) |
| 651 | responseChan <- resp |
| 652 | return nil |
| 653 | }) |
| 654 | |
| 655 | close(errorChan) |
| 656 | close(responseChan) |
| 657 | |
| 658 | if len(errorChan) == len(jobs) || err != nil { |
| 659 | return true |
| 660 | } |
| 661 | |
| 662 | for resp := range responseChan { |
| 663 | if services.State(resp.GetState()) == services.Running { |
| 664 | return false |
| 665 | } |
| 666 | } |
| 667 | return true |
| 668 | } |
| 669 | |
| 670 | func (r *Ruler) ServeHTTP(w http.ResponseWriter, req *http.Request) { |
| 671 | if r.cfg.EnableSharding { |
no test coverage detected