diff --git a/pkg/services/ngalert/state/historian/core.go b/pkg/services/ngalert/state/historian/core.go index d59181cf317..35b9e3ba40a 100644 --- a/pkg/services/ngalert/state/historian/core.go +++ b/pkg/services/ngalert/state/historian/core.go @@ -57,3 +57,10 @@ func parsePanelKey(rule history_model.RuleMeta, logger log.Logger) *panelKey { } return nil } + +func mergeLabels(base, into data.Labels) data.Labels { + for k, v := range into { + base[k] = v + } + return base +} diff --git a/pkg/services/ngalert/state/historian/loki.go b/pkg/services/ngalert/state/historian/loki.go index 7f56c668922..1f9662a41e7 100644 --- a/pkg/services/ngalert/state/historian/loki.go +++ b/pkg/services/ngalert/state/historian/loki.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "fmt" - "sort" "github.com/grafana/grafana-plugin-sdk-go/data" @@ -49,7 +48,7 @@ func (h *RemoteLokiBackend) TestConnection(ctx context.Context) error { func (h *RemoteLokiBackend) RecordStatesAsync(ctx context.Context, rule history_model.RuleMeta, states []state.StateTransition) <-chan error { logger := h.log.FromContext(ctx) - streams := h.statesToStreams(rule, states, logger) + streams := statesToStreams(rule, states, h.externalLabels, logger) errCh := make(chan error, 1) go func() { defer close(errCh) @@ -65,14 +64,14 @@ func (h *RemoteLokiBackend) QueryStates(ctx context.Context, query models.Histor return data.NewFrame("states"), nil } -func (h *RemoteLokiBackend) statesToStreams(rule history_model.RuleMeta, states []state.StateTransition, logger log.Logger) []stream { +func statesToStreams(rule history_model.RuleMeta, states []state.StateTransition, externalLabels map[string]string, logger log.Logger) []stream { buckets := make(map[string][]row) // label repr -> entries for _, state := range states { if !shouldRecord(state) { continue } - labels := h.addExternalLabels(removePrivateLabels(state.State.Labels)) + labels := mergeLabels(removePrivateLabels(state.State.Labels), externalLabels) labels[OrgIDLabel] = fmt.Sprint(rule.OrgID) labels[RuleUIDLabel] = fmt.Sprint(rule.UID) labels[GroupLabel] = fmt.Sprint(rule.Group) @@ -84,7 +83,13 @@ func (h *RemoteLokiBackend) statesToStreams(rule history_model.RuleMeta, states Previous: state.PreviousFormatted(), Current: state.Formatted(), Values: valuesAsDataBlob(state.State), + DashboardUID: rule.DashboardUID, + PanelID: rule.PanelID, } + if state.State.State == eval.Error { + entry.Error = state.Error.Error() + } + jsn, err := json.Marshal(entry) if err != nil { logger.Error("Failed to construct history record for state, skipping", "error", err) @@ -122,39 +127,20 @@ func (h *RemoteLokiBackend) recordStreams(ctx context.Context, streams []stream, return nil } -func (h *RemoteLokiBackend) addExternalLabels(labels data.Labels) data.Labels { - for k, v := range h.externalLabels { - labels[k] = v - } - return labels -} - type lokiEntry struct { SchemaVersion int `json:"schemaVersion"` Previous string `json:"previous"` Current string `json:"current"` + Error string `json:"error,omitempty"` Values *simplejson.Json `json:"values"` + DashboardUID string `json:"dashboardUID"` + PanelID int64 `json:"panelID"` } func valuesAsDataBlob(state *state.State) *simplejson.Json { - jsonData := simplejson.New() - - switch state.State { - case eval.Error: - if state.Error == nil { - jsonData.Set("error", nil) - } else { - jsonData.Set("error", state.Error.Error()) - } - case eval.NoData: - jsonData.Set("noData", true) - default: - keys := make([]string, 0, len(state.Values)) - for k := range state.Values { - keys = append(keys, k) - } - sort.Strings(keys) - jsonData.Set("values", simplejson.NewFromAny(state.Values)) + if state.State == eval.Error || state.State == eval.NoData { + return simplejson.New() } - return jsonData + + return jsonifyValues(state.Values) } diff --git a/pkg/services/ngalert/state/historian/loki_test.go b/pkg/services/ngalert/state/historian/loki_test.go new file mode 100644 index 00000000000..0357bac6cc9 --- /dev/null +++ b/pkg/services/ngalert/state/historian/loki_test.go @@ -0,0 +1,175 @@ +package historian + +import ( + "encoding/json" + "fmt" + "sort" + "testing" + + "github.com/grafana/grafana-plugin-sdk-go/data" + "github.com/grafana/grafana/pkg/infra/log" + "github.com/grafana/grafana/pkg/services/ngalert/eval" + "github.com/grafana/grafana/pkg/services/ngalert/state" + history_model "github.com/grafana/grafana/pkg/services/ngalert/state/historian/model" + "github.com/stretchr/testify/require" +) + +func TestRemoteLokiBackend(t *testing.T) { + t.Run("statesToStreams", func(t *testing.T) { + t.Run("skips non-transitory states", func(t *testing.T) { + rule := createTestRule() + l := log.NewNopLogger() + states := singleFromNormal(&state.State{State: eval.Normal}) + + res := statesToStreams(rule, states, nil, l) + + require.Empty(t, res) + }) + + t.Run("maps evaluation errors", func(t *testing.T) { + rule := createTestRule() + l := log.NewNopLogger() + states := singleFromNormal(&state.State{State: eval.Error, Error: fmt.Errorf("oh no")}) + + res := statesToStreams(rule, states, nil, l) + + entry := requireSingleEntry(t, res) + require.Contains(t, entry.Error, "oh no") + }) + + t.Run("maps NoData results", func(t *testing.T) { + rule := createTestRule() + l := log.NewNopLogger() + states := singleFromNormal(&state.State{State: eval.NoData}) + + res := statesToStreams(rule, states, nil, l) + + _ = requireSingleEntry(t, res) + }) + + t.Run("produces expected stream identifier", func(t *testing.T) { + rule := createTestRule() + l := log.NewNopLogger() + states := singleFromNormal(&state.State{ + State: eval.Alerting, + Labels: data.Labels{"a": "b"}, + }) + + res := statesToStreams(rule, states, nil, l) + + require.Len(t, res, 1) + exp := map[string]string{ + "folderUID": rule.NamespaceUID, + "group": rule.Group, + "orgID": fmt.Sprint(rule.OrgID), + "ruleUID": rule.UID, + "a": "b", + } + require.Equal(t, exp, res[0].Stream) + }) + + t.Run("groups streams based on combined labels", func(t *testing.T) { + rule := createTestRule() + l := log.NewNopLogger() + states := []state.StateTransition{ + { + PreviousState: eval.Normal, + State: &state.State{ + State: eval.Alerting, + Labels: data.Labels{"a": "b"}, + }, + }, + { + PreviousState: eval.Normal, + State: &state.State{ + State: eval.Alerting, + Labels: data.Labels{"a": "b"}, + }, + }, + { + PreviousState: eval.Normal, + State: &state.State{ + State: eval.Alerting, + Labels: data.Labels{"c": "d"}, + }, + }, + } + + res := statesToStreams(rule, states, nil, l) + + require.Len(t, res, 2) + sort.Slice(res, func(i, j int) bool { return len(res[i].Values) > len(res[j].Values) }) + require.Contains(t, res[0].Stream, "a") + require.Len(t, res[0].Values, 2) + require.Contains(t, res[1].Stream, "c") + require.Len(t, res[1].Values, 1) + }) + + t.Run("excludes private labels", func(t *testing.T) { + rule := createTestRule() + l := log.NewNopLogger() + states := singleFromNormal(&state.State{ + State: eval.Alerting, + Labels: data.Labels{"__private__": "b"}, + }) + + res := statesToStreams(rule, states, nil, l) + + require.Len(t, res, 1) + require.NotContains(t, res[0].Stream, "__private__") + }) + + t.Run("serializes values when regular", func(t *testing.T) { + rule := createTestRule() + l := log.NewNopLogger() + states := singleFromNormal(&state.State{ + State: eval.Alerting, + Values: map[string]float64{"A": 2.0, "B": 5.5}, + }) + + res := statesToStreams(rule, states, nil, l) + + entry := requireSingleEntry(t, res) + require.NotNil(t, entry.Values) + require.NotNil(t, entry.Values.Get("A")) + require.NotNil(t, entry.Values.Get("B")) + require.InDelta(t, 2.0, entry.Values.Get("A").MustFloat64(), 1e-4) + require.InDelta(t, 5.5, entry.Values.Get("B").MustFloat64(), 1e-4) + }) + }) +} + +func singleFromNormal(st *state.State) []state.StateTransition { + return []state.StateTransition{ + { + PreviousState: eval.Normal, + State: st, + }, + } +} + +func createTestRule() history_model.RuleMeta { + return history_model.RuleMeta{ + OrgID: 1, + UID: "rule-uid", + Group: "my-group", + NamespaceUID: "my-folder", + DashboardUID: "dash-uid", + PanelID: 123, + } +} + +func requireSingleEntry(t *testing.T, res []stream) lokiEntry { + require.Len(t, res, 1) + require.Len(t, res[0].Values, 1) + return requireEntry(t, res[0].Values[0]) +} + +func requireEntry(t *testing.T, row row) lokiEntry { + t.Helper() + + var entry lokiEntry + err := json.Unmarshal([]byte(row.Val), &entry) + require.NoError(t, err) + return entry +}