telegraf/plugins/inputs/mysql/mysql.go

238 lines
4.2 KiB
Go
Raw Normal View History

2015-05-18 18:54:11 +00:00
package mysql
import (
"database/sql"
2016-02-29 16:52:58 +00:00
"net/url"
2015-05-18 18:54:11 +00:00
"strconv"
"strings"
2016-02-29 16:52:58 +00:00
"time"
2015-05-18 18:54:11 +00:00
_ "github.com/go-sql-driver/mysql"
"github.com/influxdata/telegraf"
2016-01-20 18:57:35 +00:00
"github.com/influxdata/telegraf/plugins/inputs"
2015-05-18 18:54:11 +00:00
)
type Mysql struct {
2015-05-18 22:22:04 +00:00
Servers []string
}
var sampleConfig = `
## specify servers via a url matching:
## [username[:password]@][protocol[(address)]]/[?tls=[true|false|skip-verify]]
## see https://github.com/go-sql-driver/mysql#dsn-data-source-name
## e.g.
## root:passwd@tcp(127.0.0.1:3306)/?tls=false
## root@tcp(127.0.0.1:3306)/?tls=false
##
## If no servers are specified, then localhost is used as the host.
servers = ["tcp(127.0.0.1:3306)/"]
2015-08-26 15:21:39 +00:00
`
2016-02-29 16:52:58 +00:00
var defaultTimeout = time.Second * time.Duration(5)
func (m *Mysql) SampleConfig() string {
return sampleConfig
}
func (m *Mysql) Description() string {
return "Read metrics from one or many mysql servers"
2015-05-18 18:54:11 +00:00
}
2015-05-18 22:22:04 +00:00
var localhost = ""
2015-05-18 18:54:11 +00:00
func (m *Mysql) Gather(acc telegraf.Accumulator) error {
2015-05-18 18:54:11 +00:00
if len(m.Servers) == 0 {
// if we can't get stats in this case, thats fine, don't report
// an error.
m.gatherServer(localhost, acc)
return nil
}
for _, serv := range m.Servers {
err := m.gatherServer(serv, acc)
if err != nil {
return err
}
}
return nil
}
type mapping struct {
onServer string
inExport string
}
var mappings = []*mapping{
2015-09-01 17:30:56 +00:00
{
onServer: "Aborted_",
inExport: "aborted_",
},
2015-05-18 18:54:11 +00:00
{
onServer: "Bytes_",
inExport: "bytes_",
2015-05-18 18:54:11 +00:00
},
{
onServer: "Com_",
inExport: "commands_",
2015-05-18 18:54:11 +00:00
},
2015-09-01 17:30:56 +00:00
{
onServer: "Created_",
inExport: "created_",
},
2015-05-18 18:54:11 +00:00
{
onServer: "Handler_",
inExport: "handler_",
2015-05-18 18:54:11 +00:00
},
{
onServer: "Innodb_",
inExport: "innodb_",
2015-05-18 18:54:11 +00:00
},
2015-09-01 17:30:56 +00:00
{
onServer: "Key_",
inExport: "key_",
},
{
onServer: "Open_",
inExport: "open_",
},
{
onServer: "Opened_",
inExport: "opened_",
},
{
onServer: "Qcache_",
inExport: "qcache_",
},
{
onServer: "Table_",
inExport: "table_",
},
2015-07-19 20:01:45 +00:00
{
2015-08-04 22:09:59 +00:00
onServer: "Tokudb_",
inExport: "tokudb_",
},
2015-05-18 18:54:11 +00:00
{
onServer: "Threads_",
inExport: "threads_",
2015-05-18 18:54:11 +00:00
},
}
func (m *Mysql) gatherServer(serv string, acc telegraf.Accumulator) error {
// If user forgot the '/', add it
if strings.HasSuffix(serv, ")") {
serv = serv + "/"
} else if serv == "localhost" {
2015-05-18 23:08:22 +00:00
serv = ""
}
2016-02-29 16:52:58 +00:00
serv, err := dsnAddTimeout(serv)
if err != nil {
return err
}
2015-05-18 22:22:04 +00:00
db, err := sql.Open("mysql", serv)
2015-05-18 18:54:11 +00:00
if err != nil {
return err
}
defer db.Close()
rows, err := db.Query(`SHOW /*!50002 GLOBAL */ STATUS`)
if err != nil {
return err
2015-05-18 18:54:11 +00:00
}
var servtag string
servtag, err = parseDSN(serv)
if err != nil {
servtag = "localhost"
}
2015-12-15 00:03:33 +00:00
tags := map[string]string{"server": servtag}
fields := make(map[string]interface{})
2015-05-18 18:54:11 +00:00
for rows.Next() {
var name string
var val interface{}
err = rows.Scan(&name, &val)
if err != nil {
return err
}
var found bool
for _, mapped := range mappings {
if strings.HasPrefix(name, mapped.onServer) {
i, _ := strconv.Atoi(string(val.([]byte)))
2015-12-15 00:03:33 +00:00
fields[mapped.inExport+name[len(mapped.onServer):]] = i
2015-05-18 18:54:11 +00:00
found = true
}
}
if found {
continue
}
switch name {
case "Queries":
i, err := strconv.ParseInt(string(val.([]byte)), 10, 64)
if err != nil {
return err
}
2015-12-15 00:03:33 +00:00
fields["queries"] = i
2015-05-18 18:54:11 +00:00
case "Slow_queries":
i, err := strconv.ParseInt(string(val.([]byte)), 10, 64)
if err != nil {
return err
}
2015-12-15 00:03:33 +00:00
fields["slow_queries"] = i
2015-05-18 18:54:11 +00:00
}
}
2015-12-15 00:03:33 +00:00
acc.AddFields("mysql", fields, tags)
2015-05-18 18:54:11 +00:00
conn_rows, err := db.Query("SELECT user, sum(1) FROM INFORMATION_SCHEMA.PROCESSLIST GROUP BY user")
for conn_rows.Next() {
var user string
var connections int64
err = conn_rows.Scan(&user, &connections)
if err != nil {
return err
}
tags := map[string]string{"server": servtag, "user": user}
2015-12-15 00:03:33 +00:00
fields := make(map[string]interface{})
if err != nil {
return err
}
2015-12-15 00:03:33 +00:00
fields["connections"] = connections
acc.AddFields("mysql_users", fields, tags)
}
2015-05-18 18:54:11 +00:00
return nil
}
2016-02-29 16:52:58 +00:00
func dsnAddTimeout(dsn string) (string, error) {
u, err := url.Parse(dsn)
if err != nil {
return "", err
}
v := u.Query()
// Only override timeout if not already defined
if _, ok := v["timeout"]; ok == false {
v.Add("timeout", defaultTimeout.String())
u.RawQuery = v.Encode()
}
return u.String(), nil
}
2015-05-18 18:54:11 +00:00
func init() {
inputs.Add("mysql", func() telegraf.Input {
2015-05-18 18:54:11 +00:00
return &Mysql{}
})
}