diff --git a/pkg/services/ngalert/metrics/notification_historian.go b/pkg/services/ngalert/metrics/notification_historian.go index bb3dd4f0977..880e156a83c 100644 --- a/pkg/services/ngalert/metrics/notification_historian.go +++ b/pkg/services/ngalert/metrics/notification_historian.go @@ -8,8 +8,8 @@ import ( type NotificationHistorian struct { Info prometheus.Gauge - WritesTotal *prometheus.CounterVec - WritesFailed *prometheus.CounterVec + WritesTotal prometheus.Counter + WritesFailed prometheus.Counter WriteDuration *instrument.HistogramCollector BytesWritten prometheus.Counter } @@ -22,18 +22,18 @@ func NewNotificationHistorianMetrics(r prometheus.Registerer) *NotificationHisto Name: "notification_history_info", Help: "Information about the notification history store.", }), - WritesTotal: promauto.With(r).NewCounterVec(prometheus.CounterOpts{ + WritesTotal: promauto.With(r).NewCounter(prometheus.CounterOpts{ Namespace: Namespace, Subsystem: Subsystem, Name: "notification_history_writes_total", Help: "The total number of notification history batches that were attempted to be written.", - }, []string{"org"}), - WritesFailed: promauto.With(r).NewCounterVec(prometheus.CounterOpts{ + }), + WritesFailed: promauto.With(r).NewCounter(prometheus.CounterOpts{ Namespace: Namespace, Subsystem: Subsystem, Name: "notification_history_writes_failed_total", Help: "The total number of failed writes of notification history batches.", - }, []string{"org"}), + }), WriteDuration: instrument.NewHistogramCollector(promauto.With(r).NewHistogramVec(prometheus.HistogramOpts{ Namespace: Namespace, Subsystem: Subsystem, diff --git a/pkg/services/ngalert/notifier/historian.go b/pkg/services/ngalert/notifier/historian.go index bdaa631d6ef..30274ca924e 100644 --- a/pkg/services/ngalert/notifier/historian.go +++ b/pkg/services/ngalert/notifier/historian.go @@ -8,7 +8,7 @@ import ( "strings" "time" - models2 "github.com/grafana/alerting/models" + alertingModels "github.com/grafana/alerting/models" "github.com/grafana/grafana/pkg/infra/log" "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/services/ngalert/client" @@ -17,9 +17,14 @@ import ( "github.com/grafana/grafana/pkg/setting" "github.com/prometheus/alertmanager/notify" "github.com/prometheus/alertmanager/types" + prometheusModel "github.com/prometheus/common/model" + "go.opentelemetry.io/otel/trace" ) const LokiClientSpanName = "ngalert.notification-historian.client" +const NotificationHistoryWriteTimeout = time.Minute +const NotificationHistoryKey = "from" +const NotificationHistoryLabelValue = "notify-history" type NotificationHistoryLokiEntry struct { SchemaVersion int `json:"schemaVersion"` @@ -29,19 +34,6 @@ type NotificationHistoryLokiEntry struct { GroupLabels map[string]string `json:"groupLabels"` Alerts []NotificationHistoryLokiEntryAlert `json:"alerts"` Error string `json:"error,omitempty"` - - // TODO ? - //RuleIDs string `json:"ruleIDs"` - - //DashboardUID string `json:"dashboardUID"` - //PanelID int64 `json:"panelID"` - //Fingerprint string `json:"fingerprint"` - //RuleTitle string `json:"ruleTitle"` - //RuleID int64 `json:"ruleID"` - //RuleUID string `json:"ruleUID"` - //// InstanceLabels is exactly the set of labels associated with the alert instance in Alertmanager. - //// These should not be conflated with labels associated with log streams. - //InstanceLabels map[string]string `json:"labels"` } type NotificationHistoryLokiEntryAlert struct { @@ -50,17 +42,6 @@ type NotificationHistoryLokiEntryAlert struct { Annotations map[string]string `json:"annotations"` StartsAt time.Time `json:"startsAt"` EndsAt time.Time `json:"endsAt"` - //GeneratorURL string `json:"generatorURL"` - //Fingerprint string `json:"fingerprint"` - //SilenceURL string `json:"silenceURL"` - //DashboardURL string `json:"dashboardURL"` - //PanelURL string `json:"panelURL"` - //Values map[string]float64 `json:"values"` - //ValueString string `json:"valueString"` // TODO: Remove in Grafana 10 - //ImageURL string `json:"imageURL,omitempty"` - //EmbeddedImage string `json:"embeddedImage,omitempty"` - //OrgID *int64 `json:"orgId,omitempty"` - } type remoteLokiClient interface { @@ -73,10 +54,6 @@ type NotificationHistorian struct { externalLabels map[string]string metrics *metrics.NotificationHistorian log log.Logger - - // TODO - //ac AccessControl - //ruleStore RuleStore } func NewNotificationHistorian(logger log.Logger, cfg lokiclient.LokiConfig, req client.Requester, metrics *metrics.NotificationHistorian, tracer tracing.Tracer) *NotificationHistorian { @@ -86,127 +63,54 @@ func NewNotificationHistorian(logger log.Logger, cfg lokiclient.LokiConfig, req metrics: metrics, log: logger, } - -} - -func (h *NotificationHistorian) NotificationHistoryLogToStream(ctx context.Context, alerts []*types.Alert, notificationError error) (lokiclient.Stream, error) { - receiverName, _ := notify.ReceiverName(ctx) - now, _ := notify.Now(ctx) - - labels := make(map[string]string) - labels["from"] = "notify-history" - // TODO - //labels[OrgIDLabel] = fmt.Sprint(orgID) - //labels[GroupLabel] = fmt.Sprint(rule.Group) - //labels[FolderUIDLabel] = fmt.Sprint(rule.NamespaceUID) - //labels["receiver"] = receiver - //labels["status"] = status - - ruleUIDs := make([]string, 0) - seen := make(map[string]bool) - for _, alert := range alerts { - ruleUID := string(alert.Labels[models2.RuleUIDLabel]) - if ruleUID == "" { - continue - } - if _, exists := seen[ruleUID]; !exists { - seen[ruleUID] = true - ruleUIDs = append(ruleUIDs, ruleUID) - } - } - - // TODO: rename - alertss := make([]NotificationHistoryLokiEntryAlert, len(alerts)) - for i, alert := range alerts { - alertss[i] = NotificationHistoryLokiEntryAlert{ - Labels: make(map[string]string), - Annotations: make(map[string]string), - Status: string(alert.Status()), - StartsAt: alert.StartsAt, - EndsAt: alert.EndsAt, - } - for k, v := range alert.Labels { - if !strings.HasPrefix(string(k), "__") && !strings.HasSuffix(string(k), "__") { - alertss[i].Labels[string(k)] = string(v) - } - } - for k, v := range alert.Annotations { - if !strings.HasPrefix(string(k), "__") && !strings.HasSuffix(string(k), "__") { - alertss[i].Annotations[string(k)] = string(v) - } - } - } - - notificationErrorStr := "" - if notificationError != nil { - notificationErrorStr = notificationError.Error() - } - - entry := NotificationHistoryLokiEntry{ - SchemaVersion: 1, - RuleUIDs: ruleUIDs, - Receiver: receiverName, - Status: "?", - // TODO: add group labels - GroupLabels: make(map[string]string), - Alerts: alertss, - Error: notificationErrorStr, - } - - // Marshal to JSON for the historian - entryJSON, jsonErr := json.Marshal(entry) - if jsonErr != nil { - return lokiclient.Stream{}, jsonErr - } - - return lokiclient.Stream{ - Stream: labels, - Values: []lokiclient.Sample{ - { - T: now, - V: string(entryJSON), - }}, - }, nil -} - -func (h *NotificationHistorian) Record(ctx context.Context, alerts []*types.Alert, notificationErr error) error { - // TODO: goroutine - - stream, err := h.NotificationHistoryLogToStream(ctx, alerts, notificationErr) - if err != nil { - h.log.FromContext(ctx).Error("Failed to convert notification history to stream", "error", err) - return fmt.Errorf("failed to convert notification history to stream: %w", err) - } - - logger := h.log.FromContext(ctx) - logger.Debug("Saving notification history", "samples", len(stream.Values)) - - // TODO metrics - //org := fmt.Sprint(orgID) - //h.metrics.WritesTotal.WithLabelValues(org, "loki").Inc() - - if err := h.client.Push(ctx, []lokiclient.Stream{stream}); err != nil { - // TODO metrics - //h.metrics.WritesFailed.WithLabelValues(org, "loki").Inc() - logger.Error("Failed to save notification history", "error", err) - return err - } - - receiverName, _ := notify.ReceiverName(ctx) - groupLabels, _ := notify.GroupLabels(ctx) - groupKey, _ := notify.GroupKey(ctx) - now, _ := notify.Now(ctx) - - logger.Debug("ctx", "receiverName", receiverName, "groupLabels", groupLabels, "groupKey", groupKey, "now", now) - logger.Debug("Done saving notification history", "samples", len(stream.Values)) - return nil } func (h *NotificationHistorian) TestConnection(ctx context.Context) error { return h.client.Ping(ctx) } -// TODO: reuse the one from state historian? +func (h *NotificationHistorian) Record(ctx context.Context, alerts []*types.Alert, notificationErr error) <-chan error { + stream, err := h.prepareStream(ctx, alerts, notificationErr) + logger := h.log.FromContext(ctx) + errCh := make(chan error, 1) + if err != nil { + logger.Error("Failed to convert notification history to stream", "error", err) + errCh <- fmt.Errorf("failed to convert notification history to stream: %w", err) + close(errCh) + return errCh + } + + if len(stream.Values) == 0 { + close(errCh) + return errCh + } + + // This is a new background job, so let's create a new context for it. + // We want it to be isolated, i.e. we don't want grafana shutdowns to interrupt this work + // immediately but rather try to flush writes. + // This also prevents timeouts or other lingering objects (like transactions) from being + // incorrectly propagated here from other areas. + writeCtx := context.Background() + writeCtx, cancel := context.WithTimeout(writeCtx, NotificationHistoryWriteTimeout) + writeCtx = trace.ContextWithSpan(writeCtx, trace.SpanFromContext(ctx)) + + go func(ctx context.Context) { + defer cancel() + defer close(errCh) + logger := h.log.FromContext(ctx) + logger.Debug("Saving notification history batch", "samples", len(stream.Values)) + h.metrics.WritesTotal.Inc() + + if err := h.recordStream(ctx, stream, logger); err != nil { + logger.Error("Failed to save notification history batch", "error", err) + h.metrics.WritesFailed.Inc() + errCh <- fmt.Errorf("failed to save notification history batch: %w", err) + } + }(writeCtx) + return errCh +} + +// NewLokiConfig is a copy of historian.NewLokiConfig func NewLokiConfig(cfg setting.UnifiedAlertingNotificationHistorySettings) (lokiclient.LokiConfig, error) { read, write := cfg.LokiReadURL, cfg.LokiWriteURL if read == "" { @@ -245,3 +149,99 @@ func NewLokiConfig(cfg setting.UnifiedAlertingNotificationHistorySettings) (loki Encoder: lokiclient.SnappyProtoEncoder{}, }, nil } + +func (h *NotificationHistorian) prepareStream(ctx context.Context, alerts []*types.Alert, notificationError error) (lokiclient.Stream, error) { + receiverName, ok := notify.ReceiverName(ctx) + if !ok { + return lokiclient.Stream{}, fmt.Errorf("receiver name not found in context") + } + groupLabels, ok := notify.GroupLabels(ctx) + if !ok { + return lokiclient.Stream{}, fmt.Errorf("group labels not found in context") + } + now, ok := notify.Now(ctx) + if !ok { + return lokiclient.Stream{}, fmt.Errorf("now not found in context") + } + + // Get unique rule UIDs from the alerts + ruleUIDs := make([]string, 0) + seen := make(map[string]bool) + for _, alert := range alerts { + ruleUID := string(alert.Labels[alertingModels.RuleUIDLabel]) + if ruleUID == "" { + return lokiclient.Stream{}, fmt.Errorf("alert missing rule UID label: %v", alert.Labels) + } + if _, exists := seen[ruleUID]; !exists { + seen[ruleUID] = true + ruleUIDs = append(ruleUIDs, ruleUID) + } + } + + entryAlerts := make([]NotificationHistoryLokiEntryAlert, len(alerts)) + for i, alert := range alerts { + labels := prepareLabels(alert.Labels) + annotations := prepareLabels(alert.Annotations) + entryAlerts[i] = NotificationHistoryLokiEntryAlert{ + Labels: labels, + Annotations: annotations, + Status: string(alert.StatusAt(now)), + StartsAt: alert.StartsAt, + EndsAt: alert.EndsAt, + } + } + + notificationErrorStr := "" + if notificationError != nil { + notificationErrorStr = notificationError.Error() + } + + entry := NotificationHistoryLokiEntry{ + SchemaVersion: 1, + RuleUIDs: ruleUIDs, + Receiver: receiverName, + Status: string(types.Alerts(alerts...).StatusAt(now)), + GroupLabels: prepareLabels(groupLabels), + Alerts: entryAlerts, + Error: notificationErrorStr, + } + + entryJSON, err := json.Marshal(entry) + if err != nil { + return lokiclient.Stream{}, err + } + + streamLabels := make(map[string]string) + streamLabels[NotificationHistoryKey] = NotificationHistoryLabelValue + for k, v := range h.externalLabels { + streamLabels[k] = v + } + + return lokiclient.Stream{ + Stream: streamLabels, + Values: []lokiclient.Sample{ + { + T: now, + V: string(entryJSON), + }}, + }, nil +} + +func (h *NotificationHistorian) recordStream(ctx context.Context, stream lokiclient.Stream, logger log.Logger) error { + if err := h.client.Push(ctx, []lokiclient.Stream{stream}); err != nil { + return err + } + logger.Debug("Done saving notification history batch", "samples", len(stream.Values)) + return nil +} + +func prepareLabels(labels prometheusModel.LabelSet) map[string]string { + result := make(map[string]string) + for k, v := range labels { + // Remove private labels + if !strings.HasPrefix(string(k), "__") && !strings.HasSuffix(string(k), "__") { + result[string(k)] = string(v) + } + } + return result +}