Add new webhooks plugin that superseed github and rollbar plugins.

closes #1289

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

Rename internals struct.

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

Update changelog.

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

Update READMEs and CHANGELOG.

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

Update SampleConfig.

Update the config format.

Update telegraf config.

Update the webhooks README.

Update changelog.

Update the changelog with an upgrade path.

Update default ports.

Fix indent.

Check for nil value on AvailableWebhooks.

Check for CanInterface.

plugins/inputs/all/all.go

@@ -19,7 +19,6 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/inputs/elasticsearch"
 	_ "github.com/influxdata/telegraf/plugins/inputs/exec"
 	_ "github.com/influxdata/telegraf/plugins/inputs/filestat"
-	_ "github.com/influxdata/telegraf/plugins/inputs/github_webhooks"
 	_ "github.com/influxdata/telegraf/plugins/inputs/graylog"
 	_ "github.com/influxdata/telegraf/plugins/inputs/haproxy"
 	_ "github.com/influxdata/telegraf/plugins/inputs/http_response"
@@ -57,7 +56,6 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/inputs/redis"
 	_ "github.com/influxdata/telegraf/plugins/inputs/rethinkdb"
 	_ "github.com/influxdata/telegraf/plugins/inputs/riak"
-	_ "github.com/influxdata/telegraf/plugins/inputs/rollbar_webhooks"
 	_ "github.com/influxdata/telegraf/plugins/inputs/sensors"
 	_ "github.com/influxdata/telegraf/plugins/inputs/snmp"
 	_ "github.com/influxdata/telegraf/plugins/inputs/sqlserver"
@@ -70,6 +68,7 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/inputs/twemproxy"
 	_ "github.com/influxdata/telegraf/plugins/inputs/udp_listener"
 	_ "github.com/influxdata/telegraf/plugins/inputs/varnish"
+	_ "github.com/influxdata/telegraf/plugins/inputs/webhooks"
 	_ "github.com/influxdata/telegraf/plugins/inputs/win_perf_counters"
 	_ "github.com/influxdata/telegraf/plugins/inputs/zfs"
 	_ "github.com/influxdata/telegraf/plugins/inputs/zookeeper"

plugins/inputs/rollbar_webhooks/rollbar_webhooks.go

@@ -1,119 +0,0 @@
-package rollbar_webhooks
-
-import (
-	"encoding/json"
-	"errors"
-	"fmt"
-	"io/ioutil"
-	"log"
-	"net/http"
-	"sync"
-	"time"
-
-	"github.com/gorilla/mux"
-	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/plugins/inputs"
-)
-
-func init() {
-	inputs.Add("rollbar_webhooks", func() telegraf.Input { return NewRollbarWebhooks() })
-}
-
-type RollbarWebhooks struct {
-	ServiceAddress string
-	// Lock for the struct
-	sync.Mutex
-	// Events buffer to store events between Gather calls
-	events []Event
-}
-
-func NewRollbarWebhooks() *RollbarWebhooks {
-	return &RollbarWebhooks{}
-}
-
-func (rb *RollbarWebhooks) SampleConfig() string {
-	return `
-  ## Address and port to host Webhook listener on
-  service_address = ":1619"
-`
-}
-
-func (rb *RollbarWebhooks) Description() string {
-	return "A Rollbar Webhook Event collector"
-}
-
-func (rb *RollbarWebhooks) Gather(acc telegraf.Accumulator) error {
-	rb.Lock()
-	defer rb.Unlock()
-	for _, event := range rb.events {
-		acc.AddFields("rollbar_webhooks", event.Fields(), event.Tags(), time.Now())
-	}
-	rb.events = make([]Event, 0)
-	return nil
-}
-
-func (rb *RollbarWebhooks) Listen() {
-	r := mux.NewRouter()
-	r.HandleFunc("/", rb.eventHandler).Methods("POST")
-	err := http.ListenAndServe(fmt.Sprintf("%s", rb.ServiceAddress), r)
-	if err != nil {
-		log.Printf("Error starting server: %v", err)
-	}
-}
-
-func (rb *RollbarWebhooks) Start(_ telegraf.Accumulator) error {
-	go rb.Listen()
-	log.Printf("Started the rollbar_webhooks service on %s\n", rb.ServiceAddress)
-	return nil
-}
-
-func (rb *RollbarWebhooks) Stop() {
-	log.Println("Stopping the rbWebhooks service")
-}
-
-func (rb *RollbarWebhooks) eventHandler(w http.ResponseWriter, r *http.Request) {
-	defer r.Body.Close()
-	data, err := ioutil.ReadAll(r.Body)
-	if err != nil {
-		w.WriteHeader(http.StatusBadRequest)
-		return
-	}
-
-	dummyEvent := &DummyEvent{}
-	err = json.Unmarshal(data, dummyEvent)
-	if err != nil {
-		w.WriteHeader(http.StatusBadRequest)
-		return
-	}
-
-	event, err := NewEvent(dummyEvent, data)
-	if err != nil {
-		w.WriteHeader(http.StatusOK)
-		return
-	}
-
-	rb.Lock()
-	rb.events = append(rb.events, event)
-	rb.Unlock()
-
-	w.WriteHeader(http.StatusOK)
-}
-
-func generateEvent(event Event, data []byte) (Event, error) {
-	err := json.Unmarshal(data, event)
-	if err != nil {
-		return nil, err
-	}
-	return event, nil
-}
-
-func NewEvent(dummyEvent *DummyEvent, data []byte) (Event, error) {
-	switch dummyEvent.EventName {
-	case "new_item":
-		return generateEvent(&NewItem{}, data)
-	case "deploy":
-		return generateEvent(&Deploy{}, data)
-	default:
-		return nil, errors.New("Not implemented type: " + dummyEvent.EventName)
-	}
-}

plugins/inputs/webhooks/README.md

@@ -0,0 +1,27 @@
+# Webhooks
+
+This is a Telegraf service plugin that start an http server and register multiple webhook listeners.
+
+```sh
+$ telegraf -sample-config -input-filter webhooks -output-filter influxdb > config.conf.new
+```
+
+Change the config file to point to the InfluxDB server you are using and adjust the settings to match your environment. Once that is complete:
+
+```sh
+$ cp config.conf.new /etc/telegraf/telegraf.conf
+$ sudo service telegraf start
+```
+
+## Available webhooks
+
+- [Github](github/)
+- [Rollbar](rollbar/)
+
+## Adding new webhooks plugin
+
+1. Add your webhook plugin inside the `webhooks` folder
+1. Your plugin must implement the `Webhook` interface
+1. Import your plugin in the `webhooks.go` file and add it to the `Webhooks` struct
+
+Both [Github](github/) and [Rollbar](rollbar/) are good example to follow.

plugins/inputs/webhooks/github/README.md

@@ -1,15 +1,6 @@
-# github_webhooks
+# github webhooks
 
-This is a Telegraf service plugin that listens for events kicked off by Github's Webhooks service and persists data from them into configured outputs. To set up the listener first generate the proper configuration:
-```sh
-$ telegraf -sample-config -input-filter github_webhooks -output-filter influxdb > config.conf.new
-```
-Change the config file to point to the InfluxDB server you are using and adjust the settings to match your environment. Once that is complete:
-```sh
-$ cp config.conf.new /etc/telegraf/telegraf.conf
-$ sudo service telegraf start
-```
-Once the server is running you should configure your Organization's Webhooks to point at the `github_webhooks` service. To do this go to `github.com/{my_organization}` and click `Settings > Webhooks > Add webhook`. In the resulting menu set `Payload URL` to `http://<my_ip>:1618`, `Content type` to `application/json` and under the section `Which events would you like to trigger this webhook?` select 'Send me <b>everything</b>'. By default all of the events will write to the `github_webhooks` measurement, this is configurable by setting the `measurement_name` in the config file.
+You should configure your Organization's Webhooks to point at the `webhooks` service. To do this go to `github.com/{my_organization}` and click `Settings > Webhooks > Add webhook`. In the resulting menu set `Payload URL` to `http://<my_ip>:1619/github`, `Content type` to `application/json` and under the section `Which events would you like to trigger this webhook?` select 'Send me <b>everything</b>'. By default all of the events will write to the `github_webhooks` measurement, this is configurable by setting the `measurement_name` in the config file.
 
 ## Events
 

plugins/inputs/webhooks/github/github_webhooks.go

@@ -1,78 +1,27 @@
-package github_webhooks
+package github
 
 import (
 	"encoding/json"
-	"fmt"
 	"io/ioutil"
 	"log"
 	"net/http"
-	"sync"
 
 	"github.com/gorilla/mux"
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
-func init() {
-	inputs.Add("github_webhooks", func() telegraf.Input { return &GithubWebhooks{} })
+type GithubWebhook struct {
+	Path string
+	acc  telegraf.Accumulator
 }
 
-type GithubWebhooks struct {
-	ServiceAddress string
-	// Lock for the struct
-	sync.Mutex
-	// Events buffer to store events between Gather calls
-	events []Event
+func (gh *GithubWebhook) Register(router *mux.Router, acc telegraf.Accumulator) {
+	router.HandleFunc(gh.Path, gh.eventHandler).Methods("POST")
+	log.Printf("Started the webhooks_github on %s\n", gh.Path)
+	gh.acc = acc
 }
 
-func NewGithubWebhooks() *GithubWebhooks {
-	return &GithubWebhooks{}
-}
-
-func (gh *GithubWebhooks) SampleConfig() string {
-	return `
-  ## Address and port to host Webhook listener on
-  service_address = ":1618"
-`
-}
-
-func (gh *GithubWebhooks) Description() string {
-	return "A Github Webhook Event collector"
-}
-
-// Writes the points from <-gh.in to the Accumulator
-func (gh *GithubWebhooks) Gather(acc telegraf.Accumulator) error {
-	gh.Lock()
-	defer gh.Unlock()
-	for _, event := range gh.events {
-		p := event.NewMetric()
-		acc.AddFields("github_webhooks", p.Fields(), p.Tags(), p.Time())
-	}
-	gh.events = make([]Event, 0)
-	return nil
-}
-
-func (gh *GithubWebhooks) Listen() {
-	r := mux.NewRouter()
-	r.HandleFunc("/", gh.eventHandler).Methods("POST")
-	err := http.ListenAndServe(fmt.Sprintf("%s", gh.ServiceAddress), r)
-	if err != nil {
-		log.Printf("Error starting server: %v", err)
-	}
-}
-
-func (gh *GithubWebhooks) Start(_ telegraf.Accumulator) error {
-	go gh.Listen()
-	log.Printf("Started the github_webhooks service on %s\n", gh.ServiceAddress)
-	return nil
-}
-
-func (gh *GithubWebhooks) Stop() {
-	log.Println("Stopping the ghWebhooks service")
-}
-
-// Handles the / route
-func (gh *GithubWebhooks) eventHandler(w http.ResponseWriter, r *http.Request) {
+func (gh *GithubWebhook) eventHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	eventType := r.Header["X-Github-Event"][0]
 	data, err := ioutil.ReadAll(r.Body)
@@ -85,9 +34,10 @@ func (gh *GithubWebhooks) eventHandler(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusBadRequest)
 		return
 	}
-	gh.Lock()
-	gh.events = append(gh.events, e)
-	gh.Unlock()
+
+	p := e.NewMetric()
+	gh.acc.AddFields("github_webhooks", p.Fields(), p.Tags(), p.Time())
+
 	w.WriteHeader(http.StatusOK)
 }
 

plugins/inputs/webhooks/github/github_webhooks_mock_json.go

@@ -1,4 +1,4 @@
-package github_webhooks
+package github
 
 func CommitCommentEventJSON() string {
 	return `{

plugins/inputs/webhooks/github/github_webhooks_models.go

@@ -1,4 +1,4 @@
-package github_webhooks
+package github
 
 import (
 	"fmt"

plugins/inputs/webhooks/github/github_webhooks_test.go

@@ -1,15 +1,18 @@
-package github_webhooks
+package github
 
 import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
 	"testing"
+
+	"github.com/influxdata/telegraf/testutil"
 )
 
 func GithubWebhookRequest(event string, jsonString string, t *testing.T) {
-	gh := NewGithubWebhooks()
-	req, _ := http.NewRequest("POST", "/", strings.NewReader(jsonString))
+	var acc testutil.Accumulator
+	gh := &GithubWebhook{Path: "/github", acc: &acc}
+	req, _ := http.NewRequest("POST", "/github", strings.NewReader(jsonString))
 	req.Header.Add("X-Github-Event", event)
 	w := httptest.NewRecorder()
 	gh.eventHandler(w, req)

plugins/inputs/webhooks/rollbar/README.md

@@ -1,15 +1,6 @@
-# rollbar_webhooks
+# rollbar webhooks
 
-This is a Telegraf service plugin that listens for events kicked off by Rollbar Webhooks service and persists data from them into configured outputs. To set up the listener first generate the proper configuration:
-```sh
-$ telegraf -sample-config -input-filter rollbar_webhooks -output-filter influxdb > config.conf.new
-```
-Change the config file to point to the InfluxDB server you are using and adjust the settings to match your environment. Once that is complete:
-```sh
-$ cp config.conf.new /etc/telegraf/telegraf.conf
-$ sudo service telegraf start
-```
-Once the server is running you should configure your Rollbar's Webhooks to point at the `rollbar_webhooks` service. To do this go to `rollbar.com/` and click `Settings > Notifications > Webhook`. In the resulting page set `URL` to `http://<my_ip>:1619`, and click on `Enable Webhook Integration`.
+You should configure your Rollbar's Webhooks to point at the `webhooks` service. To do this go to `rollbar.com/` and click `Settings > Notifications > Webhook`. In the resulting page set `URL` to `http://<my_ip>:1619/rollbar`, and click on `Enable Webhook Integration`.
 
 ## Events
 

plugins/inputs/webhooks/rollbar/rollbar_webhooks.go

@@ -0,0 +1,69 @@
+package rollbar
+
+import (
+	"encoding/json"
+	"errors"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/influxdata/telegraf"
+)
+
+type RollbarWebhook struct {
+	Path string
+	acc  telegraf.Accumulator
+}
+
+func (rb *RollbarWebhook) Register(router *mux.Router, acc telegraf.Accumulator) {
+	router.HandleFunc(rb.Path, rb.eventHandler).Methods("POST")
+	log.Printf("Started the webhooks_rollbar on %s\n", rb.Path)
+	rb.acc = acc
+}
+
+func (rb *RollbarWebhook) eventHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	data, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	dummyEvent := &DummyEvent{}
+	err = json.Unmarshal(data, dummyEvent)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	event, err := NewEvent(dummyEvent, data)
+	if err != nil {
+		w.WriteHeader(http.StatusOK)
+		return
+	}
+
+	rb.acc.AddFields("rollbar_webhooks", event.Fields(), event.Tags(), time.Now())
+
+	w.WriteHeader(http.StatusOK)
+}
+
+func generateEvent(event Event, data []byte) (Event, error) {
+	err := json.Unmarshal(data, event)
+	if err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func NewEvent(dummyEvent *DummyEvent, data []byte) (Event, error) {
+	switch dummyEvent.EventName {
+	case "new_item":
+		return generateEvent(&NewItem{}, data)
+	case "deploy":
+		return generateEvent(&Deploy{}, data)
+	default:
+		return nil, errors.New("Not implemented type: " + dummyEvent.EventName)
+	}
+}

plugins/inputs/webhooks/rollbar/rollbar_webhooks_events.go

@@ -1,4 +1,4 @@
-package rollbar_webhooks
+package rollbar
 
 import "strconv"
 

plugins/inputs/webhooks/rollbar/rollbar_webhooks_events_json_test.go

@@ -1,4 +1,4 @@
-package rollbar_webhooks
+package rollbar
 
 func NewItemJSON() string {
 	return `

plugins/inputs/webhooks/rollbar/rollbar_webhooks_test.go

@@ -1,4 +1,4 @@
-package rollbar_webhooks
+package rollbar
 
 import (
 	"net/http"
@@ -9,7 +9,7 @@ import (
 	"github.com/influxdata/telegraf/testutil"
 )
 
-func postWebhooks(rb *RollbarWebhooks, eventBody string) *httptest.ResponseRecorder {
+func postWebhooks(rb *RollbarWebhook, eventBody string) *httptest.ResponseRecorder {
 	req, _ := http.NewRequest("POST", "/", strings.NewReader(eventBody))
 	w := httptest.NewRecorder()
 	w.Code = 500
@@ -21,12 +21,11 @@ func postWebhooks(rb *RollbarWebhooks, eventBody string) *httptest.ResponseRecor
 
 func TestNewItem(t *testing.T) {
 	var acc testutil.Accumulator
-	rb := NewRollbarWebhooks()
+	rb := &RollbarWebhook{Path: "/rollbar", acc: &acc}
 	resp := postWebhooks(rb, NewItemJSON())
 	if resp.Code != http.StatusOK {
 		t.Errorf("POST new_item returned HTTP status code %v.\nExpected %v", resp.Code, http.StatusOK)
 	}
-	rb.Gather(&acc)
 
 	fields := map[string]interface{}{
 		"id": 272716944,
@@ -45,12 +44,11 @@ func TestNewItem(t *testing.T) {
 
 func TestDeploy(t *testing.T) {
 	var acc testutil.Accumulator
-	rb := NewRollbarWebhooks()
+	rb := &RollbarWebhook{Path: "/rollbar", acc: &acc}
 	resp := postWebhooks(rb, DeployJSON())
 	if resp.Code != http.StatusOK {
 		t.Errorf("POST deploy returned HTTP status code %v.\nExpected %v", resp.Code, http.StatusOK)
 	}
-	rb.Gather(&acc)
 
 	fields := map[string]interface{}{
 		"id": 187585,
@@ -66,7 +64,7 @@ func TestDeploy(t *testing.T) {
 }
 
 func TestUnknowItem(t *testing.T) {
-	rb := NewRollbarWebhooks()
+	rb := &RollbarWebhook{Path: "/rollbar"}
 	resp := postWebhooks(rb, UnknowJSON())
 	if resp.Code != http.StatusOK {
 		t.Errorf("POST unknow returned HTTP status code %v.\nExpected %v", resp.Code, http.StatusOK)

plugins/inputs/webhooks/webhooks.go

@@ -0,0 +1,99 @@
+package webhooks
+
+import (
+	"fmt"
+	"log"
+	"net/http"
+	"reflect"
+
+	"github.com/gorilla/mux"
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/inputs"
+
+	"github.com/influxdata/telegraf/plugins/inputs/webhooks/github"
+	"github.com/influxdata/telegraf/plugins/inputs/webhooks/rollbar"
+)
+
+type Webhook interface {
+	Register(router *mux.Router, acc telegraf.Accumulator)
+}
+
+func init() {
+	inputs.Add("webhooks", func() telegraf.Input { return NewWebhooks() })
+}
+
+type Webhooks struct {
+	ServiceAddress string
+
+	Github  *github.GithubWebhook
+	Rollbar *rollbar.RollbarWebhook
+}
+
+func NewWebhooks() *Webhooks {
+	return &Webhooks{}
+}
+
+func (wb *Webhooks) SampleConfig() string {
+	return `
+  ## Address and port to host Webhook listener on
+  service_address = ":1619"
+
+  [inputs.webhooks.github]
+    path = "/github"
+
+  [inputs.webhooks.rollbar]
+    path = "/rollbar"
+ `
+}
+
+func (wb *Webhooks) Description() string {
+	return "A Webhooks Event collector"
+}
+
+func (wb *Webhooks) Gather(_ telegraf.Accumulator) error {
+	return nil
+}
+
+func (wb *Webhooks) Listen(acc telegraf.Accumulator) {
+	r := mux.NewRouter()
+
+	for _, webhook := range wb.AvailableWebhooks() {
+		webhook.Register(r, acc)
+	}
+
+	err := http.ListenAndServe(fmt.Sprintf("%s", wb.ServiceAddress), r)
+	if err != nil {
+		log.Printf("Error starting server: %v", err)
+	}
+}
+
+// Looks for fields which implement Webhook interface
+func (wb *Webhooks) AvailableWebhooks() []Webhook {
+	webhooks := make([]Webhook, 0)
+	s := reflect.ValueOf(wb).Elem()
+	for i := 0; i < s.NumField(); i++ {
+		f := s.Field(i)
+
+		if !f.CanInterface() {
+			continue
+		}
+
+		if wbPlugin, ok := f.Interface().(Webhook); ok {
+			if !reflect.ValueOf(wbPlugin).IsNil() {
+				webhooks = append(webhooks, wbPlugin)
+			}
+		}
+	}
+
+	return webhooks
+}
+
+func (wb *Webhooks) Start(acc telegraf.Accumulator) error {
+	go wb.Listen(acc)
+	log.Printf("Started the webhooks service on %s\n", wb.ServiceAddress)
+	return nil
+}
+
+func (rb *Webhooks) Stop() {
+	log.Println("Stopping the Webhooks service")
+}

plugins/inputs/webhooks/webhooks_test.go

@@ -0,0 +1,29 @@
+package webhooks
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/influxdata/telegraf/plugins/inputs/webhooks/github"
+	"github.com/influxdata/telegraf/plugins/inputs/webhooks/rollbar"
+)
+
+func TestAvailableWebhooks(t *testing.T) {
+	wb := NewWebhooks()
+	expected := make([]Webhook, 0)
+	if !reflect.DeepEqual(wb.AvailableWebhooks(), expected) {
+		t.Errorf("expected to %v.\nGot %v", expected, wb.AvailableWebhooks())
+	}
+
+	wb.Github = &github.GithubWebhook{Path: "/github"}
+	expected = append(expected, wb.Github)
+	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) {
+		t.Errorf("expected to be %v.\nGot %v", expected, wb.AvailableWebhooks())
+	}
+}