Add HTTP listener service input plugin

CHANGELOG.md

@@ -67,6 +67,7 @@ consistent with the behavior of `collection_jitter`.
 - [#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.
+- [#1407](https://github.com/influxdata/telegraf/pull/1407): HTTP service listener input plugin.
 - [#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.
 - [#1369](https://github.com/influxdata/telegraf/pull/1480): add ability to read redis from a socket.

README.md

@@ -216,6 +216,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

@@ -23,6 +23,7 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/inputs/graylog"
 	_ "github.com/influxdata/telegraf/plugins/inputs/haproxy"
 	_ "github.com/influxdata/telegraf/plugins/inputs/hddtemp"
+	_ "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,24 @@
+# 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.
+When chaining Telegraf instances using this plugin, CREATE DATABASE requests receive a 200 OK response with message body `{"results":[]}` but they are not relayed. The output configuration of the Telegraf instance which ultimately submits data to InfluxDB determines the destination database.
+
+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
+  read_timeout = "10s"
+  write_timeout = "10s"
+```

plugins/inputs/http_listener/http_listener.go

@@ -0,0 +1,156 @@
+package http_listener
+
+import (
+	"io/ioutil"
+	"log"
+	"net"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/internal"
+	"github.com/influxdata/telegraf/plugins/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs/http_listener/stoppableListener"
+	"github.com/influxdata/telegraf/plugins/parsers"
+)
+
+type HttpListener struct {
+	ServiceAddress string
+	ReadTimeout    internal.Duration
+	WriteTimeout   internal.Duration
+
+	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
+  read_timeout = "10s"
+  write_timeout = "10s"
+`
+
+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)
+	if err != nil {
+		return err
+	}
+	t.listener, err = stoppableListener.New(rawListener)
+	if err != nil {
+		return err
+	}
+
+	go t.httpListen()
+
+	log.Printf("Started HTTP listener service on %s\n", t.ServiceAddress)
+
+	return nil
+}
+
+// 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 {
+
+	if t.ReadTimeout.Duration < time.Second {
+		t.ReadTimeout.Duration = time.Second * 10
+	}
+	if t.WriteTimeout.Duration < time.Second {
+		t.WriteTimeout.Duration = time.Second * 10
+	}
+
+	var server = http.Server{
+		Handler:      t,
+		ReadTimeout:  t.ReadTimeout.Duration,
+		WriteTimeout: t.WriteTimeout.Duration,
+	}
+
+	return server.Serve(t.listener)
+}
+
+func (t *HttpListener) ServeHTTP(res http.ResponseWriter, req *http.Request) {
+	body, err := ioutil.ReadAll(req.Body)
+
+	if err != nil {
+		log.Printf("Problem reading request: [%s], Error: %s\n", string(body), err)
+		http.Error(res, "ERROR reading request", http.StatusInternalServerError)
+		return
+	}
+
+	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)
+			res.WriteHeader(http.StatusNoContent)
+		} else {
+			log.Printf("Problem parsing body: [%s], Error: %s\n", string(body), err)
+			http.Error(res, "ERROR parsing metrics", http.StatusInternalServerError)
+		}
+	} 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(http.StatusOK)
+		res.Write([]byte("{\"results\":[]}"))
+	} else {
+		// Don't know how to respond to calls to other endpoints
+		http.NotFound(res, req)
+	}
+}
+
+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,128 @@
+package http_listener
+
+import (
+	"testing"
+	"time"
+
+	"github.com/influxdata/telegraf/internal"
+	"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:    internal.Duration{Duration: time.Second * 10},
+		WriteTimeout:   internal.Duration{Duration: time.Second * 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)
+}

plugins/inputs/http_listener/stoppableListener/LICENSE

@@ -0,0 +1,10 @@
+Copyright (c) 2014, Eric Urban
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:
+
+1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.