Merge branch 'master' into develop

pkg/services/alerting/notifiers/kafka.go

@@ -0,0 +1,120 @@
+package notifiers
+
+import (
+	"strconv"
+
+	"fmt"
+
+	"github.com/grafana/grafana/pkg/bus"
+	"github.com/grafana/grafana/pkg/components/simplejson"
+	"github.com/grafana/grafana/pkg/log"
+	m "github.com/grafana/grafana/pkg/models"
+	"github.com/grafana/grafana/pkg/services/alerting"
+)
+
+func init() {
+	alerting.RegisterNotifier(&alerting.NotifierPlugin{
+		Type:        "kafka",
+		Name:        "Kafka REST Proxy",
+		Description: "Sends notifications to Kafka Rest Proxy",
+		Factory:     NewKafkaNotifier,
+		OptionsTemplate: `
+      <h3 class="page-heading">Kafka settings</h3>
+      <div class="gf-form">
+        <span class="gf-form-label width-14">Kafka REST Proxy</span>
+        <input type="text" required class="gf-form-input max-width-22" ng-model="ctrl.model.settings.kafkaRestProxy" placeholder="http://localhost:8082"></input>
+      </div>
+      <div class="gf-form">
+        <span class="gf-form-label width-14">Topic</span>
+        <input type="text" required class="gf-form-input max-width-22" ng-model="ctrl.model.settings.kafkaTopic" placeholder="topic1"></input>
+      </div>
+    `,
+	})
+}
+
+func NewKafkaNotifier(model *m.AlertNotification) (alerting.Notifier, 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: NewNotifierBase(model.Id, model.IsDefault, model.Name, model.Type, model.Settings),
+		Endpoint:     endpoint,
+		Topic:        topic,
+		log:          log.New("alerting.notifier.kafka"),
+	}, nil
+}
+
+type KafkaNotifier struct {
+	NotifierBase
+	Endpoint string
+	Topic    string
+	log      log.Logger
+}
+
+func (this *KafkaNotifier) Notify(evalContext *alerting.EvalContext) error {
+
+	state := evalContext.Rule.State
+
+	customData := "Triggered metrics:\n\n"
+	for _, evt := range evalContext.EvalMatches {
+		customData = customData + fmt.Sprintf("%s: %v\n", evt.Metric, evt.Value)
+	}
+
+	this.log.Info("Notifying Kafka", "alert_state", state)
+
+	recordJSON := simplejson.New()
+	records := make([]interface{}, 1)
+
+	bodyJSON := simplejson.New()
+	bodyJSON.Set("description", evalContext.Rule.Name+" - "+evalContext.Rule.Message)
+	bodyJSON.Set("client", "Grafana")
+	bodyJSON.Set("details", customData)
+	bodyJSON.Set("incident_key", "alertId-"+strconv.FormatInt(evalContext.Rule.Id, 10))
+
+	ruleUrl, err := evalContext.GetRuleUrl()
+	if err != nil {
+		this.log.Error("Failed get rule link", "error", err)
+		return err
+	}
+	bodyJSON.Set("client_url", ruleUrl)
+
+	if evalContext.ImagePublicUrl != "" {
+		contexts := make([]interface{}, 1)
+		imageJSON := simplejson.New()
+		imageJSON.Set("type", "image")
+		imageJSON.Set("src", evalContext.ImagePublicUrl)
+		contexts[0] = imageJSON
+		bodyJSON.Set("contexts", contexts)
+	}
+
+	valueJSON := simplejson.New()
+	valueJSON.Set("value", bodyJSON)
+	records[0] = valueJSON
+	recordJSON.Set("records", records)
+	body, _ := recordJSON.MarshalJSON()
+
+	topicUrl := this.Endpoint + "/topics/" + this.Topic
+
+	cmd := &m.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(evalContext.Ctx, cmd); err != nil {
+		this.log.Error("Failed to send notification to Kafka", "error", err, "body", string(body))
+		return err
+	}
+
+	return nil
+}

pkg/services/alerting/notifiers/kafka_test.go

@@ -0,0 +1,55 @@
+package notifiers
+
+import (
+	"testing"
+
+	"github.com/grafana/grafana/pkg/components/simplejson"
+	m "github.com/grafana/grafana/pkg/models"
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func TestKafkaNotifier(t *testing.T) {
+	Convey("Kafka notifier tests", t, func() {
+
+		Convey("Parsing alert notification from settings", func() {
+			Convey("empty settings should return error", func() {
+				json := `{ }`
+
+				settingsJSON, _ := simplejson.NewJson([]byte(json))
+				model := &m.AlertNotification{
+					Name:     "kafka_testing",
+					Type:     "kafka",
+					Settings: settingsJSON,
+				}
+
+				_, err := NewKafkaNotifier(model)
+				So(err, ShouldNotBeNil)
+			})
+
+			Convey("settings should send an event to kafka", func() {
+				json := `
+				{
+					"kafkaRestProxy": "http://localhost:8082",
+					"kafkaTopic": "topic1"
+				}`
+
+				settingsJSON, _ := simplejson.NewJson([]byte(json))
+				model := &m.AlertNotification{
+					Name:     "kafka_testing",
+					Type:     "kafka",
+					Settings: settingsJSON,
+				}
+
+				not, err := NewKafkaNotifier(model)
+				kafkaNotifier := not.(*KafkaNotifier)
+
+				So(err, ShouldBeNil)
+				So(kafkaNotifier.Name, ShouldEqual, "kafka_testing")
+				So(kafkaNotifier.Type, ShouldEqual, "kafka")
+				So(kafkaNotifier.Endpoint, ShouldEqual, "http://localhost:8082")
+				So(kafkaNotifier.Topic, ShouldEqual, "topic1")
+			})
+
+		})
+	})
+}

pkg/services/sqlstore/alert.go

@@ -100,13 +100,13 @@ func HandleAlertsQuery(query *m.GetAlertsQuery) error {
 		sql.WriteString(")")
 	}
 
+	sql.WriteString(" ORDER BY name ASC")
+
 	if query.Limit != 0 {
 		sql.WriteString(" LIMIT ?")
 		params = append(params, query.Limit)
 	}
 
-	sql.WriteString(" ORDER BY name ASC")
-
 	alerts := make([]*m.Alert, 0)
 	if err := x.Sql(sql.String(), params...).Find(&alerts); err != nil {
 		return err