dom
/
telegraf
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Webhooks plugin: add mandrill (#1408) 

* Add mandrill webhook.

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

Signed-off-by: Cyril Duez <cyril@stormz.me>
Signed-off-by: François de Metz <francois@stormz.me>

* Decode body to get the mandrill_events.

Signed-off-by: Cyril Duez <cyril@stormz.me>
Signed-off-by: François de Metz <francois@stormz.me>

* Handle HEAD request.

Signed-off-by: Cyril Duez <cyril@stormz.me>
Signed-off-by: François de Metz <francois@stormz.me>

* Add the README.

Signed-off-by: Cyril Duez <cyril@stormz.me>
Signed-off-by: François de Metz <francois@stormz.me>

* Add mandrill_webhooks to the README.

Signed-off-by: Cyril Duez <cyril@stormz.me>
Signed-off-by: François de Metz <francois@stormz.me>

* Update changelog.

Signed-off-by: Cyril Duez <cyril@stormz.me>
Signed-off-by: François de Metz <francois@stormz.me>

* Run gofmt.

Signed-off-by: Cyril Duez <cyril@stormz.me>
Signed-off-by: François de Metz <francois@stormz.me>
This commit is contained in:
François de Metz 2016-07-18 13:41:13 +02:00 committed by Cameron Sparr
parent 5dc4cce157
commit 1c2965703d
9 changed files with 248 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@ -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.

1
README.md
View File

@ -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)

1
plugins/inputs/webhooks/README.md
View File

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

15
plugins/inputs/webhooks/mandrill/README.md Normal file
View File

@ -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

56
plugins/inputs/webhooks/mandrill/mandrill_webhooks.go Normal file
View File

@ -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)
}

24
plugins/inputs/webhooks/mandrill/mandrill_webhooks_events.go Normal file
View File

@ -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,
}
}

58
plugins/inputs/webhooks/mandrill/mandrill_webhooks_events_json_test.go Normal file
View File

@ -0,0 +1,58 @@
package mandrill
func SendEventJSON() string {
return `
{
"event": "send",
"msg": {
"ts": 1365109999,
"subject": "This an example webhook message",
"email": "example.webhook@mandrillapp.com",
"sender": "example.sender@mandrillapp.com",
"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": "example.webhook@mandrillapp.com",
"sender": "example.sender@mandrillapp.com",
"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
}`
}

85
plugins/inputs/webhooks/mandrill/mandrill_webhooks_test.go Normal file
View File

@ -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)
}

9
plugins/inputs/webhooks/webhooks.go
View File

@ -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"
`