Alert State History: Skip invalid entries when merging streams (#111387)

pkg/services/ngalert/state/historian/loki.go

@@ -175,11 +175,11 @@ func (h *RemoteLokiBackend) Query(ctx context.Context, query models.HistoryQuery
 		}
 		res = append(res, r.Data.Result...)
 	}
-	return merge(res, uids)
+	return h.merge(res, uids)
 }
 
 // merge will put all the results in one array sorted by timestamp.
-func merge(res []lokiclient.Stream, folderUIDToFilter []string) (*data.Frame, error) {
+func (h RemoteLokiBackend) merge(res []lokiclient.Stream, folderUIDToFilter []string) (*data.Frame, error) {
 	filterByFolderUIDMap := make(map[string]struct{}, len(folderUIDToFilter))
 	for _, uid := range folderUIDToFilter {
 		filterByFolderUIDMap[uid] = struct{}{}
@@ -244,19 +244,25 @@ func merge(res []lokiclient.Stream, folderUIDToFilter []string) (*data.Frame, er
 		var entry LokiEntry
 		err := json.Unmarshal([]byte(minEl.V), &entry)
 		if err != nil {
-			return nil, fmt.Errorf("failed to unmarshal entry: %w", err)
+			h.log.Warn("failed to unmarshal entry, continuing", "err", err, "entry", minEl.V)
+			pointers[minElStreamIdx]++
+			continue
 		}
 		// Append the minimum element to the merged slice and move the pointer.
 		tsNano := minEl.T.UnixNano()
-		// TODO: In general, perhaps we should omit the offending line and log, rather than failing the request entirely.
 		streamLbls := res[minElStreamIdx].Stream
 		lblsJson, err := json.Marshal(streamLbls)
 		if err != nil {
-			return nil, fmt.Errorf("failed to serialize stream labels: %w", err)
+			// This should in theory never happen, as we're marshalling a map[string]string.
+			h.log.Warn("failed to serialize stream labels, continuing", "err", err, "labels", streamLbls)
+			pointers[minElStreamIdx]++
+			continue
 		}
 		line, err := jsonifyRow(minEl.V)
 		if err != nil {
-			return nil, fmt.Errorf("a line was in an invalid format: %w", err)
+			h.log.Warn("a line was in an invalid format, continuing", "err", err, "line", minEl.V)
+			pointers[minElStreamIdx]++
+			continue
 		}
 
 		times = append(times, time.Unix(0, tsNano))

pkg/services/ngalert/state/historian/loki_test.go

@@ -538,6 +538,62 @@ func TestMerge(t *testing.T) {
 				}),
 			),
 		},
+		{
+			name: "Should handle bad values",
+			res: lokiclient.QueryRes{
+				Data: lokiclient.QueryData{
+					Result: []lokiclient.Stream{
+						{
+							Stream: map[string]string{
+								"from":      "state-history",
+								"orgID":     "1",
+								"group":     "test-group-1",
+								"folderUID": "test-folder-1",
+							},
+							Values: []lokiclient.Sample{
+								{T: time.Unix(1, 0), V: `{"schemaVersion": 1, "previous": "normal", "current": "pending", "values":{"a": 1.5}, "ruleUID": "test-rule-1"}`},
+								{T: time.Unix(5, 0), V: `{"schemaVersion": 1, "previous": "pending", "current": "normal", "values":{"a": 0.5}, "ruleUID": "test-rule-2", "bad_label": "\e"}`},
+							},
+						},
+						{
+							Stream: map[string]string{
+								"from":      "state-history",
+								"orgID":     "1",
+								"group":     "test-group-2",
+								"folderUID": "test-folder-1",
+							},
+							Values: []lokiclient.Sample{
+								{T: time.Unix(2, 0), V: `{"schemaVersion": 1, "previous": "pending", "current": "firing", "values":{"a": 2.5}, "ruleUID": "test-rule-3"}`},
+							},
+						},
+					},
+				},
+			},
+			expected: data.NewFrame("states",
+				data.NewField(dfTime, data.Labels{}, []time.Time{
+					time.Unix(1, 0),
+					time.Unix(2, 0),
+				}),
+				data.NewField(dfLine, data.Labels{}, []json.RawMessage{
+					toJson(LokiEntry{RuleUID: "test-rule-1", SchemaVersion: 1, Previous: "normal", Current: "pending", Values: jsonifyValues(map[string]float64{"a": 1.5})}),
+					toJson(LokiEntry{RuleUID: "test-rule-3", SchemaVersion: 1, Previous: "pending", Current: "firing", Values: jsonifyValues(map[string]float64{"a": 2.5})}),
+				}),
+				data.NewField(dfLabels, data.Labels{}, []json.RawMessage{
+					toJson(map[string]string{
+						StateHistoryLabelKey: "state-history",
+						OrgIDLabel:           "1",
+						GroupLabel:           "test-group-1",
+						FolderUIDLabel:       "test-folder-1",
+					}),
+					toJson(map[string]string{
+						StateHistoryLabelKey: "state-history",
+						OrgIDLabel:           "1",
+						GroupLabel:           "test-group-2",
+						FolderUIDLabel:       "test-folder-1",
+					}),
+				}),
+			),
+		},
 		{
 			name:       "should filter streams by folder UID",
 			folderUIDs: []string{"test-folder-1"},
@@ -654,9 +710,11 @@ func TestMerge(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
+			req := instrumenttest.NewFakeRequester()
+			loki := createTestLokiBackend(t, req, metrics.NewHistorianMetrics(prometheus.NewRegistry(), metrics.Subsystem))
 			expectedJson, err := tc.expected.MarshalJSON()
 			require.NoError(t, err)
-			m, err := merge(tc.res.Data.Result, tc.folderUIDs)
+			m, err := loki.merge(tc.res.Data.Result, tc.folderUIDs)
 			require.NoError(t, err)
 			actualJson, err := m.MarshalJSON()
 			assert.NoError(t, err)