/* ex: set tabstop=4 shiftwidth=4 autoindent: +-------------------------------------------------------------------------+ | Copyright (C) 2004-2026 The Cacti Group | | | | This program is free software; you can redistribute it and/or | | modify it under the terms of the GNU Lesser General Public | | License as published by the Free Software Foundation; either | | version 2.1 of the License, or (at your option) any later version. | | | | This program is distributed in the hope that it will be useful, | | but WITHOUT ANY WARRANTY; without even the implied warranty of | | MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | | GNU Lesser General Public License for more details. | | | | You should have received a copy of the GNU Lesser General Public | | License along with this library; if not, write to the Free Software | | Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA | | 02110-1301, USA | | | +-------------------------------------------------------------------------+ | spine: a backend data gatherer for cacti | +-------------------------------------------------------------------------+ | This poller would not have been possible without: | | - Larry Adams (current development and enhancements) | | - Rivo Nurges (rrd support, mysql poller cache, misc functions) | | - RTG (core poller code, pthreads, snmp, autoconf examples) | | - Brady Alleman/Doug Warner (threading ideas, implementation details) | +-------------------------------------------------------------------------+ | - Cacti - http://www.cacti.net/ | +-------------------------------------------------------------------------+ */ #include "common.h" #include "spine.h" /*! \fn int db_insert(MYSQL *mysql, int type, const char *query) * \brief inserts a row or rows in a database table. * \param mysql the database connection object * \param type the database to connect to local or remote * \param query the database query to execute * * Unless the SQL_readonly boolean is set to TRUE, the function will execute * the SQL statement specified in the query variable. * * \return TRUE if successful, or FALSE if not. * */ static void retry_disconnected_query(MYSQL *mysql, int type, int error, const char *function, int *error_count) { if (errno == EINTR) { spine_sleep_usec(50000); return; } db_reconnect(mysql, type, error, function); if (++*error_count > 30) die("FATAL: Too many Reconnect Attempts!"); } int db_insert(MYSQL *mysql, int type, const char *query) { int error_count = 0; char query_frag[LRG_BUFSIZE]; snprintf(query_frag, sizeof(query_frag), "%s", query); SPINE_LOG_DEVDBG(("DEVDBG: SQL:%s", query_frag)); if (set.poller.SQL_readonly != FALSE) return TRUE; while (mysql_query(mysql, query) != 0) { int error = mysql_errno(mysql); if (error == 2013 || error == 2006) { retry_disconnected_query(mysql, type, error, "db_insert", &error_count); continue; } if (error == 1213 || error == 1205) { spine_sleep_usec(50000); if (++error_count > 30) { SPINE_LOG(("ERROR: Too many Lock/Deadlock errors occurred!, SQL Fragment:'%s'", query_frag)); return FALSE; } continue; } SPINE_LOG(("ERROR: SQL Failed! Error:'%i', Message:'%s', SQL Fragment:'%s'", error, mysql_error(mysql), query_frag)); return FALSE; } return TRUE; } int db_reconnect(MYSQL *mysql, int type, int error, const char *function) { unsigned long mysql_thread = 0; char query[100]; int ping_status; mysql_thread = mysql_thread_id(mysql); ping_status = mysql_ping(mysql); if (mysql_thread_id(mysql) != mysql_thread) { SPINE_LOG(("WARNING: Connection Broken in Function %s with Error %i. Reconnect via mysql_ping() successful.", function, error)); snprintf(query, 100, "KILL %lu;", mysql_thread); mysql_query(mysql, query); mysql_query(mysql, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'NO_ZERO_DATE', ''))"); mysql_query(mysql, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'NO_ZERO_IN_DATE', ''))"); mysql_query(mysql, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'ONLY_FULL_GROUP_BY', ''))"); mysql_query(mysql, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'NO_AUTO_VALUE_ON_ZERO', ''))"); mysql_query(mysql, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'TRADITIONAL', ''))"); mysql_query(mysql, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'STRICT_ALL_TABLES', ''))"); sleep(1); return TRUE; } /* A live session on the same thread has nothing to reconnect. */ if (ping_status == 0) { return FALSE; } /* mysql_ping() did not reconnect; do it explicitly */ SPINE_LOG(("WARNING: Connection Broken in Function %s with Error %i. Attempting explicit reconnect.", function, error)); mysql_close(mysql); db_connect(type, mysql); if (mysql_thread_id(mysql) > 0) { SPINE_LOG(("WARNING: Explicit reconnect successful in Function %s.", function)); return TRUE; } SPINE_LOG(("WARNING: Connection Broken with Error %i. Reconnect failed.", error)); return FALSE; } /*! \fn MYSQL_RES *db_query(MYSQL *mysql, int type, const char *query) * \brief executes a query and returns a pointer to the result set. * \param mysql the database connection object * \param query the database query to execute * * This function will execute the SQL statement specified in the query variable. * * \return MYSQL_RES a MySQL result structure * */ MYSQL_RES *db_query(MYSQL *mysql, int type, const char *query) { int error_count = 0; char query_frag[LRG_BUFSIZE]; snprintf(query_frag, sizeof(query_frag), "%s", query); SPINE_LOG_DEVDBG(("DEVDBG: SQL:%s", query_frag)); while (mysql_query(mysql, query) != 0) { int error = mysql_errno(mysql); if (error == 2013 || error == 2006) { retry_disconnected_query(mysql, type, error, "db_query", &error_count); continue; } if (error == 1213 || error == 1205) { spine_sleep_usec(50000); if (++error_count > 30) { SPINE_LOG(("FATAL: Too many Lock/Deadlock errors occurred!, SQL Fragment:'%s'", query_frag)); exit(1); } continue; } SPINE_LOG(("FATAL: Database Error:'%i', Message:'%s'", error, mysql_error(mysql))); SPINE_LOG(("ERROR: The Query Was:'%s'", query)); exit(1); } return mysql_store_result(mysql); } /*! \fn void db_connect(char *database, MYSQL *mysql) * \brief opens a connection to a MySQL database. * \param database a string pointer to the database name * \param mysql a pointer to a mysql database connection object * * This function will attempt to open a connection to a MySQL database and then * return the connection object to the calling function. If the database connection * fails more than 20 times, the function will fail and Spine will terminate. * */ void db_address_init(db_address_t *address, const char *value, bool parse_socket) { address->storage = strdup(value); if (address->storage == NULL) die("ERROR: Fatal malloc error: database address!"); address->hostname = address->storage; address->socket = NULL; if (!parse_socket) return; struct stat socket_stat; if (stat(address->storage, &socket_stat) == 0) { if (S_ISSOCK(socket_stat.st_mode)) { address->socket = address->storage; address->hostname = NULL; } return; } address->socket = strchr(address->storage, ':'); if (address->socket != NULL) *address->socket++ = '\0'; } void db_address_release(db_address_t *address) { free(address->storage); address->storage = NULL; address->hostname = NULL; address->socket = NULL; } void db_set_option(MYSQL *mysql, enum mysql_option option, const void *value, const char *description) { if (mysql_options(mysql, option, value) != 0) { set.exit.exit_code = EXIT_FAILURE; die("FATAL: MySQL options unable to set %s option", description); } } #ifdef HAS_MYSQL_OPT_SSL_KEY static void db_set_ssl_options(MYSQL *mysql, int type) { #ifdef HAS_MYSQL_OPT_SSL_VERIFY_SERVER_CERT int ssl_enabled = type == LOCAL ? set.database.ssl : set.remote_database.ssl; if (ssl_enabled == 0) { bool ssl_enforce = false; db_set_option(mysql, MYSQL_OPT_SSL_VERIFY_SERVER_CERT, &ssl_enforce, "ssl disable"); } #endif const char *key = type == REMOTE ? set.remote_database.ssl_key : set.database.ssl_key; const char *ca = type == REMOTE ? set.remote_database.ssl_ca : set.database.ssl_ca; const char *cert = type == REMOTE ? set.remote_database.ssl_cert : set.database.ssl_cert; if (key[0] != '\0') db_set_option(mysql, MYSQL_OPT_SSL_KEY, key, "ssl key"); if (ca[0] != '\0') db_set_option(mysql, MYSQL_OPT_SSL_CA, ca, "ssl ca"); if (cert[0] != '\0') db_set_option(mysql, MYSQL_OPT_SSL_CERT, cert, "ssl cert"); } #endif void db_connect(int type, MYSQL *mysql) { int tries; int attempts; int timeout; int rtimeout; int wtimeout; int success; int error = 0; const MYSQL *connect_error; db_address_t address; static int connections = 0; bool local_address = set.poller.poller_id <= 1 || type == LOCAL; db_address_init(&address, local_address ? set.database.host : set.remote_database.host, local_address); /* initialalize variables */ tries = 2; success = FALSE; timeout = 5; rtimeout = 30; wtimeout = 30; attempts = 1; if (mysql_init(mysql) == NULL) { db_address_release(&address); printf("FATAL: Database unable to allocate memory and therefore can not connect\n"); exit(1); } db_set_option(mysql, MYSQL_OPT_READ_TIMEOUT, &rtimeout, "read timeout"); db_set_option(mysql, MYSQL_OPT_WRITE_TIMEOUT, &wtimeout, "write timeout"); db_set_option(mysql, MYSQL_OPT_CONNECT_TIMEOUT, &timeout, "general timeout"); #ifdef HAS_MYSQL_OPT_RETRY_COUNT db_set_option(mysql, MYSQL_OPT_RETRY_COUNT, &tries, "retry count"); #endif #ifdef HAS_MYSQL_OPT_SSL_KEY db_set_ssl_options(mysql, type); #endif while (tries > 0) { tries--; if (set.poller.poller_id > 1) { if (type == LOCAL) { connect_error = mysql_real_connect(mysql, address.hostname, set.database.user, set.database.password, set.database.database, set.database.port, address.socket, 0); } else { connect_error = mysql_real_connect(mysql, address.hostname, set.remote_database.user, set.remote_database.password, set.remote_database.database, set.remote_database.port, address.socket, 0); } } else { connect_error = mysql_real_connect(mysql, address.hostname, set.database.user, set.database.password, set.database.database, set.database.port, address.socket, 0); } if (!connect_error) { error = mysql_errno(mysql); if ((error == 2002 || error == 2003 || error == 2006 || error == 2013) && errno == EINTR) { spine_sleep_usec(5000); tries++; success = FALSE; } else if (error == 2002) { printf("Database: Connection Failed: Attempt:'%d', Error:'%u', Message:'%s'\n", attempts, mysql_errno(mysql), mysql_error(mysql)); sleep(1); success = FALSE; } else if (error != 1049 && error != 2005 && error != 1045) { printf("Database: Connection Failed: Error:'%d', Message:'%s'\n", error, mysql_error(mysql)); success = FALSE; spine_sleep_usec(50000); } else { tries = 0; success = FALSE; } } else { success = TRUE; break; } attempts++; } db_address_release(&address); if (!success){ printf("FATAL: Connection Failed, Error:'%i', Message:'%s'\n", error, mysql_error(mysql)); exit(1); } SPINE_LOG_DEBUG(("DEBUG: Total Connections made %i", connections)); connections++; } /*! \fn void db_disconnect(MYSQL *mysql) * \brief closes connection to MySQL database * \param mysql the database connection object * */ void db_disconnect(MYSQL *mysql) { if (mysql != NULL) { mysql_close(mysql); } } /*! \fn void db_create_connection_pool(int type) * \brief Creates a connection pool for spine * \param type the connection type, LOCAL or REMOTE * */ void db_create_connection_pool(int type) { int id; if (type == LOCAL) { SPINE_LOG_DEBUG(("DEBUG: Creating Local Connection Pool of %i threads.", set.poller.threads)); for(id = 0; id < set.poller.threads; id++) { SPINE_LOG_DEBUG(("DEBUG: Creating Local Connection %i.", id)); db_connect(type, &db_pool_local[id].mysql); db_insert(&db_pool_local[id].mysql, LOCAL, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'NO_ZERO_DATE', ''))"); db_insert(&db_pool_local[id].mysql, LOCAL, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'NO_ZERO_IN_DATE', ''))"); db_insert(&db_pool_local[id].mysql, LOCAL, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'ONLY_FULL_GROUP_BY', ''))"); db_insert(&db_pool_local[id].mysql, LOCAL, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'NO_AUTO_VALUE_ON_ZERO', ''))"); db_insert(&db_pool_local[id].mysql, LOCAL, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'TRADITIONAL', ''))"); db_insert(&db_pool_local[id].mysql, LOCAL, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'STRICT_ALL_TABLES', ''))"); db_insert(&db_pool_local[id].mysql, LOCAL, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'STRICT_TRANS_TABLES', ''))"); db_pool_local[id].free = TRUE; db_pool_local[id].id = id; } } else { SPINE_LOG_DEBUG(("DEBUG: Creating Remote Connection Pool of %i threads.", set.poller.threads)); for(id = 0; id < set.poller.threads; id++) { SPINE_LOG_DEBUG(("DEBUG: Creating Remote Connection %i.", id)); db_connect(type, &db_pool_remote[id].mysql); db_insert(&db_pool_remote[id].mysql, LOCAL, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'NO_ZERO_DATE', ''))"); db_insert(&db_pool_remote[id].mysql, LOCAL, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'NO_ZERO_IN_DATE', ''))"); db_insert(&db_pool_remote[id].mysql, LOCAL, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'ONLY_FULL_GROUP_BY', ''))"); db_insert(&db_pool_remote[id].mysql, LOCAL, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'NO_AUTO_VALUE_ON_ZERO', ''))"); db_insert(&db_pool_remote[id].mysql, LOCAL, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'TRADITIONAL', ''))"); db_insert(&db_pool_remote[id].mysql, LOCAL, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'STRICT_ALL_TABLES', ''))"); db_insert(&db_pool_remote[id].mysql, LOCAL, "SET SESSION sql_mode = (SELECT REPLACE(@@sql_mode,'STRICT_TRANS_TABLES', ''))"); db_pool_remote[id].free = TRUE; db_pool_remote[id].id = id; } } } /*! \fn void db_close_connection_pool(int type) * \brief Closes a connection pool for spine * \param type the connection type, LOCAL or REMOTE * */ void db_close_connection_pool(int type) { int id; if (type == LOCAL) { for(id = 0; id < set.poller.threads; id++) { SPINE_LOG_DEBUG(("DEBUG: Closing Local Connection Pool ID %i", id)); db_disconnect(&db_pool_local[id].mysql); } free(db_pool_local); } else { for(id = 0; id < set.poller.threads; id++) { SPINE_LOG_DEBUG(("DEBUG: Closing Remote Connection Pool ID %i", id)); db_disconnect(&db_pool_remote[id].mysql); } free(db_pool_remote); } } /*! \fn pool_t db_get_connection(int type) * \brief returns a free mysql connection from the pool * \param type the connection type, LOCAL or REMOTE * */ pool_t *db_get_connection(int type) { int id; thread_mutex_lock(LOCK_POOL); if (type == LOCAL) { SPINE_LOG_DEBUG(("DEBUG: Traversing Local Connection Pool for free connection.")); for (id = 0; id < set.poller.threads; id++) { SPINE_LOG_DEBUG(("DEBUG: Checking Local Pool ID %i.", id)); if (db_pool_local[id].free == TRUE) { SPINE_LOG_DEBUG(("DEBUG: Allocating Local Pool ID %i.", id)); db_pool_local[id].free = FALSE; thread_mutex_unlock(LOCK_POOL); return &db_pool_local[id]; } } } else { SPINE_LOG_DEBUG(("DEBUG: Traversing Remote Connection Pool for free connection.")); for (id = 0; id < set.poller.threads; id++) { SPINE_LOG_DEBUG(("DEBUG: Checking Remote Pool ID %i.", id)); if (db_pool_remote[id].free == TRUE) { SPINE_LOG_DEBUG(("DEBUG: Allocating Remote Pool ID %i.", id)); db_pool_remote[id].free = FALSE; thread_mutex_unlock(LOCK_POOL); return &db_pool_remote[id]; } } } SPINE_LOG(("FATAL: Connection Pool Fatal Error.")); thread_mutex_unlock(LOCK_POOL); return NULL; } /*! \fn voi db_release_connection(int id) * \brief marks a database connection as free * \param id the connection id * */ void db_release_connection(int type, int id) { thread_mutex_lock(LOCK_POOL); if (type == LOCAL) { SPINE_LOG_DEBUG(("DEBUG: Freeing Local Pool ID %i", id)); db_pool_local[id].free = TRUE; } else { SPINE_LOG_DEBUG(("DEBUG: Freeing Remote Pool ID %i", id)); db_pool_remote[id].free = TRUE; } thread_mutex_unlock(LOCK_POOL); } /*! \fn int append_hostrange(char *obuf, const char *colname, const config_t *set) * \brief appends a host range to a sql select statement * \param obuf the sql select statement to have the host range appended * \param colname the sql column name that will have the host range checked * \param set global runtime settings * * Several places in the code need to limit the range of hosts to * those with a certain ID range, but only if those range values * are actually nonzero. * * This appends the SQL clause if necessary, returning the # of * characters added to the buffer. Else return 0. * * \return the number of characters added to the end of the character buffer * */ int append_hostrange(char *obuf, size_t capacity, const char *colname) { if (HOSTID_DEFINED(set.hosts.start_host_id) && HOSTID_DEFINED(set.hosts.end_host_id)) { return spine_snprintf(obuf, capacity, " AND %s BETWEEN %d AND %d", colname, set.hosts.start_host_id, set.hosts.end_host_id); } else { return 0; } } /*! \fn void db_escape(MYSQL *mysql, char *output, int max_size, const char *input) * \brief Escapes a text string to make it safe for mysql insert/updates * \param mysql the connection object * \param output a pointer to the output string * \param a pointer to the input string * * A simple implementation of the mysql_real_escape_string that one * day should be portable. * * \return void * */ void db_escape(MYSQL *mysql, char *output, int max_size, const char *input) { size_t input_len; size_t max_input; if (input == NULL || output == NULL) return; if (max_size <= 1) { if (max_size == 1) { *output = '\0'; } return; } /* Escaping can double every byte and adds a NUL terminator. Derive the * input limit from the caller's destination rather than a fixed staging * buffer, so full RESULTS_BUFFER values survive in a 2N+1 destination. */ max_input = ((size_t) max_size - 1) / 2; input_len = strlen(input); if (input_len > max_input) { input_len = max_input; } mysql_real_escape_string(mysql, output, input, (unsigned long) input_len); } void db_free_result(MYSQL_RES *result) { mysql_free_result(result); } int db_column_exists(MYSQL *mysql, int type, const char *table, const char *column) { char query_frag[BUFSIZE]; MYSQL_RES *result; int exists; /* save a fragment just in case */ memset(query_frag, 0, BUFSIZE); snprintf(query_frag, BUFSIZE, "SHOW COLUMNS FROM `%s` LIKE '%s'", table, column); /* show the sql query */ SPINE_LOG_DEVDBG(("DEVDBG: db_column_exists('%s','%s'): %s", table, column, query_frag)); result = db_query(mysql, type, query_frag); if (mysql_num_rows(result)) { exists = TRUE; } else { exists = FALSE; } db_free_result(result); return exists; }