Httpjson

plugins/all/all.go

@@ -5,6 +5,7 @@ import (
 	_ "github.com/influxdb/telegraf/plugins/elasticsearch"
 	_ "github.com/influxdb/telegraf/plugins/exec"
 	_ "github.com/influxdb/telegraf/plugins/haproxy"
+	_ "github.com/influxdb/telegraf/plugins/httpjson"
 	_ "github.com/influxdb/telegraf/plugins/kafka_consumer"
 	_ "github.com/influxdb/telegraf/plugins/leofs"
 	_ "github.com/influxdb/telegraf/plugins/lustre2"

plugins/httpjson/httpjson.go

@@ -0,0 +1,214 @@
+package httpjson
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"strings"
+	"sync"
+
+	"github.com/influxdb/telegraf/plugins"
+)
+
+type HttpJson struct {
+	Services []Service
+	client   HTTPClient
+}
+
+type Service struct {
+	Name       string
+	Servers    []string
+	Method     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
+}
+
+func (c RealHTTPClient) MakeRequest(req *http.Request) (*http.Response, error) {
+	return c.client.Do(req)
+}
+
+var sampleConfig = `
+# Specify services via an array of tables
+[[httpjson.services]]
+
+	# a name for the service being polled
+	name = "webserver_stats"
+
+	# URL of each server in the service's cluster
+	servers = [
+		"http://localhost:9999/stats/",
+		"http://localhost:9998/stats/",
+	]
+
+	# HTTP method to use (case-sensitive)
+	method = "GET"
+
+	# HTTP parameters (all values must be strings)
+	[httpjson.services.parameters]
+		event_type = "cpu_spike"
+		threshold = "0.75"
+`
+
+func (h *HttpJson) SampleConfig() string {
+	return sampleConfig
+}
+
+func (h *HttpJson) Description() string {
+	return "Read flattened metrics from one or more JSON HTTP endpoints"
+}
+
+// Gathers data for all servers.
+func (h *HttpJson) Gather(acc plugins.Accumulator) error {
+	var wg sync.WaitGroup
+
+	totalServers := 0
+	for _, service := range h.Services {
+		totalServers += len(service.Servers)
+	}
+	errorChannel := make(chan error, totalServers)
+
+	for _, service := range h.Services {
+		for _, server := range service.Servers {
+			wg.Add(1)
+			go func(service Service, server string) {
+				defer wg.Done()
+				if err := h.gatherServer(acc, service, server); err != nil {
+					errorChannel <- err
+				}
+			}(service, server)
+		}
+	}
+
+	wg.Wait()
+	close(errorChannel)
+
+	// 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"))
+}
+
+// Gathers data from a particular server
+// 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 {
+		return err
+	}
+
+	var jsonOut interface{}
+	if err = json.Unmarshal([]byte(resp), &jsonOut); err != nil {
+		return errors.New("Error decoding JSON response")
+	}
+
+	tags := map[string]string{
+		"server": serverURL,
+	}
+
+	processResponse(acc, service.Name, tags, jsonOut)
+	return nil
+}
+
+// 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) {
+	case map[string]interface{}:
+		for k, v := range t {
+			processResponse(acc, prefix+"_"+k, tags, v)
+		}
+	case float64:
+		acc.Add(prefix, v, tags)
+	}
+}
+
+func init() {
+	plugins.Add("httpjson", func() plugins.Plugin {
+		return &HttpJson{client: RealHTTPClient{client: &http.Client{}}}
+	})
+}