fix various usability issues with Alerta integration

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## v0.11.0 [unreleased]
+
+### Release Notes
+
+### Features
+
+### Bugfixes
+- [#199](https://github.com/influxdata/kapacitor/issues/199): BREAKING: Various fixes for the Alerta integration.
+    The `event` property has been removed from the Alerta node and is now set as the value of the alert ID.
+
 ## v0.10.1 [2013-02-08]
 
 ### Release Notes

alert.go

@@ -81,6 +81,9 @@ type AlertData struct {
 	Time    time.Time       `json:"time"`
 	Level   AlertLevel      `json:"level"`
 	Data    influxql.Result `json:"data"`
+
+	// Info for custom templates
+	info detailsInfo
 }
 
 type AlertNode struct {
@@ -203,8 +206,26 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, l *log.Logger) (an *
 	}
 
 	for _, alerta := range n.AlertaHandlers {
-		alerta := alerta
-		an.handlers = append(an.handlers, func(ad *AlertData) { an.handleAlerta(alerta, ad) })
+		// Validate alerta templates
+		rtmpl, err := text.New("resource").Parse(alerta.Resource)
+		if err != nil {
+			return nil, err
+		}
+		gtmpl, err := text.New("group").Parse(alerta.Group)
+		if err != nil {
+			return nil, err
+		}
+		vtmpl, err := text.New("value").Parse(alerta.Value)
+		if err != nil {
+			return nil, err
+		}
+		ai := alertaHandler{
+			AlertaHandler: alerta,
+			resourceTmpl:  rtmpl,
+			groupTmpl:     gtmpl,
+			valueTmpl:     vtmpl,
+		}
+		an.handlers = append(an.handlers, func(ad *AlertData) { an.handleAlerta(ai, ad) })
 	}
 
 	for _, og := range n.OpsGenieHandlers {
@@ -353,7 +374,7 @@ func (a *AlertNode) alertData(
 	if err != nil {
 		return nil, err
 	}
-	msg, details, err := a.renderMessageAndDetails(id, name, group, tags, fields, level)
+	msg, details, info, err := a.renderMessageAndDetails(id, name, group, tags, fields, level)
 	if err != nil {
 		return nil, err
 	}
@@ -364,6 +385,7 @@ func (a *AlertNode) alertData(
 		Time:    t,
 		Level:   level,
 		Data:    a.batchToResult(b),
+		info:    info,
 	}
 	return ad, nil
 }
@@ -480,7 +502,7 @@ func (a *AlertNode) renderID(name string, group models.GroupID, tags models.Tags
 	return id.String(), nil
 }
 
-func (a *AlertNode) renderMessageAndDetails(id, name string, group models.GroupID, tags models.Tags, fields models.Fields, level AlertLevel) (string, string, error) {
+func (a *AlertNode) renderMessageAndDetails(id, name string, group models.GroupID, tags models.Tags, fields models.Fields, level AlertLevel) (string, string, detailsInfo, error) {
 	g := string(group)
 	if group == models.NilGroup {
 		g = "nil"
@@ -499,7 +521,7 @@ func (a *AlertNode) renderMessageAndDetails(id, name string, group models.GroupI
 	var msg bytes.Buffer
 	err := a.messageTmpl.Execute(&msg, minfo)
 	if err != nil {
-		return "", "", err
+		return "", "", detailsInfo{}, err
 	}
 	dinfo := detailsInfo{
 		messageInfo: minfo,
@@ -508,9 +530,9 @@ func (a *AlertNode) renderMessageAndDetails(id, name string, group models.GroupI
 	var details bytes.Buffer
 	err = a.detailsTmpl.Execute(&details, dinfo)
 	if err != nil {
-		return "", "", err
+		return "", "", dinfo, err
 	}
-	return msg.String(), details.String(), nil
+	return msg.String(), details.String(), dinfo, nil
 }
 
 //--------------------------------
@@ -672,7 +694,15 @@ func (a *AlertNode) handleHipChat(hipchat *pipeline.HipChatHandler, ad *AlertDat
 	}
 }
 
-func (a *AlertNode) handleAlerta(alerta *pipeline.AlertaHandler, ad *AlertData) {
+type alertaHandler struct {
+	*pipeline.AlertaHandler
+
+	resourceTmpl *text.Template
+	valueTmpl    *text.Template
+	groupTmpl    *text.Template
+}
+
+func (a *AlertNode) handleAlerta(alerta alertaHandler, ad *AlertData) {
 	if a.et.tm.AlertaService == nil {
 		a.logger.Println("E! failed to send Alerta message. Alerta is not enabled")
 		return
@@ -698,18 +728,44 @@ func (a *AlertNode) handleAlerta(alerta *pipeline.AlertaHandler, ad *AlertData)
 		severity = "unknown"
 		status = "unknown"
 	}
+	var buf bytes.Buffer
+	err := alerta.resourceTmpl.Execute(&buf, ad.info)
+	if err != nil {
+		a.logger.Printf("E! failed to evaluate Alerta Resource template %s", alerta.Resource)
+		return
+	}
+	resource := buf.String()
+	buf.Reset()
+	err = alerta.groupTmpl.Execute(&buf, ad.info)
+	if err != nil {
+		a.logger.Printf("E! failed to evaluate Alerta Group template %s", alerta.Group)
+		return
+	}
+	group := buf.String()
+	buf.Reset()
+	err = alerta.valueTmpl.Execute(&buf, ad.info)
+	if err != nil {
+		a.logger.Printf("E! failed to evaluate Alerta Value template %s", alerta.Value)
+		return
+	}
+	value := buf.String()
+	service := alerta.Service
+	if len(alerta.Service) == 0 {
+		service = []string{ad.info.Name}
+	}
 
-	err := a.et.tm.AlertaService.Alert(
+	err = a.et.tm.AlertaService.Alert(
 		alerta.Token,
-		alerta.Resource,
-		alerta.Event,
+		resource,
+		ad.ID,
 		alerta.Environment,
 		severity,
 		status,
-		alerta.Group,
-		alerta.Value,
+		group,
+		value,
 		ad.Message,
 		alerta.Origin,
+		service,
 		ad.Data,
 	)
 	if err != nil {

integrations/streamer_test.go

@@ -1581,11 +1581,14 @@ func TestStream_AlertAlerta(t *testing.T) {
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		requestCount++
 		type postData struct {
-			Resource    string `json:"resource"`
-			Event       string `json:"event"`
-			Environment string `json:"environment"`
-			Text        string `json:"text"`
-			Origin      string `json:"origin"`
+			Resource    string   `json:"resource"`
+			Event       string   `json:"event"`
+			Group       string   `json:"group"`
+			Environment string   `json:"environment"`
+			Text        string   `json:"text"`
+			Origin      string   `json:"origin"`
+			Service     []string `json:"service"`
+			Value       string   `json:"value"`
 		}
 		pd := postData{}
 		dec := json.NewDecoder(r.Body)
@@ -1595,27 +1598,48 @@ func TestStream_AlertAlerta(t *testing.T) {
 			if exp := "/alert?api-key=testtoken1234567"; r.URL.String() != exp {
 				t.Errorf("unexpected url got %s exp %s", r.URL.String(), exp)
 			}
+			if exp := "cpu"; pd.Resource != exp {
+				t.Errorf("unexpected resource got %s exp %s", pd.Resource, exp)
+			}
 			if exp := "production"; pd.Environment != exp {
 				t.Errorf("unexpected environment got %s exp %s", pd.Environment, exp)
 			}
+			if exp := "host=serverA,"; pd.Group != exp {
+				t.Errorf("unexpected group got %s exp %s", pd.Group, exp)
+			}
+			if exp := ""; pd.Value != exp {
+				t.Errorf("unexpected value got %s exp %s", pd.Value, exp)
+			}
+			if exp := []string{"cpu"}; !reflect.DeepEqual(pd.Service, exp) {
+				t.Errorf("unexpected service got %s exp %s", pd.Service, exp)
+			}
 			if exp := "Kapacitor"; pd.Origin != exp {
 				t.Errorf("unexpected origin got %s exp %s", pd.Origin, exp)
 			}
 		} else {
 			if exp := "/alert?api-key=anothertesttoken"; r.URL.String() != exp {
 				t.Errorf("unexpected url got %s exp %s", r.URL.String(), exp)
 			}
+			if exp := "resource: cpu"; pd.Resource != exp {
+				t.Errorf("unexpected resource got %s exp %s", pd.Resource, exp)
+			}
 			if exp := "development"; pd.Environment != exp {
 				t.Errorf("unexpected environment got %s exp %s", pd.Environment, exp)
 			}
+			if exp := "serverA"; pd.Group != exp {
+				t.Errorf("unexpected group got %s exp %s", pd.Group, exp)
+			}
+			if exp := "10"; pd.Value != exp {
+				t.Errorf("unexpected value got %s exp %s", pd.Value, exp)
+			}
+			if exp := []string{"serviceA", "serviceB"}; !reflect.DeepEqual(pd.Service, exp) {
+				t.Errorf("unexpected service got %s exp %s", pd.Service, exp)
+			}
 			if exp := "override"; pd.Origin != exp {
 				t.Errorf("unexpected origin got %s exp %s", pd.Origin, exp)
 			}
 		}
-		if exp := "serverA"; pd.Resource != exp {
-			t.Errorf("unexpected resource got %s exp %s", pd.Resource, exp)
-		}
-		if exp := "CPU Idle"; pd.Event != exp {
+		if exp := "serverA"; pd.Event != exp {
 			t.Errorf("unexpected event got %s exp %s", pd.Event, exp)
 		}
 		if exp := "kapacitor/cpu/serverA is CRITICAL"; pd.Text != exp {
@@ -1634,22 +1658,22 @@ stream
 		.every(10s)
 	.mapReduce(influxql.count('value'))
 	.alert()
-                .id('{{ index .Tags "host" }}')
+		.id('{{ index .Tags "host" }}')
 		.message('kapacitor/{{ .Name }}/{{ index .Tags "host" }} is {{ .Level }}')
 		.info(lambda: "count" > 6.0)
 		.warn(lambda: "count" > 7.0)
 		.crit(lambda: "count" > 8.0)
 		.alerta()
 			.token('testtoken1234567')
-			.resource('serverA')
-                        .event('CPU Idle')
-                        .environment('production')
+			.environment('production')
 		.alerta()
 			.token('anothertesttoken')
-			.resource('serverA')
-                        .event('CPU Idle')
-                        .environment('development')
-                        .origin('override')
+			.resource('resource: {{ .Name }}')
+			.environment('development')
+			.origin('override')
+			.group('{{ .ID }}')
+			.value('{{ index .Fields "count" | printf "%0.0f" }}')
+			.services('serviceA', 'serviceB')
 `
 
 	clock, et, replayErr, tm := testStreamer(t, "TestStream_Alert", script, nil)

pipeline/alert.go

@@ -664,11 +664,16 @@ type HipChatHandler struct {
 func (a *AlertNode) Alerta() *AlertaHandler {
 	alerta := &AlertaHandler{
 		AlertNode: a,
+		Resource:  defaultAlertaResource,
+		Group:     defaultAlertaGroup,
 	}
 	a.AlertaHandlers = append(a.AlertaHandlers, alerta)
 	return alerta
 }
 
+const defaultAlertaResource = "{{ .Name }}"
+const defaultAlertaGroup = "{{ .Group }}"
+
 // tick:embedded:AlertNode.Alerta
 type AlertaHandler struct {
 	*AlertNode
@@ -678,26 +683,38 @@ type AlertaHandler struct {
 	Token string
 
 	// Alerta resource.
-	// This is a required field.
+	// Can be a template and has access to the same data as the AlertNode.Details property.
+	// Default: {{ .Name }}
 	Resource string
 
-	// Alerta event.
-	// This is a required field.
-	Event string
-
 	// Alerta environment.
 	// If empty uses the environment from the configuration.
 	Environment string
 
 	// Alerta group.
+	// Can be a template and has access to the same data as the AlertNode.Details property.
+	// Default: {{ .Group }}
 	Group string
 
 	// Alerta value.
+	// Can be a template and has access to the same data as the AlertNode.Details property.
+	// Default is an empty string.
 	Value string
 
 	// Alerta origin.
 	// If empty uses the origin from the configuration.
 	Origin string
+
+	// List of effected Services
+	// tick:ignore
+	Service []string
+}
+
+// List of effected services.
+// If not specified defaults to the Name of the stream.
+func (a *AlertaHandler) Services(service ...string) *AlertaHandler {
+	a.Service = service
+	return a
 }
 
 // Send the alert to Sensu.

services/alerta/service.go

@@ -36,7 +36,7 @@ func (s *Service) Close() error {
 	return nil
 }
 
-func (s *Service) Alert(token, resource, event, environment, severity, status, group, value, message, origin string, data interface{}) error {
+func (s *Service) Alert(token, resource, event, environment, severity, status, group, value, message, origin string, service []string, data interface{}) error {
 	if resource == "" || event == "" {
 		return errors.New("Resource and Event are required to send an alert")
 	}
@@ -70,6 +70,9 @@ func (s *Service) Alert(token, resource, event, environment, severity, status, g
 	postData["text"] = message
 	postData["origin"] = origin
 	postData["data"] = data
+	if len(service) > 0 {
+		postData["service"] = service
+	}
 
 	var post bytes.Buffer
 	enc := json.NewEncoder(&post)
@@ -83,19 +86,19 @@ func (s *Service) Alert(token, resource, event, environment, severity, status, g
 		return err
 	}
 	defer resp.Body.Close()
-	if resp.StatusCode != http.StatusOK {
+	if resp.StatusCode != http.StatusCreated {
 		body, err := ioutil.ReadAll(resp.Body)
 		if err != nil {
 			return err
 		}
 		type response struct {
-			Error string `json:"error"`
+			Message string `json:"message"`
 		}
-		r := &response{Error: "failed to understand Alerta response: " + string(body)}
+		r := &response{Message: "failed to understand Alerta response: " + string(body)}
 		b := bytes.NewReader(body)
 		dec := json.NewDecoder(b)
 		dec.Decode(r)
-		return errors.New(r.Error)
+		return errors.New(r.Message)
 	}
 	return nil
 }

task_master.go

@@ -70,7 +70,18 @@ type TaskMaster struct {
 		Alert(room, token, message string, level AlertLevel) error
 	}
 	AlertaService interface {
-		Alert(token, resource, event, environment, severity, status, group, value, message, origin string, data interface{}) error
+		Alert(token,
+			resource,
+			event,
+			environment,
+			severity,
+			status,
+			group,
+			value,
+			message,
+			origin string,
+			service []string,
+			data interface{}) error
 	}
 	SensuService interface {
 		Alert(name, output string, level AlertLevel) error