diff --git a/pkg/services/ngalert/state/historian/loki.go b/pkg/services/ngalert/state/historian/loki.go index f31b993ad9d..d5aa7997827 100644 --- a/pkg/services/ngalert/state/historian/loki.go +++ b/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)) diff --git a/pkg/services/ngalert/state/historian/loki_test.go b/pkg/services/ngalert/state/historian/loki_test.go index 99d9aff3b79..f404402c9fe 100644 --- a/pkg/services/ngalert/state/historian/loki_test.go +++ b/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)