Modifications to httpjson plugin

This commit is contained in:
Josh Palay 2015-08-14 14:33:51 -07:00 committed by Cameron Sparr
parent 06ef2a72c5
commit 7d9efd7cff
3 changed files with 344 additions and 52 deletions

View File

@ -3,6 +3,7 @@
### Features ### Features
[#112](https://github.com/influxdb/telegraf/pull/112): Datadog output. Thanks @jipperinbham! [#112](https://github.com/influxdb/telegraf/pull/112): Datadog output. Thanks @jipperinbham!
[#116](https://github.com/influxdb/telegraf/pull/116): Use godep to vendor all dependencies [#116](https://github.com/influxdb/telegraf/pull/116): Use godep to vendor all dependencies
[#120](https://github.com/influxdb/telegraf/pull/120): Httpjson plugin. Thanks @jpalay & @alvaromorales!
### Bugfixes ### Bugfixes
[#113](https://github.com/influxdb/telegraf/issues/113): Update README with Telegraf/InfluxDB compatibility [#113](https://github.com/influxdb/telegraf/issues/113): Update README with Telegraf/InfluxDB compatibility

View File

@ -2,34 +2,69 @@ package httpjson
import ( import (
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"github.com/bitly/go-simplejson" "io/ioutil"
"github.com/influxdb/telegraf/plugins"
"net/http" "net/http"
"net/url"
"strings"
"sync" "sync"
"github.com/influxdb/telegraf/plugins"
) )
type HttpJson struct { type HttpJson struct {
Services []Service
client HTTPClient
}
type Service struct {
Name string
Servers []string Servers []string
Measurements map[string]string
Method string Method string
Foo string Parameters map[string]string
}
type HTTPClient interface {
// Returns the result of an http request
//
// Parameters:
// req: HTTP request object
//
// Returns:
// http.Response: HTTP respons object
// error : Any error that may have occurred
MakeRequest(req *http.Request) (*http.Response, error)
}
type RealHTTPClient struct {
client *http.Client client *http.Client
} }
func (c RealHTTPClient) MakeRequest(req *http.Request) (*http.Response, error) {
return c.client.Do(req)
}
var sampleConfig = ` var sampleConfig = `
# stats url endpoint # Specify services via an array of tables
servers = ["http://localhost:5000"] [[httpjson.services]]
# a name for server(s) # a name for the service being polled
foo = "mycluster" name = "webserver_stats"
# HTTP method (GET or POST) # URL of each server in the service's cluster
method = "GET" servers = [
"http://localhost:9999/stats/",
"http://localhost:9998/stats/",
]
# Map of key transforms # TODO describe # HTTP method to use (case-sensitive)
[httpjson.measurements] method = "GET"
stats_measurements_measurement = "my_measurement"
# HTTP parameters (all values must be strings)
[httpjson.services.parameters]
event_type = "cpu_spike"
threshold = "0.75"
` `
func (h *HttpJson) SampleConfig() string { func (h *HttpJson) SampleConfig() string {
@ -40,71 +75,140 @@ func (h *HttpJson) Description() string {
return "Read flattened metrics from one or more JSON HTTP endpoints" return "Read flattened metrics from one or more JSON HTTP endpoints"
} }
// Gathers data for all servers.
func (h *HttpJson) Gather(acc plugins.Accumulator) error { func (h *HttpJson) Gather(acc plugins.Accumulator) error {
var wg sync.WaitGroup var wg sync.WaitGroup
var outerr error totalServers := 0
for _, service := range h.Services {
totalServers += len(service.Servers)
}
errorChannel := make(chan error, totalServers)
for _, server := range h.Servers { for _, service := range h.Services {
for _, server := range service.Servers {
wg.Add(1) wg.Add(1)
go func(server string) { go func(service Service, server string) {
defer wg.Done() defer wg.Done()
outerr = h.gatherServer(server, acc) if err := h.gatherServer(acc, service, server); err != nil {
}(server) errorChannel <- err
}
}(service, server)
}
} }
wg.Wait() wg.Wait()
close(errorChannel)
return outerr // Get all errors and return them as one giant error
errorStrings := []string{}
for err := range errorChannel {
errorStrings = append(errorStrings, err.Error())
}
if len(errorStrings) == 0 {
return nil
}
return errors.New(strings.Join(errorStrings, "\n"))
} }
func (h *HttpJson) gatherServer(url string, acc plugins.Accumulator) error { // Gathers data from a particular server
r, err := h.client.Get(url) // Parameters:
// acc : The telegraf Accumulator to use
// serverURL: endpoint to send request to
// service : the service being queried
//
// Returns:
// error: Any error that may have occurred
func (h *HttpJson) gatherServer(acc plugins.Accumulator, service Service, serverURL string) error {
resp, err := h.sendRequest(service, serverURL)
if err != nil { if err != nil {
return err return err
} }
if r.StatusCode != http.StatusOK { var jsonOut interface{}
return fmt.Errorf("httpjson: server '%s' responded with status-code %d, expected %d", r.StatusCode, http.StatusOK) if err = json.Unmarshal([]byte(resp), &jsonOut); err != nil {
} return errors.New("Error decoding JSON response")
response, err := simplejson.NewFromReader(r.Body)
if err != nil {
return err
} }
tags := map[string]string{ tags := map[string]string{
"server": url, "server": serverURL,
} }
return parseResponse(acc, h.Foo, tags, response.Interface(), h.Measurements) processResponse(acc, service.Name, tags, jsonOut)
return nil
} }
func parseResponse(acc plugins.Accumulator, prefix string, tags map[string]string, v interface{}, measurements map[string]string) error { // Sends an HTTP request to the server using the HttpJson object's HTTPClient
// Parameters:
// serverURL: endpoint to send request to
//
// Returns:
// string: body of the response
// error : Any error that may have occurred
func (h *HttpJson) sendRequest(service Service, serverURL string) (string, error) {
// Prepare URL
requestURL, err := url.Parse(serverURL)
if err != nil {
return "", fmt.Errorf("Invalid server URL \"%s\"", serverURL)
}
params := url.Values{}
for k, v := range service.Parameters {
params.Add(k, v)
}
requestURL.RawQuery = params.Encode()
// Create + send request
req, err := http.NewRequest(service.Method, requestURL.String(), nil)
if err != nil {
return "", err
}
resp, err := h.client.MakeRequest(req)
if err != nil {
return "", err
}
defer resp.Body.Close()
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return string(body), err
}
// Process response
if resp.StatusCode != http.StatusOK {
err = fmt.Errorf("Response from url \"%s\" has status code %d (%s), expected %d (%s)",
requestURL.String(),
resp.StatusCode,
http.StatusText(resp.StatusCode),
http.StatusOK,
http.StatusText(http.StatusOK))
return string(body), err
}
return string(body), err
}
// Flattens the map generated from the JSON object and stores its float values using a
// plugins.Accumulator. It ignores any non-float values.
// Parameters:
// acc: the Accumulator to use
// prefix: What the name of the measurement name should be prefixed by.
// tags: telegraf tags to
func processResponse(acc plugins.Accumulator, prefix string, tags map[string]string, v interface{}) {
switch t := v.(type) { switch t := v.(type) {
case map[string]interface{}: case map[string]interface{}:
for k, v := range t { for k, v := range t {
if err := parseResponse(acc, prefix+"_"+k, tags, v, measurements); err != nil { processResponse(acc, prefix+"_"+k, tags, v)
return err
} }
case float64:
acc.Add(prefix, v, tags)
} }
case json.Number:
if transform, ok := measurements[prefix]; ok {
prefix = transform
}
acc.Add(prefix, t, tags)
case bool, string, []interface{}:
// ignored types
return nil
default:
return fmt.Errorf("httpjson: got unexpected type %T with value %v (%s)", t, v, prefix)
}
return nil
} }
func init() { func init() {
plugins.Add("httpjson", func() plugins.Plugin { plugins.Add("httpjson", func() plugins.Plugin {
return &HttpJson{client: http.DefaultClient} return &HttpJson{client: RealHTTPClient{client: &http.Client{}}}
}) })
} }

View File

@ -0,0 +1,187 @@
package httpjson
import (
"fmt"
"io/ioutil"
"net/http"
"strings"
"testing"
"github.com/influxdb/telegraf/testutil"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
const validJSON = `
{
"parent": {
"child": 3,
"ignored_child": "hi"
},
"ignored_null": null,
"integer": 4,
"ignored_list": [3, 4],
"ignored_parent": {
"another_ignored_list": [4],
"another_ignored_null": null,
"ignored_string": "hello, world!"
}
}`
const invalidJSON = "I don't think this is JSON"
const empty = ""
type mockHTTPClient struct {
responseBody string
statusCode int
}
// Mock implementation of MakeRequest. Usually returns an http.Response with
// hard-coded responseBody and statusCode. However, if the request uses a
// nonstandard method, it uses status code 405 (method not allowed)
func (c mockHTTPClient) MakeRequest(req *http.Request) (*http.Response, error) {
resp := http.Response{}
resp.StatusCode = c.statusCode
// basic error checking on request method
allowedMethods := []string{"GET", "HEAD", "POST", "PUT", "DELETE", "TRACE", "CONNECT"}
methodValid := false
for _, method := range allowedMethods {
if req.Method == method {
methodValid = true
break
}
}
if !methodValid {
resp.StatusCode = 405 // Method not allowed
}
resp.Body = ioutil.NopCloser(strings.NewReader(c.responseBody))
return &resp, nil
}
// Generates a pointer to an HttpJson object that uses a mock HTTP client.
// Parameters:
// response : Body of the response that the mock HTTP client should return
// statusCode: HTTP status code the mock HTTP client should return
//
// Returns:
// *HttpJson: Pointer to an HttpJson object that uses the generated mock HTTP client
func genMockHttpJson(response string, statusCode int) *HttpJson {
return &HttpJson{
client: mockHTTPClient{responseBody: response, statusCode: statusCode},
Services: []Service{
Service{
Servers: []string{
"http://server1.example.com/metrics/",
"http://server2.example.com/metrics/",
},
Name: "my_webapp",
Method: "GET",
Parameters: map[string]string{
"httpParam1": "12",
"httpParam2": "the second parameter",
},
},
Service{
Servers: []string{
"http://server1.example.com/metrics/",
"http://server2.example.com/metrics/",
},
Name: "other_webapp",
Method: "POST",
Parameters: map[string]string{
"httpParam1": "12",
"httpParam2": "the second parameter",
},
},
},
}
}
// Test that the proper values are ignored or collected
func TestHttpJson200(t *testing.T) {
httpjson := genMockHttpJson(validJSON, 200)
var acc testutil.Accumulator
err := httpjson.Gather(&acc)
require.NoError(t, err)
assert.Equal(t, 8, len(acc.Points))
for _, service := range httpjson.Services {
for _, srv := range service.Servers {
require.NoError(t,
acc.ValidateTaggedValue(
fmt.Sprintf("%s_parent_child", service.Name),
3.0,
map[string]string{"server": srv},
),
)
require.NoError(t,
acc.ValidateTaggedValue(
fmt.Sprintf("%s_integer", service.Name),
4.0,
map[string]string{"server": srv},
),
)
}
}
}
// Test response to HTTP 500
func TestHttpJson500(t *testing.T) {
httpjson := genMockHttpJson(validJSON, 500)
var acc testutil.Accumulator
err := httpjson.Gather(&acc)
assert.NotNil(t, err)
// 4 error lines for (2 urls) * (2 services)
assert.Equal(t, len(strings.Split(err.Error(), "\n")), 4)
assert.Equal(t, 0, len(acc.Points))
}
// Test response to HTTP 405
func TestHttpJsonBadMethod(t *testing.T) {
httpjson := genMockHttpJson(validJSON, 200)
httpjson.Services[0].Method = "NOT_A_REAL_METHOD"
var acc testutil.Accumulator
err := httpjson.Gather(&acc)
assert.NotNil(t, err)
// 2 error lines for (2 urls) * (1 falied service)
assert.Equal(t, len(strings.Split(err.Error(), "\n")), 2)
// (2 measurements) * (2 servers) * (1 successful service)
assert.Equal(t, 4, len(acc.Points))
}
// Test response to malformed JSON
func TestHttpJsonBadJson(t *testing.T) {
httpjson := genMockHttpJson(invalidJSON, 200)
var acc testutil.Accumulator
err := httpjson.Gather(&acc)
assert.NotNil(t, err)
// 4 error lines for (2 urls) * (2 services)
assert.Equal(t, len(strings.Split(err.Error(), "\n")), 4)
assert.Equal(t, 0, len(acc.Points))
}
// Test response to empty string as response objectgT
func TestHttpJsonEmptyResponse(t *testing.T) {
httpjson := genMockHttpJson(empty, 200)
var acc testutil.Accumulator
err := httpjson.Gather(&acc)
assert.NotNil(t, err)
// 4 error lines for (2 urls) * (2 services)
assert.Equal(t, len(strings.Split(err.Error(), "\n")), 4)
assert.Equal(t, 0, len(acc.Points))
}