Add Sensu metadata support

CHANGELOG.md

@@ -9,6 +9,7 @@
 - [#2099](https://github.com/influxdata/kapacitor/issues/2099): Add `alert/persist-topics` to config
 - [#2101](https://github.com/influxdata/kapacitor/issues/2101): Add multiple field support to the change detect node.
 - [#1961](https://github.com/influxdata/kapacitor/pull/1961): Add links to pagerduty2 alerts
+- [#1974](https://github.com/influxdata/kapacitor/issues/1974): Add additional metadata to Sensu alerts.
 
 ### Bugfixes
 

alert.go

@@ -239,6 +239,7 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, d NodeDiagnostic) (a
 		c := sensu.HandlerConfig{
 			Source:   s.Source,
 			Handlers: s.HandlersList,
+			Metadata: s.MetadataMap,
 		}
 		h, err := et.tm.SensuService.Handler(c, ctx...)
 		if err != nil {

integrations/streamer_test.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"html"
 	"io/ioutil"
+	"math/rand"
 	"net/http"
 	"net/http/httptest"
 	"net/mail"
@@ -20,8 +21,6 @@ import (
 	"text/template"
 	"time"
 
-	"math/rand"
-
 	"github.com/davecgh/go-spew/spew"
 	"github.com/docker/docker/api/types/swarm"
 	"github.com/google/go-cmp/cmp"
@@ -8466,6 +8465,8 @@ stream
 		.warn(lambda: "count" > 7.0)
 		.crit(lambda: "count" > 8.0)
 		.sensu()
+			.metadata('k1', 'v1')
+			.metadata('k2', 5)
 `
 	tmInit := func(tm *kapacitor.TaskMaster) {
 		c := sensu.NewConfig()
@@ -8483,6 +8484,10 @@ stream
 			Output: "kapacitor.cpu.serverA is CRITICAL",
 			Name:   "kapacitor.cpu.serverA",
 			Status: 2,
+			Metadata: map[string]interface{}{
+				"k1": "v1",
+				"k2": float64(5),
+			},
 		},
 	}
 

pipeline/alert.go

@@ -1316,6 +1316,10 @@ type SensuHandler struct {
 	// If empty uses the handler list from the configuration
 	// tick:ignore
 	HandlersList []string `tick:"Handlers" json:"handlers"`
+
+	// MetadataMap is a map of arbitrary metadata
+	// tick:ignore
+	MetadataMap map[string]interface{} `tick:"Metadata" json:"metadata"`
 }
 
 // List of effected services.
@@ -1326,6 +1330,18 @@ func (s *SensuHandler) Handlers(handlers ...string) *SensuHandler {
 	return s
 }
 
+// Metadata adds key values pairs to the sensu request.
+// The keys are added at the root level on the sensu JSON object.
+// Metadata for standard Sensu keys will be ignored.
+// tick:property
+func (s *SensuHandler) Metadata(key string, value interface{}) *SensuHandler {
+	if s.MetadataMap == nil {
+		s.MetadataMap = make(map[string]interface{})
+	}
+	s.MetadataMap[key] = value
+	return s
+}
+
 // Send the alert to Pushover.
 // Register your application with Pushover at
 // https://pushover.net/apps/build to get a

pipeline/tick/alert.go

@@ -144,6 +144,16 @@ func (n *AlertNode) Build(a *pipeline.AlertNode) (ast.Node, error) {
 		n.Dot("sensu").
 			Dot("source", h.Source).
 			Dot("handlers", args(h.HandlersList)...)
+
+		// Use stable key order
+		keys := make([]string, 0, len(h.MetadataMap))
+		for k := range h.MetadataMap {
+			keys = append(keys, k)
+		}
+		sort.Strings(keys)
+		for _, k := range keys {
+			n.Dot("metadata", k, h.MetadataMap[k])
+		}
 	}
 
 	for _, h := range a.SlackHandlers {

pipeline/tick/alert_test.go

@@ -423,6 +423,11 @@ func TestAlertSensu(t *testing.T) {
 	handler := from.Alert().Sensu()
 	handler.Source = "Henry Hill"
 	handler.Handlers("FBI", "Witness", "Protection")
+	handler.MetadataMap = map[string]interface{}{
+		"k1": "v1",
+		"k2": int64(5),
+		"k3": float64(5),
+	}
 
 	want := `stream
     |from()
@@ -434,6 +439,9 @@ func TestAlertSensu(t *testing.T) {
         .sensu()
         .source('Henry Hill')
         .handlers('FBI', 'Witness', 'Protection')
+        .metadata('k1', 'v1')
+        .metadata('k2', 5)
+        .metadata('k3', 5.0)
 `
 	PipelineTickTestHelper(t, pipe, want)
 }

server/server_test.go

@@ -8836,6 +8836,7 @@ func TestServer_ListServiceTests(t *testing.T) {
 					"output":   "testOutput",
 					"source":   "Kapacitor",
 					"handlers": []interface{}{},
+					"metadata": map[string]interface{}{},
 					"level":    "CRITICAL",
 				},
 			},
@@ -8975,6 +8976,7 @@ func TestServer_ListServiceTests_WithPattern(t *testing.T) {
 					"output":   "testOutput",
 					"source":   "Kapacitor",
 					"handlers": []interface{}{},
+					"metadata": map[string]interface{}{},
 					"level":    "CRITICAL",
 				},
 			},
@@ -10024,6 +10026,10 @@ func TestServer_AlertHandlers(t *testing.T) {
 				Kind: "sensu",
 				Options: map[string]interface{}{
 					"source": "Kapacitor",
+					"metadata": map[string]interface{}{
+						"k1": "v1",
+						"k2": 5,
+					},
 				},
 			},
 			setup: func(c *server.Config, ha *client.TopicHandler) (context.Context, error) {
@@ -10046,6 +10052,10 @@ func TestServer_AlertHandlers(t *testing.T) {
 					Output: "message",
 					Name:   "id",
 					Status: 2,
+					Metadata: map[string]interface{}{
+						"k1": "v1",
+						"k2": float64(5),
+					},
 				}}
 				got := ts.Requests()
 				if !reflect.DeepEqual(exp, got) {

services/sensu/sensutest/sensutest.go

@@ -64,8 +64,37 @@ func (s *Server) run() {
 }
 
 type Request struct {
-	Name   string `json:"name"`
-	Source string `json:"source"`
-	Output string `json:"output"`
-	Status int    `json:"status"`
+	Name     string                 `json:"name"`
+	Source   string                 `json:"source"`
+	Output   string                 `json:"output"`
+	Status   int                    `json:"status"`
+	Handlers []string               `json:"handlers"`
+	Metadata map[string]interface{} `json:"-"`
+}
+
+func (r *Request) UnmarshalJSON(data []byte) error {
+	type Alias Request
+	raw := struct {
+		*Alias
+	}{}
+	err := json.Unmarshal(data, &raw)
+	if err != nil {
+		return err
+	}
+	if raw.Alias != nil {
+		*r = *(*Request)(raw.Alias)
+	}
+	if r.Metadata == nil {
+		r.Metadata = make(map[string]interface{})
+	}
+	json.Unmarshal(data, &r.Metadata)
+	delete(r.Metadata, "name")
+	delete(r.Metadata, "source")
+	delete(r.Metadata, "output")
+	delete(r.Metadata, "status")
+	delete(r.Metadata, "handlers")
+	if len(r.Metadata) == 0 {
+		r.Metadata = nil
+	}
+	return nil
 }

services/sensu/service.go

@@ -60,11 +60,12 @@ func (s *Service) Update(newConfig []interface{}) error {
 }
 
 type testOptions struct {
-	Name     string      `json:"name"`
-	Source   string      `json:"source"`
-	Output   string      `json:"output"`
-	Handlers []string    `json:"handlers"`
-	Level    alert.Level `json:"level"`
+	Name     string                 `json:"name"`
+	Source   string                 `json:"source"`
+	Output   string                 `json:"output"`
+	Handlers []string               `json:"handlers"`
+	Metadata map[string]interface{} `json:"metadata"`
+	Level    alert.Level            `json:"level"`
 }
 
 func (s *Service) TestOptions() interface{} {
@@ -73,6 +74,7 @@ func (s *Service) TestOptions() interface{} {
 		Source:   "Kapacitor",
 		Output:   "testOutput",
 		Handlers: []string{},
+		Metadata: map[string]interface{}{},
 		Level:    alert.Critical,
 	}
 }
@@ -87,16 +89,17 @@ func (s *Service) Test(options interface{}) error {
 		o.Source,
 		o.Output,
 		o.Handlers,
+		o.Metadata,
 		o.Level,
 	)
 }
 
-func (s *Service) Alert(name, source, output string, handlers []string, level alert.Level) error {
+func (s *Service) Alert(name, source, output string, handlers []string, metadata map[string]interface{}, level alert.Level) error {
 	if !validNamePattern.MatchString(name) {
 		return fmt.Errorf("invalid name %q for sensu alert. Must match %v", name, validNamePattern)
 	}
 
-	addr, postData, err := s.prepareData(name, source, output, handlers, level)
+	addr, postData, err := s.prepareData(name, source, output, handlers, metadata, level)
 	if err != nil {
 		return err
 	}
@@ -122,7 +125,7 @@ func (s *Service) Alert(name, source, output string, handlers []string, level al
 	return nil
 }
 
-func (s *Service) prepareData(name, source, output string, handlers []string, level alert.Level) (*net.TCPAddr, map[string]interface{}, error) {
+func (s *Service) prepareData(name, source, output string, handlers []string, metadata map[string]interface{}, level alert.Level) (*net.TCPAddr, map[string]interface{}, error) {
 
 	c := s.config()
 
@@ -162,6 +165,12 @@ func (s *Service) prepareData(name, source, output string, handlers []string, le
 		return nil, nil, err
 	}
 
+	for k, v := range metadata {
+		if _, ok := postData[k]; !ok {
+			postData[k] = v
+		}
+	}
+
 	return addr, postData, nil
 }
 
@@ -173,6 +182,9 @@ type HandlerConfig struct {
 	// Sensu handler list
 	// If empty uses the handler list from the configuration
 	Handlers []string `mapstructure:"handlers"`
+
+	// Metadata is a map of key value data to include on the sensu API request.
+	Metadata map[string]interface{} `mapstructure:"metadata"`
 }
 
 type handler struct {
@@ -211,6 +223,7 @@ func (h *handler) Handle(event alert.Event) {
 		sourceStr,
 		event.State.Message,
 		h.c.Handlers,
+		h.c.Metadata,
 		event.State.Level,
 	); err != nil {
 		h.diag.Error("failed to send event to Sensu", err)