Minimize missed rule group evaluations

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

Author: rajagopalanand

CHANGELOG.md

@@ -5,6 +5,8 @@
 * [CHANGE] Ingester: Remove `-querier.query-store-for-labels-enabled` flag. Querying long-term store for labels is always enabled. #5984
 * [CHANGE] Server: Instrument `cortex_request_duration_seconds` metric with native histogram. If `native-histograms` feature is enabled in monitoring Prometheus then the metric name needs to be updated in your dashboards. #6056
 * [CHANGE] Distributor/Ingester: Change `cortex_distributor_ingester_appends_total`, `cortex_distributor_ingester_append_failures_total`, `cortex_distributor_ingester_queries_total`, and `cortex_distributor_ingester_query_failures_total` metrics to use the ingester ID instead of its IP as the label value. #6078
+* [FEATURE] Ruler: Minimize rule group missed evaluations via `-ruler.enable-ha` flag. #6129
+* [FEATURE] Ingester: Experimental: Enable native histogram ingestion via `-blocks-storage.tsdb.enable-native-histograms` flag. #5986
 * [FEATURE] Ingester/Distributor: Experimental: Enable native histogram ingestion via `-blocks-storage.tsdb.enable-native-histograms` flag. #5986 #6010 #6020
 * [FEATURE] Querier: Enable querying native histogram chunks. #5944 #6031
 * [FEATURE] Query Frontend: Support native histogram in query frontend response. #5996 #6043

docs/configuration/config-file-reference.md

@@ -4337,6 +4337,10 @@ ring:
   # CLI flag: -ruler.ring.final-sleep
   [final_sleep: <duration> | default = 0s]
 
+  # Keep instance in the ring on shut down.
+  # CLI flag: -ruler.ring.keep-instance-in-the-ring-on-shutdown
+  [keep_instance_in_the_ring_on_shutdown: <boolean> | default = false]
+
 # Period with which to attempt to flush rule groups.
 # CLI flag: -ruler.flush-period
 [flush_period: <duration> | default = 1m]
@@ -4371,6 +4375,18 @@ ring:
 # Disable the rule_group label on exported metrics
 # CLI flag: -ruler.disable-rule-group-label
 [disable_rule_group_label: <boolean> | default = false]
+
+# Enable high availability
+# CLI flag: -ruler.enable-ha
+[enable_ha: <boolean> | default = false]
+
+# Timeout for fanout calls to other rulers
+# CLI flag: -ruler.list-rules-fanout-timeout
+[list_rules_fanout_timeout: <duration> | default = 2m]
+
+# Timeout for liveness checks performed during rule sync
+# CLI flag: -ruler.liveness-check-timeout
+[liveness_check_timeout: <duration> | default = 1s]
 ```
 
 ### `ruler_storage_config`

integration/ruler_test.go

@@ -1093,6 +1093,148 @@ func TestRulerDisablesRuleGroups(t *testing.T) {
 	})
 }
 
+func TestRulerHA(t *testing.T) {
+	const numRulesGroups = 20
+
+	random := rand.New(rand.NewSource(time.Now().UnixNano()))
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	// Generate multiple rule groups, with 1 rule each.
+	ruleGroups := make([]rulefmt.RuleGroup, numRulesGroups)
+	expectedNames := make([]string, numRulesGroups)
+	alertCount := 0
+	evalInterval, _ := model.ParseDuration("5s")
+	for i := 0; i < numRulesGroups; i++ {
+		num := random.Intn(10)
+		var ruleNode yaml.Node
+		var exprNode yaml.Node
+
+		ruleNode.SetString(fmt.Sprintf("rule_%d", i))
+		exprNode.SetString(strconv.Itoa(i))
+		ruleName := fmt.Sprintf("test_%d", i)
+
+		expectedNames[i] = ruleName
+
+		if num%2 == 0 {
+			alertCount++
+			ruleGroups[i] = rulefmt.RuleGroup{
+				Name:     ruleName,
+				Interval: evalInterval,
+				Rules: []rulefmt.RuleNode{{
+					Alert: ruleNode,
+					Expr:  exprNode,
+				}},
+			}
+		} else {
+			ruleGroups[i] = rulefmt.RuleGroup{
+				Name:     ruleName,
+				Interval: evalInterval,
+				Rules: []rulefmt.RuleNode{{
+					Record: ruleNode,
+					Expr:   exprNode,
+				}},
+			}
+		}
+	}
+
+	// Start dependencies.
+	consul := e2edb.NewConsul()
+	minio := e2edb.NewMinio(9000, rulestoreBucketName)
+	require.NoError(t, s.StartAndWaitReady(consul, minio))
+
+	// Configure the ruler.
+	overrides := map[string]string{
+		// Since we're not going to run any rule, we don't need the
+		// store-gateway to be configured to a valid address.
+		"-querier.store-gateway-addresses": "localhost:12345",
+		// Enable the bucket index so we can skip the initial bucket scan.
+		"-blocks-storage.bucket-store.bucket-index.enabled": "true",
+		"-ruler.ring.replication-factor":                    "2",
+		"-ruler.enable-ha":                                  "true",
+		"-ruler.poll-interval":                              "5s",
+		"-ruler.list-rules-fanout-timeout":                  "2s",
+		"-ruler.liveness-check-timeout":                     "50ms",
+	}
+
+	rulerFlags := mergeFlags(
+		BlocksStorageFlags(),
+		RulerFlags(),
+		RulerShardingFlags(consul.NetworkHTTPEndpoint()),
+		overrides,
+	)
+
+	// Start rulers.
+	ruler1 := e2ecortex.NewRuler("ruler-1", consul.NetworkHTTPEndpoint(), rulerFlags, "")
+	ruler2 := e2ecortex.NewRuler("ruler-2", consul.NetworkHTTPEndpoint(), rulerFlags, "")
+	ruler3 := e2ecortex.NewRuler("ruler-3", consul.NetworkHTTPEndpoint(), rulerFlags, "")
+	rulers := e2ecortex.NewCompositeCortexService(ruler1, ruler2, ruler3)
+	require.NoError(t, s.StartAndWaitReady(ruler1, ruler2, ruler3))
+
+	// Upload rule groups to one of the rulers.
+	c, err := e2ecortex.NewClient("", "", "", ruler1.HTTPEndpoint(), "user-1")
+	require.NoError(t, err)
+	namespaceNames := []string{"test1", "test2", "test3", "test4", "test5"}
+	namespaceNameCount := make([]int, 5)
+	nsRand := rand.New(rand.NewSource(time.Now().UnixNano()))
+	for _, ruleGroup := range ruleGroups {
+		index := nsRand.Intn(len(namespaceNames))
+		namespaceNameCount[index] = namespaceNameCount[index] + 1
+		require.NoError(t, c.SetRuleGroup(ruleGroup, namespaceNames[index]))
+	}
+
+	// Wait until rulers have loaded all rules.
+	require.NoError(t, rulers.WaitSumMetricsWithOptions(e2e.Equals(numRulesGroups), []string{"cortex_prometheus_rule_group_rules"}, e2e.WaitMissingMetrics))
+
+	ruler1SyncTotal, err := ruler1.SumMetrics([]string{"cortex_ruler_sync_rules_total"})
+	require.NoError(t, err)
+	ruler3SyncTotal, err := ruler3.SumMetrics([]string{"cortex_ruler_sync_rules_total"})
+	require.NoError(t, err)
+
+	err = consul.Kill() // kill consul so the rulers will operate with the tokens/instances they already have
+	require.NoError(t, err)
+
+	err = ruler2.Kill()
+	require.NoError(t, err)
+
+	// wait for another sync
+	require.NoError(t, ruler1.WaitSumMetrics(e2e.Greater(ruler1SyncTotal[0]), "cortex_ruler_sync_rules_total"))
+	require.NoError(t, ruler3.WaitSumMetrics(e2e.Greater(ruler3SyncTotal[0]), "cortex_ruler_sync_rules_total"))
+
+	rulers = e2ecortex.NewCompositeCortexService(ruler1, ruler3)
+	require.NoError(t, rulers.WaitSumMetricsWithOptions(e2e.Equals(numRulesGroups), []string{"cortex_prometheus_rule_group_rules"}, e2e.WaitMissingMetrics))
+
+	t.Log(ruler1.SumMetrics([]string{"cortex_prometheus_rule_group_rules"}))
+	t.Log(ruler3.SumMetrics([]string{"cortex_prometheus_rule_group_rules"}))
+
+	c3, err := e2ecortex.NewClient("", "", "", ruler3.HTTPEndpoint(), "user-1")
+	require.NoError(t, err)
+
+	ruler1Rules, err := c.GetRuleGroups()
+	require.NoError(t, err)
+
+	ruler3Rules, err := c3.GetRuleGroups()
+	require.NoError(t, err)
+
+	ruleCount := 0
+	countFunc := func(ruleGroups map[string][]rulefmt.RuleGroup) {
+		for _, v := range ruleGroups {
+			ruleCount += len(v)
+		}
+	}
+
+	countFunc(ruler1Rules)
+	require.Equal(t, numRulesGroups, ruleCount)
+	ruleCount = 0
+	countFunc(ruler3Rules)
+	require.Equal(t, numRulesGroups, ruleCount)
+
+	results, err := c.GetPrometheusRules(e2ecortex.RuleFilter{})
+	require.NoError(t, err)
+	require.Equal(t, numRulesGroups, len(results))
+}
+
 func TestRulerKeepFiring(t *testing.T) {
 	s, err := e2e.NewScenario(networkName)
 	require.NoError(t, err)

pkg/ruler/client_pool_test.go

@@ -5,6 +5,8 @@ import (
 	"net"
 	"testing"
 
+	"github.com/cortexproject/cortex/pkg/util/services"
+
 	"github.com/prometheus/client_golang/prometheus"
 	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/assert"
@@ -63,6 +65,12 @@ func Test_newRulerClientFactory(t *testing.T) {
 
 type mockRulerServer struct{}
 
+func (m *mockRulerServer) LivenessCheck(ctx context.Context, request *LivenessCheckRequest) (*LivenessCheckResponse, error) {
+	return &LivenessCheckResponse{
+		State: int32(services.Running),
+	}, nil
+}
+
 func (m *mockRulerServer) Rules(context.Context, *RulesRequest) (*RulesResponse, error) {
 	return &RulesResponse{}, nil
 }