dom
/
telegraf
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

postgresql_extensible: configurable measurement name 

The output measurement name can be configured per query.
This commit is contained in:
Mika Eloranta 2016-04-25 15:10:13 +03:00 committed by Cameron Sparr
parent 01b01c5969
commit ce7d852d22
1 changed files with 26 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
plugins/inputs/postgresql_extensible/postgresql_extensible.go
View File

@ -21,18 +21,20 @@ type Postgresql struct {
AdditionalTags []string AdditionalTags []string
sanitizedAddress string sanitizedAddress string
Query []struct { Query []struct {
Sqlquery string Sqlquery string
Version int Version int
Withdbname bool Withdbname bool
Tagvalue string Tagvalue string
Measurement string
} }
} }
type query []struct { type query []struct {
Sqlquery string Sqlquery string
Version int Version int
Withdbname bool Withdbname bool
Tagvalue string Tagvalue string
Measurement string
} }
var ignoredColumns = map[string]bool{"datid": true, "datname": true, "stats_reset": true} var ignoredColumns = map[string]bool{"datid": true, "datname": true, "stats_reset": true}
@ -65,24 +67,28 @@ var sampleConfig = `
## because the databases variable was set to ['postgres', 'pgbench' ] and the ## because the databases variable was set to ['postgres', 'pgbench' ] and the
## withdbname was true. Be careful that if the withdbname is set to false you ## withdbname was true. Be careful that if the withdbname is set to false you
## don't have to define the where clause (aka with the dbname) the tagvalue ## don't have to define the where clause (aka with the dbname) the tagvalue
## field is used to define custom tags (separated by comas) ## field is used to define custom tags (separated by commas)
## The optional "measurement" value can be used to override the default
## output measurement name ("postgresql").
# #
## Structure : ## Structure :
## [[inputs.postgresql_extensible.query]] ## [[inputs.postgresql_extensible.query]]
## sqlquery string ## sqlquery string
## version string ## version string
## withdbname boolean ## withdbname boolean
## tagvalue string (coma separated) ## tagvalue string (comma separated)
## measurement string
[[inputs.postgresql_extensible.query]] [[inputs.postgresql_extensible.query]]
sqlquery="SELECT * FROM pg_stat_database" sqlquery="SELECT * FROM pg_stat_database"
version=901 version=901
withdbname=false withdbname=false
tagvalue="" tagvalue=""
measurement=""
[[inputs.postgresql_extensible.query]] [[inputs.postgresql_extensible.query]]
sqlquery="SELECT * FROM pg_stat_bgwriter" sqlquery="SELECT * FROM pg_stat_bgwriter"
version=901 version=901
withdbname=false withdbname=false
tagvalue="" tagvalue="postgresql.stats"
` `
func (p *Postgresql) SampleConfig() string { func (p *Postgresql) SampleConfig() string {
@ -106,6 +112,7 @@ func (p *Postgresql) Gather(acc telegraf.Accumulator) error {
var db_version int var db_version int
var query string var query string
var tag_value string var tag_value string
var meas_name string
if p.Address == "" || p.Address == "localhost" { if p.Address == "" || p.Address == "localhost" {
p.Address = localhost p.Address = localhost
@ -131,6 +138,11 @@ func (p *Postgresql) Gather(acc telegraf.Accumulator) error {
for i := range p.Query { for i := range p.Query {
sql_query = p.Query[i].Sqlquery sql_query = p.Query[i].Sqlquery
tag_value = p.Query[i].Tagvalue tag_value = p.Query[i].Tagvalue
if p.Query[i].Measurement != "" {
meas_name = p.Query[i].Measurement
} else {
meas_name = "postgresql"
}
if p.Query[i].Withdbname { if p.Query[i].Withdbname {
if len(p.Databases) != 0 { if len(p.Databases) != 0 {
@ -170,7 +182,7 @@ func (p *Postgresql) Gather(acc telegraf.Accumulator) error {
} }
for rows.Next() { for rows.Next() {
err = p.accRow(rows, acc) err = p.accRow(meas_name, rows, acc)
if err != nil { if err != nil {
return err return err
} }
@ -201,7 +213,7 @@ func (p *Postgresql) SanitizedAddress() (_ string, err error) {
return p.sanitizedAddress, err return p.sanitizedAddress, err
} }
func (p *Postgresql) accRow(row scanner, acc telegraf.Accumulator) error { func (p *Postgresql) accRow(meas_name string, row scanner, acc telegraf.Accumulator) error {
var columnVars []interface{} var columnVars []interface{}
var dbname bytes.Buffer var dbname bytes.Buffer
@ -267,7 +279,7 @@ func (p *Postgresql) accRow(row scanner, acc telegraf.Accumulator) error {
} }
} }
} }
acc.AddFields("postgresql", fields, tags) acc.AddFields(meas_name, fields, tags)
return nil return nil
} }