X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fpostgresql.c;h=d6dd0ba20e14399e30107aa1ef3f6674825c0ea3;hb=0742dc6da5074abaf3bb27c579de12460f7db690;hp=faad16cd9b02b404c5326279bd8fa1aa85e64624;hpb=ec404735b0676bb44fdba10f3760e1ba9632405a;p=collectd.git diff --git a/src/postgresql.c b/src/postgresql.c index faad16cd..d6dd0ba2 100644 --- a/src/postgresql.c +++ b/src/postgresql.c @@ -1,6 +1,7 @@ /** * collectd - src/postgresql.c - * Copyright (C) 2008 Sebastian Harl + * Copyright (C) 2008, 2009 Sebastian Harl + * Copyright (C) 2009 Florian Forster * * This program is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License as published by the @@ -15,8 +16,9 @@ * with this program; if not, write to the Free Software Foundation, Inc., * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA * - * Author: + * Authors: * Sebastian Harl + * Florian Forster **/ /* @@ -29,6 +31,7 @@ #include "configfile.h" #include "plugin.h" +#include "utils_db_query.h" #include "utils_complain.h" #include @@ -80,39 +83,27 @@ typedef enum { C_PSQL_PARAM_HOST = 1, C_PSQL_PARAM_DB, C_PSQL_PARAM_USER, + C_PSQL_PARAM_INTERVAL, } c_psql_param_t; +/* Parameter configuration. Stored as `user data' in the query objects. */ typedef struct { - char *type; - char *type_instance; - int ds_type; -} c_psql_col_t; - -typedef struct { - char *name; - char *query; - c_psql_param_t *params; int params_num; - - c_psql_col_t *cols; - int cols_num; - - int min_pg_version; - int max_pg_version; -} c_psql_query_t; +} c_psql_user_data_t; typedef struct { PGconn *conn; c_complain_t conn_complaint; int proto_version; + int server_version; int max_params_num; /* user configuration */ - c_psql_query_t **queries; - int queries_num; + udb_query_t **queries; + size_t queries_num; char *host; char *port; @@ -138,70 +129,12 @@ static char *def_queries[] = { }; static int def_queries_num = STATIC_ARRAY_SIZE (def_queries); -static c_psql_query_t *queries = NULL; -static int queries_num = 0; +static udb_query_t **queries = NULL; +static size_t queries_num = 0; static c_psql_database_t *databases = NULL; static int databases_num = 0; -static c_psql_query_t *c_psql_query_new (const char *name) -{ - c_psql_query_t *query; - - ++queries_num; - if (NULL == (queries = (c_psql_query_t *)realloc (queries, - queries_num * sizeof (*queries)))) { - log_err ("Out of memory."); - exit (5); - } - query = queries + queries_num - 1; - - query->name = sstrdup (name); - query->query = NULL; - - query->params = NULL; - query->params_num = 0; - - query->cols = NULL; - query->cols_num = 0; - - query->min_pg_version = 0; - query->max_pg_version = INT_MAX; - return query; -} /* c_psql_query_new */ - -static void c_psql_query_delete (c_psql_query_t *query) -{ - int i; - - sfree (query->name); - sfree (query->query); - - sfree (query->params); - query->params_num = 0; - - for (i = 0; i < query->cols_num; ++i) { - sfree (query->cols[i].type); - sfree (query->cols[i].type_instance); - } - sfree (query->cols); - query->cols_num = 0; - return; -} /* c_psql_query_delete */ - -static c_psql_query_t *c_psql_query_get (const char *name, int server_version) -{ - int i; - - for (i = 0; i < queries_num; ++i) - if (0 == strcasecmp (name, queries[i].name) - && ((-1 == server_version) - || ((queries[i].min_pg_version <= server_version) - && (server_version <= queries[i].max_pg_version)))) - return queries + i; - return NULL; -} /* c_psql_query_get */ - static c_psql_database_t *c_psql_database_new (const char *name) { c_psql_database_t *db; @@ -217,15 +150,15 @@ static c_psql_database_t *c_psql_database_new (const char *name) db->conn = NULL; - db->conn_complaint.last = 0; - db->conn_complaint.interval = 0; + C_COMPLAIN_INIT (&db->conn_complaint); db->proto_version = 0; + db->server_version = 0; db->max_params_num = 0; - db->queries = NULL; - db->queries_num = 0; + db->queries = NULL; + db->queries_num = 0; db->database = sstrdup (name); db->host = NULL; @@ -263,62 +196,6 @@ static void c_psql_database_delete (c_psql_database_t *db) return; } /* c_psql_database_delete */ -static void submit (const c_psql_database_t *db, - const char *type, const char *type_instance, - value_t *values, size_t values_len) -{ - value_list_t vl = VALUE_LIST_INIT; - - vl.values = values; - vl.values_len = values_len; - vl.time = time (NULL); - - if (C_PSQL_IS_UNIX_DOMAIN_SOCKET (db->host) - || (0 == strcmp (db->host, "localhost"))) - sstrncpy (vl.host, hostname_g, sizeof (vl.host)); - else - sstrncpy (vl.host, db->host, sizeof (vl.host)); - - sstrncpy (vl.plugin, "postgresql", sizeof (vl.plugin)); - sstrncpy (vl.plugin_instance, db->database, sizeof (vl.plugin_instance)); - - sstrncpy (vl.type, type, sizeof (vl.type)); - - if (NULL != type_instance) - sstrncpy (vl.type_instance, type_instance, sizeof (vl.type_instance)); - - plugin_dispatch_values (&vl); - return; -} /* submit */ - -static void submit_counter (const c_psql_database_t *db, - const char *type, const char *type_instance, - const char *value) -{ - value_t values[1]; - - if ((NULL == value) || ('\0' == *value)) - return; - - values[0].counter = atoll (value); - submit (db, type, type_instance, values, 1); - return; -} /* submit_counter */ - -static void submit_gauge (const c_psql_database_t *db, - const char *type, const char *type_instance, - const char *value) -{ - value_t values[1]; - - if ((NULL == value) || ('\0' == *value)) - return; - - values[0].gauge = atof (value); - submit (db, type, type_instance, values, 1); - return; -} /* submit_gauge */ - static int c_psql_check_connection (c_psql_database_t *db) { /* "ping" */ @@ -344,21 +221,33 @@ static int c_psql_check_connection (c_psql_database_t *db) db->proto_version); } + db->server_version = PQserverVersion (db->conn); + c_release (LOG_INFO, &db->conn_complaint, "Successfully reconnected to database %s", PQdb (db->conn)); return 0; } /* c_psql_check_connection */ +static PGresult *c_psql_exec_query_noparams (c_psql_database_t *db, + udb_query_t *q) +{ + return PQexec (db->conn, udb_query_get_statement (q)); +} /* c_psql_exec_query_noparams */ + static PGresult *c_psql_exec_query_params (c_psql_database_t *db, - c_psql_query_t *query) + udb_query_t *q, c_psql_user_data_t *data) { char *params[db->max_params_num]; + char interval[64]; int i; - assert (db->max_params_num >= query->params_num); + if ((data == NULL) || (data->params_num == 0)) + return (c_psql_exec_query_noparams (db, q)); + + assert (db->max_params_num >= data->params_num); - for (i = 0; i < query->params_num; ++i) { - switch (query->params[i]) { + for (i = 0; i < data->params_num; ++i) { + switch (data->params[i]) { case C_PSQL_PARAM_HOST: params[i] = C_PSQL_IS_UNIX_DOMAIN_SOCKET (db->host) ? "localhost" : db->host; @@ -369,85 +258,137 @@ static PGresult *c_psql_exec_query_params (c_psql_database_t *db, case C_PSQL_PARAM_USER: params[i] = db->user; break; + case C_PSQL_PARAM_INTERVAL: + ssnprintf (interval, sizeof (interval), "%i", interval_g); + params[i] = interval; + break; default: assert (0); } } - return PQexecParams (db->conn, query->query, query->params_num, NULL, - (const char *const *)((0 == query->params_num) ? NULL : params), + return PQexecParams (db->conn, udb_query_get_statement (q), + data->params_num, NULL, + (const char *const *) params, NULL, NULL, /* return text data */ 0); } /* c_psql_exec_query_params */ -static PGresult *c_psql_exec_query_noparams (c_psql_database_t *db, - c_psql_query_t *query) +static int c_psql_exec_query (c_psql_database_t *db, udb_query_t *q) { - return PQexec (db->conn, query->query); -} /* c_psql_exec_query_noparams */ + PGresult *res; -static int c_psql_exec_query (c_psql_database_t *db, int idx) -{ - c_psql_query_t *query; - PGresult *res; + c_psql_user_data_t *data; - int rows, cols; - int i; + const char *host; - if (idx >= db->queries_num) - return -1; + char **column_names; + char **column_values; + int column_num; - query = db->queries[idx]; + int rows_num; + int status; + int row, col; + /* The user data may hold parameter information, but may be NULL. */ + data = udb_query_get_user_data (q); + + /* Versions up to `3' don't know how to handle parameters. */ if (3 <= db->proto_version) - res = c_psql_exec_query_params (db, query); - else if (0 == query->params_num) - res = c_psql_exec_query_noparams (db, query); + res = c_psql_exec_query_params (db, q, data); + else if ((NULL == data) || (0 == data->params_num)) + res = c_psql_exec_query_noparams (db, q); else { log_err ("Connection to database \"%s\" does not support parameters " "(protocol version %d) - cannot execute query \"%s\".", - db->database, db->proto_version, query->name); + db->database, db->proto_version, + udb_query_get_name (q)); return -1; } + column_names = NULL; + column_values = NULL; + +#define BAIL_OUT(status) \ + sfree (column_names); \ + sfree (column_values); \ + PQclear (res); \ + return status + if (PGRES_TUPLES_OK != PQresultStatus (res)) { log_err ("Failed to execute SQL query: %s", PQerrorMessage (db->conn)); - log_info ("SQL query was: %s", query->query); - PQclear (res); - return -1; + log_info ("SQL query was: %s", + udb_query_get_statement (q)); + BAIL_OUT (-1); } - rows = PQntuples (res); - if (1 > rows) { - PQclear (res); - return 0; + rows_num = PQntuples (res); + if (1 > rows_num) { + BAIL_OUT (0); } - cols = PQnfields (res); - if (query->cols_num != cols) { - log_err ("SQL query returned wrong number of fields " - "(expected: %i, got: %i)", query->cols_num, cols); - log_info ("SQL query was: %s", query->query); - PQclear (res); - return -1; + column_num = PQnfields (res); + column_names = (char **) calloc (column_num, sizeof (char *)); + if (NULL == column_names) { + log_err ("calloc failed."); + BAIL_OUT (-1); } - for (i = 0; i < rows; ++i) { - int j; + column_values = (char **) calloc (column_num, sizeof (char *)); + if (NULL == column_values) { + log_err ("calloc failed."); + BAIL_OUT (-1); + } + + for (col = 0; col < column_num; ++col) { + /* Pointers returned by `PQfname' are freed by `PQclear' via + * `BAIL_OUT'. */ + column_names[col] = PQfname (res, col); + if (NULL == column_names[col]) { + log_err ("Failed to resolv name of column %i.", col); + BAIL_OUT (-1); + } + } - for (j = 0; j < cols; ++j) { - c_psql_col_t col = query->cols[j]; + if (C_PSQL_IS_UNIX_DOMAIN_SOCKET (db->host) + || (0 == strcmp (db->host, "localhost"))) + host = hostname_g; + else + host = db->host; + + status = udb_query_prepare_result (q, host, "postgresql", + db->database, column_names, (size_t) column_num); + if (0 != status) { + log_err ("udb_query_prepare_result failed with status %i.", + status); + BAIL_OUT (-1); + } + + for (row = 0; row < rows_num; ++row) { + for (col = 0; col < column_num; ++col) { + /* Pointers returned by `PQgetvalue' are freed by `PQclear' via + * `BAIL_OUT'. */ + column_values[col] = PQgetvalue (res, row, col); + if (NULL == column_values[col]) { + log_err ("Failed to get value at (row = %i, col = %i).", + row, col); + break; + } + } - char *value = PQgetvalue (res, i, j); + /* check for an error */ + if (col < column_num) + continue; - if (col.ds_type == DS_TYPE_COUNTER) - submit_counter (db, col.type, col.type_instance, value); - else if (col.ds_type == DS_TYPE_GAUGE) - submit_gauge (db, col.type, col.type_instance, value); + status = udb_query_handle_result (q, column_values); + if (status != 0) { + log_err ("udb_query_handle_result failed with status %i.", + status); } - } - PQclear (res); - return 0; + } /* for (row = 0; row < rows_num; ++row) */ + + BAIL_OUT (0); +#undef BAIL_OUT } /* c_psql_exec_query */ static int c_psql_read (void) @@ -466,7 +407,17 @@ static int c_psql_read (void) continue; for (j = 0; j < db->queries_num; ++j) - c_psql_exec_query (db, j); + { + udb_query_t *q; + + q = db->queries[j]; + + if ((0 != db->server_version) + && (udb_query_check_version (q, db->server_version) <= 0)) + continue; + + c_psql_exec_query (db, q); + } ++success; } @@ -486,21 +437,16 @@ static int c_psql_shutdown (void) plugin_unregister_read ("postgresql"); plugin_unregister_shutdown ("postgresql"); - for (i = 0; i < databases_num; ++i) { - c_psql_database_t *db = databases + i; - c_psql_database_delete (db); - } + for (i = 0; i < databases_num; ++i) + c_psql_database_delete (databases + i); sfree (databases); databases_num = 0; - for (i = 0; i < queries_num; ++i) { - c_psql_query_t *query = queries + i; - c_psql_query_delete (query); - } - - sfree (queries); + udb_query_free (queries, queries_num); + queries = NULL; queries_num = 0; + return 0; } /* c_psql_shutdown */ @@ -511,33 +457,6 @@ static int c_psql_init (void) if ((NULL == databases) || (0 == databases_num)) return 0; - for (i = 0; i < queries_num; ++i) { - c_psql_query_t *query = queries + i; - int j; - - for (j = 0; j < query->cols_num; ++j) { - c_psql_col_t *col = query->cols + j; - const data_set_t *ds; - - ds = plugin_get_ds (col->type); - if (NULL == ds) { - log_err ("Column: Unknown type \"%s\".", col->type); - c_psql_shutdown (); - return -1; - } - - if (1 != ds->ds_num) { - log_err ("Column: Invalid type \"%s\" - types defining " - "one data source are supported only (got: %i).", - col->type, ds->ds_num); - c_psql_shutdown (); - return -1; - } - - col->ds_type = ds->ds[0].type; - } - } - for (i = 0; i < databases_num; ++i) { c_psql_database_t *db = databases + i; @@ -549,8 +468,6 @@ static int c_psql_init (void) char *server_host; int server_version; - int j; - /* this will happen during reinitialization */ if (NULL != db->conn) { c_psql_check_connection (db); @@ -590,33 +507,6 @@ static int c_psql_init (void) if (3 > db->proto_version) log_warn ("Protocol version %d does not support parameters.", db->proto_version); - - /* Now that we know the PostgreSQL server version, we can get the - * right version of each query definition. */ - for (j = 0; j < db->queries_num; ++j) { - c_psql_query_t *tmp; - - tmp = c_psql_query_get (db->queries[j]->name, server_version); - - if (tmp == db->queries[j]) - continue; - - if (NULL == tmp) { - log_err ("Query \"%s\" not found for server version %i - " - "please check your configuration.", - db->queries[j]->name, server_version); - - if (db->queries_num - j - 1 > 0) - memmove (db->queries + j, db->queries + j + 1, - (db->queries_num - j - 1) * sizeof (*db->queries)); - - --db->queries_num; - --j; - continue; - } - - db->queries[j] = tmp; - } } plugin_register_read ("postgresql", c_psql_read); @@ -637,185 +527,61 @@ static int config_set_s (char *name, char **var, const oconfig_item_t *ci) return 0; } /* config_set_s */ -static int config_set_i (char *name, int *var, const oconfig_item_t *ci) +static int config_query_param_add (udb_query_t *q, oconfig_item_t *ci) { - if ((0 != ci->children_num) || (1 != ci->values_num) - || (OCONFIG_TYPE_NUMBER != ci->values[0].type)) { - log_err ("%s expects a single number argument.", name); - return 1; - } + c_psql_user_data_t *data; + const char *param_str; - *var = (int)ci->values[0].value.number; - return 0; -} /* config_set_i */ + c_psql_param_t *tmp; -static int config_set_param (c_psql_query_t *query, const oconfig_item_t *ci) -{ - c_psql_param_t param; - char *param_str; + data = udb_query_get_user_data (q); + if (NULL == data) { + data = (c_psql_user_data_t *) smalloc (sizeof (*data)); + if (NULL == data) { + log_err ("Out of memory."); + return -1; + } + memset (data, 0, sizeof (*data)); + data->params = NULL; + } - if ((0 != ci->children_num) || (1 != ci->values_num) - || (OCONFIG_TYPE_STRING != ci->values[0].type)) { - log_err ("Param expects a single string argument."); - return 1; + tmp = (c_psql_param_t *) realloc (data->params, + (data->params_num + 1) * sizeof (c_psql_param_t)); + if (NULL == tmp) { + log_err ("Out of memory."); + return -1; } + data->params = tmp; param_str = ci->values[0].value.string; if (0 == strcasecmp (param_str, "hostname")) - param = C_PSQL_PARAM_HOST; + data->params[data->params_num] = C_PSQL_PARAM_HOST; else if (0 == strcasecmp (param_str, "database")) - param = C_PSQL_PARAM_DB; + data->params[data->params_num] = C_PSQL_PARAM_DB; else if (0 == strcasecmp (param_str, "username")) - param = C_PSQL_PARAM_USER; + data->params[data->params_num] = C_PSQL_PARAM_USER; + else if (0 == strcasecmp (param_str, "interval")) + data->params[data->params_num] = C_PSQL_PARAM_INTERVAL; else { log_err ("Invalid parameter \"%s\".", param_str); return 1; } - ++query->params_num; - if (NULL == (query->params = (c_psql_param_t *)realloc (query->params, - query->params_num * sizeof (*query->params)))) { - log_err ("Out of memory."); - exit (5); - } - - query->params[query->params_num - 1] = param; - return 0; -} /* config_set_param */ - -static int config_set_column (c_psql_query_t *query, const oconfig_item_t *ci) -{ - c_psql_col_t *col; - - int i; - - if ((0 != ci->children_num) - || (1 > ci->values_num) || (2 < ci->values_num)) { - log_err ("Column expects either one or two arguments."); - return 1; - } - - for (i = 0; i < ci->values_num; ++i) { - if (OCONFIG_TYPE_STRING != ci->values[i].type) { - log_err ("Column expects either one or two string arguments."); - return 1; - } - } - - ++query->cols_num; - if (NULL == (query->cols = (c_psql_col_t *)realloc (query->cols, - query->cols_num * sizeof (*query->cols)))) { - log_err ("Out of memory."); - exit (5); - } - - col = query->cols + query->cols_num - 1; - - col->ds_type = -1; - - col->type = sstrdup (ci->values[0].value.string); - col->type_instance = (2 == ci->values_num) - ? sstrdup (ci->values[1].value.string) : NULL; - return 0; -} /* config_set_column */ - -static int set_query (c_psql_database_t *db, const char *name) -{ - c_psql_query_t *query; - - query = c_psql_query_get (name, -1); - if (NULL == query) { - log_err ("Query \"%s\" not found - please check your configuration.", - name); - return 1; - } - - ++db->queries_num; - if (NULL == (db->queries = (c_psql_query_t **)realloc (db->queries, - db->queries_num * sizeof (*db->queries)))) { - log_err ("Out of memory."); - exit (5); - } - - if (query->params_num > db->max_params_num) - db->max_params_num = query->params_num; + data->params_num++; + udb_query_set_user_data (q, data); - db->queries[db->queries_num - 1] = query; - return 0; -} /* set_query */ + return (0); +} /* config_query_param_add */ -static int config_set_query (c_psql_database_t *db, const oconfig_item_t *ci) +static int config_query_callback (udb_query_t *q, oconfig_item_t *ci) { - if ((0 != ci->children_num) || (1 != ci->values_num) - || (OCONFIG_TYPE_STRING != ci->values[0].type)) { - log_err ("Query expects a single string argument."); - return 1; - } - return set_query (db, ci->values[0].value.string); -} /* config_set_query */ + if (0 == strcasecmp ("Param", ci->key)) + return config_query_param_add (q, ci); -static int c_psql_config_query (oconfig_item_t *ci) -{ - c_psql_query_t *query; + log_err ("Option not allowed within a Query block: `%s'", ci->key); - int i; - - if ((1 != ci->values_num) - || (OCONFIG_TYPE_STRING != ci->values[0].type)) { - log_err (" expects a single string argument."); - return 1; - } - - query = c_psql_query_new (ci->values[0].value.string); - - for (i = 0; i < ci->children_num; ++i) { - oconfig_item_t *c = ci->children + i; - - if (0 == strcasecmp (c->key, "Query")) - config_set_s ("Query", &query->query, c); - else if (0 == strcasecmp (c->key, "Param")) - config_set_param (query, c); - else if (0 == strcasecmp (c->key, "Column")) - config_set_column (query, c); - else if (0 == strcasecmp (c->key, "MinPGVersion")) - config_set_i ("MinPGVersion", &query->min_pg_version, c); - else if (0 == strcasecmp (c->key, "MaxPGVersion")) - config_set_i ("MaxPGVersion", &query->max_pg_version, c); - else - log_warn ("Ignoring unknown config key \"%s\".", c->key); - } - - for (i = 0; i < queries_num - 1; ++i) { - c_psql_query_t *q = queries + i; - - if ((0 == strcasecmp (q->name, query->name)) - && (q->min_pg_version <= query->max_pg_version) - && (query->min_pg_version <= q->max_pg_version)) { - log_err ("Ignoring redefinition (with overlapping version ranges) " - "of query \"%s\".", query->name); - c_psql_query_delete (query); - --queries_num; - return 1; - } - } - - if (query->min_pg_version > query->max_pg_version) { - log_err ("Query \"%s\": MinPGVersion > MaxPGVersion.", - query->name); - c_psql_query_delete (query); - --queries_num; - return 1; - } - - if (NULL == query->query) { - log_err ("Query \"%s\" does not include an SQL query string - " - "please check your configuration.", query->name); - c_psql_query_delete (query); - --queries_num; - return 1; - } - return 0; -} /* c_psql_config_query */ + return (-1); +} /* config_query_callback */ static int c_psql_config_database (oconfig_item_t *ci) { @@ -849,17 +615,29 @@ static int c_psql_config_database (oconfig_item_t *ci) else if (0 == strcasecmp (c->key, "Service")) config_set_s ("Service", &db->service, c); else if (0 == strcasecmp (c->key, "Query")) - config_set_query (db, c); + udb_query_pick_from_list (c, queries, queries_num, + &db->queries, &db->queries_num); else log_warn ("Ignoring unknown config key \"%s\".", c->key); } - if (NULL == db->queries) { - for (i = 0; i < def_queries_num; ++i) - set_query (db, def_queries[i]); + /* If no `Query' options were given, add the default queries.. */ + if (db->queries_num == 0) + { + for (i = 0; i < def_queries_num; i++) + udb_query_pick_from_list_by_name (def_queries[i], + queries, queries_num, + &db->queries, &db->queries_num); + } + + for (i = 0; (size_t)i < db->queries_num; ++i) { + c_psql_user_data_t *data; + data = udb_query_get_user_data (db->queries[i]); + if ((data != NULL) && (data->params_num > db->max_params_num)) + db->max_params_num = data->params_num; } return 0; -} +} /* c_psql_config_database */ static int c_psql_config (oconfig_item_t *ci) { @@ -887,7 +665,9 @@ static int c_psql_config (oconfig_item_t *ci) oconfig_item_t *c = ci->children + i; if (0 == strcasecmp (c->key, "Query")) - c_psql_config_query (c); + udb_query_create (&queries, &queries_num, c, + /* callback = */ config_query_callback, + /* legacy mode = */ 1); else if (0 == strcasecmp (c->key, "Database")) c_psql_config_database (c); else @@ -903,4 +683,3 @@ void module_register (void) } /* module_register */ /* vim: set sw=4 ts=4 tw=78 noexpandtab : */ -