152 lines
3.4 KiB
Go
152 lines
3.4 KiB
Go
package postgresql
|
|
|
|
import (
|
|
"bytes"
|
|
"database/sql"
|
|
"fmt"
|
|
"strings"
|
|
|
|
"github.com/influxdata/telegraf/plugins/inputs"
|
|
|
|
_ "github.com/lib/pq"
|
|
)
|
|
|
|
type Postgresql struct {
|
|
Address string
|
|
Databases []string
|
|
OrderedColumns []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 inputs.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
|
|
}
|
|
|
|
for rows.Next() {
|
|
err = p.accRow(rows, acc)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return rows.Err()
|
|
}
|
|
|
|
type scanner interface {
|
|
Scan(dest ...interface{}) error
|
|
}
|
|
|
|
func (p *Postgresql) accRow(row scanner, acc inputs.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
|
|
}
|
|
|
|
// extract the database name from the column map
|
|
dbnameChars := (*columnMap["datname"]).([]uint8)
|
|
for i := 0; i < len(dbnameChars); i++ {
|
|
dbname.WriteString(string(dbnameChars[i]))
|
|
}
|
|
|
|
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() inputs.Input {
|
|
return &Postgresql{}
|
|
})
|
|
}
|