Merge branch 'master' into feature/jsonpipe

CHANGELOG.md

@@ -3,7 +3,9 @@
 ## Unreleased
 
 ### Features
+- [#1408](https://github.com/influxdata/kapacitor/issues/1408): Add Previous state
 
+- [#1461](https://github.com/influxdata/kapacitor/issues/1461): alert.post and https_post timeouts needed.
 - [#1413](https://github.com/influxdata/kapacitor/issues/1413): Add subscriptions modes to InfluxDB subscriptions.
 - [#1436](https://github.com/influxdata/kapacitor/issues/1436): Add linear fill support for QueryNode.
 - [#1345](https://github.com/influxdata/kapacitor/issues/1345): Add MQTT Alert Handler
@@ -12,6 +14,7 @@
     The breaking change is that the Combine and Flatten nodes previously, but erroneously, operated across batch boundaries; this has been fixed.
 - [#1497](https://github.com/influxdata/kapacitor/pull/1497): Add support for Docker Swarm autoscaling services.
 - [#1485](https://github.com/influxdata/kapacitor/issues/1485): Add bools field types to UDFs.
+- [#1549](https://github.com/influxdata/kapacitor/issues/1549): Add stateless now() function to get the current local time.
 - [#1545](https://github.com/influxdata/kapacitor/pull/1545): Add support for timeout, tags and service template in the Alerta AlertNode
 - [#1568](https://github.com/influxdata/kapacitor/issues/1568): Add support for custom HTTP Post bodies via a template system.
 - [#1569](https://github.com/influxdata/kapacitor/issues/1569): Add support for add the HTTP status code as a field when using httpPost
@@ -24,6 +27,7 @@
     Load service was added; the service can load tasks/handlers from a directory.
 - [#1606](https://github.com/influxdata/kapacitor/pull/1606): Update Go version to 1.9.1
 - [#1578](https://github.com/influxdata/kapacitor/pull/1578): Add support for exposing logs via the API.
+- [#1605](https://github.com/influxdata/kapacitor/issues/1605): Add support for {{ .Duration }} on Alert Message property
 
 ### Bugfixes
 
@@ -35,6 +39,8 @@
 - [#1516](https://github.com/influxdata/kapacitor/pull/1516): Fix bad PagerDuty test the required server info.
 - [#1581](https://github.com/influxdata/kapacitor/pull/1581): Add SNMP sysUpTime to SNMP Trap service
 - [#1547](https://github.com/influxdata/kapacitor/issues/1547): Fix panic on recording replay with HTTPPostHandler.
+- [#1623](https://github.com/influxdata/kapacitor/issues/1623): Fix k8s incluster master api dns resolution
+- [#1630](https://github.com/influxdata/kapacitor/issues/1630): Remove the pidfile after the server has exited.
 
 ## v1.3.3 [2017-08-11]
 

CONTRIBUTING.md

@@ -87,7 +87,7 @@ But instead of cloning the main repo, instead clone your fork. Follow the steps
     export GOPATH=$HOME/go
     mkdir -p $GOPATH/src/github.com/influxdata
     cd $GOPATH/src/github.com/influxdata
-    git clone [email protected]:<username>/kapacitor
+    git clone [email protected]:<username>/kapacitor.git
 
 Retaining the directory structure `$GOPATH/src/github.com/influxdata` is necessary so that Go imports work correctly.
 

alert.go

@@ -360,6 +360,7 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, d NodeDiagnostic) (a
 			Endpoint:        p.Endpoint,
 			Headers:         p.Headers,
 			CaptureResponse: p.CaptureResponseFlag,
+			Timeout:         p.Timeout,
 		}
 		h := et.tm.HTTPPostService.Handler(c, ctx...)
 		an.handlers = append(an.handlers, h)
@@ -695,7 +696,7 @@ func (n *AlertNode) event(
 	d time.Duration,
 	result models.Result,
 ) (alert.Event, error) {
-	msg, details, err := n.renderMessageAndDetails(id, name, t, group, tags, fields, level)
+	msg, details, err := n.renderMessageAndDetails(id, name, t, group, tags, fields, level, d)
 	if err != nil {
 		return alert.Event{}, err
 	}
@@ -1046,6 +1047,9 @@ type messageInfo struct {
 
 	// Time
 	Time time.Time
+
+	// Duration of the alert
+	Duration time.Duration
 }
 
 type detailsInfo struct {
@@ -1087,7 +1091,7 @@ func (n *AlertNode) renderID(name string, group models.GroupID, tags models.Tags
 	return id.String(), nil
 }
 
-func (n *AlertNode) renderMessageAndDetails(id, name string, t time.Time, group models.GroupID, tags models.Tags, fields models.Fields, level alert.Level) (string, string, error) {
+func (n *AlertNode) renderMessageAndDetails(id, name string, t time.Time, group models.GroupID, tags models.Tags, fields models.Fields, level alert.Level, d time.Duration) (string, string, error) {
 	g := string(group)
 	if group == models.NilGroup {
 		g = "nil"
@@ -1100,10 +1104,11 @@ func (n *AlertNode) renderMessageAndDetails(id, name string, t time.Time, group
 			Tags:       tags,
 			ServerInfo: n.serverInfo(),
 		},
-		ID:     id,
-		Fields: fields,
-		Level:  level.String(),
-		Time:   t,
+		ID:       id,
+		Fields:   fields,
+		Level:    level.String(),
+		Time:     t,
+		Duration: d,
 	}
 
 	// Grab a buffer for the message template and the details template

alert/types.go

@@ -19,13 +19,14 @@ type Event struct {
 
 func (e Event) AlertData() Data {
 	return Data{
-		ID:       e.State.ID,
-		Message:  e.State.Message,
-		Details:  e.State.Details,
-		Time:     e.State.Time,
-		Duration: e.State.Duration,
-		Level:    e.State.Level,
-		Data:     e.Data.Result,
+		ID:            e.State.ID,
+		Message:       e.State.Message,
+		Details:       e.State.Details,
+		Time:          e.State.Time,
+		Duration:      e.State.Duration,
+		Level:         e.State.Level,
+		Data:          e.Data.Result,
+		PreviousLevel: e.previousState.Level,
 	}
 }
 
@@ -170,11 +171,12 @@ type TopicState struct {
 // Data is a structure that contains relevant data about an alert event.
 // The structure is intended to be JSON encoded, providing a consistent data format.
 type Data struct {
-	ID       string        `json:"id"`
-	Message  string        `json:"message"`
-	Details  string        `json:"details"`
-	Time     time.Time     `json:"time"`
-	Duration time.Duration `json:"duration"`
-	Level    Level         `json:"level"`
-	Data     models.Result `json:"data"`
+	ID            string        `json:"id"`
+	Message       string        `json:"message"`
+	Details       string        `json:"details"`
+	Time          time.Time     `json:"time"`
+	Duration      time.Duration `json:"duration"`
+	Level         Level         `json:"level"`
+	Data          models.Result `json:"data"`
+	PreviousLevel Level         `json:"previousLevel"`
 }

cmd/kapacitord/run/command.go

@@ -41,6 +41,7 @@ type Command struct {
 	Commit  string
 
 	closing chan struct{}
+	pidfile string
 	Closed  chan struct{}
 
 	Stdin  io.Reader
@@ -104,7 +105,7 @@ func (cmd *Command) Run(args ...string) error {
 	// Initialize Logging Services
 	cmd.diagService = diagnostic.NewService(config.Logging, cmd.Stdout, cmd.Stderr)
 	if err := cmd.diagService.Open(); err != nil {
-		return fmt.Errorf("failed to opend diagnostic service: %v", err)
+		return fmt.Errorf("failed to open diagnostic service: %v", err)
 	}
 
 	// Initialize cmd diagnostic
@@ -118,6 +119,7 @@ func (cmd *Command) Run(args ...string) error {
 	if err := cmd.writePIDFile(options.PIDFile); err != nil {
 		return fmt.Errorf("write pid file: %s", err)
 	}
+	cmd.pidfile = options.PIDFile
 
 	// Create server from config and start it.
 	buildInfo := server.BuildInfo{Version: cmd.Version, Commit: cmd.Commit, Branch: cmd.Branch}
@@ -142,6 +144,7 @@ func (cmd *Command) Run(args ...string) error {
 // Close shuts down the server.
 func (cmd *Command) Close() error {
 	defer close(cmd.Closed)
+	defer cmd.removePIDFile()
 	close(cmd.closing)
 	if cmd.Server != nil {
 		return cmd.Server.Close()
@@ -165,6 +168,14 @@ func (cmd *Command) monitorServerErrors() {
 	}
 }
 
+func (cmd *Command) removePIDFile() {
+	if cmd.pidfile != "" {
+		if err := os.Remove(cmd.pidfile); err != nil {
+			cmd.Diag.Error("unable to remove pidfile", err)
+		}
+	}
+}
+
 // ParseFlags parses the command line flags from args and returns an options set.
 func (cmd *Command) ParseFlags(args ...string) (Options, error) {
 	var options Options
@@ -191,8 +202,7 @@ func (cmd *Command) writePIDFile(path string) error {
 	}
 
 	// Ensure the required directory structure exists.
-	err := os.MkdirAll(filepath.Dir(path), 0777)
-	if err != nil {
+	if err := os.MkdirAll(filepath.Dir(path), 0777); err != nil {
 		return fmt.Errorf("mkdir: %s", err)
 	}
 

cmd/kapacitord/run/command_test.go

@@ -0,0 +1,70 @@
+package run_test
+
+import (
+	"bytes"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"testing"
+	"text/template"
+	"time"
+
+	"fmt"
+
+	"github.com/influxdata/kapacitor/cmd/kapacitord/run"
+)
+
+func TestCommand_PIDFile(t *testing.T) {
+	tmpdir, err := ioutil.TempDir(os.TempDir(), "kapacitord-test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(tmpdir)
+
+	// Write out a config file that does not attempt to connect to influxdb.
+	configFile := filepath.Join(tmpdir, "kapacitor.conf")
+	configTemplate := template.Must(template.New("config_file").Parse(`data_dir = "{{.TempDir}}/data"
+[[influxdb]]
+  enabled = false
+[replay]
+  dir = "{{.TempDir}}/replay"
+[storage]
+  boltdb = "{{.TempDir}}/kapacitor.db"
+[task]
+  dir = "{{.TempDir}}/tasks"
+[load]
+  dir = "{{.TempDir}}/load"
+`))
+	var buf bytes.Buffer
+	if err := configTemplate.Execute(&buf, map[string]string{"TempDir": tmpdir}); err != nil {
+		t.Fatalf("unable to generate config file: %s", err)
+	}
+	fmt.Println(buf.String())
+	if err := ioutil.WriteFile(configFile, buf.Bytes(), 0600); err != nil {
+		t.Fatalf("unable to write %s: %s", configFile, err)
+	}
+
+	pidFile := filepath.Join(tmpdir, "kapacitor.pid")
+
+	cmd := run.NewCommand()
+	if err := cmd.Run("-config", configFile, "-pidfile", pidFile); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+
+	if _, err := os.Stat(pidFile); err != nil {
+		t.Fatalf("could not stat pid file: %s", err)
+	}
+	go cmd.Close()
+
+	timeout := time.NewTimer(5 * time.Second)
+	select {
+	case <-timeout.C:
+		t.Fatal("unexpected timeout")
+	case <-cmd.Closed:
+		timeout.Stop()
+	}
+
+	if _, err := os.Stat(pidFile); err == nil {
+		t.Fatal("expected pid file to be removed")
+	}
+}

examples/nodes/tasks/alert.tick

@@ -0,0 +1,8 @@
+dbrp "telegraf"."autogen"
+
+stream
+    |from()
+        .measurement('cpu')
+        .groupBy(*)
+    |alert()
+        .crit(lambda: "usage_user" > 80)

examples/nodes/tasks/batch.tick

@@ -0,0 +1,6 @@
+dbrp "telegraf"."autogen"
+
+batch
+    |query('SELECT mean("usage_user") FROM "telegraf"."autogen"."cpu"')
+        .period(1m)
+        .every(10s)

examples/nodes/tasks/combine.tick

@@ -0,0 +1,8 @@
+dbrp "telegraf"."autogen"
+
+stream
+    |from()
+        .measurement('cpu')
+        .groupBy(*)
+    |combine(lambda: "cpu" == 'cpu-total', lambda: TRUE)
+        .as('total', 'cpu')

examples/nodes/tasks/default.tick

@@ -0,0 +1,9 @@
+dbrp "telegraf"."autogen"
+
+stream
+    |from()
+        .measurement('cpu')
+        .groupBy(*)
+    |default()
+        .tag('sweet_new_tag', 'this_is')
+        .field('value', 10.9)

examples/nodes/tasks/delete.tick

@@ -0,0 +1,9 @@
+dbrp "telegraf"."autogen"
+
+stream
+    |from()
+        .measurement('cpu')
+        .groupBy(*)
+    |delete()
+        .tag('cpu')
+        .field('usuage_idle')

examples/nodes/tasks/eval.tick

@@ -0,0 +1,7 @@
+dbrp "telegraf"."autogen"
+
+stream
+    |from()
+        .measurement('cpu')
+    |eval(lambda: 100.0 - "usage_idle")
+        .as('usage_not_idle')

examples/nodes/tasks/flatten.tick

@@ -0,0 +1,7 @@
+dbrp "telegraf"."autogen"
+
+stream
+    |from()
+        .measurement('cpu')
+    |flatten()
+        .on('cpu')

examples/nodes/tasks/groupby.tick

@@ -0,0 +1,6 @@
+dbrp "telegraf"."autogen"
+
+stream
+    |from()
+        .measurement('system')
+    |groupBy(*)

examples/nodes/tasks/handle_loopback.tick

@@ -0,0 +1,6 @@
+dbrp "loopback"."autogen"
+
+stream
+    |from()
+        .measurement('system')
+        .groupBy(*)

examples/nodes/tasks/httpout.tick

@@ -0,0 +1,9 @@
+dbrp "telegraf"."autogen"
+
+stream
+    |from()
+        .measurement('system')
+    |window()
+        .period(1m)
+        .every(10s)
+    |httpOut('data')

examples/nodes/tasks/httppost.tick

@@ -0,0 +1,9 @@
+dbrp "telegraf"."autogen"
+
+stream
+    |from()
+        .measurement('system')
+    |window()
+        .period(1m)
+        .every(10s)
+    |httpPost('http://localhost:8080/example')

examples/nodes/tasks/influxdbout.tick

@@ -0,0 +1,13 @@
+dbrp "telegraf"."autogen"
+
+stream
+    |from()
+        .measurement('system')
+    |window()
+        .period(1m)
+        .every(10s)
+    |mean('load1')
+        .as('load1')
+    |influxDBOut()
+        .database('mydb')
+        .measurement('derived_system')