Merge pull request #154 from influxdata/nc-issue#153

Fix panic with wrong field name for top/bottom MR functions.
influxdata · Jan 20, 2016 · 33b10fe · 33b10fe
2 parents 88cd92e + 0378df5
commit 33b10fe
Show file tree

Hide file tree

Showing 5 changed files with 19 additions and 10 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -22,6 +22,7 @@ This way you can easily tell which versions of Telegraf, InfluxDB, Chronograf an
 - [#139](https://github.com/influxdata/kapacitor/issues/139): Alerta.io support thanks! @md14454
 
 ### Bugfixes
+- [#153](https://github.com/influxdata/kapacitor/issues/153): Fix panic if referencing non existant field in MapReduce function.
 
 ## v0.2.4 [2016-01-07]
 

diff --git a/etc/kapacitor/kapacitor.conf b/etc/kapacitor/kapacitor.conf
@@ -48,7 +48,7 @@ data_dir = "/var/lib/kapacitor"
   # Id -- the alert Id, NODE_NAME will be replaced with the name of the node being monitored.
   id = "node 'NODE_NAME' in task '{{ .TaskName }}'"
   # The message of the alert. INTERVAL will be replaced by the interval.
-  message = "{{ .ID }} is dead: {{ index .Fields \"collected\" }} points/INTERVAL"
+  message = "{{ .ID }} is {{ if eq .Level \"OK\" }}alive{{ else }}dead{{ end }}: {{ index .Fields \"collected\" | printf \"%0.3f\" }} points/INTERVAL."
 
 
 [influxdb]

diff --git a/integrations/batcher_test.go b/integrations/batcher_test.go
@@ -209,7 +209,7 @@ batch
 		.period(10s)
 		.every(10s)
 		.groupBy(time(2s), 'cpu')
-	.mapReduce(influxql.count('value'))
+	.mapReduce(influxql.count('mean'))
 	.window()
 		.period(20s)
 		.every(20s)

diff --git a/integrations/streamer_test.go b/integrations/streamer_test.go
@@ -1294,7 +1294,7 @@ stream
 	.window()
 		.period(10s)
 		.every(10s)
-	.mapReduce(influxql.count('idle'))
+	.mapReduce(influxql.count('value'))
 	.alert()
 		.id('kapacitor/{{ .Name }}/{{ index .Tags "host" }}')
 		.info(lambda: "count" > infoThreshold)
@@ -1370,7 +1370,7 @@ stream
 	.window()
 		.period(10s)
 		.every(10s)
-	.mapReduce(influxql.count('idle'))
+	.mapReduce(influxql.count('value'))
 	.alert()
 		.id('kapacitor/{{ .Name }}/{{ index .Tags "host" }}')
 		.info(lambda: "count" > 6.0)
@@ -1447,7 +1447,7 @@ stream
 	.window()
 		.period(10s)
 		.every(10s)
-	.mapReduce(influxql.count('idle'))
+	.mapReduce(influxql.count('value'))
 	.alert()
 		.id('kapacitor/{{ .Name }}/{{ index .Tags "host" }}')
 		.info(lambda: "count" > 6.0)
@@ -1537,7 +1537,7 @@ stream
 	.window()
 		.period(10s)
 		.every(10s)
-	.mapReduce(influxql.count('idle'))
+	.mapReduce(influxql.count('value'))
 	.alert()
                 .id('{{ index .Tags "host" }}')
 		.message('kapacitor/{{ .Name }}/{{ index .Tags "host" }} is {{ .Level }}')
@@ -1662,7 +1662,7 @@ stream
 	.window()
 		.period(10s)
 		.every(10s)
-	.mapReduce(influxql.count('idle'))
+	.mapReduce(influxql.count('value'))
 	.alert()
 		.id('kapacitor/{{ .Name }}/{{ index .Tags "host" }}')
 		.info(lambda: "count" > 6.0)
@@ -1738,7 +1738,7 @@ stream
 	.window()
 		.period(10s)
 		.every(10s)
-	.mapReduce(influxql.count('idle'))
+	.mapReduce(influxql.count('value'))
 	.alert()
 		.id('kapacitor/{{ .Name }}/{{ index .Tags "host" }}')
 		.message('{{ .Level }} alert for {{ .ID }}')
@@ -1821,7 +1821,7 @@ stream
 	.window()
 		.period(10s)
 		.every(10s)
-	.mapReduce(influxql.count('idle'))
+	.mapReduce(influxql.count('value'))
 	.alert()
 		.id('kapacitor/{{ .Name }}/{{ index .Tags "host" }}')
 		.info(lambda: "count" > 6.0)

diff --git a/map_reduce.go b/map_reduce.go
@@ -116,9 +116,17 @@ func (m *MapNode) mapBatch(b models.Batch) error {
 	inputs := make([]tsdb.MapInput, m.parallel)
 	j := 0
 	for _, p := range b.Points {
+		value, ok := p.Fields[m.field]
+		if !ok {
+			fields := make([]string, 0, len(p.Fields))
+			for field := range p.Fields {
+				fields = append(fields, field)
+			}
+			return fmt.Errorf("unknown field %s, available fields %v", m.field, fields)
+		}
 		item := tsdb.MapItem{
 			Timestamp: p.Time.Unix(),
-			Value:     p.Fields[m.field],
+			Value:     value,
 			Fields:    p.Fields,
 			Tags:      p.Tags,
 		}