initial http_listener implementation

plugins/inputs/http_listener/README.md

@@ -0,0 +1,30 @@
+# TCP listener service input plugin
+
+The TCP listener is a service input plugin that listens for messages on a TCP
+socket and adds those messages to InfluxDB.
+The plugin expects messages in the
+[Telegraf Input Data Formats](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md).
+
+### Configuration:
+
+This is a sample configuration for the plugin.
+
+```toml
+# Generic TCP listener
+[[inputs.tcp_listener]]
+  ## Address and port to host TCP listener on
+  service_address = ":8094"
+
+  ## Number of TCP messages allowed to queue up. Once filled, the
+  ## TCP listener will start dropping packets.
+  allowed_pending_messages = 10000
+
+  ## Maximum number of concurrent TCP connections to allow
+  max_tcp_connections = 250
+
+  ## Data format to consume.
+  ## Each data format has it's own unique set of configuration options, read
+  ## more about them here:
+  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
+  data_format = "influx"
+```

plugins/inputs/http_listener/http_listener.go

@@ -0,0 +1,141 @@
+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"
+)
+
+type HttpListener struct {
+	ServiceAddress  string
+	readTimeout	int
+	writeTimeout	int
+
+	sync.Mutex
+
+	listener *stoppableListener.StoppableListener
+
+	parser parsers.Parser
+	acc    telegraf.Accumulator
+}
+
+const sampleConfig = `
+  ## Address and port to host HTTP listener on
+  service_address = ":8086"
+
+  ## 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 {
+
+	var server = http.Server{
+		Handler:        t.writeHandler,
+		ReadTimeout:    t.readTimeout * time.Second,
+		WriteTimeout:   t.writeTimeout * time.Second,
+	}
+
+	var err = server.Serve(t.listener)
+
+	return err
+}
+
+func (t *HttpListener) writeHandler(res http.ResponseWriter, req *http.Request) error {
+	body, err := ioutil.ReadAll(req.Body)
+
+	if err == nil {
+		var metrics []telegraf.Metric
+		for {
+			if len(body) == 0 {
+				continue
+			}
+			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 {
+		log.Printf("Problem reading request: [%s], Error: %s\n", string(body), err)
+		res.WriteHeader(500)
+		res.Write([]byte("ERROR reading request"))
+	}
+
+	return err
+}
+
+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,259 @@
+package http_listener
+
+import (
+	"fmt"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/testutil"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+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
+`
+)
+
+func newTestTcpListener() (*TcpListener, chan []byte) {
+	in := make(chan []byte, 1500)
+	listener := &TcpListener{
+		ServiceAddress:         ":8194",
+		AllowedPendingMessages: 10000,
+		MaxTCPConnections:      250,
+		in:                     in,
+		done:                   make(chan struct{}),
+	}
+	return listener, in
+}
+
+func TestConnectTCP(t *testing.T) {
+	listener := TcpListener{
+		ServiceAddress:         ":8194",
+		AllowedPendingMessages: 10000,
+		MaxTCPConnections:      250,
+	}
+	listener.parser, _ = parsers.NewInfluxParser()
+
+	acc := &testutil.Accumulator{}
+	require.NoError(t, listener.Start(acc))
+	defer listener.Stop()
+
+	time.Sleep(time.Millisecond * 25)
+	conn, err := net.Dial("tcp", "127.0.0.1:8194")
+	require.NoError(t, err)
+
+	// send single message to socket
+	fmt.Fprintf(conn, testMsg)
+	time.Sleep(time.Millisecond * 15)
+	acc.AssertContainsTaggedFields(t, "cpu_load_short",
+		map[string]interface{}{"value": float64(12)},
+		map[string]string{"host": "server01"},
+	)
+
+	// send multiple messages to socket
+	fmt.Fprintf(conn, testMsgs)
+	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},
+		)
+	}
+}
+
+// Test that MaxTCPConections is respected
+func TestConcurrentConns(t *testing.T) {
+	listener := TcpListener{
+		ServiceAddress:         ":8195",
+		AllowedPendingMessages: 10000,
+		MaxTCPConnections:      2,
+	}
+	listener.parser, _ = parsers.NewInfluxParser()
+
+	acc := &testutil.Accumulator{}
+	require.NoError(t, listener.Start(acc))
+	defer listener.Stop()
+
+	time.Sleep(time.Millisecond * 25)
+	_, err := net.Dial("tcp", "127.0.0.1:8195")
+	assert.NoError(t, err)
+	_, err = net.Dial("tcp", "127.0.0.1:8195")
+	assert.NoError(t, err)
+
+	// Connection over the limit:
+	conn, err := net.Dial("tcp", "127.0.0.1:8195")
+	assert.NoError(t, err)
+	net.Dial("tcp", "127.0.0.1:8195")
+	buf := make([]byte, 1500)
+	n, err := conn.Read(buf)
+	assert.NoError(t, err)
+	assert.Equal(t,
+		"Telegraf maximum concurrent TCP connections (2) reached, closing.\n"+
+			"You may want to increase max_tcp_connections in"+
+			" the Telegraf tcp listener configuration.\n",
+		string(buf[:n]))
+
+	_, err = conn.Write([]byte(testMsg))
+	assert.NoError(t, err)
+	time.Sleep(time.Millisecond * 10)
+	assert.Zero(t, acc.NFields())
+}
+
+// Test that MaxTCPConections is respected when max==1
+func TestConcurrentConns1(t *testing.T) {
+	listener := TcpListener{
+		ServiceAddress:         ":8196",
+		AllowedPendingMessages: 10000,
+		MaxTCPConnections:      1,
+	}
+	listener.parser, _ = parsers.NewInfluxParser()
+
+	acc := &testutil.Accumulator{}
+	require.NoError(t, listener.Start(acc))
+	defer listener.Stop()
+
+	time.Sleep(time.Millisecond * 25)
+	_, err := net.Dial("tcp", "127.0.0.1:8196")
+	assert.NoError(t, err)
+
+	// Connection over the limit:
+	conn, err := net.Dial("tcp", "127.0.0.1:8196")
+	assert.NoError(t, err)
+	net.Dial("tcp", "127.0.0.1:8196")
+	buf := make([]byte, 1500)
+	n, err := conn.Read(buf)
+	assert.NoError(t, err)
+	assert.Equal(t,
+		"Telegraf maximum concurrent TCP connections (1) reached, closing.\n"+
+			"You may want to increase max_tcp_connections in"+
+			" the Telegraf tcp listener configuration.\n",
+		string(buf[:n]))
+
+	_, err = conn.Write([]byte(testMsg))
+	assert.NoError(t, err)
+	time.Sleep(time.Millisecond * 10)
+	assert.Zero(t, acc.NFields())
+}
+
+// Test that MaxTCPConections is respected
+func TestCloseConcurrentConns(t *testing.T) {
+	listener := TcpListener{
+		ServiceAddress:         ":8195",
+		AllowedPendingMessages: 10000,
+		MaxTCPConnections:      2,
+	}
+	listener.parser, _ = parsers.NewInfluxParser()
+
+	acc := &testutil.Accumulator{}
+	require.NoError(t, listener.Start(acc))
+
+	time.Sleep(time.Millisecond * 25)
+	_, err := net.Dial("tcp", "127.0.0.1:8195")
+	assert.NoError(t, err)
+	_, err = net.Dial("tcp", "127.0.0.1:8195")
+	assert.NoError(t, err)
+
+	listener.Stop()
+}
+
+func TestRunParser(t *testing.T) {
+	var testmsg = []byte(testMsg)
+
+	listener, in := newTestTcpListener()
+	acc := testutil.Accumulator{}
+	listener.acc = &acc
+	defer close(listener.done)
+
+	listener.parser, _ = parsers.NewInfluxParser()
+	listener.wg.Add(1)
+	go listener.tcpParser()
+
+	in <- testmsg
+	time.Sleep(time.Millisecond * 25)
+	listener.Gather(&acc)
+
+	if a := acc.NFields(); a != 1 {
+		t.Errorf("got %v, expected %v", a, 1)
+	}
+
+	acc.AssertContainsTaggedFields(t, "cpu_load_short",
+		map[string]interface{}{"value": float64(12)},
+		map[string]string{"host": "server01"},
+	)
+}
+
+func TestRunParserInvalidMsg(t *testing.T) {
+	var testmsg = []byte("cpu_load_short")
+
+	listener, in := newTestTcpListener()
+	acc := testutil.Accumulator{}
+	listener.acc = &acc
+	defer close(listener.done)
+
+	listener.parser, _ = parsers.NewInfluxParser()
+	listener.wg.Add(1)
+	go listener.tcpParser()
+
+	in <- testmsg
+	time.Sleep(time.Millisecond * 25)
+
+	if a := acc.NFields(); a != 0 {
+		t.Errorf("got %v, expected %v", a, 0)
+	}
+}
+
+func TestRunParserGraphiteMsg(t *testing.T) {
+	var testmsg = []byte("cpu.load.graphite 12 1454780029")
+
+	listener, in := newTestTcpListener()
+	acc := testutil.Accumulator{}
+	listener.acc = &acc
+	defer close(listener.done)
+
+	listener.parser, _ = parsers.NewGraphiteParser("_", []string{}, nil)
+	listener.wg.Add(1)
+	go listener.tcpParser()
+
+	in <- testmsg
+	time.Sleep(time.Millisecond * 25)
+	listener.Gather(&acc)
+
+	acc.AssertContainsFields(t, "cpu_load_graphite",
+		map[string]interface{}{"value": float64(12)})
+}
+
+func TestRunParserJSONMsg(t *testing.T) {
+	var testmsg = []byte("{\"a\": 5, \"b\": {\"c\": 6}}\n")
+
+	listener, in := newTestTcpListener()
+	acc := testutil.Accumulator{}
+	listener.acc = &acc
+	defer close(listener.done)
+
+	listener.parser, _ = parsers.NewJSONParser("udp_json_test", []string{}, nil)
+	listener.wg.Add(1)
+	go listener.tcpParser()
+
+	in <- testmsg
+	time.Sleep(time.Millisecond * 25)
+	listener.Gather(&acc)
+
+	acc.AssertContainsFields(t, "udp_json_test",
+		map[string]interface{}{
+			"a":   float64(5),
+			"b_c": float64(6),
+		})
+}