Mark deleted rule's series as stale on next evaluation. (#5759)

Fixes #5755

Signed-off-by: Brian Brazil <brian.brazil@robustperception.io>

rules/manager.go

@@ -223,6 +223,7 @@ type Group struct {
 	interval             time.Duration
 	rules                []Rule
 	seriesInPreviousEval []map[string]labels.Labels // One per Rule.
+	staleSeries          []labels.Labels
 	opts                 *ManagerOptions
 	mtx                  sync.Mutex
 	evaluationDuration   time.Duration
@@ -480,6 +481,18 @@ func (g *Group) CopyState(from *Group) {
 			ar.active[fp] = a
 		}
 	}
+
+	// Handle deleted and unmatched duplicate rules.
+	g.staleSeries = from.staleSeries
+	for fi, fromRule := range from.rules {
+		nameAndLabels := nameAndLabels(fromRule)
+		l := ruleMap[nameAndLabels]
+		if len(l) != 0 {
+			for _, series := range from.seriesInPreviousEval[fi] {
+				g.staleSeries = append(g.staleSeries, series)
+			}
+		}
+	}
 }
 
 // Eval runs a single evaluation cycle in which all rules are evaluated sequentially.
@@ -575,6 +588,32 @@ func (g *Group) Eval(ctx context.Context, ts time.Time) {
 			}
 		}(i, rule)
 	}
+
+	if len(g.staleSeries) != 0 {
+		app, err := g.opts.Appendable.Appender()
+		if err != nil {
+			level.Warn(g.logger).Log("msg", "creating appender failed", "err", err)
+			return
+		}
+		for _, s := range g.staleSeries {
+			// Rule that produced series no longer configured, mark it stale.
+			_, err = app.Add(s, timestamp.FromTime(ts), math.Float64frombits(value.StaleNaN))
+			switch err {
+			case nil:
+			case storage.ErrOutOfOrderSample, storage.ErrDuplicateSampleForTimestamp:
+				// Do not count these in logging, as this is expected if series
+				// is exposed from a different rule.
+			default:
+				level.Warn(g.logger).Log("msg", "adding stale sample for previous configuration failed", "sample", s, "err", err)
+			}
+		}
+		if err := app.Commit(); err != nil {
+			level.Warn(g.logger).Log("msg", "stale sample appending for previous configuration failed", "err", err)
+		} else {
+			g.staleSeries = nil
+		}
+	}
+
 }
 
 // RestoreForState restores the 'for' state of the alerts

rules/manager_test.go

@@ -586,14 +586,16 @@ func TestCopyState(t *testing.T) {
 			NewRecordingRule("rule2", nil, nil),
 			NewRecordingRule("rule3", nil, labels.Labels{{Name: "l1", Value: "v1"}}),
 			NewRecordingRule("rule3", nil, labels.Labels{{Name: "l1", Value: "v2"}}),
+			NewRecordingRule("rule3", nil, labels.Labels{{Name: "l1", Value: "v3"}}),
 			NewAlertingRule("alert2", nil, 0, labels.Labels{{Name: "l2", Value: "v1"}}, nil, nil, true, nil),
 		},
 		seriesInPreviousEval: []map[string]labels.Labels{
-			{"a": nil},
+			{},
-			{"r1": nil},
+			{},
-			{"r2": nil},
+			{},
 			{"r3a": labels.Labels{{Name: "l1", Value: "v1"}}},
 			{"r3b": labels.Labels{{Name: "l1", Value: "v2"}}},
+			{"r3c": labels.Labels{{Name: "l1", Value: "v3"}}},
 			{"a2": labels.Labels{{Name: "l2", Value: "v1"}}},
 		},
 		evaluationDuration: time.Second,
@@ -618,8 +620,8 @@ func TestCopyState(t *testing.T) {
 		nil,
 		{"r3a": labels.Labels{{Name: "l1", Value: "v1"}}},
 		{"r3b": labels.Labels{{Name: "l1", Value: "v2"}}},
-		{"a": nil},
+		{},
-		{"r1": nil},
+		{},
 		nil,
 		{"a2": labels.Labels{{Name: "l2", Value: "v1"}}},
 		nil,
@@ -627,6 +629,49 @@ func TestCopyState(t *testing.T) {
 	testutil.Equals(t, want, newGroup.seriesInPreviousEval)
 	testutil.Equals(t, oldGroup.rules[0], newGroup.rules[3])
 	testutil.Equals(t, oldGroup.evaluationDuration, newGroup.evaluationDuration)
+	testutil.Equals(t, []labels.Labels{labels.Labels{{Name: "l1", Value: "v3"}}}, newGroup.staleSeries)
+}
+
+func TestDeletedRuleMarkedStale(t *testing.T) {
+	storage := testutil.NewStorage(t)
+	defer storage.Close()
+	oldGroup := &Group{
+		rules: []Rule{
+			NewRecordingRule("rule1", nil, labels.Labels{{Name: "l1", Value: "v1"}}),
+		},
+		seriesInPreviousEval: []map[string]labels.Labels{
+			{"r1": labels.Labels{{Name: "l1", Value: "v1"}}},
+		},
+	}
+	newGroup := &Group{
+		rules:                []Rule{},
+		seriesInPreviousEval: []map[string]labels.Labels{},
+		opts: &ManagerOptions{
+			Appendable: storage,
+		},
+	}
+	newGroup.CopyState(oldGroup)
+
+	newGroup.Eval(context.Background(), time.Unix(0, 0))
+
+	querier, err := storage.Querier(context.Background(), 0, 2000)
+	testutil.Ok(t, err)
+	defer querier.Close()
+
+	matcher, err := labels.NewMatcher(labels.MatchEqual, "l1", "v1")
+	testutil.Ok(t, err)
+
+	set, _, err := querier.Select(nil, matcher)
+	testutil.Ok(t, err)
+
+	samples, err := readSeriesSet(set)
+	testutil.Ok(t, err)
+
+	metric := labels.FromStrings("l1", "v1").String()
+	metricSample, ok := samples[metric]
+
+	testutil.Assert(t, ok, "Series %s not returned.", metric)
+	testutil.Assert(t, value.IsStaleNaN(metricSample[0].V), "Appended sample not as expected. Wanted: stale NaN Got: %x", math.Float64bits(metricSample[0].V))
 }
 
 func TestUpdate(t *testing.T) {