Add Particle Webhook Plugin

Godeps

@@ -23,6 +23,7 @@ github.com/golang/snappy 427fb6fc07997f43afa32f35e850833760e489a7
 github.com/gonuts/go-shellquote e842a11b24c6abfb3dd27af69a17f482e4b483c2
 github.com/gorilla/context 1ea25387ff6f684839d82767c1733ff4d4d15d0a
 github.com/gorilla/mux c9e326e2bdec29039a3761c07bece13133863e1e
+github.com/gorilla/schema 8aac656cd31cfb73a9dfd142494864fa0b84f723
 github.com/hailocab/go-hostpool e80d13ce29ede4452c43dea11e79b9bc8a15b478
 github.com/hashicorp/consul 5aa90455ce78d4d41578bafc86305e6e6b28d7d2
 github.com/hpcloud/tail b2940955ab8b26e19d43a43c4da0475dd81bdb56

Godeps_windows

@@ -24,6 +24,7 @@ github.com/golang/snappy 427fb6fc07997f43afa32f35e850833760e489a7
 github.com/gonuts/go-shellquote e842a11b24c6abfb3dd27af69a17f482e4b483c2
 github.com/gorilla/context 1ea25387ff6f684839d82767c1733ff4d4d15d0a
 github.com/gorilla/mux c9e326e2bdec29039a3761c07bece13133863e1e
+github.com/gorilla/schema 8aac656cd31cfb73a9dfd142494864fa0b84f723
 github.com/hailocab/go-hostpool e80d13ce29ede4452c43dea11e79b9bc8a15b478
 github.com/influxdata/config b79f6829346b8d6e78ba73544b1e1038f1f1c9da
 github.com/influxdata/influxdb e3fef5593c21644f2b43af55d6e17e70910b0e48

plugins/inputs/webhooks/particle/README.md

@@ -0,0 +1,52 @@
+# Particle Webhook
+[Particle](https://www.particle.io) Webhook Plug-in for Telegraf
+
+Particle events can be sent to Webhooks. Webhooks are configured using [Integrations](https://dashboard.particle.io/user/integrations).
+Assuming your Particle publishes an event called MY_EVENT, you could configure an Integration with:
+
+1. "Event Name" set to MY_EVENT
+2. "URL" set to http://MY_IP:1619/particle
+3. Leave "Request Type" set to "POST"
+4. Leave "Device" set to "Any"
+
+Replace MY_EVENT and MY_IP appropriately.
+
+Then click "Create Webhook".
+
+You may watch the stream of Particle events including the hook-sent/MY_EVENT and hook-response/MY_EVENT entries in the [Logs](https://dashboard.particle.io/user/logs)
+
+## Particle
+
+See Particle [Webhooks](https://docs.particle.io/guide/tools-and-features/webhooks/) documentation.
+
+The default data is:
+
+```
+{
+    "event": MY_EVENT,
+    "data": MY_EVENT_DATA,
+    "published_at": MY_EVENT_TIMESTAMP,
+    "coreid": DEVICE_ID
+}
+```
+
+The following Particle (trivial) sample publishes a random number as an event called "randomnumber" every 10 seconds:
+
+```
+void loop() {
+    Particle.publish("randomnumber", String(random(1000)), PRIVATE);
+    delay(10000);
+}
+```
+
+## Events
+
+**Tags:**
+* 'event' = `event` string
+* 'coreid' = `coreid` string
+
+**Fields:**
+* 'data' = `data` int
+
+**Time:**
+* 'published' = `published_at` time.Time ([ISO-8601](https://en.wikipedia.org/wiki/ISO_8601))

plugins/inputs/webhooks/particle/particle_webhooks.go

@@ -0,0 +1,49 @@
+package particle
+
+import (
+	"log"
+	"net/http"
+
+	"github.com/gorilla/mux"
+	"github.com/gorilla/schema"
+	"github.com/influxdata/telegraf"
+)
+
+type ParticleWebhook struct {
+	Path string
+	acc  telegraf.Accumulator
+}
+
+var decoder = schema.NewDecoder()
+
+func (pwh *ParticleWebhook) Register(router *mux.Router, acc telegraf.Accumulator) {
+	router.HandleFunc(pwh.Path, pwh.eventHandler).Methods("POST")
+	log.Printf("Started '%s' on %s\n", meas, pwh.Path)
+	pwh.acc = acc
+}
+
+func (pwh *ParticleWebhook) eventHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+
+	if 	err := r.ParseForm(); err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+	e, err := NewEvent(r, &ParticleEvent{})
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	p := e.NewMetric()
+	pwh.acc.AddFields(meas, p.Fields(), p.Tags(), p.Time())
+
+	w.WriteHeader(http.StatusOK)
+}
+
+func NewEvent(r *http.Request, event Event) (Event, error) {
+	if err := decoder.Decode(event, r.PostForm); err != nil {
+		return nil, err
+	}
+	return event, nil
+}

plugins/inputs/webhooks/particle/particle_webhooks_mocks.go

@@ -0,0 +1,31 @@
+package particle
+
+import (
+    "fmt"
+    "math/rand"
+    "net/url"
+    "time"
+)
+
+const hexBytes = "0123456789abcdef"
+
+func init() {
+    rand.Seed(time.Now().UnixNano())
+}
+
+func RandStringBytes(n int) string {
+    b := make([]byte, n)
+    for i := range b {
+        b[i] = hexBytes[rand.Intn(len(hexBytes))]
+    }
+    return string(b)
+}
+
+func NewEventURLEncoded() string {
+    rand.Seed(time.Now().UnixNano())
+    return fmt.Sprintf("event=%v&data=%v&published_at=%v&coreid=%v",
+        "event",
+        rand.Intn(1000),
+        url.QueryEscape(time.Now().Format(time.RFC3339)),
+        RandStringBytes(24))
+}

plugins/inputs/webhooks/particle/particle_webhooks_models.go

@@ -0,0 +1,51 @@
+package particle
+
+import (
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/influxdata/telegraf"
+)
+
+const meas = "particle"
+
+type Event interface {
+	NewMetric() telegraf.Metric
+}
+
+type ParticleEvent struct {
+  Event       string	  `schema:"event"`
+  Data	      int	      `schema:"data"`
+  PublishedAt time.Time	`schema:"published_at"`
+  CoreID	    string	  `schema:"coreid"`
+}
+
+func (pe ParticleEvent) String() string {
+  return fmt.Sprintf(`
+  Event == {
+    event: %v,
+    data: %v,
+    published: %v,
+    coreid: %v
+  }`,
+  pe.Event,
+  pe.Data,
+  pe.PublishedAt,
+  pe.CoreID)
+}
+
+func (pe ParticleEvent) NewMetric() telegraf.Metric {
+  t := map[string]string{
+    "event": pe.Event,
+    "coreid": pe.CoreID,
+  }
+  f := map[string]interface{}{
+    "data": pe.Data,
+  }
+  m, err := telegraf.NewMetric(pe.Event, t, f, pe.PublishedAt)
+  if err != nil {
+    log.Fatalf("Failed to create %v event", meas)
+  }
+  return m
+}

plugins/inputs/webhooks/particle/particle_webhooks_test.go

@@ -0,0 +1,24 @@
+package particle
+
+import (
+  "net/http"
+  "net/http/httptest"
+  "testing"
+
+  "github.com/influxdata/telegraf/testutil"
+)
+
+func ParticleWebhookRequest(urlEncodedString string, t *testing.T) {
+	var acc testutil.Accumulator
+	pwh:= &ParticleWebhook{Path: "/particle", acc: &acc}
+	req, _ := http.NewRequest("POST", "/particle", urlEncodedString)
+	w := httptest.NewRecorder()
+	pwh.eventHandler(w, req)
+	if w.Code != http.StatusOK {
+		t.Errorf("POST "+event+" returned HTTP status code %v.\nExpected %v", w.Code, http.StatusOK)
+	}
+}
+
+func TestNewEvent(t *testing.T) {
+    ParticleWebhookRequest(NewEventURLEncoded())
+}

plugins/inputs/webhooks/webhooks.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/influxdata/telegraf/plugins/inputs/webhooks/github"
 	"github.com/influxdata/telegraf/plugins/inputs/webhooks/mandrill"
+	"github.com/influxdata/telegraf/plugins/inputs/webhooks/particle"
 	"github.com/influxdata/telegraf/plugins/inputs/webhooks/rollbar"
 )
 
@@ -28,6 +29,7 @@ type Webhooks struct {
 
 	Github   *github.GithubWebhook
 	Mandrill *mandrill.MandrillWebhook
+	Particle *particle.ParticleWebhook
 	Rollbar  *rollbar.RollbarWebhook
 }
 
@@ -46,6 +48,9 @@ func (wb *Webhooks) SampleConfig() string {
   [inputs.webhooks.mandrill]
     path = "/mandrill"
 
+  [inputs.webhooks.particle]
+    path = "/particle"
+
   [inputs.webhooks.rollbar]
     path = "/rollbar"
  `

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/particle"
 	"github.com/influxdata/telegraf/plugins/inputs/webhooks/rollbar"
 )
 
@@ -21,6 +22,12 @@ func TestAvailableWebhooks(t *testing.T) {
 		t.Errorf("expected to be %v.\nGot %v", expected, wb.AvailableWebhooks())
 	}
 
+	wb.Particle = &particle.ParticleWebhook{Path: "/particle"}
+	expected = append(expected, wb.Particle)
+	if !reflect.DeepEqual(wb.AvailableWebhooks(), expected) {
+		t.Errorf("expected to be %v.\nGot %v", expected, wb.AvailableWebhooks())
+	}
+
 	wb.Rollbar = &rollbar.RollbarWebhook{Path: "/rollbar"}
 	expected = append(expected, wb.Rollbar)
 	if !reflect.DeepEqual(wb.AvailableWebhooks(), expected) {