public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Alerting: Add missing custom title and description fields in Kafka contact point (#57361)

Browse Source 
* Add description and details to Kafka notifier

* Fixed testing and add new logic testing

* Add proper description to kafka contact point UI

* Update pkg/services/ngalert/notifier/channels_config/available_channels.go

Co-authored-by: Santiago <santiagohernandez.1997@gmail.com>

* Update pkg/services/ngalert/notifier/channels_config/available_channels.go

Co-authored-by: Santiago <santiagohernandez.1997@gmail.com>

Co-authored-by: Santiago <santiagohernandez.1997@gmail.com>
This commit is contained in:
Alex Moreno
2022-10-27 15:25:54 +02:00
committed by GitHub
co-authored by Santiago
parent 91ee5fec14
commit 1dcc432537
3 changed files with 130 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/ngalert/notifier/channels/kafka.go
+90 -81
View File
@@ -21,108 +21,119 @@ import (
// alert notifications to Kafka.
type KafkaNotifier struct {
*Base
Endpoint string
Topic string
log log.Logger
images ImageStore
ns notifications.WebhookSender
tmpl *template.Template
settings kafkaSettings
}
type KafkaConfig struct {
*NotificationChannelConfig
Endpoint string
Topic string
type kafkaSettings struct {
Endpoint string
Topic string
Description string
Details string
}
func KafkaFactory(fc FactoryConfig) (NotificationChannel, error) {
cfg, err := NewKafkaConfig(fc.Config)
ch, err := newKafkaNotifier(fc)
if err != nil {
return nil, receiverInitError{
Reason: err.Error(),
Cfg: *fc.Config,
}
}
return NewKafkaNotifier(cfg, fc.ImageStore, fc.NotificationService, fc.Template), nil
return ch, nil
}
func NewKafkaConfig(config *NotificationChannelConfig) (*KafkaConfig, error) {
endpoint := config.Settings.Get("kafkaRestProxy").MustString()
// newKafkaNotifier is the constructor function for the Kafka notifier.
func newKafkaNotifier(fc FactoryConfig) (*KafkaNotifier, error) {
endpoint := fc.Config.Settings.Get("kafkaRestProxy").MustString()
if endpoint == "" {
return nil, errors.New("could not find kafka rest proxy endpoint property in settings")
}
topic := config.Settings.Get("kafkaTopic").MustString()
topic := fc.Config.Settings.Get("kafkaTopic").MustString()
if topic == "" {
return nil, errors.New("could not find kafka topic property in settings")
}
return &KafkaConfig{
NotificationChannelConfig: config,
Endpoint: endpoint,
Topic: topic,
}, nil
}
description := fc.Config.Settings.Get("description").MustString(DefaultMessageTitleEmbed)
details := fc.Config.Settings.Get("details").MustString(DefaultMessageEmbed)
// NewKafkaNotifier is the constructor function for the Kafka notifier.
func NewKafkaNotifier(config *KafkaConfig, images ImageStore, ns notifications.WebhookSender, t *template.Template) *KafkaNotifier {
return &KafkaNotifier{
Base: NewBase(&models.AlertNotification{
Uid: config.UID,
Name: config.Name,
Type: config.Type,
DisableResolveMessage: config.DisableResolveMessage,
Settings: config.Settings,
Uid: fc.Config.UID,
Name: fc.Config.Name,
Type: fc.Config.Type,
DisableResolveMessage: fc.Config.DisableResolveMessage,
Settings: fc.Config.Settings,
}),
Endpoint: config.Endpoint,
Topic: config.Topic,
log: log.New("alerting.notifier.kafka"),
images: images,
ns: ns,
tmpl: t,
}
images: fc.ImageStore,
ns: fc.NotificationService,
tmpl: fc.Template,
settings: kafkaSettings{Endpoint: endpoint, Topic: topic, Description: description, Details: details},
}, 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)
var tmplErr error
tmpl, _ := TmplText(ctx, kn.tmpl, as, kn.log, &tmplErr)
topicURL := strings.TrimRight(kn.settings.Endpoint, "/") + "/topics/" + tmpl(kn.settings.Topic)
body, err := kn.buildBody(ctx, tmpl, as...)
if err != nil {
return false, err
}
if tmplErr != nil {
kn.log.Warn("failed to template Kafka message", "error", tmplErr.Error())
}
cmd := &models.SendWebhookSync{
Url: topicURL,
Body: body,
HttpMethod: "POST",
HttpHeader: map[string]string{
"Content-Type": "application/vnd.kafka.json.v2+json",
"Accept": "application/vnd.kafka.v2+json",
},
}
if err = kn.ns.SendWebhookSync(ctx, cmd); err != nil {
kn.log.Error("Failed to send notification to Kafka", "error", err, "body", body)
return false, err
}
return true, nil
}
func (kn *KafkaNotifier) SendResolved() bool {
return !kn.GetDisableResolveMessage()
}
func (kn *KafkaNotifier) buildBody(ctx context.Context, tmpl func(string) string, as ...*types.Alert) (string, error) {
bodyJSON := simplejson.New()
bodyJSON.Set("alert_state", state)
bodyJSON.Set("description", tmpl(DefaultMessageTitleEmbed))
bodyJSON.Set("client", "Grafana")
bodyJSON.Set("details", tmpl(DefaultMessageEmbed))
bodyJSON.Set("description", tmpl(kn.settings.Description))
bodyJSON.Set("details", tmpl(kn.settings.Details))
state := buildState(as...)
kn.log.Debug("notifying Kafka", "alert_state", state)
bodyJSON.Set("alert_state", state)
ruleURL := joinUrlPath(kn.tmpl.ExternalURL.String(), "/alerting/list", kn.log)
bodyJSON.Set("client_url", ruleURL)
var contexts []interface{}
_ = withStoredImages(ctx, kn.log, kn.images,
func(_ int, image ngmodels.Image) error {
if image.URL != "" {
imageJSON := simplejson.New()
imageJSON.Set("type", "image")
imageJSON.Set("src", image.URL)
contexts = append(contexts, imageJSON)
}
return nil
}, as...)
contexts := buildContextImages(ctx, kn.log, kn.images, as...)
if len(contexts) > 0 {
bodyJSON.Set("contexts", contexts)
}
groupKey, err := notify.ExtractGroupKey(ctx)
if err != nil {
return false, err
return "", err
}
bodyJSON.Set("incident_key", groupKey.Hash())
@@ -134,33 +145,31 @@ func (kn *KafkaNotifier) Notify(ctx context.Context, as ...*types.Alert) (bool,
body, err := recordJSON.MarshalJSON()
if err != nil {
return false, err
return "", err
}
topicURL := strings.TrimRight(kn.Endpoint, "/") + "/topics/" + tmpl(kn.Topic)
if tmplErr != nil {
kn.log.Warn("failed to template Kafka message", "error", tmplErr.Error())
}
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 := kn.ns.SendWebhookSync(ctx, cmd); err != nil {
kn.log.Error("Failed to send notification to Kafka", "error", err, "body", string(body))
return false, err
}
return true, nil
return string(body), nil
}
func (kn *KafkaNotifier) SendResolved() bool {
return !kn.GetDisableResolveMessage()
func buildState(as ...*types.Alert) models.AlertStateType {
// We are using the state from 7.x to not break kafka.
// TODO: should we switch to the new ones?
if types.Alerts(as...).Status() == model.AlertResolved {
return models.AlertStateOK
}
return models.AlertStateAlerting
}
func buildContextImages(ctx context.Context, l log.Logger, imageStore ImageStore, as ...*types.Alert) []interface{} {
var contexts []interface{}
_ = withStoredImages(ctx, l, imageStore,
func(_ int, image ngmodels.Image) error {
if image.URL != "" {
imageJSON := simplejson.New()
imageJSON.Set("type", "image")
imageJSON.Set("src", image.URL)
contexts = append(contexts, imageJSON)
}
return nil
}, as...)
return contexts
}
pkg/services/ngalert/notifier/channels/kafka_test.go
+21 -12
View File
@@ -31,10 +31,12 @@ func TestKafkaNotifier(t *testing.T) {
expMsgError error
}{
{
name: "A single alert with image",
name: "A single alert with image and custom description and details",
settings: `{
"kafkaRestProxy": "http://localhost",
"kafkaTopic": "sometopic"
"kafkaTopic": "sometopic",
"description": "customDescription",
"details": "customDetails"
}`,
alerts: []*types.Alert{
{
@@ -53,8 +55,8 @@ func TestKafkaNotifier(t *testing.T) {
"client": "Grafana",
"client_url": "http://localhost/alerting/list",
"contexts": [{"type": "image", "src": "https://www.example.com/test-image-1.jpg"}],
"description": "[FIRING:1] (val1)",
"details": "**Firing**\n\nValue: [no value]\nLabels:\n - alertname = alert1\n - lbl1 = val1\nAnnotations:\n - ann1 = annv1\nSilence: http://localhost/alerting/silence/new?alertmanager=grafana&matcher=alertname%3Dalert1&matcher=lbl1%3Dval1\nDashboard: http://localhost/d/abcd\nPanel: http://localhost/d/abcd?viewPanel=efgh\n",
"description": "customDescription",
"details": "customDetails",
"incident_key": "6e3538104c14b583da237e9693b76debbc17f0f8058ef20492e5853096cf8733"
}
}
@@ -62,7 +64,7 @@ func TestKafkaNotifier(t *testing.T) {
}`,
expMsgError: nil,
}, {
name: "Multiple alerts with images",
name: "Multiple alerts with images with default description and details",
settings: `{
"kafkaRestProxy": "http://localhost",
"kafkaTopic": "sometopic"
@@ -113,14 +115,22 @@ func TestKafkaNotifier(t *testing.T) {
settingsJSON, err := simplejson.NewJson([]byte(c.settings))
require.NoError(t, err)
m := &NotificationChannelConfig{
Name: "kafka_testing",
Type: "kafka",
Settings: settingsJSON,
webhookSender := mockNotificationService()
fc := FactoryConfig{
Config: &NotificationChannelConfig{
Name: "kafka_testing",
Type: "kafka",
Settings: settingsJSON,
},
ImageStore: images,
// TODO: allow changing the associated values for different tests.
NotificationService: webhookSender,
DecryptFunc: nil,
Template: tmpl,
}
webhookSender := mockNotificationService()
cfg, err := NewKafkaConfig(m)
pn, err := newKafkaNotifier(fc)
if c.expInitError != "" {
require.Error(t, err)
require.Equal(t, c.expInitError, err.Error())
@@ -131,7 +141,6 @@ func TestKafkaNotifier(t *testing.T) {
ctx := notify.WithGroupKey(context.Background(), "alertname")
ctx = notify.WithGroupLabels(ctx, model.LabelSet{"alertname": ""})
pn := NewKafkaNotifier(cfg, images, webhookSender, tmpl)
ok, err := pn.Notify(ctx, c.alerts...)
if c.expMsgError != nil {
require.False(t, ok)
pkg/services/ngalert/notifier/channels_config/available_channels.go
+19 -4
View File
@@ -162,6 +162,21 @@ func GetAvailableNotifiers() []*NotifierPlugin {
PropertyName: "kafkaTopic",
Required: true,
},
{
Label: "Description",
Element: ElementTypeInput,
InputType: InputTypeText,
Description: "Templated description of the Kafka message",
PropertyName: "description",
Placeholder: channels.DefaultMessageTitleEmbed,
},
{
Label: "Details",
Element: ElementTypeTextArea,
Description: "Custom details to include with the message. You can use template variables.",
PropertyName: "details",
Placeholder: channels.DefaultMessageEmbed,
},
},
},
{
@@ -195,7 +210,7 @@ func GetAvailableNotifiers() []*NotifierPlugin {
InputType: InputTypeText,
Description: "Templated subject of the email",
PropertyName: "subject",
Placeholder: `{{ template "default.title" . }}`,
Placeholder: channels.DefaultMessageTitleEmbed,
},
},
},
@@ -575,7 +590,7 @@ func GetAvailableNotifiers() []*NotifierPlugin {
InputType: InputTypeText,
Description: "Templated title of the Teams message.",
PropertyName: "title",
Placeholder: `{{ template "default.title" . }}`,
Placeholder: channels.DefaultMessageTitleEmbed,
},
{
Label: "Section Title",
@@ -766,7 +781,7 @@ func GetAvailableNotifiers() []*NotifierPlugin {
InputType: InputTypeText,
Description: "Templated title of the message",
PropertyName: "title",
Placeholder: `{{ template "default.title" . }}`,
Placeholder: channels.DefaultMessageTitleEmbed,
},
{
Label: "To User",
@@ -947,7 +962,7 @@ func GetAvailableNotifiers() []*NotifierPlugin {
Description: "Alert text limited to 130 characters.",
Element: ElementTypeInput,
InputType: InputTypeText,
Placeholder: `{{ template "default.title" . }}`,
Placeholder: channels.DefaultMessageTitleEmbed,
PropertyName: "message",
},
{