telegraf/plugins/inputs/haproxy/haproxy.go

306 lines
7.4 KiB
Go
Raw Permalink Normal View History

2015-07-22 23:38:12 +00:00
package haproxy
import (
"encoding/csv"
"fmt"
"io"
"net"
2015-07-22 23:38:12 +00:00
"net/http"
"net/url"
"path/filepath"
2015-07-22 23:38:12 +00:00
"strconv"
"strings"
2015-07-22 23:38:12 +00:00
"sync"
2015-12-14 22:45:29 +00:00
"time"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/plugins/inputs"
2015-07-22 23:38:12 +00:00
)
2016-09-08 10:11:24 +00:00
//CSV format: https://cbonte.github.io/haproxy-dconv/1.5/configuration.html#9.1
2015-07-22 23:38:12 +00:00
type haproxy struct {
Servers []string
client *http.Client
KeepFieldNames bool
// Path to CA file
SSLCA string `toml:"ssl_ca"`
// Path to host cert file
SSLCert string `toml:"ssl_cert"`
// Path to cert key file
SSLKey string `toml:"ssl_key"`
// Use SSL but skip chain & host verification
InsecureSkipVerify bool
2015-07-22 23:38:12 +00:00
}
var sampleConfig = `
## An array of address to gather stats about. Specify an ip on hostname
## with optional port. ie localhost, 10.10.3.33:1936, etc.
## Make sure you specify the complete path to the stats endpoint
## including the protocol, ie http://10.10.3.33:1936/haproxy?stats
2017-04-27 18:23:37 +00:00
## If no servers are specified, then default to 127.0.0.1:1936/haproxy?stats
servers = ["http://myhaproxy.com:1936/haproxy?stats"]
2017-04-27 18:23:37 +00:00
## You can also use local socket with standard wildcard globbing.
## Server address not starting with 'http' will be treated as a possible
## socket, so both examples below are valid.
2017-04-27 18:23:37 +00:00
# servers = ["socket:/run/haproxy/admin.sock", "/run/haproxy/*.sock"]
## By default, some of the fields are renamed from what haproxy calls them.
## Setting this option to true results in the plugin keeping the original
## field names.
2017-12-14 23:50:03 +00:00
# keep_field_names = false
## Optional SSL Config
# ssl_ca = "/etc/telegraf/ca.pem"
# ssl_cert = "/etc/telegraf/cert.pem"
# ssl_key = "/etc/telegraf/key.pem"
## Use SSL but skip chain & host verification
# insecure_skip_verify = false
2015-07-22 23:38:12 +00:00
`
func (r *haproxy) SampleConfig() string {
return sampleConfig
}
func (r *haproxy) Description() string {
return "Read metrics of haproxy, via socket or csv stats page"
}
// Reads stats from all configured servers accumulates stats.
// Returns one of the errors encountered while gather stats (if any).
func (g *haproxy) Gather(acc telegraf.Accumulator) error {
2015-07-22 23:38:12 +00:00
if len(g.Servers) == 0 {
return g.gatherServer("http://127.0.0.1:1936/haproxy?stats", acc)
2015-07-22 23:38:12 +00:00
}
endpoints := make([]string, 0, len(g.Servers))
for _, endpoint := range g.Servers {
if strings.HasPrefix(endpoint, "http") {
endpoints = append(endpoints, endpoint)
continue
}
socketPath := getSocketAddr(endpoint)
matches, err := filepath.Glob(socketPath)
if err != nil {
return err
}
if len(matches) == 0 {
endpoints = append(endpoints, socketPath)
} else {
for _, match := range matches {
endpoints = append(endpoints, match)
}
}
}
2015-07-22 23:38:12 +00:00
var wg sync.WaitGroup
wg.Add(len(endpoints))
for _, server := range endpoints {
2015-07-22 23:38:12 +00:00
go func(serv string) {
defer wg.Done()
if err := g.gatherServer(serv, acc); err != nil {
acc.AddError(err)
}
}(server)
2015-07-22 23:38:12 +00:00
}
wg.Wait()
return nil
2015-07-22 23:38:12 +00:00
}
func (g *haproxy) gatherServerSocket(addr string, acc telegraf.Accumulator) error {
socketPath := getSocketAddr(addr)
c, err := net.Dial("unix", socketPath)
if err != nil {
return fmt.Errorf("Could not connect to socket '%s': %s", addr, err)
}
_, errw := c.Write([]byte("show stat\n"))
if errw != nil {
return fmt.Errorf("Could not write to socket '%s': %s", addr, errw)
}
return g.importCsvResult(c, acc, socketPath)
}
func (g *haproxy) gatherServer(addr string, acc telegraf.Accumulator) error {
if !strings.HasPrefix(addr, "http") {
return g.gatherServerSocket(addr, acc)
}
2015-07-22 23:38:12 +00:00
if g.client == nil {
tlsCfg, err := internal.GetTLSConfig(
g.SSLCert, g.SSLKey, g.SSLCA, g.InsecureSkipVerify)
if err != nil {
return err
}
tr := &http.Transport{
ResponseHeaderTimeout: time.Duration(3 * time.Second),
TLSClientConfig: tlsCfg,
}
2016-02-29 16:52:58 +00:00
client := &http.Client{
Transport: tr,
Timeout: time.Duration(4 * time.Second),
}
2015-07-22 23:38:12 +00:00
g.client = client
}
if !strings.HasSuffix(addr, ";csv") {
addr += "/;csv"
}
2015-07-22 23:38:12 +00:00
u, err := url.Parse(addr)
if err != nil {
return fmt.Errorf("Unable parse server address '%s': %s", addr, err)
}
req, err := http.NewRequest("GET", addr, nil)
2015-07-22 23:38:12 +00:00
if u.User != nil {
p, _ := u.User.Password()
req.SetBasicAuth(u.User.Username(), p)
}
res, err := g.client.Do(req)
if err != nil {
return fmt.Errorf("Unable to connect to haproxy server '%s': %s", addr, err)
}
if res.StatusCode != 200 {
return fmt.Errorf("Unable to get valid stat result from '%s', http response code : %d", addr, res.StatusCode)
2015-07-22 23:38:12 +00:00
}
if err := g.importCsvResult(res.Body, acc, u.Host); err != nil {
return fmt.Errorf("Unable to parse stat result from '%s': %s", addr, err)
}
return nil
2015-07-22 23:38:12 +00:00
}
func getSocketAddr(sock string) string {
socketAddr := strings.Split(sock, ":")
if len(socketAddr) >= 2 {
return socketAddr[1]
} else {
return socketAddr[0]
}
}
var typeNames = []string{"frontend", "backend", "server", "listener"}
var fieldRenames = map[string]string{
"pxname": "proxy",
"svname": "sv",
"act": "active_servers",
"bck": "backup_servers",
"cli_abrt": "cli_abort",
"srv_abrt": "srv_abort",
"hrsp_1xx": "http_response.1xx",
"hrsp_2xx": "http_response.2xx",
"hrsp_3xx": "http_response.3xx",
"hrsp_4xx": "http_response.4xx",
"hrsp_5xx": "http_response.5xx",
"hrsp_other": "http_response.other",
}
func (g *haproxy) importCsvResult(r io.Reader, acc telegraf.Accumulator, host string) error {
csvr := csv.NewReader(r)
2015-12-14 22:45:29 +00:00
now := time.Now()
2015-07-22 23:38:12 +00:00
headers, err := csvr.Read()
if err != nil {
return err
}
if len(headers[0]) <= 2 || headers[0][:2] != "# " {
return fmt.Errorf("did not receive standard haproxy headers")
}
headers[0] = headers[0][2:]
for {
row, err := csvr.Read()
if err == io.EOF {
break
}
if err != nil {
return err
}
2015-12-19 00:09:01 +00:00
fields := make(map[string]interface{})
tags := map[string]string{
"server": host,
}
if len(row) != len(headers) {
return fmt.Errorf("number of columns does not match number of headers. headers=%d columns=%d", len(headers), len(row))
}
for i, v := range row {
if v == "" {
continue
}
colName := headers[i]
fieldName := colName
if !g.KeepFieldNames {
if fieldRename, ok := fieldRenames[colName]; ok {
fieldName = fieldRename
2015-07-22 23:38:12 +00:00
}
}
switch colName {
case "pxname", "svname":
tags[fieldName] = v
case "type":
vi, err := strconv.ParseInt(v, 10, 64)
if err != nil {
return fmt.Errorf("unable to parse type value '%s'", v)
}
if int(vi) >= len(typeNames) {
return fmt.Errorf("received unknown type value: %d", vi)
}
tags[fieldName] = typeNames[vi]
case "check_desc", "agent_desc":
// do nothing. These fields are just a more verbose description of the check_status & agent_status fields
case "status", "check_status", "last_chk", "mode", "tracked", "agent_status", "last_agt", "addr", "cookie":
// these are string fields
fields[fieldName] = v
case "lastsess":
vi, err := strconv.ParseInt(v, 10, 64)
if err != nil {
//TODO log the error. And just once (per column) so we don't spam the log
continue
}
fields[fieldName] = vi
default:
vi, err := strconv.ParseUint(v, 10, 64)
if err != nil {
//TODO log the error. And just once (per column) so we don't spam the log
continue
}
fields[fieldName] = vi
}
2015-07-22 23:38:12 +00:00
}
2015-12-14 22:45:29 +00:00
acc.AddFields("haproxy", fields, tags, now)
2015-07-22 23:38:12 +00:00
}
2015-12-14 22:45:29 +00:00
return err
2015-07-22 23:38:12 +00:00
}
func init() {
inputs.Add("haproxy", func() telegraf.Input {
2015-07-22 23:38:12 +00:00
return &haproxy{}
})
}