Webhooks plugin: add mandrill (#1408)

* Add mandrill webhook.

* Store the id of the msg as part of event.

Signed-off-by: Cyril Duez <[email protected]>
Signed-off-by: François de Metz <[email protected]>

* Decode body to get the mandrill_events.

Signed-off-by: Cyril Duez <[email protected]>
Signed-off-by: François de Metz <[email protected]>

* Handle HEAD request.

Signed-off-by: Cyril Duez <[email protected]>
Signed-off-by: François de Metz <[email protected]>

* Add the README.

Signed-off-by: Cyril Duez <[email protected]>
Signed-off-by: François de Metz <[email protected]>

* Add mandrill_webhooks to the README.

Signed-off-by: Cyril Duez <[email protected]>
Signed-off-by: François de Metz <[email protected]>

* Update changelog.

Signed-off-by: Cyril Duez <[email protected]>
Signed-off-by: François de Metz <[email protected]>

* Run gofmt.

Signed-off-by: Cyril Duez <[email protected]>
Signed-off-by: François de Metz <[email protected]>

CHANGELOG.md

@@ -36,6 +36,7 @@ should now look like:
 
 - [#1289](https://github.com/influxdata/telegraf/pull/1289): webhooks input plugin. Thanks @francois2metz and @cduez!
 - [#1247](https://github.com/influxdata/telegraf/pull/1247): rollbar webhook plugin.
+- [#1408](https://github.com/influxdata/telegraf/pull/1408): mandrill webhook plugin.
 - [#1402](https://github.com/influxdata/telegraf/pull/1402): docker-machine/boot2docker no longer required for unit tests.
 - [#1350](https://github.com/influxdata/telegraf/pull/1350): cgroup input plugin.
 - [#1369](https://github.com/influxdata/telegraf/pull/1369): Add input plugin for consuming metrics from NSQD.

README.md

@@ -219,6 +219,7 @@ Telegraf can also collect metrics via the following service plugins:
 * [nats_consumer](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/nats_consumer)
 * [webhooks](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/webhooks)
   * [github](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/webhooks/github)
+  * [mandrill](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/webhooks/mandrill)
   * [rollbar](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/webhooks/rollbar)
 * [nsq_consumer](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/nsq_consumer)
 

plugins/inputs/webhooks/README.md

@@ -16,6 +16,7 @@ $ sudo service telegraf start
 ## Available webhooks
 
 - [Github](github/)
+- [Mandrill](mandrill/)
 - [Rollbar](rollbar/)
 
 ## Adding new webhooks plugin

plugins/inputs/webhooks/mandrill/README.md

@@ -0,0 +1,15 @@
+# mandrill webhook
+
+You should configure your Mandrill's Webhooks to point at the `webhooks` service. To do this go to `mandrillapp.com/` and click `Settings > Webhooks`. In the resulting page, click on `Add a Webhook`, select all events, and set the `URL` to `http://<my_ip>:1619/mandrill`, and click on `Create Webhook`.
+
+## Events
+
+See the [webhook doc](https://mandrill.zendesk.com/hc/en-us/articles/205583307-Message-Event-Webhook-format).
+
+All events for logs the original timestamp, the event name and the unique identifier of the message that generated the event.
+
+**Tags:**
+* 'event' = `event.event` string
+
+**Fields:**
+* 'id' = `event._id` string

plugins/inputs/webhooks/mandrill/mandrill_webhooks.go

@@ -0,0 +1,56 @@
+package mandrill
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/influxdata/telegraf"
+)
+
+type MandrillWebhook struct {
+	Path string
+	acc  telegraf.Accumulator
+}
+
+func (md *MandrillWebhook) Register(router *mux.Router, acc telegraf.Accumulator) {
+	router.HandleFunc(md.Path, md.returnOK).Methods("HEAD")
+	router.HandleFunc(md.Path, md.eventHandler).Methods("POST")
+
+	log.Printf("Started the webhooks_mandrill on %s\n", md.Path)
+	md.acc = acc
+}
+
+func (md *MandrillWebhook) returnOK(w http.ResponseWriter, r *http.Request) {
+	w.WriteHeader(http.StatusOK)
+}
+
+func (md *MandrillWebhook) eventHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+	data, err := url.ParseQuery(string(body))
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+	var events []MandrillEvent
+	err = json.Unmarshal([]byte(data.Get("mandrill_events")), &events)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	for _, event := range events {
+		md.acc.AddFields("mandrill_webhooks", event.Fields(), event.Tags(), time.Unix(event.TimeStamp, 0))
+	}
+
+	w.WriteHeader(http.StatusOK)
+}

plugins/inputs/webhooks/mandrill/mandrill_webhooks_events.go

@@ -0,0 +1,24 @@
+package mandrill
+
+type Event interface {
+	Tags() map[string]string
+	Fields() map[string]interface{}
+}
+
+type MandrillEvent struct {
+	EventName string `json:"event"`
+	TimeStamp int64  `json:"ts"`
+	Id        string `json:"_id"`
+}
+
+func (me *MandrillEvent) Tags() map[string]string {
+	return map[string]string{
+		"event": me.EventName,
+	}
+}
+
+func (me *MandrillEvent) Fields() map[string]interface{} {
+	return map[string]interface{}{
+		"id": me.Id,
+	}
+}

plugins/inputs/webhooks/mandrill/mandrill_webhooks_events_json_test.go

@@ -0,0 +1,58 @@
+package mandrill
+
+func SendEventJSON() string {
+	return `
+	{
+	    "event": "send",
+	    "msg": {
+	      "ts": 1365109999,
+	      "subject": "This an example webhook message",
+	      "email": "[email protected]",
+	      "sender": "[email protected]",
+	      "tags": [
+	        "webhook-example"
+	      ],
+	      "opens": [
+
+	      ],
+	      "clicks": [
+
+	      ],
+	      "state": "sent",
+	      "metadata": {
+	        "user_id": 111
+	      },
+	      "_id": "exampleaaaaaaaaaaaaaaaaaaaaaaaaa",
+	      "_version": "exampleaaaaaaaaaaaaaaa"
+	    },
+	    "_id": "id1",
+	    "ts": 1384954004
+	}`
+}
+
+func HardBounceEventJSON() string {
+	return `
+	{
+	    "event": "hard_bounce",
+	    "msg": {
+	      "ts": 1365109999,
+	      "subject": "This an example webhook message",
+	      "email": "[email protected]",
+	      "sender": "[email protected]",
+	      "tags": [
+	        "webhook-example"
+	      ],
+	      "state": "bounced",
+	      "metadata": {
+	        "user_id": 111
+	      },
+	      "_id": "exampleaaaaaaaaaaaaaaaaaaaaaaaaa2",
+	      "_version": "exampleaaaaaaaaaaaaaaa",
+	      "bounce_description": "bad_mailbox",
+	      "bgtools_code": 10,
+	      "diag": "smtp;550 5.1.1 The email account that you tried to reach does not exist. Please try double-checking the recipient's email address for typos or unnecessary spaces."
+	    },
+	    "_id": "id2",
+	    "ts": 1384954004
+	}`
+}

plugins/inputs/webhooks/mandrill/mandrill_webhooks_test.go

@@ -0,0 +1,85 @@
+package mandrill
+
+import (
+	"github.com/influxdata/telegraf/testutil"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+)
+
+func postWebhooks(md *MandrillWebhook, eventBody string) *httptest.ResponseRecorder {
+	body := url.Values{}
+	body.Set("mandrill_events", eventBody)
+	req, _ := http.NewRequest("POST", "/mandrill", strings.NewReader(body.Encode()))
+	w := httptest.NewRecorder()
+
+	md.eventHandler(w, req)
+
+	return w
+}
+
+func headRequest(md *MandrillWebhook) *httptest.ResponseRecorder {
+	req, _ := http.NewRequest("HEAD", "/mandrill", strings.NewReader(""))
+	w := httptest.NewRecorder()
+
+	md.returnOK(w, req)
+
+	return w
+}
+
+func TestHead(t *testing.T) {
+	md := &MandrillWebhook{Path: "/mandrill"}
+	resp := headRequest(md)
+	if resp.Code != http.StatusOK {
+		t.Errorf("HEAD returned HTTP status code %v.\nExpected %v", resp.Code, http.StatusOK)
+	}
+}
+
+func TestSendEvent(t *testing.T) {
+	var acc testutil.Accumulator
+	md := &MandrillWebhook{Path: "/mandrill", acc: &acc}
+	resp := postWebhooks(md, "["+SendEventJSON()+"]")
+	if resp.Code != http.StatusOK {
+		t.Errorf("POST send returned HTTP status code %v.\nExpected %v", resp.Code, http.StatusOK)
+	}
+
+	fields := map[string]interface{}{
+		"id": "id1",
+	}
+
+	tags := map[string]string{
+		"event": "send",
+	}
+
+	acc.AssertContainsTaggedFields(t, "mandrill_webhooks", fields, tags)
+}
+
+func TestMultipleEvents(t *testing.T) {
+	var acc testutil.Accumulator
+	md := &MandrillWebhook{Path: "/mandrill", acc: &acc}
+	resp := postWebhooks(md, "["+SendEventJSON()+","+HardBounceEventJSON()+"]")
+	if resp.Code != http.StatusOK {
+		t.Errorf("POST send returned HTTP status code %v.\nExpected %v", resp.Code, http.StatusOK)
+	}
+
+	fields := map[string]interface{}{
+		"id": "id1",
+	}
+
+	tags := map[string]string{
+		"event": "send",
+	}
+
+	acc.AssertContainsTaggedFields(t, "mandrill_webhooks", fields, tags)
+
+	fields = map[string]interface{}{
+		"id": "id2",
+	}
+
+	tags = map[string]string{
+		"event": "hard_bounce",
+	}
+	acc.AssertContainsTaggedFields(t, "mandrill_webhooks", fields, tags)
+}

plugins/inputs/webhooks/webhooks.go

@@ -11,6 +11,7 @@ import (
 	"github.com/influxdata/telegraf/plugins/inputs"
 
 	"github.com/influxdata/telegraf/plugins/inputs/webhooks/github"
+	"github.com/influxdata/telegraf/plugins/inputs/webhooks/mandrill"
 	"github.com/influxdata/telegraf/plugins/inputs/webhooks/rollbar"
 )
 
@@ -25,8 +26,9 @@ func init() {
 type Webhooks struct {
 	ServiceAddress string
 
-	Github  *github.GithubWebhook
-	Rollbar *rollbar.RollbarWebhook
+	Github   *github.GithubWebhook
+	Mandrill *mandrill.MandrillWebhook
+	Rollbar  *rollbar.RollbarWebhook
 }
 
 func NewWebhooks() *Webhooks {
@@ -41,6 +43,9 @@ func (wb *Webhooks) SampleConfig() string {
   [inputs.webhooks.github]
     path = "/github"
 
+  [inputs.webhooks.mandrill]
+    path = "/mandrill"
+
   [inputs.webhooks.rollbar]
     path = "/rollbar"
  `