Add HTTP listener service input plugin

Godeps

@@ -25,6 +25,7 @@ github.com/gorilla/mux c9e326e2bdec29039a3761c07bece13133863e1e
 github.com/hailocab/go-hostpool e80d13ce29ede4452c43dea11e79b9bc8a15b478
 github.com/hashicorp/consul 5aa90455ce78d4d41578bafc86305e6e6b28d7d2
 github.com/hpcloud/tail b2940955ab8b26e19d43a43c4da0475dd81bdb56
+github.com/hydrogen18/stoppableListener dadc9ccc400c712e5a316107a5c462863919e579
 github.com/influxdata/config b79f6829346b8d6e78ba73544b1e1038f1f1c9da
 github.com/influxdata/influxdb e094138084855d444195b252314dfee9eae34cab
 github.com/influxdata/toml af4df43894b16e3fd2b788d01bd27ad0776ef2d0

README.md

@@ -214,6 +214,7 @@ Telegraf can also collect metrics via the following service plugins:
 * [tail](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/tail)
 * [udp_listener](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/udp_listener)
 * [tcp_listener](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/tcp_listener)
+* [http_listener](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/http_listener)
 * [mqtt_consumer](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/mqtt_consumer)
 * [kafka_consumer](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/kafka_consumer)
 * [nats_consumer](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/nats_consumer)

plugins/inputs/all/all.go

@@ -21,6 +21,7 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/inputs/filestat"
 	_ "github.com/influxdata/telegraf/plugins/inputs/graylog"
 	_ "github.com/influxdata/telegraf/plugins/inputs/haproxy"
+	_ "github.com/influxdata/telegraf/plugins/inputs/http_listener"
 	_ "github.com/influxdata/telegraf/plugins/inputs/http_response"
 	_ "github.com/influxdata/telegraf/plugins/inputs/httpjson"
 	_ "github.com/influxdata/telegraf/plugins/inputs/influxdb"

plugins/inputs/http_listener/README.md

@@ -0,0 +1,23 @@
+# HTTP listener service input plugin
+
+The HTTP listener is a service input plugin that listens for messages sent via HTTP POST.
+The plugin expects messages in the InfluxDB line-protocol ONLY, other Telegraf input data formats are not supported.
+The intent of the plugin is to allow Telegraf to serve as a proxy/router for the /write endpoint of the InfluxDB HTTP API.
+
+See: [Telegraf Input Data Formats](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md#influx).
+Example:  curl -i -XPOST 'http://localhost:8186/write' --data-binary 'cpu_load_short,host=server01,region=us-west value=0.64 1434055562000000000'
+
+### Configuration:
+
+This is a sample configuration for the plugin.
+
+```toml
+# # Influx HTTP write listener
+[[inputs.http_listener]]
+  ## Address and port to host HTTP listener on
+  service_address = ":8186"
+
+  ## timeouts in seconds
+  read_timeout = "10"
+  write_timeout = "10"
+```

plugins/inputs/http_listener/http_listener.go

@@ -0,0 +1,152 @@
+package http_listener
+
+import (
+	"log"
+	"net"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/inputs"
+	"github.com/influxdata/telegraf/plugins/parsers"
+	"io/ioutil"
+
+	"github.com/hydrogen18/stoppableListener"
+	"strconv"
+)
+
+type HttpListener struct {
+	ServiceAddress string
+	ReadTimeout    string
+	WriteTimeout   string
+
+	sync.Mutex
+
+	listener *stoppableListener.StoppableListener
+
+	parser parsers.Parser
+	acc    telegraf.Accumulator
+}
+
+const sampleConfig = `
+  ## Address and port to host HTTP listener on
+  service_address = ":8186"
+
+  ## timeouts in seconds
+  read_timeout = "10"
+  write_timeout = "10"
+`
+
+func (t *HttpListener) SampleConfig() string {
+	return sampleConfig
+}
+
+func (t *HttpListener) Description() string {
+	return "Influx HTTP write listener"
+}
+
+func (t *HttpListener) Gather(_ telegraf.Accumulator) error {
+	return nil
+}
+
+func (t *HttpListener) SetParser(parser parsers.Parser) {
+	t.parser = parser
+}
+
+// Start starts the http listener service.
+func (t *HttpListener) Start(acc telegraf.Accumulator) error {
+	t.Lock()
+	defer t.Unlock()
+
+	t.acc = acc
+
+	var rawListener, err = net.Listen("tcp", t.ServiceAddress)
+	t.listener, err = stoppableListener.New(rawListener)
+
+	go t.httpListen()
+
+	log.Printf("Started HTTP listener service on %s\n", t.ServiceAddress)
+
+	return err
+}
+
+// Stop cleans up all resources
+func (t *HttpListener) Stop() {
+	t.Lock()
+	defer t.Unlock()
+
+	t.listener.Stop()
+	t.listener.Close()
+
+	log.Println("Stopped HTTP listener service on ", t.ServiceAddress)
+}
+
+// httpListen listens for HTTP requests.
+func (t *HttpListener) httpListen() error {
+
+	readTimeout, err := strconv.ParseInt(t.ReadTimeout, 10, 32)
+	writeTimeout, err := strconv.ParseInt(t.WriteTimeout, 10, 32)
+
+	var server = http.Server{
+		Handler:      t,
+		ReadTimeout:  time.Duration(readTimeout) * time.Second,
+		WriteTimeout: time.Duration(writeTimeout) * time.Second,
+	}
+
+	err = server.Serve(t.listener)
+
+	return err
+}
+
+func (t *HttpListener) ServeHTTP(res http.ResponseWriter, req *http.Request) {
+	body, err := ioutil.ReadAll(req.Body)
+
+	if err == nil {
+		var path = req.URL.Path[1:]
+
+		if path == "write" {
+			var metrics []telegraf.Metric
+			metrics, err = t.parser.Parse(body)
+			if err == nil {
+				t.storeMetrics(metrics)
+			} else {
+				log.Printf("Problem parsing body: [%s], Error: %s\n", string(body), err)
+				res.WriteHeader(500)
+				res.Write([]byte("ERROR parsing metrics"))
+			}
+			res.WriteHeader(204)
+			res.Write([]byte(""))
+		} else if path == "query" {
+			// Deliver a dummy response to the query endpoint, as some InfluxDB clients test endpoint availability with a query
+			res.Header().Set("Content-Type", "application/json")
+			res.Header().Set("X-Influxdb-Version", "1.0")
+			res.WriteHeader(200)
+			res.Write([]byte("{\"results\":[]}"))
+		} else {
+			// Don't know how to respond to calls to other endpoints
+			res.WriteHeader(404)
+			res.Write([]byte("Not Found"))
+		}
+	} else {
+		log.Printf("Problem reading request: [%s], Error: %s\n", string(body), err)
+		res.WriteHeader(500)
+		res.Write([]byte("ERROR reading request"))
+	}
+}
+
+func (t *HttpListener) storeMetrics(metrics []telegraf.Metric) error {
+	t.Lock()
+	defer t.Unlock()
+
+	for _, m := range metrics {
+		t.acc.AddFields(m.Name(), m.Fields(), m.Tags(), m.Time())
+	}
+	return nil
+}
+
+func init() {
+	inputs.Add("http_listener", func() telegraf.Input {
+		return &HttpListener{}
+	})
+}

plugins/inputs/http_listener/http_listener_test.go

@@ -0,0 +1,127 @@
+package http_listener
+
+import (
+	"testing"
+	"time"
+
+	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/testutil"
+
+	"bytes"
+	"github.com/stretchr/testify/require"
+	"net/http"
+)
+
+const (
+	testMsg = "cpu_load_short,host=server01 value=12.0 1422568543702900257\n"
+
+	testMsgs = `cpu_load_short,host=server02 value=12.0 1422568543702900257
+cpu_load_short,host=server03 value=12.0 1422568543702900257
+cpu_load_short,host=server04 value=12.0 1422568543702900257
+cpu_load_short,host=server05 value=12.0 1422568543702900257
+cpu_load_short,host=server06 value=12.0 1422568543702900257
+`
+	badMsg = "blahblahblah: 42\n"
+
+	emptyMsg = ""
+)
+
+func newTestHttpListener() *HttpListener {
+	listener := &HttpListener{
+		ServiceAddress: ":8186",
+		ReadTimeout:    "10",
+		WriteTimeout:   "10",
+	}
+	return listener
+}
+
+func TestWriteHTTP(t *testing.T) {
+	listener := newTestHttpListener()
+	listener.parser, _ = parsers.NewInfluxParser()
+
+	acc := &testutil.Accumulator{}
+	require.NoError(t, listener.Start(acc))
+	defer listener.Stop()
+
+	time.Sleep(time.Millisecond * 25)
+
+	// post single message to listener
+	var resp, err = http.Post("http://localhost:8186/write?db=mydb", "", bytes.NewBuffer([]byte(testMsg)))
+	require.NoError(t, err)
+	require.EqualValues(t, 204, resp.StatusCode)
+
+	time.Sleep(time.Millisecond * 15)
+	acc.AssertContainsTaggedFields(t, "cpu_load_short",
+		map[string]interface{}{"value": float64(12)},
+		map[string]string{"host": "server01"},
+	)
+
+	// post multiple message to listener
+	resp, err = http.Post("http://localhost:8186/write?db=mydb", "", bytes.NewBuffer([]byte(testMsgs)))
+	require.NoError(t, err)
+	require.EqualValues(t, 204, resp.StatusCode)
+
+	time.Sleep(time.Millisecond * 15)
+	hostTags := []string{"server02", "server03",
+		"server04", "server05", "server06"}
+	for _, hostTag := range hostTags {
+		acc.AssertContainsTaggedFields(t, "cpu_load_short",
+			map[string]interface{}{"value": float64(12)},
+			map[string]string{"host": hostTag},
+		)
+	}
+}
+
+func TestWriteHTTPInvalid(t *testing.T) {
+	time.Sleep(time.Millisecond * 250)
+
+	listener := newTestHttpListener()
+	listener.parser, _ = parsers.NewInfluxParser()
+
+	acc := &testutil.Accumulator{}
+	require.NoError(t, listener.Start(acc))
+	defer listener.Stop()
+
+	time.Sleep(time.Millisecond * 25)
+
+	// post single message to listener
+	var resp, err = http.Post("http://localhost:8186/write?db=mydb", "", bytes.NewBuffer([]byte(badMsg)))
+	require.NoError(t, err)
+	require.EqualValues(t, 500, resp.StatusCode)
+}
+
+func TestWriteHTTPEmpty(t *testing.T) {
+	time.Sleep(time.Millisecond * 250)
+
+	listener := newTestHttpListener()
+	listener.parser, _ = parsers.NewInfluxParser()
+
+	acc := &testutil.Accumulator{}
+	require.NoError(t, listener.Start(acc))
+	defer listener.Stop()
+
+	time.Sleep(time.Millisecond * 25)
+
+	// post single message to listener
+	var resp, err = http.Post("http://localhost:8186/write?db=mydb", "", bytes.NewBuffer([]byte(emptyMsg)))
+	require.NoError(t, err)
+	require.EqualValues(t, 204, resp.StatusCode)
+}
+
+func TestQueryHTTP(t *testing.T) {
+	time.Sleep(time.Millisecond * 250)
+
+	listener := newTestHttpListener()
+	listener.parser, _ = parsers.NewInfluxParser()
+
+	acc := &testutil.Accumulator{}
+	require.NoError(t, listener.Start(acc))
+	defer listener.Stop()
+
+	time.Sleep(time.Millisecond * 25)
+
+	// post query to listener
+	var resp, err = http.Post("http://localhost:8186/query?db=&q=CREATE+DATABASE+IF+NOT+EXISTS+%22mydb%22", "", nil)
+	require.NoError(t, err)
+	require.EqualValues(t, 200, resp.StatusCode)
+}