Add support for setting the alert message as a field (influxdata#878)

* add message field for alerts

* CHANGELOG.md

* add batch tests

CHANGELOG.md

@@ -7,6 +7,7 @@
 ### Features
 
 - [#873](https://github.com/influxdata/kapacitor/pull/873): Add TCP alert handler
+- [#869](https://github.com/influxdata/kapacitor/issues/869): Add ability to set alert message as a field
 
 ### Bugfixes
 

alert.go

@@ -450,11 +450,14 @@ func (a *AlertNode) runAlert([]byte) error {
 						p.Tags[a.a.IdTag] = ad.ID
 					}
 				}
-				if a.a.LevelField != "" || a.a.IdField != "" || a.a.DurationField != "" {
+				if a.a.LevelField != "" || a.a.IdField != "" || a.a.DurationField != "" || a.a.MessageField != "" {
 					p.Fields = p.Fields.Copy()
 					if a.a.LevelField != "" {
 						p.Fields[a.a.LevelField] = l.String()
 					}
+					if a.a.MessageField != "" {
+						p.Fields[a.a.MessageField] = ad.Message
+					}
 					if a.a.IdField != "" {
 						p.Fields[a.a.IdField] = ad.ID
 					}
@@ -541,7 +544,8 @@ func (a *AlertNode) runAlert([]byte) error {
 					a.a.LevelField != "" ||
 					a.a.IdTag != "" ||
 					a.a.IdField != "" ||
-					a.a.DurationField != "" {
+					a.a.DurationField != "" ||
+					a.a.MessageField != "" {
 					for i := range b.Points {
 						if a.a.LevelTag != "" || a.a.IdTag != "" {
 							b.Points[i].Tags = b.Points[i].Tags.Copy()
@@ -552,11 +556,14 @@ func (a *AlertNode) runAlert([]byte) error {
 								b.Points[i].Tags[a.a.IdTag] = ad.ID
 							}
 						}
-						if a.a.LevelField != "" || a.a.IdField != "" || a.a.DurationField != "" {
+						if a.a.LevelField != "" || a.a.IdField != "" || a.a.DurationField != "" || a.a.MessageField != "" {
 							b.Points[i].Fields = b.Points[i].Fields.Copy()
 							if a.a.LevelField != "" {
 								b.Points[i].Fields[a.a.LevelField] = l.String()
 							}
+							if a.a.MessageField != "" {
+								b.Points[i].Fields[a.a.MessageField] = ad.Message
+							}
 							if a.a.IdField != "" {
 								b.Points[i].Fields[a.a.IdField] = ad.ID
 							}

integrations/batcher_test.go

@@ -1243,6 +1243,64 @@ batch
 	testBatcherWithOutput(t, "TestBatch_SimpleMR", script, 30*time.Second, er, false)
 }
 
+func TestBatch_AlertMessage(t *testing.T) {
+
+	var script = `
+batch
+	|query('''
+		SELECT mean("value")
+		FROM "telegraf"."default".cpu_usage_idle
+		WHERE "host" = 'serverA' AND "cpu" != 'cpu-total'
+''')
+		.period(10s)
+		.every(10s)
+		.groupBy(time(2s), 'cpu')
+	|alert()
+		.crit(lambda:"mean" > 95)
+		.messageField('msg')
+	|httpOut('TestBatch_SimpleMR')
+`
+
+	er := kapacitor.Result{
+		Series: imodels.Rows{
+			{
+				Name:    "cpu_usage_idle",
+				Tags:    map[string]string{"cpu": "cpu1"},
+				Columns: []string{"time", "mean", "msg"},
+				Values: [][]interface{}{
+					{
+						time.Date(1971, 1, 1, 0, 0, 20, 0, time.UTC),
+						96.49999999996908,
+						"cpu_usage_idle:cpu=cpu1 is CRITICAL",
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 22, 0, time.UTC),
+						93.46464646468584,
+						"cpu_usage_idle:cpu=cpu1 is CRITICAL",
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 24, 0, time.UTC),
+						95.00950095007724,
+						"cpu_usage_idle:cpu=cpu1 is CRITICAL",
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 26, 0, time.UTC),
+						92.99999999998636,
+						"cpu_usage_idle:cpu=cpu1 is CRITICAL",
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 28, 0, time.UTC),
+						90.99999999998545,
+						"cpu_usage_idle:cpu=cpu1 is CRITICAL",
+					},
+				},
+			},
+		},
+	}
+
+	testBatcherWithOutput(t, "TestBatch_SimpleMR", script, 30*time.Second, er, false)
+}
+
 func TestBatch_AlertStateChangesOnly(t *testing.T) {
 	requestCount := int32(0)
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

integrations/streamer_test.go

@@ -4536,6 +4536,7 @@ stream
 		.idField('id')
 		.idTag('id')
 		.levelField('level')
+		.messageField('msg')
 		.levelTag('level')
 		.info(lambda: "count" > infoThreshold)
 		.warn(lambda: "count" > warnThreshold)
@@ -4549,12 +4550,13 @@ stream
 			{
 				Name:    "cpu",
 				Tags:    map[string]string{"host": "serverA", "level": "CRITICAL", "id": "kapacitor/cpu/serverA"},
-				Columns: []string{"time", "count", "id", "level"},
+				Columns: []string{"time", "count", "id", "level", "msg"},
 				Values: [][]interface{}{[]interface{}{
 					time.Date(1971, 1, 1, 0, 0, 10, 0, time.UTC),
 					10.0,
 					"kapacitor/cpu/serverA",
 					"CRITICAL",
+					"kapacitor/cpu/serverA is CRITICAL",
 				}},
 			},
 		},

pipeline/alert.go

@@ -256,6 +256,9 @@ type AlertNode struct {
 	// Optional field key to add to the data, containing the alert level as a string.
 	LevelField string
 
+	// Optional field key to add to the data, containing the alert message.
+	MessageField string
+
 	// Optional field key to add the alert duration to the data.
 	// The duration is always in units of nanoseconds.
 	DurationField string