telegraf/plugins/inputs/postgresql/postgresql.go

187 lines
4.2 KiB
Go

package postgresql
import (
"bytes"
"database/sql"
"fmt"
"sort"
"strings"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/plugins/inputs"
_ "github.com/lib/pq"
)
type Postgresql struct {
Address string
Databases []string
OrderedColumns []string
AllColumns []string
}
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.
###
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
}
db, err := sql.Open("postgres", p.Address)
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, "','"))
}
rows, err := db.Query(query)
if err != nil {
return err
}
defer rows.Close()
// 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)
}
for rows.Next() {
err = p.accRow(rows, acc)
if err != nil {
return err
}
}
//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)
}
}
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()
}
type scanner interface {
Scan(dest ...interface{}) error
}
func (p *Postgresql) accRow(row scanner, acc telegraf.Accumulator) error {
var columnVars []interface{}
var dbname bytes.Buffer
// 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{})
}
// 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]])
}
// deconstruct array of variables and send to Scan
err := row.Scan(columnVars...)
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")
}
tags := map[string]string{"server": p.Address, "db": dbname.String()}
fields := make(map[string]interface{})
for col, val := range columnMap {
_, ignore := ignoredColumns[col]
if !ignore {
fields[col] = *val
}
}
acc.AddFields("postgresql", fields, tags)
return nil
}
func init() {
inputs.Add("postgresql", func() telegraf.Input {
return &Postgresql{}
})
}