[v9.4.x] Alerting: Usability adjustments to Loki representation of state history values (#62811)

Alerting: Usability adjustments to Loki representation of state history values (#62643)

* Extract label merge, add test file

* Extract error/NoData to first class fields, remove a layer from values

* Include dashUID and panelID as line-level fields

* Drop unnecessary object receiver

* Add tests for stream building

* Drop NoData field from log lines

(cherry picked from commit 9fa28c11c5)

Co-authored-by: Alexander Weaver <weaver.alex.d@gmail.com>

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
+}

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()
+	if state.State == eval.Error || state.State == eval.NoData {
-
+		return 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))
 	}
-	return jsonData
+
+	return jsonifyValues(state.Values)
 }

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
+}