Merged pull request #1795 from sebito91/pd_v2

Support PagerDuty API v2

CHANGELOG.md

@@ -20,6 +20,7 @@ alert will auto-recover.
 
 - [#1827](https://github.com/influxdata/kapacitor/pull/1827): Fix deadlock in load service when task has an error.
 
+- [#1795](https://github.com/influxdata/kapacitor/pull/1795): Support PagerDuty API v2
 
 ## v1.4.0 [2017-12-08]
 

CONTRIBUTING.md

@@ -136,7 +136,7 @@ Kapacitor uses the golang [dep](https://github.com/golang/dep) tool.
 Install the dep tool:
 
 ```
-go get -u github.com/golang/dep
+go get -v -u github.com/golang/dep/cmd/dep
 ```
 
 See the dep help for usage and documentation.

alert.go

@@ -23,6 +23,7 @@ import (
 	"github.com/influxdata/kapacitor/services/opsgenie"
 	"github.com/influxdata/kapacitor/services/opsgenie2"
 	"github.com/influxdata/kapacitor/services/pagerduty"
+	"github.com/influxdata/kapacitor/services/pagerduty2"
 	"github.com/influxdata/kapacitor/services/pushover"
 	"github.com/influxdata/kapacitor/services/sensu"
 	"github.com/influxdata/kapacitor/services/slack"
@@ -203,6 +204,19 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, d NodeDiagnostic) (a
 		an.handlers = append(an.handlers, h)
 	}
 
+	for _, pd := range n.PagerDuty2Handlers {
+		c := pagerduty2.HandlerConfig{
+			ServiceKey: pd.ServiceKey,
+		}
+		h := et.tm.PagerDuty2Service.Handler(c, ctx...)
+		an.handlers = append(an.handlers, h)
+	}
+	if len(n.PagerDuty2Handlers) == 0 && (et.tm.PagerDuty2Service != nil && et.tm.PagerDuty2Service.Global()) {
+		c := pagerduty2.HandlerConfig{}
+		h := et.tm.PagerDuty2Service.Handler(c, ctx...)
+		an.handlers = append(an.handlers, h)
+	}
+
 	for _, s := range n.SensuHandlers {
 		c := sensu.HandlerConfig{
 			Source:   s.Source,

etc/kapacitor/kapacitor.conf

@@ -311,6 +311,17 @@ default-retention-policy = ""
   # without explicitly marking them in the TICKscript.
   global = false
 
+[pagerduty2]
+  # Configure PagerDuty API v2.
+  enabled = false 
+  # Your PagerDuty Service Key.
+  service-key = ""
+  # The PagerDuty API v2 URL should not need to be changed.
+  url = "https://events.pagerduty.com/v2/enqueue"
+  # If true the all alerts will be sent to PagerDuty
+  # without explicitly marking them in the TICKscript.
+  global = false
+
 [pushover]
   # Configure Pushover.
   enabled = false

integrations/streamer_test.go

@@ -52,6 +52,8 @@ import (
 	"github.com/influxdata/kapacitor/services/opsgenie2/opsgenie2test"
 	"github.com/influxdata/kapacitor/services/pagerduty"
 	"github.com/influxdata/kapacitor/services/pagerduty/pagerdutytest"
+	"github.com/influxdata/kapacitor/services/pagerduty2"
+	"github.com/influxdata/kapacitor/services/pagerduty2/pagerduty2test"
 	"github.com/influxdata/kapacitor/services/pushover"
 	"github.com/influxdata/kapacitor/services/pushover/pushovertest"
 	"github.com/influxdata/kapacitor/services/sensu"
@@ -9239,6 +9241,112 @@ stream
 	}
 }
 
+func TestStream_AlertPagerDuty2(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()
+		.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()
+		    .serviceKey('test_override_key')
+	`
+
+	var kapacitorURL string
+	tmInit := func(tm *kapacitor.TaskMaster) {
+		c := pagerduty2.NewConfig()
+		c.Enabled = true
+		c.URL = ts.URL
+		c.ServiceKey = "service_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: "service_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_AlertHTTPPost(t *testing.T) {
 	ts := httpposttest.NewAlertServer(nil, false)
 	defer ts.Close()

pipeline/alert.go

@@ -23,6 +23,7 @@ const defaultMessageTmpl = "{{ .ID }} is {{ .Level }}"
 // Default template for constructing a details message.
 const defaultDetailsTmpl = "{{ json . }}"
 
+// AlertNode struct wraps the default AlertNodeData
 // tick:wraps:AlertNodeData
 type AlertNode struct{ *AlertNodeData }
 
@@ -325,6 +326,10 @@ type AlertNodeData struct {
 	// tick:ignore
 	PagerDutyHandlers []*PagerDutyHandler `tick:"PagerDuty" json:"pagerDuty"`
 
+	// Send alert to PagerDuty API v2.
+	// tick:ignore
+	PagerDuty2Handlers []*PagerDuty2Handler `tick:"PagerDuty2" json:"pagerDuty2"`
+
 	// Send alert to Pushover.
 	// tick:ignore
 	PushoverHandlers []*PushoverHandler `tick:"Pushover" json:"pushover"`
@@ -903,6 +908,64 @@ type PagerDutyHandler struct {
 	ServiceKey string `json:"serviceKey"`
 }
 
+// Send the alert to PagerDuty API v2.
+// To use PagerDuty alerting you must first follow the steps to enable a new 'Generic API' service.
+// NOTE: the API v2 endpoint is different and requires a new configuration in order to process/handle alerts
+//
+// From https://developer.pagerduty.com/documentation/integration/events
+//
+//    1. In your account, under the Services tab, click "Add New Service".
+//    2. Enter a name for the service and select an escalation policy. Then, select "Generic API" for the Service Type.
+//    3. Click the "Add Service" button.
+//    4. Once the service is created, you'll be taken to the service page. On this page, you'll see the "Service key", which is needed to access the API
+//
+// Place the 'service key' into the 'pagerduty' section of the Kapacitor configuration as the option 'service-key'.
+//
+// Example:
+//    [pagerduty2]
+//      enabled = true
+//      service-key = "xxxxxxxxx"
+//
+// With the correct configuration you can now use PagerDuty in TICKscripts.
+//
+// Example:
+//    stream
+//         |alert()
+//             .pagerDuty2()
+//
+// If the 'pagerduty' section in the configuration has the option: global = true
+// then all alerts are sent to PagerDuty without the need to explicitly state it
+// in the TICKscript.
+//
+// Example:
+//    [pagerduty2]
+//      enabled = true
+//      service-key = "xxxxxxxxx"
+//      global = true
+//
+// Example:
+//    stream
+//         |alert()
+//
+// Send alert to PagerDuty API v2.
+// tick:property
+func (n *AlertNodeData) PagerDuty2() *PagerDuty2Handler {
+	pd2 := &PagerDuty2Handler{
+		AlertNodeData: n,
+	}
+	n.PagerDuty2Handlers = append(n.PagerDuty2Handlers, pd2)
+	return pd2
+}
+
+// tick:embedded:AlertNode.PagerDuty
+type PagerDuty2Handler struct {
+	*AlertNodeData `json:"-"`
+
+	// The service key to use for the alert.
+	// Defaults to the value in the configuration if empty.
+	ServiceKey string `json:"serviceKey"`
+}
+
 // Send the alert to HipChat.
 // For step-by-step instructions on setting up Kapacitor with HipChat, see the [Event Handler Setup Guide](https://docs.influxdata.com//kapacitor/latest/guides/event-handler-setup/#hipchat-setup).
 // To allow Kapacitor to post to HipChat,

pipeline/alert_test.go

@@ -65,6 +65,7 @@ func TestAlertNode_MarshalJSON(t *testing.T) {
     "log": null,
     "victorOps": null,
     "pagerDuty": null,
+    "pagerDuty2": null,
     "pushover": null,
     "sensu": null,
     "slack": null,

pipeline/json_test.go

@@ -246,6 +246,7 @@ func TestPipeline_MarshalJSON(t *testing.T) {
             "log": null,
             "victorOps": null,
             "pagerDuty": null,
+            "pagerDuty2": null,
             "pushover": null,
             "sensu": null,
             "slack": null,

pipeline/tick/alert.go

@@ -109,6 +109,11 @@ func (n *AlertNode) Build(a *pipeline.AlertNode) (ast.Node, error) {
 			Dot("serviceKey", h.ServiceKey)
 	}
 
+	for _, h := range a.PagerDuty2Handlers {
+		n.Dot("pagerDuty2").
+			Dot("serviceKey", h.ServiceKey)
+	}
+
 	for _, h := range a.PushoverHandlers {
 		n.Dot("pushover").
 			Dot("userKey", h.UserKey).

pipeline/tick/alert_test.go

@@ -193,6 +193,7 @@ func TestAlertTCPJSON(t *testing.T) {
         "log": null,
         "victorOps": null,
         "pagerDuty": null,
+        "pagerDuty2": null,
         "pushover": null,
         "sensu": null,
         "slack": null,
@@ -306,6 +307,24 @@ func TestAlertPagerDuty(t *testing.T) {
 	PipelineTickTestHelper(t, pipe, want)
 }
 
+func TestAlertPagerDuty2(t *testing.T) {
+	pipe, _, from := StreamFrom()
+	handler := from.Alert().PagerDuty2()
+	handler.ServiceKey = "LeafsNation"
+
+	want := `stream
+    |from()
+    |alert()
+        .id('{{ .Name }}:{{ .Group }}')
+        .message('{{ .ID }} is {{ .Level }}')
+        .details('{{ json . }}')
+        .history(21)
+        .pagerDuty2()
+        .serviceKey('LeafsNation')
+`
+	PipelineTickTestHelper(t, pipe, want)
+}
+
 func TestAlertPushover(t *testing.T) {
 	pipe, _, from := StreamFrom()
 	handler := from.Alert().Pushover()

server/config.go

@@ -34,6 +34,7 @@ import (
 	"github.com/influxdata/kapacitor/services/opsgenie"
 	"github.com/influxdata/kapacitor/services/opsgenie2"
 	"github.com/influxdata/kapacitor/services/pagerduty"
+	"github.com/influxdata/kapacitor/services/pagerduty2"
 	"github.com/influxdata/kapacitor/services/pushover"
 	"github.com/influxdata/kapacitor/services/replay"
 	"github.com/influxdata/kapacitor/services/reporting"
@@ -79,21 +80,22 @@ type Config struct {
 	UDP      []udp.Config      `toml:"udp"`
 
 	// Alert handlers
-	Alerta    alerta.Config    `toml:"alerta" override:"alerta"`
-	HipChat   hipchat.Config   `toml:"hipchat" override:"hipchat"`
-	MQTT      mqtt.Configs     `toml:"mqtt" override:"mqtt,element-key=name"`
-	OpsGenie  opsgenie.Config  `toml:"opsgenie" override:"opsgenie"`
-	OpsGenie2 opsgenie2.Config `toml:"opsgenie2" override:"opsgenie2"`
-	PagerDuty pagerduty.Config `toml:"pagerduty" override:"pagerduty"`
-	Pushover  pushover.Config  `toml:"pushover" override:"pushover"`
-	HTTPPost  httppost.Configs `toml:"httppost" override:"httppost,element-key=endpoint"`
-	SMTP      smtp.Config      `toml:"smtp" override:"smtp"`
-	SNMPTrap  snmptrap.Config  `toml:"snmptrap" override:"snmptrap"`
-	Sensu     sensu.Config     `toml:"sensu" override:"sensu"`
-	Slack     slack.Config     `toml:"slack" override:"slack"`
-	Talk      talk.Config      `toml:"talk" override:"talk"`
-	Telegram  telegram.Config  `toml:"telegram" override:"telegram"`
-	VictorOps victorops.Config `toml:"victorops" override:"victorops"`
+	Alerta     alerta.Config     `toml:"alerta" override:"alerta"`
+	HipChat    hipchat.Config    `toml:"hipchat" override:"hipchat"`
+	MQTT       mqtt.Configs      `toml:"mqtt" override:"mqtt,element-key=name"`
+	OpsGenie   opsgenie.Config   `toml:"opsgenie" override:"opsgenie"`
+	OpsGenie2  opsgenie2.Config  `toml:"opsgenie2" override:"opsgenie2"`
+	PagerDuty  pagerduty.Config  `toml:"pagerduty" override:"pagerduty"`
+	PagerDuty2 pagerduty2.Config `toml:"pagerduty2" override:"pagerduty2"`
+	Pushover   pushover.Config   `toml:"pushover" override:"pushover"`
+	HTTPPost   httppost.Configs  `toml:"httppost" override:"httppost,element-key=endpoint"`
+	SMTP       smtp.Config       `toml:"smtp" override:"smtp"`
+	SNMPTrap   snmptrap.Config   `toml:"snmptrap" override:"snmptrap"`
+	Sensu      sensu.Config      `toml:"sensu" override:"sensu"`
+	Slack      slack.Config      `toml:"slack" override:"slack"`
+	Talk       talk.Config       `toml:"talk" override:"talk"`
+	Telegram   telegram.Config   `toml:"telegram" override:"telegram"`
+	VictorOps  victorops.Config  `toml:"victorops" override:"victorops"`
 
 	// Discovery for scraping
 	Scraper         []scraper.Config          `toml:"scraper" override:"scraper,element-key=name"`
@@ -150,6 +152,7 @@ func NewConfig() *Config {
 	c.OpsGenie = opsgenie.NewConfig()
 	c.OpsGenie2 = opsgenie2.NewConfig()
 	c.PagerDuty = pagerduty.NewConfig()
+	c.PagerDuty2 = pagerduty2.NewConfig()
 	c.Pushover = pushover.NewConfig()
 	c.HTTPPost = httppost.Configs{httppost.NewConfig()}
 	c.SMTP = smtp.NewConfig()
@@ -272,6 +275,9 @@ func (c *Config) Validate() error {
 	if err := c.PagerDuty.Validate(); err != nil {
 		return errors.Wrap(err, "pagerduty")
 	}
+	if err := c.PagerDuty2.Validate(); err != nil {
+		return errors.Wrap(err, "pagerduty2")
+	}
 	if err := c.Pushover.Validate(); err != nil {
 		return errors.Wrap(err, "pushover")
 	}