nginx_sts.go

package nginx_sts

import (
	"bufio"
	"encoding/json"
	"fmt"
	"net"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"

	"github.com/influxdata/telegraf"
	"github.com/influxdata/telegraf/internal"
	"github.com/influxdata/telegraf/plugins/common/tls"
	"github.com/influxdata/telegraf/plugins/inputs"
)

type NginxSTS struct {
	Urls            []string          `toml:"urls"`
	ResponseTimeout internal.Duration `toml:"response_timeout"`
	tls.ClientConfig

	client *http.Client
}

var sampleConfig = `
  ## An array of ngx_http_status_module or status URI to gather stats.
  urls = ["http://localhost/status"]

  ## HTTP response timeout (default: 5s)
  response_timeout = "5s"

  ## Optional TLS Config
  # tls_ca = "/etc/telegraf/ca.pem"
  # tls_cert = "/etc/telegraf/cert.pem"
  # tls_key = "/etc/telegraf/key.pem"
  ## Use TLS but skip chain & host verification
  # insecure_skip_verify = false
`

func (n *NginxSTS) SampleConfig() string {
	return sampleConfig
}

func (n *NginxSTS) Description() string {
	return "Read Nginx virtual host traffic status module information (nginx-module-sts)"
}

func (n *NginxSTS) Gather(acc telegraf.Accumulator) error {
	var wg sync.WaitGroup

	// Create an HTTP client that is re-used for each
	// collection interval

	if n.client == nil {
		client, err := n.createHTTPClient()
		if err != nil {
			return err
		}
		n.client = client
	}

	for _, u := range n.Urls {
		addr, err := url.Parse(u)
		if err != nil {
			acc.AddError(fmt.Errorf("Unable to parse address '%s': %s", u, err))
			continue
		}

		wg.Add(1)
		go func(addr *url.URL) {
			defer wg.Done()
			acc.AddError(n.gatherURL(addr, acc))
		}(addr)
	}

	wg.Wait()
	return nil
}

func (n *NginxSTS) createHTTPClient() (*http.Client, error) {
	if n.ResponseTimeout.Duration < time.Second {
		n.ResponseTimeout.Duration = time.Second * 5
	}

	tlsConfig, err := n.ClientConfig.TLSConfig()
	if err != nil {
		return nil, err
	}

	client := &http.Client{
		Transport: &http.Transport{
			TLSClientConfig: tlsConfig,
		},
		Timeout: n.ResponseTimeout.Duration,
	}

	return client, nil
}

func (n *NginxSTS) gatherURL(addr *url.URL, acc telegraf.Accumulator) error {
	resp, err := n.client.Get(addr.String())
	if err != nil {
		return fmt.Errorf("error making HTTP request to %s: %s", addr.String(), err)
	}

	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("%s returned HTTP status %s", addr.String(), resp.Status)
	}
	contentType := strings.Split(resp.Header.Get("Content-Type"), ";")[0]
	switch contentType {
	case "application/json":
		return gatherStatusURL(bufio.NewReader(resp.Body), getTags(addr), acc)
	default:
		return fmt.Errorf("%s returned unexpected content type %s", addr.String(), contentType)
	}
}

type NginxSTSResponse struct {
	Connections struct {
		Active   uint64 `json:"active"`
		Reading  uint64 `json:"reading"`
		Writing  uint64 `json:"writing"`
		Waiting  uint64 `json:"waiting"`
		Accepted uint64 `json:"accepted"`
		Handled  uint64 `json:"handled"`
		Requests uint64 `json:"requests"`
	} `json:"connections"`
	Hostname            string                       `json:"hostName"`
	StreamFilterZones   map[string]map[string]Server `json:"streamFilterZones"`
	StreamServerZones   map[string]Server            `json:"streamServerZones"`
	StreamUpstreamZones map[string][]Upstream        `json:"streamUpstreamZones"`
}

type Server struct {
	ConnectCounter     uint64 `json:"connectCounter"`
	InBytes            uint64 `json:"inBytes"`
	OutBytes           uint64 `json:"outBytes"`
	SessionMsecCounter uint64 `json:"sessionMsecCounter"`
	SessionMsec        uint64 `json:"sessionMsec"`
	Responses          struct {
		OneXx   uint64 `json:"1xx"`
		TwoXx   uint64 `json:"2xx"`
		ThreeXx uint64 `json:"3xx"`
		FourXx  uint64 `json:"4xx"`
		FiveXx  uint64 `json:"5xx"`
	} `json:"responses"`
}

type Upstream struct {
	Server         string `json:"server"`
	ConnectCounter uint64 `json:"connectCounter"`
	InBytes        uint64 `json:"inBytes"`
	OutBytes       uint64 `json:"outBytes"`
	Responses      struct {
		OneXx   uint64 `json:"1xx"`
		TwoXx   uint64 `json:"2xx"`
		ThreeXx uint64 `json:"3xx"`
		FourXx  uint64 `json:"4xx"`
		FiveXx  uint64 `json:"5xx"`
	} `json:"responses"`
	SessionMsecCounter    uint64 `json:"sessionMsecCounter"`
	SessionMsec           uint64 `json:"sessionMsec"`
	USessionMsecCounter   uint64 `json:"uSessionMsecCounter"`
	USessionMsec          uint64 `json:"uSessionMsec"`
	UConnectMsecCounter   uint64 `json:"uConnectMsecCounter"`
	UConnectMsec          uint64 `json:"uConnectMsec"`
	UFirstByteMsecCounter uint64 `json:"uFirstByteMsecCounter"`
	UFirstByteMsec        uint64 `json:"uFirstByteMsec"`
	Weight                uint64 `json:"weight"`
	MaxFails              uint64 `json:"maxFails"`
	FailTimeout           uint64 `json:"failTimeout"`
	Backup                bool   `json:"backup"`
	Down                  bool   `json:"down"`
}

func gatherStatusURL(r *bufio.Reader, tags map[string]string, acc telegraf.Accumulator) error {
	dec := json.NewDecoder(r)
	status := &NginxSTSResponse{}
	if err := dec.Decode(status); err != nil {
		return fmt.Errorf("Error while decoding JSON response")
	}

	acc.AddFields("nginx_sts_connections", map[string]interface{}{
		"active":   status.Connections.Active,
		"reading":  status.Connections.Reading,
		"writing":  status.Connections.Writing,
		"waiting":  status.Connections.Waiting,
		"accepted": status.Connections.Accepted,
		"handled":  status.Connections.Handled,
		"requests": status.Connections.Requests,
	}, tags)

	for zoneName, zone := range status.StreamServerZones {
		zoneTags := map[string]string{}
		for k, v := range tags {
			zoneTags[k] = v
		}
		zoneTags["zone"] = zoneName

		acc.AddFields("nginx_sts_server", map[string]interface{}{
			"connects":             zone.ConnectCounter,
			"in_bytes":             zone.InBytes,
			"out_bytes":            zone.OutBytes,
			"session_msec_counter": zone.SessionMsecCounter,
			"session_msec":         zone.SessionMsec,

			"response_1xx_count": zone.Responses.OneXx,
			"response_2xx_count": zone.Responses.TwoXx,
			"response_3xx_count": zone.Responses.ThreeXx,
			"response_4xx_count": zone.Responses.FourXx,
			"response_5xx_count": zone.Responses.FiveXx,
		}, zoneTags)
	}

	for filterName, filters := range status.StreamFilterZones {
		for filterKey, upstream := range filters {
			filterTags := map[string]string{}
			for k, v := range tags {
				filterTags[k] = v
			}
			filterTags["filter_key"] = filterKey
			filterTags["filter_name"] = filterName

			acc.AddFields("nginx_sts_filter", map[string]interface{}{
				"connects":             upstream.ConnectCounter,
				"in_bytes":             upstream.InBytes,
				"out_bytes":            upstream.OutBytes,
				"session_msec_counter": upstream.SessionMsecCounter,
				"session_msec":         upstream.SessionMsec,

				"response_1xx_count": upstream.Responses.OneXx,
				"response_2xx_count": upstream.Responses.TwoXx,
				"response_3xx_count": upstream.Responses.ThreeXx,
				"response_4xx_count": upstream.Responses.FourXx,
				"response_5xx_count": upstream.Responses.FiveXx,
			}, filterTags)
		}
	}

	for upstreamName, upstreams := range status.StreamUpstreamZones {
		for _, upstream := range upstreams {
			upstreamServerTags := map[string]string{}
			for k, v := range tags {
				upstreamServerTags[k] = v
			}
			upstreamServerTags["upstream"] = upstreamName
			upstreamServerTags["upstream_address"] = upstream.Server
			acc.AddFields("nginx_sts_upstream", map[string]interface{}{
				"connects":                        upstream.ConnectCounter,
				"session_msec":                    upstream.SessionMsec,
				"session_msec_counter":            upstream.SessionMsecCounter,
				"upstream_session_msec":           upstream.USessionMsec,
				"upstream_session_msec_counter":   upstream.USessionMsecCounter,
				"upstream_connect_msec":           upstream.UConnectMsec,
				"upstream_connect_msec_counter":   upstream.UConnectMsecCounter,
				"upstream_firstbyte_msec":         upstream.UFirstByteMsec,
				"upstream_firstbyte_msec_counter": upstream.UFirstByteMsecCounter,
				"in_bytes":                        upstream.InBytes,
				"out_bytes":                       upstream.OutBytes,

				"response_1xx_count": upstream.Responses.OneXx,
				"response_2xx_count": upstream.Responses.TwoXx,
				"response_3xx_count": upstream.Responses.ThreeXx,
				"response_4xx_count": upstream.Responses.FourXx,
				"response_5xx_count": upstream.Responses.FiveXx,

				"weight":       upstream.Weight,
				"max_fails":    upstream.MaxFails,
				"fail_timeout": upstream.FailTimeout,
				"backup":       upstream.Backup,
				"down":         upstream.Down,
			}, upstreamServerTags)
		}
	}

	return nil
}

// Get tag(s) for the nginx plugin
func getTags(addr *url.URL) map[string]string {
	h := addr.Host
	host, port, err := net.SplitHostPort(h)
	if err != nil {
		host = addr.Host
		if addr.Scheme == "http" {
			port = "80"
		} else if addr.Scheme == "https" {
			port = "443"
		} else {
			port = ""
		}
	}
	return map[string]string{"source": host, "port": port}
}

func init() {
	inputs.Add("nginx_sts", func() telegraf.Input {
		return &NginxSTS{}
	})
}