Add nginx-module-vts input plugin. (#3782)
This commit is contained in:
Aleksejs Sinicins
committed by
Daniel Nelson
Daniel Nelson
parent
2ff2d03389
commit
9a864d11d2
341
plugins/inputs/nginx_vts/nginx_vts.go
Normal file
341
plugins/inputs/nginx_vts/nginx_vts.go
Normal file
@@ -0,0 +1,341 @@
|
||||
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{}
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user