Skip to content

Commit

Permalink
make sql.DB connection details configurable
Browse files Browse the repository at this point in the history
  • Loading branch information
James Lawrence authored and james-lawrence committed Sep 25, 2017
1 parent 837e6b1 commit 5eb57c0
Show file tree
Hide file tree
Showing 7 changed files with 298 additions and 282 deletions.
6 changes: 5 additions & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,10 @@
`use_random_partitionkey` options has been deprecated in favor of the
`partition` subtable. This allows for more flexible methods to set the
partition key such as by metric name or by tag.
- `postgresql` plugins will now default to using a persistent connection to the database.
`Important` In environments TCP connections are terminated when idle for periods shorter than 15 minutes
and the collection interval is longer than the termination period then max_lifetime
should be set to be less than the collection interval to pervent errors when collecting metrics.

### Features

Expand All @@ -30,7 +34,7 @@

- [#3136](https://github.com/influxdata/telegraf/issues/3136): Fix webhooks input address in use during reload.
- [#3258](https://github.com/influxdata/telegraf/issues/3258): Unlock Statsd when stopping to prevent deadlock.

- [#1977](https://github.com/influxdata/telegraf/issues/1977): make postgresql connection pool persist between intervals.

## v1.4.1 [unreleased]

Expand Down
77 changes: 0 additions & 77 deletions plugins/inputs/postgresql/connect.go

This file was deleted.

91 changes: 33 additions & 58 deletions plugins/inputs/postgresql/postgresql.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,26 +2,21 @@ package postgresql

import (
"bytes"
"database/sql"
"fmt"
"regexp"
"sort"
"strings"

// register in driver.
_ "github.com/jackc/pgx/stdlib"

"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/plugins/inputs"
)

type Postgresql struct {
Address string
Service
Databases []string
IgnoredDatabases []string
OrderedColumns []string
AllColumns []string
sanitizedAddress string
}

var ignoredColumns = map[string]bool{"stats_reset": true}
Expand All @@ -41,6 +36,15 @@ var sampleConfig = `
## to grab metrics for.
##
address = "host=localhost user=postgres sslmode=disable"
## A custom name for the database that will be used as the "server" tag in the
## measurement output. If not specified, a default one generated from
## the connection address is used.
# outputaddress = "db01"
## connection configuration.
## maxlifetime - specify the maximum lifetime of a connection.
## default is forever (0s)
max_lifetime = "0s"
## A list of databases to explicitly ignore. If not specified, metrics for all
## databases are gathered. Do NOT use with the 'databases' option.
Expand All @@ -63,24 +67,13 @@ func (p *Postgresql) IgnoredColumns() map[string]bool {
return ignoredColumns
}

var localhost = "host=localhost sslmode=disable"

func (p *Postgresql) Gather(acc telegraf.Accumulator) error {
var (
err error
db *sql.DB
query string
err error
query string
columns []string
)

if p.Address == "" || p.Address == "localhost" {
p.Address = localhost
}

if db, err = sql.Open("pgx", p.Address); err != nil {
return err
}
defer db.Close()

if len(p.Databases) == 0 && len(p.IgnoredDatabases) == 0 {
query = `SELECT * FROM pg_stat_database`
} else if len(p.IgnoredDatabases) != 0 {
Expand All @@ -91,93 +84,67 @@ func (p *Postgresql) Gather(acc telegraf.Accumulator) error {
strings.Join(p.Databases, "','"))
}

rows, err := db.Query(query)
rows, err := p.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 {
if columns, err = rows.Columns(); 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)
err = p.accRow(rows, acc, columns)
if err != nil {
return err
}
}

query = `SELECT * FROM pg_stat_bgwriter`

bg_writer_row, err := db.Query(query)
bg_writer_row, err := p.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 {
if columns, err = bg_writer_row.Columns(); 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)
err = p.accRow(bg_writer_row, acc, columns)
if err != nil {
return err
}
}
sort.Strings(p.AllColumns)

return bg_writer_row.Err()
}

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 = 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 {
func (p *Postgresql) accRow(row scanner, acc telegraf.Accumulator, columns []string) 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 {
for _, column := range columns {
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]])
columnVars = append(columnVars, columnMap[columns[i]])
}

// deconstruct array of variables and send to Scan
Expand Down Expand Up @@ -215,6 +182,14 @@ func (p *Postgresql) accRow(row scanner, acc telegraf.Accumulator) error {

func init() {
inputs.Add("postgresql", func() telegraf.Input {
return &Postgresql{}
return &Postgresql{
Service: Service{
MaxIdle: 1,
MaxOpen: 1,
MaxLifetime: internal.Duration{
Duration: 0,
},
},
}
})
}
Loading

0 comments on commit 5eb57c0

Please sign in to comment.