From f1861a08b97004b4ac67ec1fc5e8e990bf85d530 Mon Sep 17 00:00:00 2001 From: Nicolas Grange Date: Fri, 16 Feb 2018 11:00:10 +1100 Subject: [PATCH] Add http input plugin which supports any input data format (#3546) --- README.md | 1 + plugins/inputs/all/all.go | 1 + plugins/inputs/http/README.md | 42 +++++++ plugins/inputs/http/http.go | 200 +++++++++++++++++++++++++++++++ plugins/inputs/http/http_test.go | 117 ++++++++++++++++++ 5 files changed, 361 insertions(+) create mode 100644 plugins/inputs/http/README.md create mode 100644 plugins/inputs/http/http.go create mode 100644 plugins/inputs/http/http_test.go diff --git a/README.md b/README.md index d52327f4c..19a5afe7e 100644 --- a/README.md +++ b/README.md @@ -152,6 +152,7 @@ configuration options. * [graylog](./plugins/inputs/graylog) * [haproxy](./plugins/inputs/haproxy) * [hddtemp](./plugins/inputs/hddtemp) +* [http](./plugins/inputs/http) (generic HTTP plugin, supports using input data formats) * [http_response](./plugins/inputs/http_response) * [httpjson](./plugins/inputs/httpjson) (generic JSON-emitting http service plugin) * [internal](./plugins/inputs/internal) diff --git a/plugins/inputs/all/all.go b/plugins/inputs/all/all.go index a13bd6cd3..d58002089 100644 --- a/plugins/inputs/all/all.go +++ b/plugins/inputs/all/all.go @@ -29,6 +29,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" _ "github.com/influxdata/telegraf/plugins/inputs/http_listener" _ "github.com/influxdata/telegraf/plugins/inputs/http_response" _ "github.com/influxdata/telegraf/plugins/inputs/httpjson" diff --git a/plugins/inputs/http/README.md b/plugins/inputs/http/README.md new file mode 100644 index 000000000..430a19b2b --- /dev/null +++ b/plugins/inputs/http/README.md @@ -0,0 +1,42 @@ +# HTTP Input Plugin + +The HTTP input plugin collects metrics from one or more HTTP(S) endpoints. The metrics need to be formatted in one of the supported data formats. Each data format has its own unique set of configuration options, read more about them here: + https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md + + +### Configuration: + +This section contains the default TOML to configure the plugin. You can +generate it using `telegraf --usage http`. + +```toml +# Read formatted metrics from one or more HTTP endpoints +[[inputs.http]] + ## One or more URLs from which to read formatted metrics + urls = [ + "http://localhost/metrics" + ] + + ## Optional HTTP Basic Auth Credentials + # username = "username" + # password = "pa$$word" + + ## Optional SSL Config + # ssl_ca = "/etc/telegraf/ca.pem" + # ssl_cert = "/etc/telegraf/cert.pem" + # ssl_key = "/etc/telegraf/key.pem" + ## Use SSL but skip chain & host verification + # insecure_skip_verify = false + + # timeout = "5s" + + ## Data format to consume. + ## Each data format has its 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" +``` + +### Metrics: + +The metrics collected by this input plugin will depend on the configurated `data_format` and the payload returned by the HTTP endpoint(s). diff --git a/plugins/inputs/http/http.go b/plugins/inputs/http/http.go new file mode 100644 index 000000000..6a98ab4fc --- /dev/null +++ b/plugins/inputs/http/http.go @@ -0,0 +1,200 @@ +package http + +import ( + "errors" + "fmt" + "io/ioutil" + "net/http" + "strings" + "sync" + "time" + + "github.com/influxdata/telegraf" + "github.com/influxdata/telegraf/internal" + "github.com/influxdata/telegraf/plugins/inputs" + "github.com/influxdata/telegraf/plugins/parsers" +) + +type HTTP struct { + URLs []string `toml:"urls"` + + Headers map[string]string + + // HTTP Basic Auth Credentials + Username string + Password string + + // Option to add "url" tag to each metric + TagURL bool `toml:"tag_url"` + + // Path to CA file + SSLCA string `toml:"ssl_ca"` + // Path to host cert file + SSLCert string `toml:"ssl_cert"` + // Path to cert key file + SSLKey string `toml:"ssl_key"` + // Use SSL but skip chain & host verification + InsecureSkipVerify bool + + Timeout internal.Duration + + client *http.Client + + // The parser will automatically be set by Telegraf core code because + // this plugin implements the ParserInput interface (i.e. the SetParser method) + parser parsers.Parser +} + +var sampleConfig = ` + ## One or more URLs from which to read formatted metrics + urls = [ + "http://localhost/metrics" + ] + + ## Optional HTTP headers + # headers = {"X-Special-Header" = "Special-Value"} + + ## Optional HTTP Basic Auth Credentials + # username = "username" + # password = "pa$$word" + + ## Tag all metrics with the url + # tag_url = true + + ## Optional SSL Config + # ssl_ca = "/etc/telegraf/ca.pem" + # ssl_cert = "/etc/telegraf/cert.pem" + # ssl_key = "/etc/telegraf/key.pem" + ## Use SSL but skip chain & host verification + # insecure_skip_verify = false + + # timeout = "5s" + + ## Data format to consume. + ## Each data format has its 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" +` + +// SampleConfig returns the default configuration of the Input +func (*HTTP) SampleConfig() string { + return sampleConfig +} + +// Description returns a one-sentence description on the Input +func (*HTTP) Description() string { + return "Read formatted metrics from one or more HTTP endpoints" +} + +// Gather takes in an accumulator and adds the metrics that the Input +// gathers. This is called every "interval" +func (h *HTTP) Gather(acc telegraf.Accumulator) error { + if h.client == nil { + tlsCfg, err := internal.GetTLSConfig( + h.SSLCert, h.SSLKey, h.SSLCA, h.InsecureSkipVerify) + if err != nil { + return err + } + h.client = &http.Client{ + Transport: &http.Transport{ + TLSClientConfig: tlsCfg, + }, + Timeout: h.Timeout.Duration, + } + } + + var wg sync.WaitGroup + for _, u := range h.URLs { + wg.Add(1) + go func(url string) { + defer wg.Done() + if err := h.gatherURL(acc, url); err != nil { + acc.AddError(fmt.Errorf("[url=%s]: %s", url, err)) + } + }(u) + } + + wg.Wait() + + return nil +} + +// SetParser takes the data_format from the config and finds the right parser for that format +func (h *HTTP) SetParser(parser parsers.Parser) { + h.parser = parser +} + +// Gathers data from a particular URL +// Parameters: +// acc : The telegraf Accumulator to use +// url : endpoint to send request to +// +// Returns: +// error: Any error that may have occurred +func (h *HTTP) gatherURL( + acc telegraf.Accumulator, + url string, +) error { + request, err := http.NewRequest("GET", url, nil) + if err != nil { + return err + } + + for k, v := range h.Headers { + if strings.ToLower(k) == "host" { + request.Host = v + } else { + request.Header.Add(k, v) + } + } + + if h.Username != "" { + request.SetBasicAuth(h.Username, h.Password) + } + + resp, err := h.client.Do(request) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("Received status code %d (%s), expected %d (%s)", + resp.StatusCode, + http.StatusText(resp.StatusCode), + http.StatusOK, + http.StatusText(http.StatusOK)) + } + + b, err := ioutil.ReadAll(resp.Body) + if err != nil { + return err + } + + if h.parser == nil { + return errors.New("Parser is not set") + } + + metrics, err := h.parser.Parse(b) + if err != nil { + return err + } + + for _, metric := range metrics { + if h.TagURL { + metric.AddTag("url", url) + } + acc.AddFields(metric.Name(), metric.Fields(), metric.Tags(), metric.Time()) + } + + return nil +} + +func init() { + inputs.Add("http", func() telegraf.Input { + return &HTTP{ + Timeout: internal.Duration{Duration: time.Second * 5}, + } + }) +} diff --git a/plugins/inputs/http/http_test.go b/plugins/inputs/http/http_test.go new file mode 100644 index 000000000..fd352829c --- /dev/null +++ b/plugins/inputs/http/http_test.go @@ -0,0 +1,117 @@ +package http_test + +import ( + "net/http" + "net/http/httptest" + "testing" + + plugin "github.com/influxdata/telegraf/plugins/inputs/http" + "github.com/influxdata/telegraf/plugins/parsers" + "github.com/influxdata/telegraf/testutil" + "github.com/stretchr/testify/require" +) + +func TestHTTPwithJSONFormat(t *testing.T) { + fakeServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/endpoint" { + _, _ = w.Write([]byte(simpleJSON)) + } else { + w.WriteHeader(http.StatusNotFound) + } + })) + defer fakeServer.Close() + + url := fakeServer.URL + "/endpoint" + plugin := &plugin.HTTP{ + URLs: []string{url}, + TagURL: true, + } + metricName := "metricName" + p, _ := parsers.NewJSONParser(metricName, nil, nil) + plugin.SetParser(p) + + var acc testutil.Accumulator + require.NoError(t, acc.GatherError(plugin.Gather)) + + require.Len(t, acc.Metrics, 1) + + // basic check to see if we got the right field, value and tag + var metric = acc.Metrics[0] + require.Equal(t, metric.Measurement, metricName) + require.Len(t, acc.Metrics[0].Fields, 1) + require.Equal(t, acc.Metrics[0].Fields["a"], 1.2) + require.Equal(t, acc.Metrics[0].Tags["url"], url) +} + +func TestHTTPHeaders(t *testing.T) { + header := "X-Special-Header" + headerValue := "Special-Value" + fakeServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/endpoint" { + if r.Header.Get(header) == headerValue { + _, _ = w.Write([]byte(simpleJSON)) + } else { + w.WriteHeader(http.StatusForbidden) + } + } else { + w.WriteHeader(http.StatusNotFound) + } + })) + defer fakeServer.Close() + + url := fakeServer.URL + "/endpoint" + plugin := &plugin.HTTP{ + URLs: []string{url}, + Headers: map[string]string{header: headerValue}, + } + metricName := "metricName" + p, _ := parsers.NewJSONParser(metricName, nil, nil) + plugin.SetParser(p) + + var acc testutil.Accumulator + require.NoError(t, acc.GatherError(plugin.Gather)) +} + +func TestInvalidStatusCode(t *testing.T) { + fakeServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusNotFound) + })) + defer fakeServer.Close() + + url := fakeServer.URL + "/endpoint" + plugin := &plugin.HTTP{ + URLs: []string{url}, + } + + metricName := "metricName" + p, _ := parsers.NewJSONParser(metricName, nil, nil) + plugin.SetParser(p) + + var acc testutil.Accumulator + require.Error(t, acc.GatherError(plugin.Gather)) +} + +func TestParserNotSet(t *testing.T) { + fakeServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/endpoint" { + _, _ = w.Write([]byte(simpleJSON)) + } else { + w.WriteHeader(http.StatusNotFound) + } + })) + defer fakeServer.Close() + + url := fakeServer.URL + "/endpoint" + plugin := &plugin.HTTP{ + URLs: []string{url}, + } + + var acc testutil.Accumulator + require.Error(t, acc.GatherError(plugin.Gather)) +} + +const simpleJSON = ` +{ + "a": 1.2 +} +`