From b2e84277a3da26aaeded64048211ecd66efa8341 Mon Sep 17 00:00:00 2001 From: Ganesh Vernekar <15064823+codesome@users.noreply.github.com> Date: Wed, 19 May 2021 23:32:09 +0530 Subject: [PATCH] NGAlert: Add Kafka notification channel (#34156) Signed-off-by: Ganesh Vernekar --- pkg/services/ngalert/notifier/alertmanager.go | 2 + .../ngalert/notifier/available_channels.go | 24 +++ .../ngalert/notifier/channels/dingding.go | 6 +- .../notifier/channels/dingding_test.go | 2 +- .../ngalert/notifier/channels/kafka.go | 124 ++++++++++++++ .../ngalert/notifier/channels/kafka_test.go | 155 ++++++++++++++++++ .../alerting/api_available_channel_test.go | 41 +++++ 7 files changed, 350 insertions(+), 4 deletions(-) create mode 100644 pkg/services/ngalert/notifier/channels/kafka.go create mode 100644 pkg/services/ngalert/notifier/channels/kafka_test.go diff --git a/pkg/services/ngalert/notifier/alertmanager.go b/pkg/services/ngalert/notifier/alertmanager.go index 7ca367a03d8..ecc8464fc66 100644 --- a/pkg/services/ngalert/notifier/alertmanager.go +++ b/pkg/services/ngalert/notifier/alertmanager.go @@ -424,6 +424,8 @@ func (am *Alertmanager) buildReceiverIntegrations(receiver *apimodels.PostableAp n, err = channels.NewTeamsNotifier(cfg, tmpl) case "dingding": n, err = channels.NewDingDingNotifier(cfg, tmpl) + case "kafka": + n, err = channels.NewKafkaNotifier(cfg, tmpl) case "webhook": n, err = channels.NewWebHookNotifier(cfg, tmpl) case "sensugo": diff --git a/pkg/services/ngalert/notifier/available_channels.go b/pkg/services/ngalert/notifier/available_channels.go index 998aef23362..0cf6aa967bd 100644 --- a/pkg/services/ngalert/notifier/available_channels.go +++ b/pkg/services/ngalert/notifier/available_channels.go @@ -138,6 +138,30 @@ func GetAvailableNotifiers() []*alerting.NotifierPlugin { }, }, }, + { + Type: "kafka", + Name: "Kafka REST Proxy", + Description: "Sends notifications to Kafka Rest Proxy", + Heading: "Kafka settings", + Options: []alerting.NotifierOption{ + { + Label: "Kafka REST Proxy", + Element: alerting.ElementTypeInput, + InputType: alerting.InputTypeText, + Placeholder: "http://localhost:8082", + PropertyName: "kafkaRestProxy", + Required: true, + }, + { + Label: "Topic", + Element: alerting.ElementTypeInput, + InputType: alerting.InputTypeText, + Placeholder: "topic1", + PropertyName: "kafkaTopic", + Required: true, + }, + }, + }, { Type: "email", Name: "Email", diff --git a/pkg/services/ngalert/notifier/channels/dingding.go b/pkg/services/ngalert/notifier/channels/dingding.go index 865361e0f1e..5c83cbddbad 100644 --- a/pkg/services/ngalert/notifier/channels/dingding.go +++ b/pkg/services/ngalert/notifier/channels/dingding.go @@ -3,11 +3,11 @@ package channels import ( "context" "encoding/json" + "fmt" "net/url" "path" gokit_log "github.com/go-kit/kit/log" - "github.com/pkg/errors" "github.com/prometheus/alertmanager/notify" "github.com/prometheus/alertmanager/template" "github.com/prometheus/alertmanager/types" @@ -106,7 +106,7 @@ func (dd *DingDingNotifier) Notify(ctx context.Context, as ...*types.Alert) (boo } if tmplErr != nil { - return false, errors.Wrap(tmplErr, "failed to template dingding message") + return false, fmt.Errorf("failed to template DingDing message: %w", tmplErr) } body, err := json.Marshal(bodyMsg) @@ -120,7 +120,7 @@ func (dd *DingDingNotifier) Notify(ctx context.Context, as ...*types.Alert) (boo } if err := bus.DispatchCtx(ctx, cmd); err != nil { - return false, errors.Wrap(err, "send notification to dingding") + return false, fmt.Errorf("send notification to dingding: %w", err) } return true, nil diff --git a/pkg/services/ngalert/notifier/channels/dingding_test.go b/pkg/services/ngalert/notifier/channels/dingding_test.go index c594937ecf2..547b707e781 100644 --- a/pkg/services/ngalert/notifier/channels/dingding_test.go +++ b/pkg/services/ngalert/notifier/channels/dingding_test.go @@ -95,7 +95,7 @@ func TestDingdingNotifier(t *testing.T) { "url": "http://localhost", "message": "{{ .Status }" }`, - expMsgError: errors.New("failed to template dingding message: template: :1: unexpected \"}\" in operand"), + expMsgError: errors.New("failed to template DingDing message: template: :1: unexpected \"}\" in operand"), }, } diff --git a/pkg/services/ngalert/notifier/channels/kafka.go b/pkg/services/ngalert/notifier/channels/kafka.go new file mode 100644 index 00000000000..45a2d241518 --- /dev/null +++ b/pkg/services/ngalert/notifier/channels/kafka.go @@ -0,0 +1,124 @@ +package channels + +import ( + "context" + "fmt" + "path" + + gokit_log "github.com/go-kit/kit/log" + "github.com/prometheus/alertmanager/notify" + "github.com/prometheus/alertmanager/template" + "github.com/prometheus/alertmanager/types" + "github.com/prometheus/common/model" + + "github.com/grafana/grafana/pkg/bus" + "github.com/grafana/grafana/pkg/components/simplejson" + "github.com/grafana/grafana/pkg/infra/log" + "github.com/grafana/grafana/pkg/models" + "github.com/grafana/grafana/pkg/services/alerting" + old_notifiers "github.com/grafana/grafana/pkg/services/alerting/notifiers" +) + +// KafkaNotifier is responsible for sending +// alert notifications to Kafka. +type KafkaNotifier struct { + old_notifiers.NotifierBase + Endpoint string + Topic string + log log.Logger + tmpl *template.Template +} + +// NewKafkaNotifier is the constructor function for the Kafka notifier. +func NewKafkaNotifier(model *NotificationChannelConfig, t *template.Template) (*KafkaNotifier, error) { + endpoint := model.Settings.Get("kafkaRestProxy").MustString() + if endpoint == "" { + return nil, alerting.ValidationError{Reason: "Could not find kafka rest proxy endpoint property in settings"} + } + topic := model.Settings.Get("kafkaTopic").MustString() + if topic == "" { + return nil, alerting.ValidationError{Reason: "Could not find kafka topic property in settings"} + } + + return &KafkaNotifier{ + NotifierBase: old_notifiers.NewNotifierBase(&models.AlertNotification{ + Uid: model.UID, + Name: model.Name, + Type: model.Type, + DisableResolveMessage: model.DisableResolveMessage, + Settings: model.Settings, + }), + Endpoint: endpoint, + Topic: topic, + log: log.New("alerting.notifier.kafka"), + tmpl: t, + }, nil +} + +// Notify sends the alert notification. +func (kn *KafkaNotifier) Notify(ctx context.Context, as ...*types.Alert) (bool, error) { + // We are using the state from 7.x to not break kafka. + // TODO: should we switch to the new ones? + alerts := types.Alerts(as...) + state := models.AlertStateAlerting + if alerts.Status() == model.AlertResolved { + state = models.AlertStateOK + } + + kn.log.Debug("Notifying Kafka", "alert_state", state) + + data := notify.GetTemplateData(ctx, kn.tmpl, as, gokit_log.NewNopLogger()) + var tmplErr error + tmpl := notify.TmplText(kn.tmpl, data, &tmplErr) + + bodyJSON := simplejson.New() + bodyJSON.Set("alert_state", state) + bodyJSON.Set("description", tmpl(`{{ template "default.title" . }}`)) + bodyJSON.Set("client", "Grafana") + bodyJSON.Set("details", tmpl(`{{ template "default.message" . }}`)) + bodyJSON.Set("client_url", path.Join(kn.tmpl.ExternalURL.String(), "/alerting/list")) + + groupKey, err := notify.ExtractGroupKey(ctx) + if err != nil { + return false, err + } + bodyJSON.Set("incident_key", groupKey.Hash()) + + valueJSON := simplejson.New() + valueJSON.Set("value", bodyJSON) + + recordJSON := simplejson.New() + recordJSON.Set("records", []interface{}{valueJSON}) + + if tmplErr != nil { + return false, fmt.Errorf("failed to template Kafka message: %w", tmplErr) + } + + body, err := recordJSON.MarshalJSON() + if err != nil { + return false, err + } + + topicURL := kn.Endpoint + "/topics/" + kn.Topic + + cmd := &models.SendWebhookSync{ + Url: topicURL, + Body: string(body), + HttpMethod: "POST", + HttpHeader: map[string]string{ + "Content-Type": "application/vnd.kafka.json.v2+json", + "Accept": "application/vnd.kafka.v2+json", + }, + } + + if err := bus.DispatchCtx(ctx, cmd); err != nil { + kn.log.Error("Failed to send notification to Kafka", "error", err, "body", string(body)) + return false, err + } + + return true, nil +} + +func (kn *KafkaNotifier) SendResolved() bool { + return !kn.GetDisableResolveMessage() +} diff --git a/pkg/services/ngalert/notifier/channels/kafka_test.go b/pkg/services/ngalert/notifier/channels/kafka_test.go new file mode 100644 index 00000000000..590c20aaf5a --- /dev/null +++ b/pkg/services/ngalert/notifier/channels/kafka_test.go @@ -0,0 +1,155 @@ +package channels + +import ( + "context" + "net/url" + "testing" + + "github.com/prometheus/alertmanager/notify" + "github.com/prometheus/alertmanager/types" + "github.com/prometheus/common/model" + "github.com/stretchr/testify/require" + + "github.com/grafana/grafana/pkg/bus" + "github.com/grafana/grafana/pkg/components/simplejson" + "github.com/grafana/grafana/pkg/models" + "github.com/grafana/grafana/pkg/services/alerting" +) + +func TestKafkaNotifier(t *testing.T) { + tmpl := templateForTests(t) + + externalURL, err := url.Parse("http://localhost") + require.NoError(t, err) + tmpl.ExternalURL = externalURL + + cases := []struct { + name string + settings string + alerts []*types.Alert + expUrl, expMsg string + expInitError error + expMsgError error + }{ + { + name: "One alert", + settings: `{ + "kafkaRestProxy": "http://localhost", + "kafkaTopic": "sometopic" + }`, + alerts: []*types.Alert{ + { + Alert: model.Alert{ + Labels: model.LabelSet{"alertname": "alert1", "lbl1": "val1"}, + Annotations: model.LabelSet{"ann1": "annv1"}, + }, + }, + }, + expUrl: "http://localhost/topics/sometopic", + expMsg: `{ + "records": [ + { + "value": { + "alert_state": "alerting", + "client": "Grafana", + "client_url": "http:/localhost/alerting/list", + "description": "[FIRING:1] (val1)", + "details": "\n**Firing**\nLabels:\n - alertname = alert1\n - lbl1 = val1\nAnnotations:\n - ann1 = annv1\nSource: \n\n\n\n\n", + "incident_key": "6e3538104c14b583da237e9693b76debbc17f0f8058ef20492e5853096cf8733" + } + } + ] + }`, + expInitError: nil, + expMsgError: nil, + }, { + name: "Multiple alerts", + settings: `{ + "kafkaRestProxy": "http://localhost", + "kafkaTopic": "sometopic" + }`, + alerts: []*types.Alert{ + { + Alert: model.Alert{ + Labels: model.LabelSet{"alertname": "alert1", "lbl1": "val1"}, + Annotations: model.LabelSet{"ann1": "annv1"}, + }, + }, { + Alert: model.Alert{ + Labels: model.LabelSet{"alertname": "alert1", "lbl1": "val2"}, + Annotations: model.LabelSet{"ann1": "annv2"}, + }, + }, + }, + expUrl: "http://localhost/topics/sometopic", + expMsg: `{ + "records": [ + { + "value": { + "alert_state": "alerting", + "client": "Grafana", + "client_url": "http:/localhost/alerting/list", + "description": "[FIRING:2] ", + "details": "\n**Firing**\nLabels:\n - alertname = alert1\n - lbl1 = val1\nAnnotations:\n - ann1 = annv1\nSource: \nLabels:\n - alertname = alert1\n - lbl1 = val2\nAnnotations:\n - ann1 = annv2\nSource: \n\n\n\n\n", + "incident_key": "6e3538104c14b583da237e9693b76debbc17f0f8058ef20492e5853096cf8733" + } + } + ] + }`, + expInitError: nil, + expMsgError: nil, + }, { + name: "Endpoint missing", + settings: `{"kafkaTopic": "sometopic"}`, + expInitError: alerting.ValidationError{Reason: "Could not find kafka rest proxy endpoint property in settings"}, + }, { + name: "Topic missing", + settings: `{"kafkaRestProxy": "http://localhost"}`, + expInitError: alerting.ValidationError{Reason: "Could not find kafka topic property in settings"}, + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + settingsJSON, err := simplejson.NewJson([]byte(c.settings)) + require.NoError(t, err) + + m := &NotificationChannelConfig{ + Name: "kafka_testing", + Type: "kafka", + Settings: settingsJSON, + } + + pn, err := NewKafkaNotifier(m, tmpl) + if c.expInitError != nil { + require.Error(t, err) + require.Equal(t, c.expInitError.Error(), err.Error()) + return + } + require.NoError(t, err) + + body := "" + actUrl := "" + bus.AddHandlerCtx("test", func(ctx context.Context, webhook *models.SendWebhookSync) error { + body = webhook.Body + actUrl = webhook.Url + return nil + }) + + ctx := notify.WithGroupKey(context.Background(), "alertname") + ctx = notify.WithGroupLabels(ctx, model.LabelSet{"alertname": ""}) + ok, err := pn.Notify(ctx, c.alerts...) + if c.expMsgError != nil { + require.False(t, ok) + require.Error(t, err) + require.Equal(t, c.expMsgError.Error(), err.Error()) + return + } + require.NoError(t, err) + require.True(t, ok) + + require.Equal(t, c.expUrl, actUrl) + require.JSONEq(t, c.expMsg, body) + }) + } +} diff --git a/pkg/tests/api/alerting/api_available_channel_test.go b/pkg/tests/api/alerting/api_available_channel_test.go index 6e969c4ae40..e5fe45d595a 100644 --- a/pkg/tests/api/alerting/api_available_channel_test.go +++ b/pkg/tests/api/alerting/api_available_channel_test.go @@ -108,6 +108,47 @@ var expAvailableChannelJsonOutput = ` } ] }, + { + "type": "kafka", + "name": "Kafka REST Proxy", + "heading": "Kafka settings", + "description": "Sends notifications to Kafka Rest Proxy", + "info": "", + "options": [ + { + "element": "input", + "inputType": "text", + "label": "Kafka REST Proxy", + "description": "", + "placeholder": "http://localhost:8082", + "propertyName": "kafkaRestProxy", + "selectOptions": null, + "showWhen": { + "field": "", + "is": "" + }, + "required": true, + "validationRule": "", + "secure": false + }, + { + "element": "input", + "inputType": "text", + "label": "Topic", + "description": "", + "placeholder": "topic1", + "propertyName": "kafkaTopic", + "selectOptions": null, + "showWhen": { + "field": "", + "is": "" + }, + "required": true, + "validationRule": "", + "secure": false + } + ] + }, { "type": "email", "name": "Email",