Update the InfluxDB dep to 1.1

CHANGELOG.md

@@ -39,6 +39,7 @@ See the API docs for more details.
 - [#980](https://github.com/influxdata/kapacitor/pull/980): Upgrade to using go 1.7
 - [#957](https://github.com/influxdata/kapacitor/issues/957): Add API endpoints for testing service integrations.
 - [#958](https://github.com/influxdata/kapacitor/issues/958): Add support for Slack icon emojis and custom usernames.
+- [#991](https://github.com/influxdata/kapacitor/pull/991): Bring Kapacitor up to parity with available InfluxQL functions in 1.1
 
 ### Bugfixes
 

influxdb/client.go

@@ -510,7 +510,7 @@ type Point struct {
 
 // Returns byte array of a line protocol representation of the point
 func (p Point) Bytes(precision string) []byte {
-	key := imodels.MakeKey([]byte(p.Name), imodels.Tags(p.Tags))
+	key := imodels.MakeKey([]byte(p.Name), imodels.NewTags(p.Tags))
 	fields := imodels.Fields(p.Fields).MarshalBinary()
 	kl := len(key)
 	fl := len(fields)

integrations/batcher_test.go

@@ -374,6 +374,54 @@ batch
 
 	testBatcherWithOutput(t, "TestBatch_MovingAverage", script, 21*time.Second, er, false)
 }
+func TestBatch_CumulativeSum(t *testing.T) {
+
+	var script = `
+batch
+	|query('''
+		SELECT "value"
+		FROM "telegraf"."default".packets
+''')
+		.period(10s)
+		.every(10s)
+	|cumulativeSum('value')
+	|httpOut('TestBatch_CumulativeSum')
+`
+
+	er := kapacitor.Result{
+		Series: imodels.Rows{
+			{
+				Name:    "packets",
+				Tags:    nil,
+				Columns: []string{"time", "cumulativeSum"},
+				Values: [][]interface{}{
+					{
+						time.Date(1971, 1, 1, 0, 0, 10, 0, time.UTC),
+						0.0,
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 12, 0, time.UTC),
+						10.0,
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 14, 0, time.UTC),
+						30.0,
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 16, 0, time.UTC),
+						60.0,
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 18, 0, time.UTC),
+						100.0,
+					},
+				},
+			},
+		},
+	}
+
+	testBatcherWithOutput(t, "TestBatch_CumulativeSum", script, 31*time.Second, er, false)
+}
 
 func TestBatch_SimpleMR(t *testing.T) {
 

integrations/data/TestBatch_CumulativeSum.0.brpl

@@ -0,0 +1,50 @@
+{
+    "name":"packets",
+    "points":[
+        {
+            "fields":{"value":1000},
+            "time":"2015-10-18T00:00:00Z"
+        },
+        {
+            "fields":{"value":1005},
+            "time":"2015-10-18T00:00:02Z"
+        },
+        {
+            "fields":{"value":1008},
+            "time":"2015-10-18T00:00:04Z"
+        },
+        {
+            "fields":{"value":1009},
+            "time":"2015-10-18T00:00:06Z"
+        },
+        {
+            "fields":{"value":1004},
+            "time":"2015-10-18T00:00:08Z"
+        }
+    ]
+}
+{
+    "name":"packets",
+    "points":[
+        {
+            "fields":{"value":0},
+            "time":"2015-10-18T00:00:10Z"
+        },
+        {
+            "fields":{"value":10},
+            "time":"2015-10-18T00:00:12Z"
+        },
+        {
+            "fields":{"value":20},
+            "time":"2015-10-18T00:00:14Z"
+        },
+        {
+            "fields":{"value":30},
+            "time":"2015-10-18T00:00:16Z"
+        },
+        {
+            "fields":{"value":40},
+            "time":"2015-10-18T00:00:18Z"
+        }
+    ]
+}

integrations/data/TestStream_CumulativeSum.srpl

@@ -0,0 +1,33 @@
+dbname
+rpname
+packets value=00 0000000000
+dbname
+rpname
+packets value=00 0000000001
+dbname
+rpname
+packets value=01 0000000002
+dbname
+rpname
+packets value=02 0000000003
+dbname
+rpname
+packets value=03 0000000004
+dbname
+rpname
+packets value=04 0000000005
+dbname
+rpname
+packets value=05 0000000006
+dbname
+rpname
+packets value=06 0000000007
+dbname
+rpname
+packets value=07 0000000008
+dbname
+rpname
+packets value=08 0000000009
+dbname
+rpname
+packets value=09 0000000010

integrations/streamer_test.go

@@ -457,6 +457,72 @@ stream
 	testStreamerWithOutput(t, "TestStream_MovingAverage", script, 16*time.Second, er, false, nil)
 }
 
+func TestStream_CumulativeSum(t *testing.T) {
+	var script = `
+stream
+	|from()
+		.measurement('packets')
+	|cumulativeSum('value')
+	|window()
+		.period(10s)
+		.every(10s)
+	|httpOut('TestStream_CumulativeSum')
+`
+	er := kapacitor.Result{
+		Series: imodels.Rows{
+			{
+				Name:    "packets",
+				Tags:    nil,
+				Columns: []string{"time", "cumulativeSum"},
+				Values: [][]interface{}{
+					{
+						time.Date(1971, 1, 1, 0, 0, 0, 0, time.UTC),
+						0.0,
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 1, 0, time.UTC),
+						0.0,
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 2, 0, time.UTC),
+						1.0,
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 3, 0, time.UTC),
+						3.0,
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 4, 0, time.UTC),
+						6.0,
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 5, 0, time.UTC),
+						10.0,
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 6, 0, time.UTC),
+						15.0,
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 7, 0, time.UTC),
+						21.0,
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 8, 0, time.UTC),
+						28.0,
+					},
+					{
+						time.Date(1971, 1, 1, 0, 0, 9, 0, time.UTC),
+						36.0,
+					},
+				},
+			},
+		},
+	}
+
+	testStreamerWithOutput(t, "TestStream_CumulativeSum", script, 13*time.Second, er, false, nil)
+}
+
 func TestStream_WindowMissing(t *testing.T) {
 
 	var script = `
@@ -7084,7 +7150,7 @@ stream
 		if len(p.Tags()) != 1 {
 			t.Errorf("got %v exp %v", len(p.Tags()), 1)
 		}
-		if got, exp := p.Tags()["key"], "value"; got != exp {
+		if got, exp := p.Tags().GetString("key"), "value"; got != exp {
 			t.Errorf("got %s exp %s", got, exp)
 		}
 		tm := time.Date(1971, 1, 1, 0, 0, 10, 0, time.UTC)

metaclient.go

@@ -15,10 +15,10 @@ func (m *NoopMetaClient) WaitForLeader(d time.Duration) error {
 func (m *NoopMetaClient) CreateDatabase(name string) (*meta.DatabaseInfo, error) {
 	return nil, nil
 }
-func (m *NoopMetaClient) CreateDatabaseWithRetentionPolicy(name string, rpi *meta.RetentionPolicyInfo) (*meta.DatabaseInfo, error) {
+func (m *NoopMetaClient) CreateDatabaseWithRetentionPolicy(name string, rpi *meta.RetentionPolicySpec) (*meta.DatabaseInfo, error) {
 	return nil, nil
 }
-func (m *NoopMetaClient) CreateRetentionPolicy(database string, rpi *meta.RetentionPolicyInfo) (*meta.RetentionPolicyInfo, error) {
+func (m *NoopMetaClient) CreateRetentionPolicy(database string, rpi *meta.RetentionPolicySpec) (*meta.RetentionPolicyInfo, error) {
 	return nil, nil
 }
 func (m *NoopMetaClient) Database(name string) *meta.DatabaseInfo {

models/point.go

@@ -166,7 +166,7 @@ func ToGroupID(name string, tags map[string]string, dims Dimensions) GroupID {
 
 // Returns byte array of a line protocol representation of the point
 func (p Point) Bytes(precision string) []byte {
-	key := models.MakeKey([]byte(p.Name), models.Tags(p.Tags))
+	key := models.MakeKey([]byte(p.Name), models.NewTags(p.Tags))
 	fields := models.Fields(p.Fields).MarshalBinary()
 	kl := len(key)
 	fl := len(fields)

pipeline/influxql.go

@@ -480,3 +480,21 @@ func (n *chainnode) holtWinters(field string, h, m int64, interval time.Duration
 	n.linkChild(i)
 	return i
 }
+
+// Compute a cumulative sum of each point that is received.
+// A point is emitted for every point collected.
+func (n *chainnode) CumulativeSum(field string) *InfluxQLNode {
+	i := newInfluxQLNode("cumulativeSum", field, n.Provides(), n.Provides(), ReduceCreater{
+		CreateFloatReducer: func() (influxql.FloatPointAggregator, influxql.FloatPointEmitter) {
+			fn := influxql.NewFloatCumulativeSumReducer()
+			return fn, fn
+		},
+		CreateIntegerReducer: func() (influxql.IntegerPointAggregator, influxql.IntegerPointEmitter) {
+			fn := influxql.NewIntegerCumulativeSumReducer()
+			return fn, fn
+		},
+		IsStreamTransformation: true,
+	})
+	n.linkChild(i)
+	return i
+}

replay.go

@@ -98,7 +98,7 @@ func readPointsFromIO(data io.ReadCloser, points chan<- models.Point, precision
 			RetentionPolicy: rp,
 			Name:            mp.Name(),
 			Group:           models.NilGroup,
-			Tags:            models.Tags(mp.Tags()),
+			Tags:            models.Tags(mp.Tags().Map()),
 			Fields:          models.Fields(mp.Fields()),
 			Time:            mp.Time().UTC(),
 		}

task_master.go

@@ -556,7 +556,7 @@ func (tm *TaskMaster) WritePoints(database, retentionPolicy string, consistencyL
 			RetentionPolicy: retentionPolicy,
 			Name:            mp.Name(),
 			Group:           models.NilGroup,
-			Tags:            models.Tags(mp.Tags()),
+			Tags:            models.Tags(mp.Tags().Map()),
 			Fields:          models.Fields(mp.Fields()),
 			Time:            mp.Time(),
 		}

vendor.list

@@ -6,7 +6,7 @@ github.com/dustin/go-humanize
 github.com/gogo/protobuf
 github.com/golang/protobuf
 github.com/gorhill/cronexpr
-github.com/influxdata/influxdb 1.0
+github.com/influxdata/influxdb master
 github.com/influxdata/usage-client
 github.com/kimor79/gollectd
 github.com/mattn/go-runewidth