src/postgresql.c

   1 /**
   2  * collectd - src/postgresql.c
   3  * Copyright (C) 2008  Sebastian Harl
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License as published by the
   7  * Free Software Foundation; only version 2 of the License is applicable.
   8  *
   9  * This program is distributed in the hope that it will be useful, but
  10  * WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along
  15  * with this program; if not, write to the Free Software Foundation, Inc.,
  16  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  17  *
  18  * Author:
  19  *   Sebastian Harl <sh at tokkee.org>
  20  **/
  21
  22 /*
  23  * This module collects PostgreSQL database statistics.
  24  */
  25
  26 #include "collectd.h"
  27 #include "common.h"
  28
  29 #include "configfile.h"
  30 #include "plugin.h"
  31
  32 #include "utils_complain.h"
  33
  34 #include <pg_config_manual.h>
  35 #include <libpq-fe.h>
  36
  37 #define log_err(...) ERROR ("postgresql: " __VA_ARGS__)
  38 #define log_warn(...) WARNING ("postgresql: " __VA_ARGS__)
  39 #define log_info(...) INFO ("postgresql: " __VA_ARGS__)
  40
  41 /* Appends the (parameter, value) pair to the string
  42  * pointed to by 'buf' suitable to be used as argument
  43  * for PQconnectdb(). If value equals NULL, the pair
  44  * is ignored. */
  45 #define C_PSQL_PAR_APPEND(buf, buf_len, parameter, value) \
  46         if ((0 < (buf_len)) && (NULL != (value)) && ('\0' != *(value))) { \
  47                 int s = ssnprintf (buf, buf_len, " %s = '%s'", parameter, value); \
  48                 if (0 < s) { \
  49                         buf     += s; \
  50                         buf_len -= s; \
  51                 } \
  52         }
  53
  54 /* Returns the tuple (major, minor, patchlevel)
  55  * for the given version number. */
  56 #define C_PSQL_SERVER_VERSION3(server_version) \
  57         (server_version) / 10000, \
  58         (server_version) / 100 - (int)((server_version) / 10000) * 100, \
  59         (server_version) - (int)((server_version) / 100) * 100
  60
  61 /* Returns true if the given host specifies a
  62  * UNIX domain socket. */
  63 #define C_PSQL_IS_UNIX_DOMAIN_SOCKET(host) \
  64         ((NULL == (host)) || ('\0' == *(host)) || ('/' == *(host)))
  65
  66 /* Returns the tuple (host, delimiter, port) for a
  67  * given (host, port) pair. Depending on the value of
  68  * 'host' a UNIX domain socket or a TCP socket is
  69  * assumed. */
  70 #define C_PSQL_SOCKET3(host, port) \
  71         ((NULL == (host)) || ('\0' == *(host))) ? DEFAULT_PGSOCKET_DIR : host, \
  72         C_PSQL_IS_UNIX_DOMAIN_SOCKET (host) ? "/.s.PGSQL." : ":", \
  73         port
  74
  75 typedef struct {
  76         PGconn      *conn;
  77         c_complain_t conn_complaint;
  78
  79         /* user configuration */
  80         char *host;
  81         char *port;
  82         char *database;
  83         char *user;
  84         char *password;
  85
  86         char *sslmode;
  87
  88         char *krbsrvname;
  89
  90         char *service;
  91 } c_psql_database_t;
  92
  93 static c_psql_database_t *databases     = NULL;
  94 static int                databases_num = 0;
  95
  96 static void submit (const c_psql_database_t *db,
  97                 const char *type, const char *type_instance,
  98                 value_t *values, size_t values_len)
  99 {
 100         value_list_t vl = VALUE_LIST_INIT;
 101
 102         vl.values     = values;
 103         vl.values_len = values_len;
 104         vl.time       = time (NULL);
 105
 106         if (C_PSQL_IS_UNIX_DOMAIN_SOCKET (db->host)
 107                         || (0 == strcmp (db->host, "localhost")))
 108                 sstrncpy (vl.host, hostname_g, sizeof (vl.host));
 109         else
 110                 sstrncpy (vl.host, db->host, sizeof (vl.host));
 111
 112         sstrncpy (vl.plugin, "postgresql", sizeof (vl.plugin));
 113         sstrncpy (vl.plugin_instance, db->database, sizeof (vl.plugin_instance));
 114
 115         sstrncpy (vl.type, type, sizeof (vl.type));
 116
 117         if (NULL != type_instance)
 118                 sstrncpy (vl.type_instance, type_instance, sizeof (vl.type_instance));
 119
 120         plugin_dispatch_values (&vl);
 121         return;
 122 } /* submit */
 123
 124 static void submit_counter (const c_psql_database_t *db,
 125                 const char *type, const char *type_instance,
 126                 const char *value)
 127 {
 128         value_t values[1];
 129
 130         if ((NULL == value) || ('\0' == *value))
 131                 return;
 132
 133         values[0].counter = atoll (value);
 134         submit (db, type, type_instance, values, 1);
 135         return;
 136 } /* submit_counter */
 137
 138 static void submit_gauge (const c_psql_database_t *db,
 139                 const char *type, const char *type_instance,
 140                 const char *value)
 141 {
 142         value_t values[1];
 143
 144         if ((NULL == value) || ('\0' == *value))
 145                 return;
 146
 147         values[0].gauge = atof (value);
 148         submit (db, type, type_instance, values, 1);
 149         return;
 150 } /* submit_gauge */
 151
 152 static int c_psql_check_connection (c_psql_database_t *db)
 153 {
 154         /* "ping" */
 155         PQclear (PQexec (db->conn, "SELECT 42;"));
 156
 157         if (CONNECTION_OK != PQstatus (db->conn)) {
 158                 PQreset (db->conn);
 159
 160                 /* trigger c_release() */
 161                 if (0 == db->conn_complaint.interval)
 162                         db->conn_complaint.interval = 1;
 163
 164                 if (CONNECTION_OK != PQstatus (db->conn)) {
 165                         c_complain (LOG_ERR, &db->conn_complaint,
 166                                         "Failed to connect to database %s: %s",
 167                                         db->database, PQerrorMessage (db->conn));
 168                         return -1;
 169                 }
 170         }
 171
 172         c_release (LOG_INFO, &db->conn_complaint,
 173                         "Successfully reconnected to database %s", PQdb (db->conn));
 174         return 0;
 175 } /* c_psql_check_connection */
 176
 177 static int c_psql_stat_database (c_psql_database_t *db)
 178 {
 179         const char *const query =
 180                 "SELECT numbackends, xact_commit, xact_rollback "
 181                         "FROM pg_stat_database "
 182                         "WHERE datname = $1;";
 183
 184         PGresult *res;
 185
 186         int n;
 187
 188         res = PQexecParams (db->conn, query, /* number of parameters */ 1,
 189                         NULL, (const char *const *)&db->database, NULL, NULL,
 190                         /* return text data */ 0);
 191
 192         if (PGRES_TUPLES_OK != PQresultStatus (res)) {
 193                 log_err ("Failed to execute SQL query: %s",
 194                                 PQerrorMessage (db->conn));
 195                 log_info ("SQL query was: %s", query);
 196                 PQclear (res);
 197                 return -1;
 198         }
 199
 200         n = PQntuples (res);
 201         if (1 < n) {
 202                 log_warn ("pg_stat_database has more than one entry "
 203                                 "for database %s - ignoring additional results.",
 204                                 db->database);
 205         }
 206         else if (1 > n) {
 207                 log_err ("pg_stat_database has no entry for database %s",
 208                                 db->database);
 209                 PQclear (res);
 210                 return -1;
 211         }
 212
 213         submit_gauge (db, "pg_numbackends", NULL,  PQgetvalue (res, 0, 0));
 214
 215         submit_counter (db, "pg_xact", "commit",   PQgetvalue (res, 0, 1));
 216         submit_counter (db, "pg_xact", "rollback", PQgetvalue (res, 0, 2));
 217
 218         PQclear (res);
 219         return 0;
 220 } /* c_psql_stat_database */
 221
 222 static int c_psql_stat_user_tables (c_psql_database_t *db)
 223 {
 224         const char *const query =
 225                 "SELECT sum(seq_scan), sum(seq_tup_read), "
 226                                 "sum(idx_scan), sum(idx_tup_fetch), "
 227                                 "sum(n_tup_ins), sum(n_tup_upd), sum(n_tup_del), "
 228                                 "sum(n_tup_hot_upd), sum(n_live_tup), sum(n_dead_tup) "
 229                         "FROM pg_stat_user_tables;";
 230
 231         PGresult *res;
 232
 233         int n;
 234
 235         res = PQexec (db->conn, query);
 236
 237         if (PGRES_TUPLES_OK != PQresultStatus (res)) {
 238                 log_err ("Failed to execute SQL query: %s",
 239                                 PQerrorMessage (db->conn));
 240                 log_info ("SQL query was: %s", query);
 241                 PQclear (res);
 242                 return -1;
 243         }
 244
 245         n = PQntuples (res);
 246         assert (1 >= n);
 247
 248         if (1 > n) /* no user tables */
 249                 return 0;
 250
 251         submit_counter (db, "pg_scan", "seq",           PQgetvalue (res, 0, 0));
 252         submit_counter (db, "pg_scan", "seq_tup_read",  PQgetvalue (res, 0, 1));
 253         submit_counter (db, "pg_scan", "idx",           PQgetvalue (res, 0, 2));
 254         submit_counter (db, "pg_scan", "idx_tup_fetch", PQgetvalue (res, 0, 3));
 255
 256         submit_counter (db, "pg_n_tup_c", "ins",        PQgetvalue (res, 0, 4));
 257         submit_counter (db, "pg_n_tup_c", "upd",        PQgetvalue (res, 0, 5));
 258         submit_counter (db, "pg_n_tup_c", "del",        PQgetvalue (res, 0, 6));
 259         submit_counter (db, "pg_n_tup_c", "hot_upd",    PQgetvalue (res, 0, 7));
 260
 261         submit_gauge (db, "pg_n_tup_g", "live",         PQgetvalue (res, 0, 8));
 262         submit_gauge (db, "pg_n_tup_g", "dead",         PQgetvalue (res, 0, 9));
 263
 264         PQclear (res);
 265         return 0;
 266 } /* c_psql_stat_user_tables */
 267
 268 static int c_psql_statio_user_tables (c_psql_database_t *db)
 269 {
 270         const char *const query =
 271                 "SELECT sum(heap_blks_read), sum(heap_blks_hit), "
 272                                 "sum(idx_blks_read), sum(idx_blks_hit), "
 273                                 "sum(toast_blks_read), sum(toast_blks_hit), "
 274                                 "sum(tidx_blks_read), sum(tidx_blks_hit) "
 275                         "FROM pg_statio_user_tables;";
 276
 277         PGresult *res;
 278
 279         int n;
 280
 281         res = PQexec (db->conn, query);
 282
 283         if (PGRES_TUPLES_OK != PQresultStatus (res)) {
 284                 log_err ("Failed to execute SQL query: %s",
 285                                 PQerrorMessage (db->conn));
 286                 log_info ("SQL query was: %s", query);
 287                 PQclear (res);
 288                 return -1;
 289         }
 290
 291         n = PQntuples (res);
 292         assert (1 >= n);
 293
 294         if (1 > n) /* no user tables */
 295                 return 0;
 296
 297         submit_counter (db, "pg_blks", "heap_read",  PQgetvalue (res, 0, 0));
 298         submit_counter (db, "pg_blks", "heap_hit",   PQgetvalue (res, 0, 1));
 299
 300         submit_counter (db, "pg_blks", "idx_read",   PQgetvalue (res, 0, 2));
 301         submit_counter (db, "pg_blks", "idx_hit",    PQgetvalue (res, 0, 3));
 302
 303         submit_counter (db, "pg_blks", "toast_read", PQgetvalue (res, 0, 4));
 304         submit_counter (db, "pg_blks", "toast_hit",  PQgetvalue (res, 0, 5));
 305
 306         submit_counter (db, "pg_blks", "tidx_read",  PQgetvalue (res, 0, 6));
 307         submit_counter (db, "pg_blks", "tidx_hit",   PQgetvalue (res, 0, 7));
 308
 309         PQclear (res);
 310         return 0;
 311 } /* c_psql_statio_user_tables */
 312
 313 static int c_psql_read (void)
 314 {
 315         int success = 0;
 316         int i;
 317
 318         for (i = 0; i < databases_num; ++i) {
 319                 c_psql_database_t *db = databases + i;
 320
 321                 assert (NULL != db->database);
 322
 323                 if (0 != c_psql_check_connection (db))
 324                         continue;
 325
 326                 c_psql_stat_database (db);
 327                 c_psql_stat_user_tables (db);
 328                 c_psql_statio_user_tables (db);
 329
 330                 ++success;
 331         }
 332
 333         if (! success)
 334                 return -1;
 335         return 0;
 336 } /* c_psql_read */
 337
 338 static int c_psql_shutdown (void)
 339 {
 340         int i;
 341
 342         if ((NULL == databases) || (0 == databases_num))
 343                 return 0;
 344
 345         plugin_unregister_read ("postgresql");
 346         plugin_unregister_shutdown ("postgresql");
 347
 348         for (i = 0; i < databases_num; ++i) {
 349                 c_psql_database_t *db = databases + i;
 350
 351                 PQfinish (db->conn);
 352
 353                 sfree (db->database);
 354                 sfree (db->host);
 355                 sfree (db->port);
 356                 sfree (db->user);
 357                 sfree (db->password);
 358
 359                 sfree (db->sslmode);
 360
 361                 sfree (db->krbsrvname);
 362
 363                 sfree (db->service);
 364         }
 365
 366         sfree (databases);
 367         databases_num = 0;
 368         return 0;
 369 } /* c_psql_shutdown */
 370
 371 static int c_psql_init (void)
 372 {
 373         int i;
 374
 375         if ((NULL == databases) || (0 == databases_num))
 376                 return 0;
 377
 378         for (i = 0; i < databases_num; ++i) {
 379                 c_psql_database_t *db = databases + i;
 380
 381                 char  conninfo[4096];
 382                 char *buf     = conninfo;
 383                 int   buf_len = sizeof (conninfo);
 384                 int   status;
 385
 386                 char *server_host;
 387                 int   server_version;
 388
 389                 status = ssnprintf (buf, buf_len, "dbname = '%s'", db->database);
 390                 if (0 < status) {
 391                         buf     += status;
 392                         buf_len -= status;
 393                 }
 394
 395                 C_PSQL_PAR_APPEND (buf, buf_len, "host",       db->host);
 396                 C_PSQL_PAR_APPEND (buf, buf_len, "port",       db->port);
 397                 C_PSQL_PAR_APPEND (buf, buf_len, "user",       db->user);
 398                 C_PSQL_PAR_APPEND (buf, buf_len, "password",   db->password);
 399                 C_PSQL_PAR_APPEND (buf, buf_len, "sslmode",    db->sslmode);
 400                 C_PSQL_PAR_APPEND (buf, buf_len, "krbsrvname", db->krbsrvname);
 401                 C_PSQL_PAR_APPEND (buf, buf_len, "service",    db->service);
 402
 403                 db->conn = PQconnectdb (conninfo);
 404                 if (0 != c_psql_check_connection (db))
 405                         continue;
 406
 407                 server_host    = PQhost (db->conn);
 408                 server_version = PQserverVersion (db->conn);
 409                 log_info ("Sucessfully connected to database %s (user %s) "
 410                                 "at server %s%s%s (server version: %d.%d.%d, "
 411                                 "protocol version: %d, pid: %d)",
 412                                 PQdb (db->conn), PQuser (db->conn),
 413                                 C_PSQL_SOCKET3(server_host, PQport (db->conn)),
 414                                 C_PSQL_SERVER_VERSION3 (server_version),
 415                                 PQprotocolVersion (db->conn), PQbackendPID (db->conn));
 416         }
 417
 418         plugin_register_read ("postgresql", c_psql_read);
 419         plugin_register_shutdown ("postgresql", c_psql_shutdown);
 420         return 0;
 421 } /* c_psql_init */
 422
 423 static int config_set (char *name, char **var, const oconfig_item_t *ci)
 424 {
 425         if ((0 != ci->children_num) || (1 != ci->values_num)
 426                         || (OCONFIG_TYPE_STRING != ci->values[0].type)) {
 427                 log_err ("%s expects a single string argument.", name);
 428                 return 1;
 429         }
 430
 431         sfree (*var);
 432         *var = sstrdup (ci->values[0].value.string);
 433         return 0;
 434 } /* config_set */
 435
 436 static int c_psql_config_database (oconfig_item_t *ci)
 437 {
 438         c_psql_database_t *db;
 439
 440         int i;
 441
 442         if ((1 != ci->values_num)
 443                         || (OCONFIG_TYPE_STRING != ci->values[0].type)) {
 444                 log_err ("<Database> expects a single string argument.");
 445                 return 1;
 446         }
 447
 448         ++databases_num;
 449         if (NULL == (databases = (c_psql_database_t *)realloc (databases,
 450                                 databases_num * sizeof (*databases)))) {
 451                 log_err ("Out of memory.");
 452                 exit (5);
 453         }
 454
 455         db = databases + (databases_num - 1);
 456
 457         db->conn = NULL;
 458
 459         db->conn_complaint.last     = 0;
 460         db->conn_complaint.interval = 0;
 461
 462         db->database   = sstrdup (ci->values[0].value.string);
 463         db->host       = NULL;
 464         db->port       = NULL;
 465         db->user       = NULL;
 466         db->password   = NULL;
 467
 468         db->sslmode    = NULL;
 469
 470         db->krbsrvname = NULL;
 471
 472         db->service    = NULL;
 473
 474         for (i = 0; i < ci->children_num; ++i) {
 475                 oconfig_item_t *c = ci->children + i;
 476
 477                 if (0 == strcasecmp (c->key, "Host"))
 478                         config_set ("Host", &db->host, c);
 479                 else if (0 == strcasecmp (c->key, "Port"))
 480                         config_set ("Port", &db->port, c);
 481                 else if (0 == strcasecmp (c->key, "User"))
 482                         config_set ("User", &db->user, c);
 483                 else if (0 == strcasecmp (c->key, "Password"))
 484                         config_set ("Password", &db->password, c);
 485                 else if (0 == strcasecmp (c->key, "SSLMode"))
 486                         config_set ("SSLMode", &db->sslmode, c);
 487                 else if (0 == strcasecmp (c->key, "KRBSrvName"))
 488                         config_set ("KRBSrvName", &db->krbsrvname, c);
 489                 else if (0 == strcasecmp (c->key, "Service"))
 490                         config_set ("Service", &db->service, c);
 491                 else
 492                         log_warn ("Ignoring unknown config key \"%s\".", c->key);
 493         }
 494         return 0;
 495 }
 496
 497 static int c_psql_config (oconfig_item_t *ci)
 498 {
 499         int i;
 500
 501         for (i = 0; i < ci->children_num; ++i) {
 502                 oconfig_item_t *c = ci->children + i;
 503
 504                 if (0 == strcasecmp (c->key, "Database"))
 505                         c_psql_config_database (c);
 506                 else
 507                         log_warn ("Ignoring unknown config key \"%s\".", c->key);
 508         }
 509         return 0;
 510 } /* c_psql_config */
 511
 512 void module_register (void)
 513 {
 514         plugin_register_complex_config ("postgresql", c_psql_config);
 515         plugin_register_init ("postgresql", c_psql_init);
 516 } /* module_register */
 517
 518 /* vim: set sw=4 ts=4 tw=78 noexpandtab : */
 519