telegraf/plugins/inputs/postgresql/postgresql.go

212 lines
4.8 KiB
Go
Raw Normal View History

2015-05-18 17:46:44 +00:00
package postgresql
import (
"bytes"
2015-05-18 17:46:44 +00:00
"database/sql"
"fmt"
"regexp"
"sort"
"strings"
2015-05-18 17:46:44 +00:00
"github.com/influxdata/telegraf"
2016-01-20 18:57:35 +00:00
"github.com/influxdata/telegraf/plugins/inputs"
2015-05-18 17:46:44 +00:00
"github.com/lib/pq"
2015-05-18 17:46:44 +00:00
)
type Postgresql struct {
Address string
Databases []string
OrderedColumns []string
AllColumns []string
sanitizedAddress string
2015-05-18 17:46:44 +00:00
}
var ignoredColumns = map[string]bool{"datid": true, "datname": true, "stats_reset": true}
var sampleConfig = `
## specify address via a url matching:
## postgres://[pqgotest[:password]]@localhost[/dbname]?sslmode=[disable|verify-ca|verify-full]
## or a simple string:
## host=localhost user=pqotest password=... sslmode=... dbname=app_production
##
## All connection parameters are optional.
##
## Without the dbname parameter, the driver will default to a database
## with the same name as the user. This dbname is just for instantiating a
## connection with the server and doesn't restrict the databases we are trying
## to grab metrics for.
##
2015-12-19 00:09:01 +00:00
address = "host=localhost user=postgres sslmode=disable"
## A list of databases to pull metrics about. If not specified, metrics for all
## databases are gathered.
# databases = ["app_production", "testing"]
`
func (p *Postgresql) SampleConfig() string {
return sampleConfig
}
func (p *Postgresql) Description() string {
return "Read metrics from one or many postgresql servers"
}
func (p *Postgresql) IgnoredColumns() map[string]bool {
return ignoredColumns
}
var localhost = "host=localhost sslmode=disable"
func (p *Postgresql) Gather(acc telegraf.Accumulator) error {
var query string
if p.Address == "" || p.Address == "localhost" {
p.Address = localhost
2015-05-18 22:22:04 +00:00
}
db, err := sql.Open("postgres", p.Address)
2015-05-18 17:46:44 +00:00
if err != nil {
return err
}
defer db.Close()
if len(p.Databases) == 0 {
query = `SELECT * FROM pg_stat_database`
} else {
query = fmt.Sprintf(`SELECT * FROM pg_stat_database WHERE datname IN ('%s')`,
strings.Join(p.Databases, "','"))
}
2015-05-18 17:46:44 +00:00
rows, err := db.Query(query)
if err != nil {
return err
}
2015-05-18 17:46:44 +00:00
defer rows.Close()
2015-05-18 17:46:44 +00:00
2015-09-13 16:51:50 +00:00
// grab the column information from the result
p.OrderedColumns, err = rows.Columns()
if err != nil {
return err
} else {
p.AllColumns = make([]string, len(p.OrderedColumns))
copy(p.AllColumns, p.OrderedColumns)
}
2015-05-18 17:46:44 +00:00
for rows.Next() {
err = p.accRow(rows, acc)
if err != nil {
return err
2015-05-18 17:46:44 +00:00
}
}
//return rows.Err()
query = `SELECT * FROM pg_stat_bgwriter`
bg_writer_row, err := db.Query(query)
if err != nil {
return err
}
defer bg_writer_row.Close()
// grab the column information from the result
p.OrderedColumns, err = bg_writer_row.Columns()
if err != nil {
return err
} else {
for _, v := range p.OrderedColumns {
p.AllColumns = append(p.AllColumns, v)
}
}
2015-05-18 17:46:44 +00:00
for bg_writer_row.Next() {
err = p.accRow(bg_writer_row, acc)
if err != nil {
return err
}
}
sort.Strings(p.AllColumns)
return bg_writer_row.Err()
2015-05-18 17:46:44 +00:00
}
type scanner interface {
Scan(dest ...interface{}) error
}
var passwordKVMatcher, _ = regexp.Compile("password=\\S+ ?")
func (p *Postgresql) SanitizedAddress() (_ string, err error) {
var canonicalizedAddress string
if strings.HasPrefix(p.Address, "postgres://") || strings.HasPrefix(p.Address, "postgresql://") {
canonicalizedAddress, err = pq.ParseURL(p.Address)
if err != nil {
return p.sanitizedAddress, err
}
} else {
canonicalizedAddress = p.Address
}
p.sanitizedAddress = passwordKVMatcher.ReplaceAllString(canonicalizedAddress, "")
return p.sanitizedAddress, err
}
func (p *Postgresql) accRow(row scanner, acc telegraf.Accumulator) error {
var columnVars []interface{}
var dbname bytes.Buffer
2015-09-13 16:51:50 +00:00
// this is where we'll store the column name with its *interface{}
columnMap := make(map[string]*interface{})
for _, column := range p.OrderedColumns {
columnMap[column] = new(interface{})
}
2015-09-13 16:51:50 +00:00
// populate the array of interface{} with the pointers in the right order
for i := 0; i < len(columnMap); i++ {
columnVars = append(columnVars, columnMap[p.OrderedColumns[i]])
}
2015-09-13 16:51:50 +00:00
// deconstruct array of variables and send to Scan
err := row.Scan(columnVars...)
2015-05-18 17:46:44 +00:00
if err != nil {
return err
}
if columnMap["datname"] != nil {
// extract the database name from the column map
dbnameChars := (*columnMap["datname"]).([]uint8)
for i := 0; i < len(dbnameChars); i++ {
dbname.WriteString(string(dbnameChars[i]))
}
} else {
dbname.WriteString("postgres")
}
var tagAddress string
tagAddress, err = p.SanitizedAddress()
if err != nil {
return err
}
tags := map[string]string{"server": tagAddress, "db": dbname.String()}
2015-12-19 00:09:01 +00:00
fields := make(map[string]interface{})
for col, val := range columnMap {
_, ignore := ignoredColumns[col]
if !ignore {
2015-12-19 00:09:01 +00:00
fields[col] = *val
}
}
2015-12-19 00:09:01 +00:00
acc.AddFields("postgresql", fields, tags)
2015-05-18 17:46:44 +00:00
return nil
}
func init() {
inputs.Add("postgresql", func() telegraf.Input {
2015-05-18 17:46:44 +00:00
return &Postgresql{}
})
}