Bagelswitch master

CHANGELOG.md

@@ -17,6 +17,7 @@
 - [#1542](https://github.com/influxdata/telegraf/pull/1542): Add filestack webhook plugin.
 - [#1599](https://github.com/influxdata/telegraf/pull/1599): Add server hostname for each docker measurements.
 - [#1697](https://github.com/influxdata/telegraf/pull/1697): Add NATS output plugin.
+- [#1407](https://github.com/influxdata/telegraf/pull/1407): HTTP service listener input plugin.
 
 ### Bugfixes
 

README.md

@@ -217,6 +217,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,154 @@
+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
+	wg sync.WaitGroup
+
+	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()
+
+	t.wg.Wait()
+
+	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) {
+	t.wg.Add(1)
+	defer t.wg.Done()
+	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
+	}
+
+	switch req.URL.Path {
+	case "/write":
+		var metrics []telegraf.Metric
+		metrics, err = t.parser.Parse(body)
+		if err == nil {
+			for _, m := range metrics {
+				t.acc.AddFields(m.Name(), m.Fields(), m.Tags(), m.Time())
+			}
+			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)
+		}
+	case "/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\":[]}"))
+	case "/ping":
+		// respond to ping requests
+		res.WriteHeader(http.StatusNoContent)
+	default:
+		// Don't know how to respond to calls to other endpoints
+		http.NotFound(res, req)
+	}
+}
+
+func init() {
+	inputs.Add("http_listener", func() telegraf.Input {
+		return &HttpListener{}
+	})
+}

plugins/inputs/http_listener/http_listener_test.go

@@ -0,0 +1,181 @@
+package http_listener
+
+import (
+	"sync"
+	"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",
+	}
+	return listener
+}
+
+func TestWriteHTTP(t *testing.T) {
+	listener := newTestHttpListener()
+	parser, _ := parsers.NewInfluxParser()
+	listener.SetParser(parser)
+
+	acc := &testutil.Accumulator{}
+	require.NoError(t, listener.Start(acc))
+	defer listener.Stop()
+
+	time.Sleep(time.Millisecond * 25)
+
+	// post single message to listener
+	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},
+		)
+	}
+}
+
+// writes 25,000 metrics to the listener with 10 different writers
+func TestWriteHTTPHighTraffic(t *testing.T) {
+	listener := &HttpListener{ServiceAddress: ":8286"}
+	parser, _ := parsers.NewInfluxParser()
+	listener.SetParser(parser)
+
+	acc := &testutil.Accumulator{}
+	require.NoError(t, listener.Start(acc))
+	defer listener.Stop()
+
+	time.Sleep(time.Millisecond * 25)
+
+	// post many messages to listener
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			for i := 0; i < 500; i++ {
+				resp, err := http.Post("http://localhost:8286/write?db=mydb", "", bytes.NewBuffer([]byte(testMsgs)))
+				require.NoError(t, err)
+				require.EqualValues(t, 204, resp.StatusCode)
+			}
+			wg.Done()
+		}()
+	}
+
+	wg.Wait()
+	time.Sleep(time.Millisecond * 50)
+	listener.Gather(acc)
+
+	require.Equal(t, int64(25000), int64(acc.NMetrics()))
+}
+
+func TestReceive404ForInvalidEndpoint(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
+	resp, err := http.Post("http://localhost:8186/foobar", "", bytes.NewBuffer([]byte(testMsg)))
+	require.NoError(t, err)
+	require.EqualValues(t, 404, resp.StatusCode)
+}
+
+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
+	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
+	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 TestQueryAndPingHTTP(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
+	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)
+
+	// post ping to listener
+	resp, err = http.Post("http://localhost:8186/ping", "", nil)
+	require.NoError(t, err)
+	require.EqualValues(t, 204, 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.