telegraf/plugins/inputs/haproxy/haproxy.go

package haproxy

import (
        "encoding/csv"
        "fmt"
        "github.com/influxdata/telegraf"
        "github.com/influxdata/telegraf/plugins/inputs"
        "io"
        "net"
        "net/http"
        "net/url"
        "strconv"
        "strings"
        "sync"
        "time"
)

type haproxy struct {
        Servers []string

        client *http.Client
}

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.
  ## If no servers are specified, then default to 127.0.0.1:1936
  servers = ["http://myhaproxy.com:1936", "http://anotherhaproxy.com:1936"]
  ## Or you can also use local socket
  ## servers = ["socket:/run/haproxy/admin.sock"]
`

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 {
        if len(g.Servers) == 0 {
                return g.gatherServer("http://127.0.0.1:1936", acc)
        }

        var wg sync.WaitGroup

        var outerr error

        for _, serv := range g.Servers {
                wg.Add(1)
                go func(serv string) {
                        defer wg.Done()
                        outerr = g.gatherServer(serv, acc)
                }(serv)
        }

        wg.Wait()

        return outerr
}

func (g *haproxy) gatherServerSocket(addr string, acc telegraf.Accumulator) error {
        var socketPath string
        socketAddr := strings.Split(addr, ":")

        if len(socketAddr) >= 2 {
                socketPath = socketAddr[1]
        } else {
                socketPath = socketAddr[0]
        }

        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 importCsvResult(c, acc, socketPath)
}

func (g *haproxy) gatherServer(addr string, acc telegraf.Accumulator) error {
        if !strings.HasPrefix(addr, "http") {
                return g.gatherServerSocket(addr, acc)
        }

        if g.client == nil {
                tr := &http.Transport{ResponseHeaderTimeout: time.Duration(3 * time.Second)}
                client := &http.Client{
                        Transport: tr,
                        Timeout:   time.Duration(4 * time.Second),
                }
                g.client = client
        }

        u, err := url.Parse(addr)
        if err != nil {
                return fmt.Errorf("Unable parse server address '%s': %s", addr, err)
        }

        req, err := http.NewRequest("GET", fmt.Sprintf("%s://%s%s/;csv", u.Scheme, u.Host, u.Path), nil)
        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': %s", addr, err)
        }

        return importCsvResult(res.Body, acc, u.Host)
}

func importCsvResult(r io.Reader, acc telegraf.Accumulator, host string) error {
        csv := csv.NewReader(r)
        result, err := csv.ReadAll()
        now := time.Now()

        var keys []string
        px_sv_status := make(map[string]map[string]int)
        non_stat_fields := map[string]bool{
                "pxname":       true,
                "svname":       true,
                "status":       true,
                "tracked":      true,
                "check_status": true,
                "last_chk":     true,
                "pid":          true,
                "iid":          true,
                "sid":          true,
                "lastchg":      true,
                "type":         true,
                "check_code":   true,
        }

        for i := range result {
                if i == 0 {
                        keys = result[i]
                        keys[0] = strings.Replace(keys[0], "# ", "", -1)
                } else {

                        row := make(map[string]string, len(result[i]))

                        for f, v := range result[i] {
                                row[keys[f]] = v
                        }

                        tags := map[string]string{
                                "server": host,
                                "proxy":  row["pxname"],
                                "sv":     row["svname"],
                        }

                        if row["svname"] != "BACKEND" && row["svname"] != "FRONTEND" {
                                if len(px_sv_status[row["pxname"]]) == 0 {
                                        px_sv_status[row["pxname"]] = map[string]int{
                                                "_status_act":       0,
                                                "_status_act_up":    0,
                                                "_status_act_down":  0,
                                                "_status_act_maint": 0,
                                                "_status_act_drain": 0,
                                                "_status_act_other": 0,
                                                "_status_bck":       0,
                                                "_status_bck_up":    0,
                                                "_status_bck_down":  0,
                                                "_status_bck_maint": 0,
                                                "_status_bck_drain": 0,
                                                "_status_bck_other": 0,
                                        }
                                }

                                if row["act"] == "1" {
                                        px_sv_status[row["pxname"]]["_status_act"]++
                                        switch row["status"] {
                                        case "UP":
                                                px_sv_status[row["pxname"]]["_status_act_up"]++
                                        case "DOWN":
                                                px_sv_status[row["pxname"]]["_status_act_down"]++
                                        case "MAINT":
                                                px_sv_status[row["pxname"]]["_status_act_maint"]++
                                        case "DRAIN":
                                                px_sv_status[row["pxname"]]["_status_act_drain"]++
                                        default:
                                                px_sv_status[row["pxname"]]["_status_act_other"]++
                                        }

                                } else {
                                        px_sv_status[row["pxname"]]["_status_bck"]++
                                        switch row["status"] {
                                        case "UP":
                                                px_sv_status[row["pxname"]]["_status_bck_up"]++
                                        case "DOWN":
                                                px_sv_status[row["pxname"]]["_status_bck_down"]++
                                        case "MAINT":
                                                px_sv_status[row["pxname"]]["_status_bck_maint"]++
                                        case "DRAIN":
                                                px_sv_status[row["pxname"]]["_status_bck_drain"]++
                                        default:
                                                px_sv_status[row["pxname"]]["_status_bck_other"]++
                                        }
                                }
                        }

                        if row["svname"] == "BACKEND" && len(px_sv_status[row["pxname"]]) > 0 {
                                for s, c := range px_sv_status[row["pxname"]] {
                                        row[s] = strconv.Itoa(c)
                                }
                        }

                        fields := make(map[string]interface{})
                        for field, v := range row {
                                if non_stat_fields[field] {
                                        continue
                                }

                                ival, err := strconv.ParseUint(v, 10, 64)
                                if err == nil {
                                        fields[field] = ival
                                }
                        }
                        acc.AddFields("haproxy", fields, tags, now)
                }
        }
        return err
}

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