Add a papertrail webhook input.

plugins/inputs/webhooks/papertrail/README.md

@@ -0,0 +1,33 @@
+# papertrail webhooks
+
+Enables Telegraf to act as a [Papertrail Webhook](http://help.papertrailapp.com/kb/how-it-works/web-hooks/).
+
+## Events
+
+See [here](http://help.papertrailapp.com/kb/how-it-works/web-hooks/#callback) for full documentation.
+
+Events from Papertrail come in two forms:
+
+* The event-based callback (shown in the example
+  [here](http://help.papertrailapp.com/kb/how-it-works/web-hooks/#callback)):
+
+  * A point is created per event, with the timestamp as "received_at"
+  * Each point has a field counter (`count`), which is set to 1 (signifying the event occurred)
+  * Each event "hostname" object is converted to a `host` tag
+  * The "saved_search" name in the payload is added as an `event` tag
+
+* The count-based callback (described [here](http://help.papertrailapp.com/kb/how-it-works/web-hooks/#count-only-webhooks))
+
+  * A point is created per timeseries object per count, with the timestamp as the "timeseries" key (the unix epoch of the event)
+  * Each point has a field counter (`count`), which is set to the value of each "timeseries" object
+  * Each count "source_name" object is converted to a `host` tag
+  * The "saved_search" name in the payload is added as an `event` tag
+
+The current functionality is very basic, however this allows you to
+track the number of events by host and saved search.
+
+When an event is received, any point will look similar to:
+
+```
+papertrail,host=myserver.example.com,event=saved_search_name count=3 1453248892
+```

plugins/inputs/webhooks/papertrail/papertrail_test.go

@@ -0,0 +1,144 @@
+package papertrail
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+
+	"github.com/influxdata/telegraf/testutil"
+)
+
+func postWebhooks(pt *PapertrailWebhook, payloadBody string) *httptest.ResponseRecorder {
+	req, _ := http.NewRequest("POST", "/", strings.NewReader(payloadBody))
+	w := httptest.NewRecorder()
+	w.Code = 500
+	pt.eventHandler(w, req)
+
+	return w
+}
+
+func TestEventPayload(t *testing.T) {
+	var acc testutil.Accumulator
+	pt := &PapertrailWebhook{Path: "/papertrail", acc: &acc}
+	payload := url.QueryEscape(sampleEventPayload)
+	resp := postWebhooks(pt, payload)
+	if resp.Code != http.StatusOK {
+		t.Errorf("POST new_item returned HTTP status code %v.\nExpected %v", resp.Code, http.StatusOK)
+	}
+
+	fields := map[string]interface{}{
+		"count": 1,
+	}
+
+	tags1 := map[string]string{
+		"event": "Important stuff",
+		"host":  "abc",
+	}
+	tags2 := map[string]string{
+		"event": "Important stuff",
+		"host":  "def",
+	}
+
+	t.Logf("%v", acc.Metrics)
+	acc.AssertContainsTaggedFields(t, "papertrail", fields, tags1)
+	acc.AssertContainsTaggedFields(t, "papertrail", fields, tags2)
+}
+
+func TestCountPayload(t *testing.T) {
+	var acc testutil.Accumulator
+	pt := &PapertrailWebhook{Path: "/papertrail", acc: &acc}
+	payload := url.QueryEscape(sampleCountPayload)
+	resp := postWebhooks(pt, payload)
+	if resp.Code != http.StatusOK {
+		t.Errorf("POST new_item returned HTTP status code %v.\nExpected %v", resp.Code, http.StatusOK)
+	}
+
+	fields1 := map[string]interface{}{
+		"count": 5,
+	}
+	fields2 := map[string]interface{}{
+		"count": 3,
+	}
+
+	tags1 := map[string]string{
+		"event": "Important stuff",
+		"host":  "arthur",
+	}
+	tags2 := map[string]string{
+		"event": "Important stuff",
+		"host":  "ford",
+	}
+
+	acc.AssertContainsTaggedFields(t, "papertrail", fields1, tags1)
+	acc.AssertContainsTaggedFields(t, "papertrail", fields2, tags2)
+}
+
+const sampleEventPayload = `payload={
+  "events": [
+    {
+      "id": 7711561783320576,
+      "received_at": "2011-05-18T20:30:02-07:00",
+      "display_received_at": "May 18 20:30:02",
+      "source_ip": "208.75.57.121",
+      "source_name": "abc",
+      "source_id": 2,
+      "hostname": "abc",
+      "program": "CROND",
+      "severity": "Info",
+      "facility": "Cron",
+      "message": "message body"
+    },
+    {
+      "id": 7711562567655424,
+      "received_at": "2011-05-18T20:30:02-07:00",
+      "display_received_at": "May 18 20:30:02",
+      "source_ip": "208.75.57.120",
+      "source_name": "server1",
+      "source_id": 19,
+      "hostname": "def",
+      "program": "CROND",
+      "severity": "Info",
+      "facility": "Cron",
+      "message": "A short event"
+    }
+  ],
+  "saved_search": {
+    "id": 42,
+    "name": "Important stuff",
+    "query": "cron OR server1",
+    "html_edit_url": "https://papertrailapp.com/searches/42/edit",
+    "html_search_url": "https://papertrailapp.com/searches/42"
+  },
+  "max_id": "7711582041804800",
+  "min_id": "7711561783320576"
+}`
+
+const sampleCountPayload = `payload={
+   "counts": [
+     {
+       "source_name": "arthur",
+       "source_id": 4,
+       "timeseries": {
+         "1453248895": 5
+       }
+     },
+     {
+       "source_name": "ford",
+       "source_id": 3,
+       "timeseries": {
+         "1453248927": 3
+       }
+     }
+   ],
+   "saved_search": {
+     "id": 42,
+     "name": "Important stuff",
+     "query": "cron OR server1",
+     "html_edit_url": "https://papertrailapp.com/searches/42/edit",
+     "html_search_url": "https://papertrailapp.com/searches/42"
+   },
+   "max_id": "7711582041804800",
+   "min_id": "7711561783320576"
+}`

plugins/inputs/webhooks/papertrail/papertrail_webhooks.go

@@ -0,0 +1,94 @@
+package papertrail
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/influxdata/telegraf"
+)
+
+type PapertrailWebhook struct {
+	Path string
+	acc  telegraf.Accumulator
+}
+
+func (pt *PapertrailWebhook) Register(router *mux.Router, acc telegraf.Accumulator) {
+	router.HandleFunc(pt.Path, pt.eventHandler).Methods("POST")
+	log.Printf("I! Started the papertrail_webhook on %s\n", pt.Path)
+	pt.acc = acc
+}
+
+func (pt *PapertrailWebhook) eventHandler(w http.ResponseWriter, r *http.Request) {
+	if r.Method != "POST" {
+		http.NotFound(w, r)
+	}
+
+	defer r.Body.Close()
+	reqBody, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		http.Error(w, "Invalid request", 400)
+		return
+	}
+
+	data, err := url.QueryUnescape(string(reqBody))
+	if err != nil {
+		http.Error(w, "Invalid request", 400)
+		return
+	}
+
+	var payload Payload
+	// JSON payload is x-www-form-urlencoded, remove this string when unmarshaling
+	remove := "payload="
+	if len(data) > 0 && data[0:len(remove)] == remove {
+		err = json.Unmarshal([]byte(data[len(remove):len(data)]), &payload)
+		if err != nil {
+			http.Error(w, "Unable to parse request body", 400)
+			return
+		}
+	} else {
+		http.Error(w, "Invalid request", 400)
+		return
+	}
+
+	if payload.Events != nil {
+
+		// Handle event-based payload
+		for _, e := range payload.Events {
+			// FIXME: Duplicate event timestamps will overwrite each other
+			tags := map[string]string{
+				"host":  e.Hostname,
+				"event": payload.SavedSearch.Name,
+			}
+			fields := map[string]interface{}{
+				"count": 1,
+			}
+			pt.acc.AddFields("papertrail", fields, tags, e.ReceivedAt)
+		}
+
+	} else if payload.Counts != nil {
+
+		// Handle count-based payload
+		for _, c := range payload.Counts {
+			for ts, count := range *c.TimeSeries {
+				tags := map[string]string{
+					"host":  c.SourceName,
+					"event": payload.SavedSearch.Name,
+				}
+				fields := map[string]interface{}{
+					"count": count,
+				}
+				pt.acc.AddFields("papertrail", fields, tags, time.Unix(int64(ts), 0))
+			}
+		}
+	} else {
+		http.Error(w, "Invalid request", 400)
+		return
+	}
+
+	w.WriteHeader(http.StatusOK)
+}

plugins/inputs/webhooks/papertrail/papertrail_webhooks_models.go

@@ -0,0 +1,41 @@
+package papertrail
+
+import (
+	"time"
+)
+
+type Event struct {
+	ID                int64     `json:"id"`
+	ReceivedAt        time.Time `json:"received_at"`
+	DisplayReceivedAt string    `json:"display_received_at"`
+	SourceIP          string    `json:"source_ip"`
+	SourceName        string    `json:"source_name"`
+	SourceID          int       `json:"source_id"`
+	Hostname          string    `json:"hostname"`
+	Program           string    `json:"program"`
+	Severity          string    `json:"severity"`
+	Facility          string    `json:"facility"`
+	Message           string    `json:"message"`
+}
+
+type Count struct {
+	SourceName string       `json:"source_name"`
+	SourceID   int64        `json:"source_id"`
+	TimeSeries *map[int]int `json:"timeseries"`
+}
+
+type SavedSearch struct {
+	ID        int64  `json:"id"`
+	Name      string `json:"name"`
+	Query     string `json:"query"`
+	EditURL   string `json:"html_edit_url"`
+	SearchURL string `json:"html_search_url"`
+}
+
+type Payload struct {
+	Events      []*Event     `json:"events"`
+	Counts      []*Count     `json:"counts"`
+	SavedSearch *SavedSearch `json:"saved_search"`
+	MaxID       string       `json:"max_id"`
+	MinID       string       `json:"min_id"`
+}

plugins/inputs/webhooks/webhooks.go

@@ -13,6 +13,7 @@ import (
 	"github.com/influxdata/telegraf/plugins/inputs/webhooks/filestack"
 	"github.com/influxdata/telegraf/plugins/inputs/webhooks/github"
 	"github.com/influxdata/telegraf/plugins/inputs/webhooks/mandrill"
+	"github.com/influxdata/telegraf/plugins/inputs/webhooks/papertrail"
 	"github.com/influxdata/telegraf/plugins/inputs/webhooks/rollbar"
 )
 
@@ -27,10 +28,11 @@ func init() {
 type Webhooks struct {
 	ServiceAddress string
 
-	Github    *github.GithubWebhook
-	Filestack *filestack.FilestackWebhook
-	Mandrill  *mandrill.MandrillWebhook
-	Rollbar   *rollbar.RollbarWebhook
+	Github     *github.GithubWebhook
+	Filestack  *filestack.FilestackWebhook
+	Mandrill   *mandrill.MandrillWebhook
+	Rollbar    *rollbar.RollbarWebhook
+	Papertrail *papertrail.PapertrailWebhook
 }
 
 func NewWebhooks() *Webhooks {
@@ -53,6 +55,9 @@ func (wb *Webhooks) SampleConfig() string {
 
   [inputs.webhooks.rollbar]
     path = "/rollbar"
+
+  [inputs.webhooks.papertrail]
+    path = "/papertrail"
  `
 }
 

plugins/inputs/webhooks/webhooks_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	"github.com/influxdata/telegraf/plugins/inputs/webhooks/github"
+	"github.com/influxdata/telegraf/plugins/inputs/webhooks/papertrail"
 	"github.com/influxdata/telegraf/plugins/inputs/webhooks/rollbar"
 )
 
@@ -26,4 +27,10 @@ func TestAvailableWebhooks(t *testing.T) {
 	if !reflect.DeepEqual(wb.AvailableWebhooks(), expected) {
 		t.Errorf("expected to be %v.\nGot %v", expected, wb.AvailableWebhooks())
 	}
+
+	wb.Papertrail = &papertrail.PapertrailWebhook{Path: "/papertrail"}
+	expected = append(expected, wb.Papertrail)
+	if !reflect.DeepEqual(wb.AvailableWebhooks(), expected) {
+		t.Errorf("expected to be %v.\nGot %v", expected, wb.AvailableWebhooks())
+	}
 }