nginx_vts.go

package nginx_vts

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/inputs"
)

type NginxVTS struct {
	Urls []string

	client *http.Client

	ResponseTimeout internal.Duration
}

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"
`

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

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

func (n *NginxVTS) 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 *NginxVTS) createHTTPClient() (*http.Client, error) {
	if n.ResponseTimeout.Duration < time.Second {
		n.ResponseTimeout.Duration = time.Second * 5
	}

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

	return client, nil
}

func (n *NginxVTS) 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 NginxVTSResponse 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"`
	ServerZones   map[string]Server            `json:"serverZones"`
	FilterZones   map[string]map[string]Server `json:"filterZones"`
	UpstreamZones map[string][]Upstream        `json:"upstreamZones"`
	CacheZones    map[string]Cache             `json:"cacheZones"`
}

type Server struct {
	RequestCounter uint64 `json:"requestCounter"`
	InBytes        uint64 `json:"inBytes"`
	OutBytes       uint64 `json:"outBytes"`
	RequestMsec    uint64 `json:"requestMsec"`
	Responses      struct {
		OneXx       uint64 `json:"1xx"`
		TwoXx       uint64 `json:"2xx"`
		ThreeXx     uint64 `json:"3xx"`
		FourXx      uint64 `json:"4xx"`
		FiveXx      uint64 `json:"5xx"`
		Miss        uint64 `json:"miss"`
		Bypass      uint64 `json:"bypass"`
		Expired     uint64 `json:"expired"`
		Stale       uint64 `json:"stale"`
		Updating    uint64 `json:"updating"`
		Revalidated uint64 `json:"revalidated"`
		Hit         uint64 `json:"hit"`
		Scarce      uint64 `json:"scarce"`
	} `json:"responses"`
}

type Upstream struct {
	Server         string `json:"server"`
	RequestCounter uint64 `json:"requestCounter"`
	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"`
	ResponseMsec uint64 `json:"responseMsec"`
	RequestMsec  uint64 `json:"requestMsec"`
	Weight       uint64 `json:"weight"`
	MaxFails     uint64 `json:"maxFails"`
	FailTimeout  uint64 `json:"failTimeout"`
	Backup       bool   `json:"backup"`
	Down         bool   `json:"down"`
}

type Cache struct {
	MaxSize   uint64 `json:"maxSize"`
	UsedSize  uint64 `json:"usedSize"`
	InBytes   uint64 `json:"inBytes"`
	OutBytes  uint64 `json:"outBytes"`
	Responses struct {
		Miss        uint64 `json:"miss"`
		Bypass      uint64 `json:"bypass"`
		Expired     uint64 `json:"expired"`
		Stale       uint64 `json:"stale"`
		Updating    uint64 `json:"updating"`
		Revalidated uint64 `json:"revalidated"`
		Hit         uint64 `json:"hit"`
		Scarce      uint64 `json:"scarce"`
	} `json:"responses"`
}

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

	acc.AddFields("nginx_vts_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.ServerZones {
		zoneTags := map[string]string{}
		for k, v := range tags {
			zoneTags[k] = v
		}
		zoneTags["zone"] = zoneName

		acc.AddFields("nginx_vts_server", map[string]interface{}{
			"requests":     zone.RequestCounter,
			"request_time": zone.RequestMsec,
			"in_bytes":     zone.InBytes,
			"out_bytes":    zone.OutBytes,

			"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,

			"cache_miss":        zone.Responses.Miss,
			"cache_bypass":      zone.Responses.Bypass,
			"cache_expired":     zone.Responses.Expired,
			"cache_stale":       zone.Responses.Stale,
			"cache_updating":    zone.Responses.Updating,
			"cache_revalidated": zone.Responses.Revalidated,
			"cache_hit":         zone.Responses.Hit,
			"cache_scarce":      zone.Responses.Scarce,
		}, zoneTags)
	}

	for filterName, filters := range status.FilterZones {
		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_vts_filter", map[string]interface{}{
				"requests":     upstream.RequestCounter,
				"request_time": upstream.RequestMsec,
				"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,

				"cache_miss":        upstream.Responses.Miss,
				"cache_bypass":      upstream.Responses.Bypass,
				"cache_expired":     upstream.Responses.Expired,
				"cache_stale":       upstream.Responses.Stale,
				"cache_updating":    upstream.Responses.Updating,
				"cache_revalidated": upstream.Responses.Revalidated,
				"cache_hit":         upstream.Responses.Hit,
				"cache_scarce":      upstream.Responses.Scarce,
			}, filterTags)
		}
	}

	for upstreamName, upstreams := range status.UpstreamZones {
		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_vts_upstream", map[string]interface{}{
				"requests":      upstream.RequestCounter,
				"request_time":  upstream.RequestMsec,
				"response_time": upstream.ResponseMsec,
				"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)
		}
	}

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

		acc.AddFields("nginx_vts_cache", map[string]interface{}{
			"max_bytes":  zone.MaxSize,
			"used_bytes": zone.UsedSize,
			"in_bytes":   zone.InBytes,
			"out_bytes":  zone.OutBytes,

			"miss":        zone.Responses.Miss,
			"bypass":      zone.Responses.Bypass,
			"expired":     zone.Responses.Expired,
			"stale":       zone.Responses.Stale,
			"updating":    zone.Responses.Updating,
			"revalidated": zone.Responses.Revalidated,
			"hit":         zone.Responses.Hit,
			"scarce":      zone.Responses.Scarce,
		}, zoneTags)
	}

	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_vts", func() telegraf.Input {
		return &NginxVTS{}
	})
}