diff --git a/CHANGELOG.md b/CHANGELOG.md index e8dee5175b171..e3084fc3a756f 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 0415d89f74504..98a3972d18b90 100644 --- a/README.md +++ b/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) diff --git a/plugins/inputs/all/all.go b/plugins/inputs/all/all.go index 96fbdffe1b0b0..058b230d80604 100644 --- a/plugins/inputs/all/all.go +++ b/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" diff --git a/plugins/inputs/http_listener/README.md b/plugins/inputs/http_listener/README.md new file mode 100644 index 0000000000000..9643f6a2ed8e5 --- /dev/null +++ b/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" +``` diff --git a/plugins/inputs/http_listener/http_listener.go b/plugins/inputs/http_listener/http_listener.go new file mode 100644 index 0000000000000..9110fd106dd14 --- /dev/null +++ b/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{} + }) +} diff --git a/plugins/inputs/http_listener/http_listener_test.go b/plugins/inputs/http_listener/http_listener_test.go new file mode 100644 index 0000000000000..270e8264a1808 --- /dev/null +++ b/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) +} diff --git a/plugins/inputs/http_listener/stoppableListener/LICENSE b/plugins/inputs/http_listener/stoppableListener/LICENSE new file mode 100644 index 0000000000000..eb07824517a4d --- /dev/null +++ b/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. diff --git a/plugins/inputs/http_listener/stoppableListener/listener.go b/plugins/inputs/http_listener/stoppableListener/listener.go new file mode 100644 index 0000000000000..69a9f33cc3b46 --- /dev/null +++ b/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) +}