src/postgresql.c

   1 /**
   2  * collectd - src/postgresql.c
   3  * Copyright (C) 2008-2012  Sebastian Harl
   4  * Copyright (C) 2009       Florian Forster
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining a
   7  * copy of this software and associated documentation files (the "Software"),
   8  * to deal in the Software without restriction, including without limitation
   9  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
  10  * and/or sell copies of the Software, and to permit persons to whom the
  11  * Software is furnished to do so, subject to the following conditions:
  12  *
  13  * The above copyright notice and this permission notice shall be included in
  14  * all copies or substantial portions of the Software.
  15  *
  16  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  17  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  18  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  19  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  20  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  21  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  22  * DEALINGS IN THE SOFTWARE.
  23  *
  24  * Authors:
  25  *   Sebastian Harl <sh at tokkee.org>
  26  *   Florian Forster <octo at collectd.org>
  27  **/
  28
  29 /*
  30  * This module collects PostgreSQL database statistics.
  31  */
  32
  33 #include "collectd.h"
  34
  35 #include "common.h"
  36
  37 #include "plugin.h"
  38
  39 #include "utils_cache.h"
  40 #include "utils_complain.h"
  41 #include "utils_db_query.h"
  42
  43 #include <libpq-fe.h>
  44 #include <pg_config_manual.h>
  45
  46 #define log_err(...) ERROR("postgresql: " __VA_ARGS__)
  47 #define log_warn(...) WARNING("postgresql: " __VA_ARGS__)
  48 #define log_info(...) INFO("postgresql: " __VA_ARGS__)
  49 #define log_debug(...) DEBUG("postgresql: " __VA_ARGS__)
  50
  51 #ifndef C_PSQL_DEFAULT_CONF
  52 #define C_PSQL_DEFAULT_CONF PKGDATADIR "/postgresql_default.conf"
  53 #endif
  54
  55 /* Appends the (parameter, value) pair to the string
  56  * pointed to by 'buf' suitable to be used as argument
  57  * for PQconnectdb(). If value equals NULL, the pair
  58  * is ignored. */
  59 #define C_PSQL_PAR_APPEND(buf, buf_len, parameter, value)                      \
  60   if ((0 < (buf_len)) && (NULL != (value)) && ('\0' != *(value))) {            \
  61     int s = snprintf(buf, buf_len, " %s = '%s'", parameter, value);            \
  62     if (0 < s) {                                                               \
  63       buf += s;                                                                \
  64       buf_len -= s;                                                            \
  65     }                                                                          \
  66   }
  67
  68 /* Returns the tuple (major, minor, patchlevel)
  69  * for the given version number. */
  70 #define C_PSQL_SERVER_VERSION3(server_version)                                 \
  71   (server_version) / 10000,                                                    \
  72       (server_version) / 100 - (int)((server_version) / 10000) * 100,          \
  73       (server_version) - (int)((server_version) / 100) * 100
  74
  75 /* Returns true if the given host specifies a
  76  * UNIX domain socket. */
  77 #define C_PSQL_IS_UNIX_DOMAIN_SOCKET(host)                                     \
  78   ((NULL == (host)) || ('\0' == *(host)) || ('/' == *(host)))
  79
  80 /* Returns the tuple (host, delimiter, port) for a
  81  * given (host, port) pair. Depending on the value of
  82  * 'host' a UNIX domain socket or a TCP socket is
  83  * assumed. */
  84 #define C_PSQL_SOCKET3(host, port)                                             \
  85   ((NULL == (host)) || ('\0' == *(host))) ? DEFAULT_PGSOCKET_DIR : host,       \
  86       C_PSQL_IS_UNIX_DOMAIN_SOCKET(host) ? "/.s.PGSQL." : ":", port
  87
  88 typedef enum {
  89   C_PSQL_PARAM_HOST = 1,
  90   C_PSQL_PARAM_DB,
  91   C_PSQL_PARAM_USER,
  92   C_PSQL_PARAM_INTERVAL,
  93   C_PSQL_PARAM_INSTANCE,
  94 } c_psql_param_t;
  95
  96 /* Parameter configuration. Stored as `user data' in the query objects. */
  97 typedef struct {
  98   c_psql_param_t *params;
  99   int params_num;
 100 } c_psql_user_data_t;
 101
 102 typedef struct {
 103   char *name;
 104   char *statement;
 105   _Bool store_rates;
 106 } c_psql_writer_t;
 107
 108 typedef struct {
 109   PGconn *conn;
 110   c_complain_t conn_complaint;
 111
 112   int proto_version;
 113   int server_version;
 114
 115   int max_params_num;
 116
 117   /* user configuration */
 118   udb_query_preparation_area_t **q_prep_areas;
 119   udb_query_t **queries;
 120   size_t queries_num;
 121
 122   c_psql_writer_t **writers;
 123   size_t writers_num;
 124
 125   /* make sure we don't access the database object in parallel */
 126   pthread_mutex_t db_lock;
 127
 128   cdtime_t interval;
 129
 130   /* writer "caching" settings */
 131   cdtime_t commit_interval;
 132   cdtime_t next_commit;
 133   cdtime_t expire_delay;
 134
 135   char *host;
 136   char *port;
 137   char *database;
 138   char *user;
 139   char *password;
 140
 141   char *instance;
 142   char *plugin_name;
 143
 144   char *sslmode;
 145
 146   char *krbsrvname;
 147
 148   char *service;
 149
 150   int ref_cnt;
 151 } c_psql_database_t;
 152
 153 static const char *const def_queries[] = {
 154     "backends",     "transactions", "queries",   "query_plans",
 155     "table_states", "disk_io",      "disk_usage"};
 156 static int def_queries_num = STATIC_ARRAY_SIZE(def_queries);
 157
 158 static c_psql_database_t **databases = NULL;
 159 static size_t databases_num = 0;
 160
 161 static udb_query_t **queries = NULL;
 162 static size_t queries_num = 0;
 163
 164 static c_psql_writer_t *writers = NULL;
 165 static size_t writers_num = 0;
 166
 167 static int c_psql_begin(c_psql_database_t *db) {
 168   PGresult *r = PQexec(db->conn, "BEGIN");
 169
 170   int status = 1;
 171
 172   if (r != NULL) {
 173     if (PGRES_COMMAND_OK == PQresultStatus(r)) {
 174       db->next_commit = cdtime() + db->commit_interval;
 175       status = 0;
 176     } else
 177       log_warn("Failed to initiate ('BEGIN') transaction: %s",
 178                PQerrorMessage(db->conn));
 179     PQclear(r);
 180   }
 181   return status;
 182 } /* c_psql_begin */
 183
 184 static int c_psql_commit(c_psql_database_t *db) {
 185   PGresult *r = PQexec(db->conn, "COMMIT");
 186
 187   int status = 1;
 188
 189   if (r != NULL) {
 190     if (PGRES_COMMAND_OK == PQresultStatus(r)) {
 191       db->next_commit = 0;
 192       log_debug("Successfully committed transaction.");
 193       status = 0;
 194     } else
 195       log_warn("Failed to commit transaction: %s", PQerrorMessage(db->conn));
 196     PQclear(r);
 197   }
 198   return status;
 199 } /* c_psql_commit */
 200
 201 static c_psql_database_t *c_psql_database_new(const char *name) {
 202   c_psql_database_t **tmp;
 203   c_psql_database_t *db;
 204
 205   db = malloc(sizeof(*db));
 206   if (NULL == db) {
 207     log_err("Out of memory.");
 208     return NULL;
 209   }
 210
 211   tmp = realloc(databases, (databases_num + 1) * sizeof(*databases));
 212   if (NULL == tmp) {
 213     log_err("Out of memory.");
 214     sfree(db);
 215     return NULL;
 216   }
 217
 218   databases = tmp;
 219   databases[databases_num] = db;
 220   ++databases_num;
 221
 222   db->conn = NULL;
 223
 224   C_COMPLAIN_INIT(&db->conn_complaint);
 225
 226   db->proto_version = 0;
 227   db->server_version = 0;
 228
 229   db->max_params_num = 0;
 230
 231   db->q_prep_areas = NULL;
 232   db->queries = NULL;
 233   db->queries_num = 0;
 234
 235   db->writers = NULL;
 236   db->writers_num = 0;
 237
 238   pthread_mutex_init(&db->db_lock, /* attrs = */ NULL);
 239
 240   db->interval = 0;
 241
 242   db->commit_interval = 0;
 243   db->next_commit = 0;
 244   db->expire_delay = 0;
 245
 246   db->database = sstrdup(name);
 247   db->host = NULL;
 248   db->port = NULL;
 249   db->user = NULL;
 250   db->password = NULL;
 251
 252   db->instance = sstrdup(name);
 253
 254   db->plugin_name = NULL;
 255
 256   db->sslmode = NULL;
 257
 258   db->krbsrvname = NULL;
 259
 260   db->service = NULL;
 261
 262   db->ref_cnt = 0;
 263   return db;
 264 } /* c_psql_database_new */
 265
 266 static void c_psql_database_delete(void *data) {
 267   c_psql_database_t *db = data;
 268
 269   --db->ref_cnt;
 270   /* readers and writers may access this database */
 271   if (db->ref_cnt > 0)
 272     return;
 273
 274   /* wait for the lock to be released by the last writer */
 275   pthread_mutex_lock(&db->db_lock);
 276
 277   if (db->next_commit > 0)
 278     c_psql_commit(db);
 279
 280   PQfinish(db->conn);
 281   db->conn = NULL;
 282
 283   if (db->q_prep_areas)
 284     for (size_t i = 0; i < db->queries_num; ++i)
 285       udb_query_delete_preparation_area(db->q_prep_areas[i]);
 286   free(db->q_prep_areas);
 287
 288   sfree(db->queries);
 289   db->queries_num = 0;
 290
 291   sfree(db->writers);
 292   db->writers_num = 0;
 293
 294   pthread_mutex_unlock(&db->db_lock);
 295
 296   pthread_mutex_destroy(&db->db_lock);
 297
 298   sfree(db->database);
 299   sfree(db->host);
 300   sfree(db->port);
 301   sfree(db->user);
 302   sfree(db->password);
 303
 304   sfree(db->instance);
 305
 306   sfree(db->plugin_name);
 307
 308   sfree(db->sslmode);
 309
 310   sfree(db->krbsrvname);
 311
 312   sfree(db->service);
 313
 314   /* don't care about freeing or reordering the 'databases' array
 315    * this is done in 'shutdown'; also, don't free the database instance
 316    * object just to make sure that in case anybody accesses it before
 317    * shutdown won't segfault */
 318   return;
 319 } /* c_psql_database_delete */
 320
 321 static int c_psql_connect(c_psql_database_t *db) {
 322   char conninfo[4096];
 323   char *buf = conninfo;
 324   int buf_len = sizeof(conninfo);
 325   int status;
 326
 327   if ((!db) || (!db->database))
 328     return -1;
 329
 330   status = snprintf(buf, buf_len, "dbname = '%s'", db->database);
 331   if (0 < status) {
 332     buf += status;
 333     buf_len -= status;
 334   }
 335
 336   C_PSQL_PAR_APPEND(buf, buf_len, "host", db->host);
 337   C_PSQL_PAR_APPEND(buf, buf_len, "port", db->port);
 338   C_PSQL_PAR_APPEND(buf, buf_len, "user", db->user);
 339   C_PSQL_PAR_APPEND(buf, buf_len, "password", db->password);
 340   C_PSQL_PAR_APPEND(buf, buf_len, "sslmode", db->sslmode);
 341   C_PSQL_PAR_APPEND(buf, buf_len, "krbsrvname", db->krbsrvname);
 342   C_PSQL_PAR_APPEND(buf, buf_len, "service", db->service);
 343
 344   db->conn = PQconnectdb(conninfo);
 345   db->proto_version = PQprotocolVersion(db->conn);
 346   return 0;
 347 } /* c_psql_connect */
 348
 349 static int c_psql_check_connection(c_psql_database_t *db) {
 350   _Bool init = 0;
 351
 352   if (!db->conn) {
 353     init = 1;
 354
 355     /* trigger c_release() */
 356     if (0 == db->conn_complaint.interval)
 357       db->conn_complaint.interval = 1;
 358
 359     c_psql_connect(db);
 360   }
 361
 362   if (CONNECTION_OK != PQstatus(db->conn)) {
 363     PQreset(db->conn);
 364
 365     /* trigger c_release() */
 366     if (0 == db->conn_complaint.interval)
 367       db->conn_complaint.interval = 1;
 368
 369     if (CONNECTION_OK != PQstatus(db->conn)) {
 370       c_complain(LOG_ERR, &db->conn_complaint,
 371                  "Failed to connect to database %s (%s): %s", db->database,
 372                  db->instance, PQerrorMessage(db->conn));
 373       return -1;
 374     }
 375
 376     db->proto_version = PQprotocolVersion(db->conn);
 377   }
 378
 379   db->server_version = PQserverVersion(db->conn);
 380
 381   if (c_would_release(&db->conn_complaint)) {
 382     char *server_host;
 383     int server_version;
 384
 385     server_host = PQhost(db->conn);
 386     server_version = PQserverVersion(db->conn);
 387
 388     c_do_release(LOG_INFO, &db->conn_complaint,
 389                  "Successfully %sconnected to database %s (user %s) "
 390                  "at server %s%s%s (server version: %d.%d.%d, "
 391                  "protocol version: %d, pid: %d)",
 392                  init ? "" : "re", PQdb(db->conn), PQuser(db->conn),
 393                  C_PSQL_SOCKET3(server_host, PQport(db->conn)),
 394                  C_PSQL_SERVER_VERSION3(server_version), db->proto_version,
 395                  PQbackendPID(db->conn));
 396
 397     if (3 > db->proto_version)
 398       log_warn("Protocol version %d does not support parameters.",
 399                db->proto_version);
 400   }
 401   return 0;
 402 } /* c_psql_check_connection */
 403
 404 static PGresult *c_psql_exec_query_noparams(c_psql_database_t *db,
 405                                             udb_query_t *q) {
 406   return PQexec(db->conn, udb_query_get_statement(q));
 407 } /* c_psql_exec_query_noparams */
 408
 409 static PGresult *c_psql_exec_query_params(c_psql_database_t *db, udb_query_t *q,
 410                                           c_psql_user_data_t *data) {
 411   const char *params[db->max_params_num];
 412   char interval[64];
 413
 414   if ((data == NULL) || (data->params_num == 0))
 415     return c_psql_exec_query_noparams(db, q);
 416
 417   assert(db->max_params_num >= data->params_num);
 418
 419   for (int i = 0; i < data->params_num; ++i) {
 420     switch (data->params[i]) {
 421     case C_PSQL_PARAM_HOST:
 422       params[i] =
 423           C_PSQL_IS_UNIX_DOMAIN_SOCKET(db->host) ? "localhost" : db->host;
 424       break;
 425     case C_PSQL_PARAM_DB:
 426       params[i] = db->database;
 427       break;
 428     case C_PSQL_PARAM_USER:
 429       params[i] = db->user;
 430       break;
 431     case C_PSQL_PARAM_INTERVAL:
 432       snprintf(interval, sizeof(interval), "%.3f",
 433                (db->interval > 0) ? CDTIME_T_TO_DOUBLE(db->interval)
 434                                   : plugin_get_interval());
 435       params[i] = interval;
 436       break;
 437     case C_PSQL_PARAM_INSTANCE:
 438       params[i] = db->instance;
 439       break;
 440     default:
 441       assert(0);
 442     }
 443   }
 444
 445   return PQexecParams(db->conn, udb_query_get_statement(q), data->params_num,
 446                       NULL, (const char *const *)params, NULL, NULL, 0);
 447 } /* c_psql_exec_query_params */
 448
 449 /* db->db_lock must be locked when calling this function */
 450 static int c_psql_exec_query(c_psql_database_t *db, udb_query_t *q,
 451                              udb_query_preparation_area_t *prep_area) {
 452   PGresult *res;
 453
 454   c_psql_user_data_t *data;
 455
 456   const char *host;
 457
 458   char **column_names;
 459   char **column_values;
 460   int column_num;
 461
 462   int rows_num;
 463   int status;
 464
 465   /* The user data may hold parameter information, but may be NULL. */
 466   data = udb_query_get_user_data(q);
 467
 468   /* Versions up to `3' don't know how to handle parameters. */
 469   if (3 <= db->proto_version)
 470     res = c_psql_exec_query_params(db, q, data);
 471   else if ((NULL == data) || (0 == data->params_num))
 472     res = c_psql_exec_query_noparams(db, q);
 473   else {
 474     log_err("Connection to database \"%s\" (%s) does not support "
 475             "parameters (protocol version %d) - "
 476             "cannot execute query \"%s\".",
 477             db->database, db->instance, db->proto_version,
 478             udb_query_get_name(q));
 479     return -1;
 480   }
 481
 482   /* give c_psql_write() a chance to acquire the lock if called recursively
 483    * through dispatch_values(); this will happen if, both, queries and
 484    * writers are configured for a single connection */
 485   pthread_mutex_unlock(&db->db_lock);
 486
 487   column_names = NULL;
 488   column_values = NULL;
 489
 490   if (PGRES_TUPLES_OK != PQresultStatus(res)) {
 491     pthread_mutex_lock(&db->db_lock);
 492
 493     if ((CONNECTION_OK != PQstatus(db->conn)) &&
 494         (0 == c_psql_check_connection(db))) {
 495       PQclear(res);
 496       return c_psql_exec_query(db, q, prep_area);
 497     }
 498
 499     log_err("Failed to execute SQL query: %s", PQerrorMessage(db->conn));
 500     log_info("SQL query was: %s", udb_query_get_statement(q));
 501     PQclear(res);
 502     return -1;
 503   }
 504
 505 #define BAIL_OUT(status)                                                       \
 506   sfree(column_names);                                                         \
 507   sfree(column_values);                                                        \
 508   PQclear(res);                                                                \
 509   pthread_mutex_lock(&db->db_lock);                                            \
 510   return status
 511
 512   rows_num = PQntuples(res);
 513   if (1 > rows_num) {
 514     BAIL_OUT(0);
 515   }
 516
 517   column_num = PQnfields(res);
 518   column_names = (char **)calloc(column_num, sizeof(char *));
 519   if (NULL == column_names) {
 520     log_err("calloc failed.");
 521     BAIL_OUT(-1);
 522   }
 523
 524   column_values = (char **)calloc(column_num, sizeof(char *));
 525   if (NULL == column_values) {
 526     log_err("calloc failed.");
 527     BAIL_OUT(-1);
 528   }
 529
 530   for (int col = 0; col < column_num; ++col) {
 531     /* Pointers returned by `PQfname' are freed by `PQclear' via
 532      * `BAIL_OUT'. */
 533     column_names[col] = PQfname(res, col);
 534     if (NULL == column_names[col]) {
 535       log_err("Failed to resolve name of column %i.", col);
 536       BAIL_OUT(-1);
 537     }
 538   }
 539
 540   if (C_PSQL_IS_UNIX_DOMAIN_SOCKET(db->host) ||
 541       (0 == strcmp(db->host, "127.0.0.1")) ||
 542       (0 == strcmp(db->host, "localhost")))
 543     host = hostname_g;
 544   else
 545     host = db->host;
 546
 547   status = udb_query_prepare_result(
 548       q, prep_area, host,
 549       (db->plugin_name != NULL) ? db->plugin_name : "postgresql",
 550       db->instance, column_names, (size_t)column_num, db->interval);
 551
 552   if (0 != status) {
 553     log_err("udb_query_prepare_result failed with status %i.", status);
 554     BAIL_OUT(-1);
 555   }
 556
 557   for (int row = 0; row < rows_num; ++row) {
 558     int col;
 559     for (col = 0; col < column_num; ++col) {
 560       /* Pointers returned by `PQgetvalue' are freed by `PQclear' via
 561        * `BAIL_OUT'. */
 562       column_values[col] = PQgetvalue(res, row, col);
 563       if (NULL == column_values[col]) {
 564         log_err("Failed to get value at (row = %i, col = %i).", row, col);
 565         break;
 566       }
 567     }
 568
 569     /* check for an error */
 570     if (col < column_num)
 571       continue;
 572
 573     status = udb_query_handle_result(q, prep_area, column_values);
 574     if (status != 0) {
 575       log_err("udb_query_handle_result failed with status %i.", status);
 576     }
 577   } /* for (row = 0; row < rows_num; ++row) */
 578
 579   udb_query_finish_result(q, prep_area);
 580
 581   BAIL_OUT(0);
 582 #undef BAIL_OUT
 583 } /* c_psql_exec_query */
 584
 585 static int c_psql_read(user_data_t *ud) {
 586   c_psql_database_t *db;
 587
 588   int success = 0;
 589
 590   if ((ud == NULL) || (ud->data == NULL)) {
 591     log_err("c_psql_read: Invalid user data.");
 592     return -1;
 593   }
 594
 595   db = ud->data;
 596
 597   assert(NULL != db->database);
 598   assert(NULL != db->instance);
 599   assert(NULL != db->queries);
 600
 601   pthread_mutex_lock(&db->db_lock);
 602
 603   if (0 != c_psql_check_connection(db)) {
 604     pthread_mutex_unlock(&db->db_lock);
 605     return -1;
 606   }
 607
 608   for (size_t i = 0; i < db->queries_num; ++i) {
 609     udb_query_preparation_area_t *prep_area;
 610     udb_query_t *q;
 611
 612     prep_area = db->q_prep_areas[i];
 613     q = db->queries[i];
 614
 615     if ((0 != db->server_version) &&
 616         (udb_query_check_version(q, db->server_version) <= 0))
 617       continue;
 618
 619     if (0 == c_psql_exec_query(db, q, prep_area))
 620       success = 1;
 621   }
 622
 623   pthread_mutex_unlock(&db->db_lock);
 624
 625   if (!success)
 626     return -1;
 627   return 0;
 628 } /* c_psql_read */
 629
 630 static char *values_name_to_sqlarray(const data_set_t *ds, char *string,
 631                                      size_t string_len) {
 632   char *str_ptr;
 633   size_t str_len;
 634
 635   str_ptr = string;
 636   str_len = string_len;
 637
 638   for (size_t i = 0; i < ds->ds_num; ++i) {
 639     int status = snprintf(str_ptr, str_len, ",'%s'", ds->ds[i].name);
 640
 641     if (status < 1)
 642       return NULL;
 643     else if ((size_t)status >= str_len) {
 644       str_len = 0;
 645       break;
 646     } else {
 647       str_ptr += status;
 648       str_len -= (size_t)status;
 649     }
 650   }
 651
 652   if (str_len <= 2) {
 653     log_err("c_psql_write: Failed to stringify value names");
 654     return NULL;
 655   }
 656
 657   /* overwrite the first comma */
 658   string[0] = '{';
 659   str_ptr[0] = '}';
 660   str_ptr[1] = '\0';
 661
 662   return string;
 663 } /* values_name_to_sqlarray */
 664
 665 static char *values_type_to_sqlarray(const data_set_t *ds, char *string,
 666                                      size_t string_len, _Bool store_rates) {
 667   char *str_ptr;
 668   size_t str_len;
 669
 670   str_ptr = string;
 671   str_len = string_len;
 672
 673   for (size_t i = 0; i < ds->ds_num; ++i) {
 674     int status;
 675
 676     if (store_rates)
 677       status = snprintf(str_ptr, str_len, ",'gauge'");
 678     else
 679       status = snprintf(str_ptr, str_len, ",'%s'",
 680                         DS_TYPE_TO_STRING(ds->ds[i].type));
 681
 682     if (status < 1) {
 683       str_len = 0;
 684       break;
 685     } else if ((size_t)status >= str_len) {
 686       str_len = 0;
 687       break;
 688     } else {
 689       str_ptr += status;
 690       str_len -= (size_t)status;
 691     }
 692   }
 693
 694   if (str_len <= 2) {
 695     log_err("c_psql_write: Failed to stringify value types");
 696     return NULL;
 697   }
 698
 699   /* overwrite the first comma */
 700   string[0] = '{';
 701   str_ptr[0] = '}';
 702   str_ptr[1] = '\0';
 703
 704   return string;
 705 } /* values_type_to_sqlarray */
 706
 707 static char *values_to_sqlarray(const data_set_t *ds, const value_list_t *vl,
 708                                 char *string, size_t string_len,
 709                                 _Bool store_rates) {
 710   char *str_ptr;
 711   size_t str_len;
 712
 713   gauge_t *rates = NULL;
 714
 715   str_ptr = string;
 716   str_len = string_len;
 717
 718   for (size_t i = 0; i < vl->values_len; ++i) {
 719     int status = 0;
 720
 721     if ((ds->ds[i].type != DS_TYPE_GAUGE) &&
 722         (ds->ds[i].type != DS_TYPE_COUNTER) &&
 723         (ds->ds[i].type != DS_TYPE_DERIVE) &&
 724         (ds->ds[i].type != DS_TYPE_ABSOLUTE)) {
 725       log_err("c_psql_write: Unknown data source type: %i", ds->ds[i].type);
 726       sfree(rates);
 727       return NULL;
 728     }
 729
 730     if (ds->ds[i].type == DS_TYPE_GAUGE)
 731       status =
 732           snprintf(str_ptr, str_len, "," GAUGE_FORMAT, vl->values[i].gauge);
 733     else if (store_rates) {
 734       if (rates == NULL)
 735         rates = uc_get_rate(ds, vl);
 736
 737       if (rates == NULL) {
 738         log_err("c_psql_write: Failed to determine rate");
 739         return NULL;
 740       }
 741
 742       status = snprintf(str_ptr, str_len, ",%lf", rates[i]);
 743     } else if (ds->ds[i].type == DS_TYPE_COUNTER)
 744       status = snprintf(str_ptr, str_len, ",%llu", vl->values[i].counter);
 745     else if (ds->ds[i].type == DS_TYPE_DERIVE)
 746       status = snprintf(str_ptr, str_len, ",%" PRIi64, vl->values[i].derive);
 747     else if (ds->ds[i].type == DS_TYPE_ABSOLUTE)
 748       status = snprintf(str_ptr, str_len, ",%" PRIu64, vl->values[i].absolute);
 749
 750     if (status < 1) {
 751       str_len = 0;
 752       break;
 753     } else if ((size_t)status >= str_len) {
 754       str_len = 0;
 755       break;
 756     } else {
 757       str_ptr += status;
 758       str_len -= (size_t)status;
 759     }
 760   }
 761
 762   sfree(rates);
 763
 764   if (str_len <= 2) {
 765     log_err("c_psql_write: Failed to stringify value list");
 766     return NULL;
 767   }
 768
 769   /* overwrite the first comma */
 770   string[0] = '{';
 771   str_ptr[0] = '}';
 772   str_ptr[1] = '\0';
 773
 774   return string;
 775 } /* values_to_sqlarray */
 776
 777 static int c_psql_write(const data_set_t *ds, const value_list_t *vl,
 778                         user_data_t *ud) {
 779   c_psql_database_t *db;
 780
 781   char time_str[RFC3339NANO_SIZE];
 782   char values_name_str[1024];
 783   char values_type_str[1024];
 784   char values_str[1024];
 785
 786   const char *params[9];
 787
 788   int success = 0;
 789
 790   if ((ud == NULL) || (ud->data == NULL)) {
 791     log_err("c_psql_write: Invalid user data.");
 792     return -1;
 793   }
 794
 795   db = ud->data;
 796   assert(db->database != NULL);
 797   assert(db->writers != NULL);
 798
 799   if (rfc3339nano_local(time_str, sizeof(time_str), vl->time) != 0) {
 800     log_err("c_psql_write: Failed to convert time to RFC 3339 format");
 801     return -1;
 802   }
 803
 804   if (values_name_to_sqlarray(ds, values_name_str, sizeof(values_name_str)) ==
 805       NULL)
 806     return -1;
 807
 808 #define VALUE_OR_NULL(v) ((((v) == NULL) || (*(v) == '\0')) ? NULL : (v))
 809
 810   params[0] = time_str;
 811   params[1] = vl->host;
 812   params[2] = vl->plugin;
 813   params[3] = VALUE_OR_NULL(vl->plugin_instance);
 814   params[4] = vl->type;
 815   params[5] = VALUE_OR_NULL(vl->type_instance);
 816   params[6] = values_name_str;
 817
 818 #undef VALUE_OR_NULL
 819
 820   if (db->expire_delay > 0 &&
 821       vl->time < (cdtime() - vl->interval - db->expire_delay)) {
 822     log_info("c_psql_write: Skipped expired value @ %s - %s/%s-%s/%s-%s/%s",
 823              params[0], params[1], params[2], params[3], params[4], params[5],
 824              params[6]);
 825     return 0;
 826   }
 827
 828   pthread_mutex_lock(&db->db_lock);
 829
 830   if (0 != c_psql_check_connection(db)) {
 831     pthread_mutex_unlock(&db->db_lock);
 832     return -1;
 833   }
 834
 835   if ((db->commit_interval > 0) && (db->next_commit == 0))
 836     c_psql_begin(db);
 837
 838   for (size_t i = 0; i < db->writers_num; ++i) {
 839     c_psql_writer_t *writer;
 840     PGresult *res;
 841
 842     writer = db->writers[i];
 843
 844     if (values_type_to_sqlarray(ds, values_type_str, sizeof(values_type_str),
 845                                 writer->store_rates) == NULL) {
 846       pthread_mutex_unlock(&db->db_lock);
 847       return -1;
 848     }
 849
 850     if (values_to_sqlarray(ds, vl, values_str, sizeof(values_str),
 851                            writer->store_rates) == NULL) {
 852       pthread_mutex_unlock(&db->db_lock);
 853       return -1;
 854     }
 855
 856     params[7] = values_type_str;
 857     params[8] = values_str;
 858
 859     res = PQexecParams(db->conn, writer->statement, STATIC_ARRAY_SIZE(params),
 860                        NULL, (const char *const *)params, NULL, NULL,
 861                        /* return text data */ 0);
 862
 863     if ((PGRES_COMMAND_OK != PQresultStatus(res)) &&
 864         (PGRES_TUPLES_OK != PQresultStatus(res))) {
 865       PQclear(res);
 866
 867       if ((CONNECTION_OK != PQstatus(db->conn)) &&
 868           (0 == c_psql_check_connection(db))) {
 869         /* try again */
 870         res = PQexecParams(
 871             db->conn, writer->statement, STATIC_ARRAY_SIZE(params), NULL,
 872             (const char *const *)params, NULL, NULL, /* return text data */ 0);
 873
 874         if ((PGRES_COMMAND_OK == PQresultStatus(res)) ||
 875             (PGRES_TUPLES_OK == PQresultStatus(res))) {
 876           PQclear(res);
 877           success = 1;
 878           continue;
 879         }
 880       }
 881
 882       log_err("Failed to execute SQL query: %s", PQerrorMessage(db->conn));
 883       log_info("SQL query was: '%s', "
 884                "params: %s, %s, %s, %s, %s, %s, %s, %s",
 885                writer->statement, params[0], params[1], params[2], params[3],
 886                params[4], params[5], params[6], params[7]);
 887
 888       /* this will abort any current transaction -> restart */
 889       if (db->next_commit > 0)
 890         c_psql_commit(db);
 891
 892       pthread_mutex_unlock(&db->db_lock);
 893       return -1;
 894     }
 895
 896     PQclear(res);
 897     success = 1;
 898   }
 899
 900   if ((db->next_commit > 0) && (cdtime() > db->next_commit))
 901     c_psql_commit(db);
 902
 903   pthread_mutex_unlock(&db->db_lock);
 904
 905   if (!success)
 906     return -1;
 907   return 0;
 908 } /* c_psql_write */
 909
 910 /* We cannot flush single identifiers as all we do is to commit the currently
 911  * running transaction, thus making sure that all written data is actually
 912  * visible to everybody. */
 913 static int c_psql_flush(cdtime_t timeout,
 914                         __attribute__((unused)) const char *ident,
 915                         user_data_t *ud) {
 916   c_psql_database_t **dbs = databases;
 917   size_t dbs_num = databases_num;
 918
 919   if ((ud != NULL) && (ud->data != NULL)) {
 920     dbs = (void *)&ud->data;
 921     dbs_num = 1;
 922   }
 923
 924   for (size_t i = 0; i < dbs_num; ++i) {
 925     c_psql_database_t *db = dbs[i];
 926
 927     /* don't commit if the timeout is larger than the regular commit
 928      * interval as in that case all requested data has already been
 929      * committed */
 930     if ((db->next_commit > 0) && (db->commit_interval > timeout))
 931       c_psql_commit(db);
 932   }
 933   return 0;
 934 } /* c_psql_flush */
 935
 936 static int c_psql_shutdown(void) {
 937   _Bool had_flush = 0;
 938
 939   plugin_unregister_read_group("postgresql");
 940
 941   for (size_t i = 0; i < databases_num; ++i) {
 942     c_psql_database_t *db = databases[i];
 943
 944     if (db->writers_num > 0) {
 945       char cb_name[DATA_MAX_NAME_LEN];
 946       snprintf(cb_name, sizeof(cb_name), "postgresql-%s", db->database);
 947
 948       if (!had_flush) {
 949         plugin_unregister_flush("postgresql");
 950         had_flush = 1;
 951       }
 952
 953       plugin_unregister_flush(cb_name);
 954       plugin_unregister_write(cb_name);
 955     }
 956
 957     sfree(db);
 958   }
 959
 960   udb_query_free(queries, queries_num);
 961   queries = NULL;
 962   queries_num = 0;
 963
 964   sfree(writers);
 965   writers = NULL;
 966   writers_num = 0;
 967
 968   sfree(databases);
 969   databases = NULL;
 970   databases_num = 0;
 971
 972   return 0;
 973 } /* c_psql_shutdown */
 974
 975 static int config_query_param_add(udb_query_t *q, oconfig_item_t *ci) {
 976   c_psql_user_data_t *data;
 977   const char *param_str;
 978
 979   c_psql_param_t *tmp;
 980
 981   data = udb_query_get_user_data(q);
 982   if (NULL == data) {
 983     data = calloc(1, sizeof(*data));
 984     if (NULL == data) {
 985       log_err("Out of memory.");
 986       return -1;
 987     }
 988     data->params = NULL;
 989     data->params_num = 0;
 990
 991     udb_query_set_user_data(q, data);
 992   }
 993
 994   tmp = realloc(data->params, (data->params_num + 1) * sizeof(*data->params));
 995   if (NULL == tmp) {
 996     log_err("Out of memory.");
 997     return -1;
 998   }
 999   data->params = tmp;
1000
1001   param_str = ci->values[0].value.string;
1002   if (0 == strcasecmp(param_str, "hostname"))
1003     data->params[data->params_num] = C_PSQL_PARAM_HOST;
1004   else if (0 == strcasecmp(param_str, "database"))
1005     data->params[data->params_num] = C_PSQL_PARAM_DB;
1006   else if (0 == strcasecmp(param_str, "username"))
1007     data->params[data->params_num] = C_PSQL_PARAM_USER;
1008   else if (0 == strcasecmp(param_str, "interval"))
1009     data->params[data->params_num] = C_PSQL_PARAM_INTERVAL;
1010   else if (0 == strcasecmp(param_str, "instance"))
1011     data->params[data->params_num] = C_PSQL_PARAM_INSTANCE;
1012   else {
1013     log_err("Invalid parameter \"%s\".", param_str);
1014     return 1;
1015   }
1016
1017   data->params_num++;
1018   return 0;
1019 } /* config_query_param_add */
1020
1021 static int config_query_callback(udb_query_t *q, oconfig_item_t *ci) {
1022   if (0 == strcasecmp("Param", ci->key))
1023     return config_query_param_add(q, ci);
1024
1025   log_err("Option not allowed within a Query block: `%s'", ci->key);
1026
1027   return -1;
1028 } /* config_query_callback */
1029
1030 static int config_add_writer(oconfig_item_t *ci, c_psql_writer_t *src_writers,
1031                              size_t src_writers_num,
1032                              c_psql_writer_t ***dst_writers,
1033                              size_t *dst_writers_num) {
1034   char *name;
1035
1036   size_t i;
1037
1038   if ((ci == NULL) || (dst_writers == NULL) || (dst_writers_num == NULL))
1039     return -1;
1040
1041   if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_STRING)) {
1042     log_err("`Writer' expects a single string argument.");
1043     return 1;
1044   }
1045
1046   name = ci->values[0].value.string;
1047
1048   for (i = 0; i < src_writers_num; ++i) {
1049     c_psql_writer_t **tmp;
1050
1051     if (strcasecmp(name, src_writers[i].name) != 0)
1052       continue;
1053
1054     tmp = realloc(*dst_writers, sizeof(**dst_writers) * (*dst_writers_num + 1));
1055     if (tmp == NULL) {
1056       log_err("Out of memory.");
1057       return -1;
1058     }
1059
1060     tmp[*dst_writers_num] = src_writers + i;
1061
1062     *dst_writers = tmp;
1063     ++(*dst_writers_num);
1064     break;
1065   }
1066
1067   if (i >= src_writers_num) {
1068     log_err("No such writer: `%s'", name);
1069     return -1;
1070   }
1071
1072   return 0;
1073 } /* config_add_writer */
1074
1075 static int c_psql_config_writer(oconfig_item_t *ci) {
1076   c_psql_writer_t *writer;
1077   c_psql_writer_t *tmp;
1078
1079   int status = 0;
1080
1081   if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_STRING)) {
1082     log_err("<Writer> expects a single string argument.");
1083     return 1;
1084   }
1085
1086   tmp = realloc(writers, sizeof(*writers) * (writers_num + 1));
1087   if (tmp == NULL) {
1088     log_err("Out of memory.");
1089     return -1;
1090   }
1091
1092   writers = tmp;
1093   writer = writers + writers_num;
1094   memset(writer, 0, sizeof(*writer));
1095
1096   writer->name = sstrdup(ci->values[0].value.string);
1097   writer->statement = NULL;
1098   writer->store_rates = 1;
1099
1100   for (int i = 0; i < ci->children_num; ++i) {
1101     oconfig_item_t *c = ci->children + i;
1102
1103     if (strcasecmp("Statement", c->key) == 0)
1104       status = cf_util_get_string(c, &writer->statement);
1105     else if (strcasecmp("StoreRates", c->key) == 0)
1106       status = cf_util_get_boolean(c, &writer->store_rates);
1107     else
1108       log_warn("Ignoring unknown config key \"%s\".", c->key);
1109   }
1110
1111   if (status != 0) {
1112     sfree(writer->statement);
1113     sfree(writer->name);
1114     return status;
1115   }
1116
1117   ++writers_num;
1118   return 0;
1119 } /* c_psql_config_writer */
1120
1121 static int c_psql_config_database(oconfig_item_t *ci) {
1122   c_psql_database_t *db;
1123
1124   char cb_name[DATA_MAX_NAME_LEN];
1125   static _Bool have_flush = 0;
1126
1127   if ((1 != ci->values_num) || (OCONFIG_TYPE_STRING != ci->values[0].type)) {
1128     log_err("<Database> expects a single string argument.");
1129     return 1;
1130   }
1131
1132   db = c_psql_database_new(ci->values[0].value.string);
1133   if (db == NULL)
1134     return -1;
1135
1136   for (int i = 0; i < ci->children_num; ++i) {
1137     oconfig_item_t *c = ci->children + i;
1138
1139     if (0 == strcasecmp(c->key, "Host"))
1140       cf_util_get_string(c, &db->host);
1141     else if (0 == strcasecmp(c->key, "Port"))
1142       cf_util_get_service(c, &db->port);
1143     else if (0 == strcasecmp(c->key, "User"))
1144       cf_util_get_string(c, &db->user);
1145     else if (0 == strcasecmp(c->key, "Password"))
1146       cf_util_get_string(c, &db->password);
1147     else if (0 == strcasecmp(c->key, "Instance"))
1148       cf_util_get_string(c, &db->instance);
1149     else if (0 == strcasecmp (c->key, "Plugin"))
1150       cf_util_get_string (c, &db->plugin_name);
1151     else if (0 == strcasecmp(c->key, "SSLMode"))
1152       cf_util_get_string(c, &db->sslmode);
1153     else if (0 == strcasecmp(c->key, "KRBSrvName"))
1154       cf_util_get_string(c, &db->krbsrvname);
1155     else if (0 == strcasecmp(c->key, "Service"))
1156       cf_util_get_string(c, &db->service);
1157     else if (0 == strcasecmp(c->key, "Query"))
1158       udb_query_pick_from_list(c, queries, queries_num, &db->queries,
1159                                &db->queries_num);
1160     else if (0 == strcasecmp(c->key, "Writer"))
1161       config_add_writer(c, writers, writers_num, &db->writers,
1162                         &db->writers_num);
1163     else if (0 == strcasecmp(c->key, "Interval"))
1164       cf_util_get_cdtime(c, &db->interval);
1165     else if (strcasecmp("CommitInterval", c->key) == 0)
1166       cf_util_get_cdtime(c, &db->commit_interval);
1167     else if (strcasecmp("ExpireDelay", c->key) == 0)
1168       cf_util_get_cdtime(c, &db->expire_delay);
1169     else
1170       log_warn("Ignoring unknown config key \"%s\".", c->key);
1171   }
1172
1173   /* If no `Query' options were given, add the default queries.. */
1174   if ((db->queries_num == 0) && (db->writers_num == 0)) {
1175     for (int i = 0; i < def_queries_num; i++)
1176       udb_query_pick_from_list_by_name(def_queries[i], queries, queries_num,
1177                                        &db->queries, &db->queries_num);
1178   }
1179
1180   if (db->queries_num > 0) {
1181     db->q_prep_areas = (udb_query_preparation_area_t **)calloc(
1182         db->queries_num, sizeof(*db->q_prep_areas));
1183
1184     if (db->q_prep_areas == NULL) {
1185       log_err("Out of memory.");
1186       c_psql_database_delete(db);
1187       return -1;
1188     }
1189   }
1190
1191   for (int i = 0; (size_t)i < db->queries_num; ++i) {
1192     c_psql_user_data_t *data;
1193     data = udb_query_get_user_data(db->queries[i]);
1194     if ((data != NULL) && (data->params_num > db->max_params_num))
1195       db->max_params_num = data->params_num;
1196
1197     db->q_prep_areas[i] = udb_query_allocate_preparation_area(db->queries[i]);
1198
1199     if (db->q_prep_areas[i] == NULL) {
1200       log_err("Out of memory.");
1201       c_psql_database_delete(db);
1202       return -1;
1203     }
1204   }
1205
1206   snprintf(cb_name, sizeof(cb_name), "postgresql-%s", db->instance);
1207
1208   user_data_t ud = {.data = db, .free_func = c_psql_database_delete};
1209
1210   if (db->queries_num > 0) {
1211     ++db->ref_cnt;
1212     plugin_register_complex_read("postgresql", cb_name, c_psql_read,
1213                                  /* interval = */ db->interval, &ud);
1214   }
1215   if (db->writers_num > 0) {
1216     ++db->ref_cnt;
1217     plugin_register_write(cb_name, c_psql_write, &ud);
1218
1219     if (!have_flush) {
1220       /* flush all */
1221       plugin_register_flush("postgresql", c_psql_flush, /* user data = */ NULL);
1222       have_flush = 1;
1223     }
1224
1225     /* flush this connection only */
1226     ++db->ref_cnt;
1227     plugin_register_flush(cb_name, c_psql_flush, &ud);
1228   } else if (db->commit_interval > 0) {
1229     log_warn("Database '%s': You do not have any writers assigned to "
1230              "this database connection. Setting 'CommitInterval' does "
1231              "not have any effect.",
1232              db->database);
1233   }
1234   return 0;
1235 } /* c_psql_config_database */
1236
1237 static int c_psql_config(oconfig_item_t *ci) {
1238   static int have_def_config = 0;
1239
1240   if (0 == have_def_config) {
1241     oconfig_item_t *c;
1242
1243     have_def_config = 1;
1244
1245     c = oconfig_parse_file(C_PSQL_DEFAULT_CONF);
1246     if (NULL == c)
1247       log_err("Failed to read default config (" C_PSQL_DEFAULT_CONF ").");
1248     else
1249       c_psql_config(c);
1250
1251     if (NULL == queries)
1252       log_err("Default config (" C_PSQL_DEFAULT_CONF ") did not define "
1253               "any queries - please check your installation.");
1254   }
1255
1256   for (int i = 0; i < ci->children_num; ++i) {
1257     oconfig_item_t *c = ci->children + i;
1258
1259     if (0 == strcasecmp(c->key, "Query"))
1260       udb_query_create(&queries, &queries_num, c,
1261                        /* callback = */ config_query_callback);
1262     else if (0 == strcasecmp(c->key, "Writer"))
1263       c_psql_config_writer(c);
1264     else if (0 == strcasecmp(c->key, "Database"))
1265       c_psql_config_database(c);
1266     else
1267       log_warn("Ignoring unknown config key \"%s\".", c->key);
1268   }
1269   return 0;
1270 } /* c_psql_config */
1271
1272 void module_register(void) {
1273   plugin_register_complex_config("postgresql", c_psql_config);
1274   plugin_register_shutdown("postgresql", c_psql_shutdown);
1275 } /* module_register */