Merge branch 'master' into pagerduty2_routingKey

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### 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
 
 ## v1.5.0 [2018-05-17]
 

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

@@ -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,70 @@ 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()},
+		}}
+		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",
+		},
+	}
+
+	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()

server/server_test.go

@@ -9624,7 +9624,7 @@ func TestServer_AlertHandlers(t *testing.T) {
 					Message:   string(adJSON) + "\n",
 				}}
 				if !cmp.Equal(exp, got) {
-					return fmt.Errorf("unexpected kafak messages -exp/+got:\n%s", cmp.Diff(exp, got))
+					return fmt.Errorf("unexpected kafka messages -exp/+got:\n%s", cmp.Diff(exp, got))
 				}
 				return nil
 			},

services/alert/dao.go

@@ -1,5 +1,7 @@
 package alert
 
+//go:generate easyjson dao.go
+
 import (
 	"encoding/json"
 	"fmt"
@@ -9,6 +11,7 @@ import (
 
 	"github.com/influxdata/kapacitor/alert"
 	"github.com/influxdata/kapacitor/services/storage"
+	"github.com/mailru/easyjson/jlexer"
 	"github.com/pkg/errors"
 )
 
@@ -243,11 +246,13 @@ type TopicStateDAO interface {
 
 const topicStateVersion = 1
 
+//easyjson:json
 type TopicState struct {
 	Topic       string                `json:"topic"`
 	EventStates map[string]EventState `json:"event-states"`
 }
 
+//easyjson:json
 type EventState struct {
 	Message  string        `json:"message"`
 	Details  string        `json:"details"`
@@ -265,8 +270,9 @@ func (t TopicState) MarshalBinary() ([]byte, error) {
 }
 
 func (t *TopicState) UnmarshalBinary(data []byte) error {
-	return storage.VersionJSONDecode(data, func(version int, dec *json.Decoder) error {
-		return dec.Decode(&t)
+	return storage.VersionEasyJSONDecode(data, func(version int, dec *jlexer.Lexer) error {
+		t.UnmarshalEasyJSON(dec)
+		return dec.Error()
 	})
 }