Merge remote-tracking branch 'upstream/master' into feature/sideload-…

…http

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## v1.5.1 [2018-08-06]
+
+### Bugfixes
+
+- [#1938](https://github.com/influxdata/kapacitor/issues/1938): pagerduty2 should use routingKey rather than serviceKey
+- [#1982](https://github.com/influxdata/kapacitor/pull/1982): Fix KafkaTopic not working from TICKscript
+- [#1989](https://github.com/influxdata/kapacitor/pull/1989): Improve Kafka alert throughput.
+
 ## v1.5.0 [2018-05-17]
 
 ### Features

Dockerfile_build_ubuntu32

@@ -19,7 +19,7 @@ RUN apt-get -qq update && apt-get -qq install -y \
     curl
 
 # Install protobuf3 protoc binary
-ENV PROTO_VERSION 3.0.0
+ENV PROTO_VERSION 3.4.0
 RUN wget -q https://github.com/google/protobuf/releases/download/v${PROTO_VERSION}/protoc-${PROTO_VERSION}-linux-x86_32.zip\
     && unzip -j protoc-${PROTO_VERSION}-linux-x86_32.zip bin/protoc -d /bin \
     rm protoc-${PROTO_VERSION}-linux-x86_64.zip

Dockerfile_build_ubuntu64

@@ -29,7 +29,7 @@ RUN apt-get -qq update && apt-get -qq install -y \
 RUN gem install fpm
 
 # Install protobuf3 protoc binary
-ENV PROTO_VERSION 3.0.0
+ENV PROTO_VERSION 3.4.0
 RUN wget -q https://github.com/google/protobuf/releases/download/v${PROTO_VERSION}/protoc-${PROTO_VERSION}-linux-x86_64.zip \
     && unzip -j protoc-${PROTO_VERSION}-linux-x86_64.zip bin/protoc -d /bin \
     rm protoc-${PROTO_VERSION}-linux-x86_64.zip

Gopkg.toml

@@ -1,4 +1,8 @@
-required = ["github.com/benbjohnson/tmpl","github.com/golang/protobuf/protoc-gen-go"]
+required = [
+    "github.com/benbjohnson/tmpl",
+    "github.com/golang/protobuf/protoc-gen-go",
+    "github.com/mailru/easyjson/easyjson",
+]
 
 [prune]
   unused-packages = true
@@ -9,6 +13,10 @@ required = ["github.com/benbjohnson/tmpl","github.com/golang/protobuf/protoc-gen
   branch = "master"
   name = "github.com/davecgh/go-spew"
 
+[[constraint]]
+  branch = "master"
+  name = "github.com/mailru/easyjson"
+
 [[constraint]]
   branch = "master"
   name = "github.com/evanphx/json-patch"
@@ -57,3 +65,11 @@ required = ["github.com/benbjohnson/tmpl","github.com/golang/protobuf/protoc-gen
   name = "github.com/Azure/go-autorest"
   revision = "a2fdd780c9a50455cecd249b00bdc3eb73a78e31"
 
+[[override]]
+  name= "gopkg.in/fsnotify.v1"
+  revision = "629574ca2a5df945712d3079857300b5e4da0236"
+  source = "[email protected]:fsnotify/fsnotify"
+
+[[override]]
+  name= "github.com/mailru/easyjson"
+  revision = "3fdea8d05856a0c8df22ed4bc71b3219245e4485"

LICENSE_OF_DEPENDENCIES.md

@@ -23,3 +23,4 @@ Dependencies
 * github.com/shurcooL/sanitized\_anchor\_name [MIT](https://github.com/shurcooL/sanitized_anchor_name/blob/master/LICENSE)
 * github.com/stretchr/testify [MIT](https://github.com/stretchr/testify/blob/master/LICENSE)
 * gopkg.in/gomail.v2 [MIT](https://github.com/go-gomail/gomail/blob/v2/LICENSE)
+* github.com/mailru/easyjson [MIT](https://github.com/mailru/easyjson/blob/3fdea8d05856a0c8df22ed4bc71b3219245e4485/LICENSE)

alert.go

@@ -209,7 +209,7 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, d NodeDiagnostic) (a
 
 	for _, pd := range n.PagerDuty2Handlers {
 		c := pagerduty2.HandlerConfig{
-			RoutingKey: pd.ServiceKey,
+			RoutingKey: pd.RoutingKey,
 		}
 		h := et.tm.PagerDuty2Service.Handler(c, ctx...)
 		an.handlers = append(an.handlers, h)
@@ -319,7 +319,7 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, d NodeDiagnostic) (a
 	for _, k := range n.KafkaHandlers {
 		c := kafka.HandlerConfig{
 			Cluster:  k.Cluster,
-			Topic:    k.Topic,
+			Topic:    k.KafkaTopic,
 			Template: k.Template,
 		}
 		h, err := et.tm.KafkaService.Handler(c, ctx...)

build.py

@@ -153,6 +153,7 @@ def run_generate():
     logging.info("Running generate...")
     run("go install ./vendor/github.com/golang/protobuf/protoc-gen-go")
     run("go install ./vendor/github.com/benbjohnson/tmpl")
+    run("go install ./vendor/github.com/mailru/easyjson/easyjson")
     generate_cmd = ["go", "generate"]
     generate_cmd.extend(go_list())
     p = subprocess.Popen(generate_cmd)

integrations/streamer_test.go

@@ -46,6 +46,8 @@ import (
 	"github.com/influxdata/kapacitor/services/httppost/httpposttest"
 	k8s "github.com/influxdata/kapacitor/services/k8s/client"
 	"github.com/influxdata/kapacitor/services/k8s/k8stest"
+	"github.com/influxdata/kapacitor/services/kafka"
+	"github.com/influxdata/kapacitor/services/kafka/kafkatest"
 	"github.com/influxdata/kapacitor/services/opsgenie"
 	"github.com/influxdata/kapacitor/services/opsgenie/opsgenietest"
 	"github.com/influxdata/kapacitor/services/opsgenie2"
@@ -8543,6 +8545,74 @@ stream
 	}
 }
 
+func TestStream_AlertKafka(t *testing.T) {
+	ts, err := kafkatest.NewServer()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ts.Close()
+
+	var script = `
+stream
+	|from()
+		.measurement('cpu')
+		.where(lambda: "host" == 'serverA')
+		.groupBy('host')
+	|window()
+		.period(10s)
+		.every(10s)
+	|count('value')
+	|alert()
+		.id('kapacitor/{{ .Name }}/{{ index .Tags "host" }}')
+		.info(lambda: "count" > 6.0)
+		.warn(lambda: "count" > 7.0)
+		.crit(lambda: "count" > 8.0)
+		.kafka()
+		.cluster('default')
+		.kafkaTopic('testTopic')
+		.template('{{.Message}}')
+`
+
+	tmInit := func(tm *kapacitor.TaskMaster) {
+		configs := kafka.Configs{{
+			Enabled:   true,
+			ID:        "default",
+			Brokers:   []string{ts.Addr.String()},
+			BatchSize: 1,
+		}}
+		d := diagService.NewKafkaHandler().WithContext(keyvalue.KV("test", "kafka"))
+		tm.KafkaService = kafka.NewService(configs, d)
+	}
+	testStreamerNoOutput(t, "TestStream_Alert", script, 13*time.Second, tmInit)
+
+	exp := []interface{}{
+		kafkatest.Message{
+			Topic:     "testTopic",
+			Partition: 1,
+			Offset:    0,
+			Key:       "kapacitor/cpu/serverA",
+			Message:   "kapacitor/cpu/serverA is CRITICAL",
+		},
+	}
+
+	// Wait for kakfa messages to be written
+	time.Sleep(time.Second)
+
+	ts.Close()
+	msgs, err := ts.Messages()
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := make([]interface{}, len(msgs))
+	for i, m := range msgs {
+		got[i] = m
+	}
+
+	if err := compareListIgnoreOrder(got, exp, nil); err != nil {
+		t.Error(err)
+	}
+}
+
 func TestStream_AlertTelegram(t *testing.T) {
 	ts := telegramtest.NewServer()
 	defer ts.Close()
@@ -9269,6 +9339,112 @@ func TestStream_AlertPagerDuty2(t *testing.T) {
 		},
 	}
 
+	var script = `
+stream
+	|from()
+		.measurement('cpu')
+		.where(lambda: "host" == 'serverA')
+		.groupBy('host')
+	|window()
+		.period(10s)
+		.every(10s)
+	|count('value')
+	|alert()
+		.id('kapacitor/{{ .Name }}/{{ index .Tags "host" }}')
+		.message('{{ .Level }} alert for {{ .ID }}')
+		.info(lambda: "count" > 6.0)
+		.warn(lambda: "count" > 7.0)
+		.crit(lambda: "count" > 8.0)
+		.pagerDuty2()
+		.pagerDuty2()
+		    .routingKey('test_override_key')
+	`
+
+	var kapacitorURL string
+	tmInit := func(tm *kapacitor.TaskMaster) {
+		c := pagerduty2.NewConfig()
+		c.Enabled = true
+		c.URL = ts.URL
+		c.RoutingKey = "routing_key"
+		pd := pagerduty2.NewService(c, diagService.NewPagerDuty2Handler())
+		pd.HTTPDService = tm.HTTPDService
+		tm.PagerDuty2Service = pd
+
+		kapacitorURL = tm.HTTPDService.URL()
+	}
+	testStreamerNoOutput(t, "TestStream_Alert", script, 13*time.Second, tmInit)
+
+	exp := []interface{}{
+		pagerduty2test.Request{
+			URL: "/",
+			PostData: pagerduty2test.PostData{
+				Client:      "kapacitor",
+				ClientURL:   kapacitorURL,
+				EventAction: "trigger",
+				DedupKey:    "kapacitor/cpu/serverA",
+				Payload: &pagerduty2test.PDCEF{
+					Summary:       "CRITICAL alert for kapacitor/cpu/serverA",
+					Source:        "serverA",
+					Severity:      "critical",
+					Class:         "TestStream_Alert",
+					CustomDetails: detailsTmpl,
+					Timestamp:     "1971-01-01T00:00:10.000000000Z",
+				},
+				RoutingKey: "routing_key",
+			},
+		},
+		pagerduty2test.Request{
+			URL: "/",
+			PostData: pagerduty2test.PostData{
+				Client:      "kapacitor",
+				ClientURL:   kapacitorURL,
+				EventAction: "trigger",
+				DedupKey:    "kapacitor/cpu/serverA",
+				Payload: &pagerduty2test.PDCEF{
+					Summary:       "CRITICAL alert for kapacitor/cpu/serverA",
+					Source:        "serverA",
+					Severity:      "critical",
+					Class:         "TestStream_Alert",
+					CustomDetails: detailsTmpl,
+					Timestamp:     "1971-01-01T00:00:10.000000000Z",
+				},
+				RoutingKey: "test_override_key",
+			},
+		},
+	}
+
+	ts.Close()
+	var got []interface{}
+	for _, g := range ts.Requests() {
+		got = append(got, g)
+	}
+
+	if err := compareListIgnoreOrder(got, exp, nil); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestStream_AlertPagerDuty2_ServiceKey(t *testing.T) {
+	ts := pagerduty2test.NewServer()
+	defer ts.Close()
+
+	detailsTmpl := map[string]interface{}{
+		"result": map[string]interface{}{
+			"series": []interface{}{
+				map[string]interface{}{
+					"name": "cpu",
+					"tags": map[string]interface{}{
+						"host": "serverA",
+					},
+					"columns": []interface{}{"time", "count"},
+					"values": []interface{}{
+						[]interface{}{"1971-01-01T00:00:10Z", float64(10)},
+					},
+				},
+			},
+		},
+	}
+
 	var script = `
 stream
 	|from()
@@ -9295,7 +9471,7 @@ stream
 		c := pagerduty2.NewConfig()
 		c.Enabled = true
 		c.URL = ts.URL
-		c.RoutingKey = "service_key"
+		c.RoutingKey = "routing_key"
 		pd := pagerduty2.NewService(c, diagService.NewPagerDuty2Handler())
 		pd.HTTPDService = tm.HTTPDService
 		tm.PagerDuty2Service = pd
@@ -9320,7 +9496,7 @@ stream
 					CustomDetails: detailsTmpl,
 					Timestamp:     "1971-01-01T00:00:10.000000000Z",
 				},
-				RoutingKey: "service_key",
+				RoutingKey: "routing_key",
 			},
 		},
 		pagerduty2test.Request{