fix missing tags on batch influxql (influxdata#657)

add tests for batch InfluxQL and HoltWinters

CHANGELOG.md

@@ -1,6 +1,6 @@
 # Changelog
 
-## v1.0.0-beta2 [unreleased]
+## v1.0.0-beta3 [unreleased]
 
 ### Release Notes
 
@@ -9,6 +9,7 @@
 ### Bugfixes
 
 - [#656](https://github.com/influxdata/kapacitor/pull/656): Fix issues where an expression could not be passed as a function parameter in TICKscript.
+- [#627](https://github.com/influxdata/kapacitor/issue/327): Fix where InfluxQL functions that returned a batch could drop tags.
 
 ## v1.0.0-beta2 [2016-06-17]
 

influxql.gen.go

@@ -224,9 +224,22 @@ func (e *floatPointEmitter) EmitBatch() models.Batch {
 		} else {
 			t = e.time
 		}
+		var tags models.Tags
+		if l := len(ap.Tags.KeyValues()); l > 0 {
+			// Merge batch and point specific tags
+			tags = make(models.Tags, len(e.tags)+l)
+			for k, v := range e.tags {
+				tags[k] = v
+			}
+			for k, v := range ap.Tags.KeyValues() {
+				tags[k] = v
+			}
+		} else {
+			tags = e.tags
+		}
 		b.Points[i] = models.BatchPoint{
 			Time:   t,
-			Tags:   ap.Tags.KeyValues(),
+			Tags:   tags,
 			Fields: map[string]interface{}{e.as: ap.Value},
 		}
 	}
@@ -442,9 +455,22 @@ func (e *integerPointEmitter) EmitBatch() models.Batch {
 		} else {
 			t = e.time
 		}
+		var tags models.Tags
+		if l := len(ap.Tags.KeyValues()); l > 0 {
+			// Merge batch and point specific tags
+			tags = make(models.Tags, len(e.tags)+l)
+			for k, v := range e.tags {
+				tags[k] = v
+			}
+			for k, v := range ap.Tags.KeyValues() {
+				tags[k] = v
+			}
+		} else {
+			tags = e.tags
+		}
 		b.Points[i] = models.BatchPoint{
 			Time:   t,
-			Tags:   ap.Tags.KeyValues(),
+			Tags:   tags,
 			Fields: map[string]interface{}{e.as: ap.Value},
 		}
 	}
@@ -660,9 +686,22 @@ func (e *stringPointEmitter) EmitBatch() models.Batch {
 		} else {
 			t = e.time
 		}
+		var tags models.Tags
+		if l := len(ap.Tags.KeyValues()); l > 0 {
+			// Merge batch and point specific tags
+			tags = make(models.Tags, len(e.tags)+l)
+			for k, v := range e.tags {
+				tags[k] = v
+			}
+			for k, v := range ap.Tags.KeyValues() {
+				tags[k] = v
+			}
+		} else {
+			tags = e.tags
+		}
 		b.Points[i] = models.BatchPoint{
 			Time:   t,
-			Tags:   ap.Tags.KeyValues(),
+			Tags:   tags,
 			Fields: map[string]interface{}{e.as: ap.Value},
 		}
 	}
@@ -878,9 +917,22 @@ func (e *booleanPointEmitter) EmitBatch() models.Batch {
 		} else {
 			t = e.time
 		}
+		var tags models.Tags
+		if l := len(ap.Tags.KeyValues()); l > 0 {
+			// Merge batch and point specific tags
+			tags = make(models.Tags, len(e.tags)+l)
+			for k, v := range e.tags {
+				tags[k] = v
+			}
+			for k, v := range ap.Tags.KeyValues() {
+				tags[k] = v
+			}
+		} else {
+			tags = e.tags
+		}
 		b.Points[i] = models.BatchPoint{
 			Time:   t,
-			Tags:   ap.Tags.KeyValues(),
+			Tags:   tags,
 			Fields: map[string]interface{}{e.as: ap.Value},
 		}
 	}

influxql.gen.go.tmpl

@@ -224,9 +224,22 @@ func (e *{{.name}}PointEmitter) EmitBatch() models.Batch {
 		} else {
 			t = e.time
 		}
+		var tags models.Tags
+		if l := len(ap.Tags.KeyValues()); l > 0 {
+			// Merge batch and point specific tags
+			tags = make(models.Tags, len(e.tags)+l)
+			for k, v := range e.tags {
+				tags[k] = v
+			}
+			for k, v := range ap.Tags.KeyValues() {
+				tags[k] = v
+			}
+		} else {
+			tags = e.tags
+		}
 		b.Points[i] = models.BatchPoint{
 			Time:   t,
-			Tags:   ap.Tags.KeyValues(),
+			Tags:   tags,
 			Fields: map[string]interface{}{e.as: ap.Value},
 		}
 	}

integrations/data/TestStream_HoltWinters.srpl

@@ -0,0 +1,36 @@
+dbname
+rpname
+packets,host=serverA value=1000 0000000001
+dbname
+rpname
+packets,host=serverA value=1001 0000000002
+dbname
+rpname
+packets,host=serverA value=1002 0000000003
+dbname
+rpname
+packets,host=serverA value=1003 0000000004
+dbname
+rpname
+packets,host=serverA value=1004 0000000005
+dbname
+rpname
+packets,host=serverA value=1006 0000000006
+dbname
+rpname
+packets,host=serverA value=1007 0000000007
+dbname
+rpname
+packets,host=serverA value=1007 0000000008
+dbname
+rpname
+packets,host=serverA value=1008 0000000009
+dbname
+rpname
+packets,host=serverA value=1009 0000000010
+dbname
+rpname
+packets,host=serverA value=1010 0000000011
+dbname
+rpname
+packets,host=serverA value=1011 0000000012