func()

in plugins/inputs/postgresql_extensible/postgresql_extensible.go [114:199]

• 70 lines of code
• 18 McCabe index (conditional complexity)

func (p *Postgresql) Gather(acc telegraf.Accumulator) error {

	var sql_query string
	var query_addon string
	var db_version int
	var query string
	var tag_value string
	var meas_name string

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

	db, err := postgresql.Connect(p.Address)
	if err != nil {
		return err
	}

	defer db.Close()

	// Retreiving the database version

	query = `select substring(setting from 1 for 3) as version from pg_settings where name='server_version_num'`
	err = db.QueryRow(query).Scan(&db_version)
	if err != nil {
		return err
	}
	// We loop in order to process each query
	// Query is not run if Database version does not match the query version.

	for i := range p.Query {
		sql_query = p.Query[i].Sqlquery
		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 len(p.Databases) != 0 {
				query_addon = fmt.Sprintf(` IN ('%s')`,
					strings.Join(p.Databases, "','"))
			} else {
				query_addon = " is not null"
			}
		} else {
			query_addon = ""
		}
		sql_query += query_addon

		if p.Query[i].Version <= db_version {
			rows, err := db.Query(sql_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 {
				for _, v := range p.OrderedColumns {
					p.AllColumns = append(p.AllColumns, v)
				}
			}
			p.AdditionalTags = nil
			if tag_value != "" {
				tag_list := strings.Split(tag_value, ",")
				for t := range tag_list {
					p.AdditionalTags = append(p.AdditionalTags, tag_list[t])
				}
			}

			for rows.Next() {
				err = p.accRow(meas_name, rows, acc)
				if err != nil {
					return err
				}
			}
		}
	}
	return nil
}