Add HTTP listener service input plugin

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,151 @@
+package http_listener
+
+import (
+	"io/ioutil"
+	"log"
+	"net"
+	"net/http"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/influxdata/telegraf"
+	"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    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)
+}

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.

plugins/inputs/http_listener/stoppableListener/listener.go

@@ -0,0 +1,62 @@
+package stoppableListener
+
+import (
+	"errors"
+	"net"
+	"time"
+)
+
+type StoppableListener struct {
+	*net.TCPListener          //Wrapped listener
+	stop             chan int //Channel used only to indicate listener should shutdown
+}
+
+func New(l net.Listener) (*StoppableListener, error) {
+	tcpL, ok := l.(*net.TCPListener)
+
+	if !ok {
+		return nil, errors.New("Cannot wrap listener")
+	}
+
+	retval := &StoppableListener{}
+	retval.TCPListener = tcpL
+	retval.stop = make(chan int)
+
+	return retval, nil
+}
+
+var StoppedError = errors.New("Listener stopped")
+
+func (sl *StoppableListener) Accept() (net.Conn, error) {
+
+	for {
+		//Wait up to one second for a new connection
+		sl.SetDeadline(time.Now().Add(time.Second))
+
+		newConn, err := sl.TCPListener.Accept()
+
+		//Check for the channel being closed
+		select {
+		case <-sl.stop:
+			return nil, StoppedError
+		default:
+			//If the channel is still open, continue as normal
+		}
+
+		if err != nil {
+			netErr, ok := err.(net.Error)
+
+			//If this is a timeout, then continue to wait for
+			//new connections
+			if ok && netErr.Timeout() && netErr.Temporary() {
+				continue
+			}
+		}
+
+		return newConn, err
+	}
+}
+
+func (sl *StoppableListener) Stop() {
+	close(sl.stop)
+}