/*****************************************************************************\
 *  mysql_common.c - common functions for the mysql storage plugin.
 *****************************************************************************
 *  Copyright (C) 2004-2007 The Regents of the University of California.
 *  Produced at Lawrence Livermore National Laboratory (cf, DISCLAIMER).
 *  Written by Danny Auble <da@llnl.gov>
 *
 *  This file is part of Slurm, a resource management program.
 *  For details, see <https://slurm.schedmd.com/>.
 *  Please also read the included file: DISCLAIMER.
 *
 *  Slurm is free software; you can redistribute it and/or modify it under
 *  the terms of the GNU General Public License as published by the Free
 *  Software Foundation; either version 2 of the License, or (at your option)
 *  any later version.
 *
 *  In addition, as a special exception, the copyright holders give permission
 *  to link the code of portions of this program with the OpenSSL library under
 *  certain conditions as described in each individual source file, and
 *  distribute linked combinations including the two. You must obey the GNU
 *  General Public License in all respects for all of the code used other than
 *  OpenSSL. If you modify file(s) with this exception, you may extend this
 *  exception to your version of the file(s), but you are not obligated to do
 *  so. If you do not wish to do so, delete this exception statement from your
 *  version.  If you delete this exception statement from all source files in
 *  the program, then also delete it here.
 *
 *  Slurm 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 General Public License for more
 *  details.
 *
 *  You should have received a copy of the GNU General Public License along
 *  with Slurm; if not, write to the Free Software Foundation, Inc.,
 *  51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA.
 *
 *  This file is patterned after jobcomp_linux.c, written by Morris Jette and
 *  Copyright (C) 2002 The Regents of the University of California.
\*****************************************************************************/

#include "config.h"

#include "mysql_common.h"
#include "src/common/log.h"
#include "src/common/xstring.h"
#include "src/common/xmalloc.h"
#include "src/common/timers.h"
#include "src/common/slurm_protocol_api.h"
#include "src/common/read_config.h"
#include "src/slurmdbd/read_config.h"

#define MAX_DEADLOCK_ATTEMPTS 10

static char *table_defs_table = "table_defs_table";

typedef struct {
	char *name;
	char *columns;
	bool non_unique;
} db_key_t;

static void _destroy_db_key(void *arg)
{
	db_key_t *db_key = (db_key_t *)arg;

	if (db_key) {
		xfree(db_key->name);
		xfree(db_key->columns);
		xfree(db_key);
	}
}

static int _find_db_key(void *x, void *key)
{
	db_key_t * db_key = (db_key_t *) x;

	if (xstrcmp(db_key->name, (char *) key))
		return 0;
	return 1;
}

/* NOTE: Ensure that mysql_conn->lock is set on function entry */
static int _clear_results(MYSQL *db_conn)
{
	MYSQL_RES *result = NULL;
	int rc = 0;

	do {
		/* did current statement return data? */
		if ((result = mysql_store_result(db_conn)))
			mysql_free_result(result);

		/* more results? -1 = no, >0 = error, 0 = yes (keep looping) */
		if ((rc = mysql_next_result(db_conn)) > 0)
			error("Could not execute statement %d %s",
			      mysql_errno(db_conn),
			      mysql_error(db_conn));
	} while (rc == 0);

	if (rc > 0) {
		errno = rc;
		return SLURM_ERROR;
	}
	return SLURM_SUCCESS;
}

/* NOTE: Ensure that mysql_conn->lock is set on function entry */
static MYSQL_RES *_get_first_result(MYSQL *db_conn)
{
	MYSQL_RES *result = NULL;
	int rc = 0;
	do {
		/* did current statement return data? */
		if ((result = mysql_store_result(db_conn)))
			return result;

		/* more results? -1 = no, >0 = error, 0 = yes (keep looping) */
		if ((rc = mysql_next_result(db_conn)) > 0)
			debug3("error: Could not execute statement %d", rc);

	} while (rc == 0);

	return NULL;
}

/* NOTE: Ensure that mysql_conn->lock is set on function entry */
static MYSQL_RES *_get_last_result(MYSQL *db_conn)
{
	MYSQL_RES *result = NULL;
	MYSQL_RES *last_result = NULL;
	int rc = 0;
	do {
		/* did current statement return data? */
		if ((result = mysql_store_result(db_conn))) {
			if (last_result)
				mysql_free_result(last_result);
			last_result = result;
		}
		/* more results? -1 = no, >0 = error, 0 = yes (keep looping) */
		if ((rc = mysql_next_result(db_conn)) > 0)
			debug3("error: Could not execute statement %d", rc);
	} while (rc == 0);

	return last_result;
}

/* NOTE: Ensure that mysql_conn->lock is set on function entry */
static int _mysql_query_internal(MYSQL *db_conn, char *query)
{
	int rc = SLURM_SUCCESS;
	int deadlock_attempt = 0;

try_again:
	if (!db_conn)
		fatal("You haven't inited this storage yet.");

	/* clear out the old results so we don't get a 2014 error */
	_clear_results(db_conn);
	if (mysql_query(db_conn, query)) {
		const char *err_str = mysql_error(db_conn);
		errno = mysql_errno(db_conn);
		if (errno == ER_NO_SUCH_TABLE) {
			debug4("This could happen often and is expected.\n"
			       "mysql_query failed: %d %s\n%s",
			       errno, err_str, query);
			errno = 0;
			goto end_it;
		}
		if (errno == ER_LOCK_DEADLOCK) {
			/*
			 * Mysql detected a deadlock and we should retry
			 * a few times since this is mainly a race condition
			 */
			deadlock_attempt++;

			if (deadlock_attempt < MAX_DEADLOCK_ATTEMPTS) {
				error("%s: deadlock detected attempt %u/%u: %d %s",
				      __func__, deadlock_attempt,
				      MAX_DEADLOCK_ATTEMPTS, errno, err_str);
				goto try_again;
			} else {
				fatal("%s: unable to resolve deadlock with attempts %u/%u: %d %s\nPlease call 'show engine innodb status;' in MySQL/MariaDB and open a bug report with SchedMD.",
				      __func__, deadlock_attempt,
				      MAX_DEADLOCK_ATTEMPTS, errno, err_str);
			}
		} else if (errno == ER_LOCK_WAIT_TIMEOUT) {
			/* FIXME: If we get ER_LOCK_WAIT_TIMEOUT here we need
			 * to restart the connections, but it appears restarting
			 * the calling program is the only way to handle this.
			 * If anyone in the future figures out a way to handle
			 * this, super.  Until then we will need to restart the
			 * calling program if you ever get this error.
			 */
			fatal("mysql gave ER_LOCK_WAIT_TIMEOUT as an error. "
			      "The only way to fix this is restart the "
			      "calling program");
		} else if (errno == ER_HOST_IS_BLOCKED) {
			fatal("MySQL gave ER_HOST_IS_BLOCKED as an error. "
			      "You will need to call 'mysqladmin flush-hosts' "
			      "to regain connectivity.");
		}
		error("mysql_query failed: %d %s\n%s", errno, err_str, query);
		rc = SLURM_ERROR;
	}
end_it:
	/*
	 * Starting in MariaDB 10.2 many of the api commands started
	 * setting errno erroneously.
	 */
	if (!rc)
		errno = 0;

	return rc;
}

/*
 * Determine if a database server upgrade has taken place and if so, check to
 * see if the candidate table alteration query should be used to alter the table
 * to its expected settings. Returns true if so.
 *
 * Background:
 *
 * From the MariaDB docs:
 *  Before MariaDB 10.2.1, BLOB and TEXT columns could not be assigned a DEFAULT
 *  value. This restriction was lifted in MariaDB 10.2.1.
 *
 * If a site begins using MariaDB >= 10.2.1 and is either using an existing
 * Slurm database from an earlier version or has restored one from a dump from
 * an earlier version or from any version of MySQL, some text/blob default
 * values will need to be altered to avoid failures from subsequent queries from
 * slurmdbd that set affected fields to DEFAULT (see bug#13606).
 *
 * Note that only one column from one table ('preempt' from qos_table) is
 * checked to determine if an upgrade has taken place with the assumption that
 * if its default value is not correct then the same is true for similar
 * text/blob columns from other tables and they will also need to be altered.
 *
 * The qos_table has been chosen for this check because it is the last table
 * with condition to be created. If that condition changes this should be
 * re-evaluated.
 */
static bool _alter_table_after_upgrade(mysql_conn_t *mysql_conn,
				       char *table_alter_query)
{
	static bool have_value = false, upgraded = false;
	MYSQL_RES *result = NULL;
	MYSQL_ROW row;

	/* check to see if upgrade has happened */
	if (!have_value) {
		const char *tmp_char;
		char *query;
		/*
		 * confirm MariaDB is being used to avoid any ambiguity with
		 * MySQL versions
		 */
		tmp_char = mysql_get_server_info(mysql_conn->db_conn);
		if (xstrcasestr(tmp_char, "mariadb") &&
		    (mysql_get_server_version(mysql_conn->db_conn) >= 100201)) {
			query = "show columns from `qos_table` like 'preempt'";
			result = mysql_db_query_ret(mysql_conn, query, 0);
			if (result) {
				/*
				 * row[4] holds the column's default value and
				 * if it's not empty ('') then it will need to
				 * be altered and an upgrade is assumed
				 */
				if ((row = mysql_fetch_row(result)) &&
				    !xstrcasecmp(row[1], "text") &&
				    xstrcmp(row[4], "''"))
					upgraded = true;
				mysql_free_result(result);
			}
		}
		have_value = true;
	}

	/*
	 * If upgrade detected and the table alter query string contains an
	 * empty string default then the query should be executed. The latter
	 * check avoids unnecessary table alterations.
	 */
	if (upgraded && xstrcasestr(table_alter_query, "default ''"))
		return true;

	return false;
}

/* NOTE: Ensure that mysql_conn->lock is NOT set on function entry */
static int _mysql_make_table_current(mysql_conn_t *mysql_conn, char *table_name,
				     storage_field_t *fields, char *ending)
{
	char *query = NULL;
	char *correct_query = NULL;
	MYSQL_RES *result = NULL;
	MYSQL_ROW row;
	int i = 0;
	list_t *columns = NULL;
	list_itr_t *itr = NULL;
	char *col = NULL;
	int adding = 0;
	int run_update = 0;
	char *primary_key = NULL;
	char *unique_index = NULL;
	int old_primary = 0;
	char *temp = NULL, *temp2 = NULL;
	list_t *keys_list = NULL;
	db_key_t *db_key = NULL;

	DEF_TIMERS;

	/* figure out the unique keys in the table */
	query = xstrdup_printf("show index from %s where non_unique=0",
			       table_name);
	if (!(result = mysql_db_query_ret(mysql_conn, query, 0))) {
		xfree(query);
		return SLURM_ERROR;
	}
	xfree(query);

	keys_list = list_create(_destroy_db_key);
	while ((row = mysql_fetch_row(result))) {
		/*
		 * row[2] = key name
		 * row[4] = column name
		 */
		if (!xstrcasecmp(row[2], "PRIMARY")) {
			old_primary = 1;
			continue;
		}

		db_key = list_find_first(keys_list, _find_db_key, row[2]);

		if (db_key) {
			xstrfmtcat(db_key->columns, ", %s", row[4]);
		} else {
			db_key = xmalloc(sizeof(db_key_t));
			db_key->name = xstrdup(row[2]);
			db_key->columns = xstrdup(row[4]);
			db_key->non_unique = false;
			list_append(keys_list, db_key);
		}
	}
	mysql_free_result(result);

	/* figure out the non-unique keys in the table */
	query = xstrdup_printf("show index from %s where non_unique=1",
			       table_name);
	if (!(result = mysql_db_query_ret(mysql_conn, query, 0))) {
		xfree(query);
		return SLURM_ERROR;
	}
	xfree(query);

	while ((row = mysql_fetch_row(result))) {
		db_key = list_find_first(keys_list, _find_db_key, row[2]);

		if (db_key) {
			xstrfmtcat(db_key->columns, ", %s", row[4]);
		} else {
			db_key = xmalloc(sizeof(db_key_t));
			db_key->name = xstrdup(row[2]); // name
			db_key->columns = xstrdup(row[4]); // column name
			db_key->non_unique = true;
			list_append(keys_list, db_key); // don't use list_push
		}
	}
	mysql_free_result(result);

	/* figure out the existing columns in the table */
	query = xstrdup_printf("show columns from %s", table_name);
	if (!(result = mysql_db_query_ret(mysql_conn, query, 0))) {
		xfree(query);
		FREE_NULL_LIST(keys_list);
		return SLURM_ERROR;
	}
	xfree(query);
	columns = list_create(xfree_ptr);
	while ((row = mysql_fetch_row(result))) {
		col = xstrdup(row[0]); //Field
		list_append(columns, col);
	}
	mysql_free_result(result);


	itr = list_iterator_create(columns);
	/* In MySQL 5.7.4 we lost the ability to run 'alter ignore'.  This was
	 * needed when converting old tables to new schemas.  If people convert
	 * in the future from an older version of Slurm that needed the ignore
	 * to work they will have to downgrade mysql to <= 5.7.3 to make things
	 * work correctly or manually edit the database to get things to work.
	 */
	/*
	 * `query` is compared against the current table_defs_table.definition
	 * and run if they are different. `correct_query` is inserted into the
	 * table, so it must be what future `query` schemas will be.
	 * In other words, `query` transitions the table to the new schema,
	 * `correct_query` represents the new schema
	 */
	query = xstrdup_printf("alter table %s", table_name);
	correct_query = xstrdup(query);
	START_TIMER;
	while (fields[i].name) {
		int found = 0;

		list_iterator_reset(itr);
		while ((col = list_next(itr))) {
			if (!xstrcmp(col, fields[i].name)) {
				xstrfmtcat(query, " modify `%s` %s,",
					   fields[i].name,
					   fields[i].options);
				xstrfmtcat(correct_query, " modify `%s` %s,",
					   fields[i].name,
					   fields[i].options);
				list_delete_item(itr);
				found = 1;
				break;
			}
		}
		if (!found) {
			if (i) {
				info("adding column %s after %s in table %s",
				     fields[i].name,
				     fields[i-1].name,
				     table_name);
				xstrfmtcat(query, " add `%s` %s after %s,",
					   fields[i].name,
					   fields[i].options,
					   fields[i-1].name);
				xstrfmtcat(correct_query, " modify `%s` %s,",
					   fields[i].name,
					   fields[i].options);
			} else {
				info("adding column %s at the beginning "
				     "of table %s",
				     fields[i].name,
				     table_name);
				xstrfmtcat(query, " add `%s` %s first,",
					   fields[i].name,
					   fields[i].options);
				xstrfmtcat(correct_query, " modify `%s` %s,",
					   fields[i].name,
					   fields[i].options);
			}
			adding = 1;
		}

		i++;
	}

	list_iterator_reset(itr);
	while ((col = list_next(itr))) {
		adding = 1;
		info("dropping column %s from table %s", col, table_name);
		xstrfmtcat(query, " drop %s,", col);
	}

	list_iterator_destroy(itr);
	FREE_NULL_LIST(columns);

	if ((temp = strstr(ending, "primary key ("))) {
		int open = 0, close =0;
		int end = 0;
		while (temp[end++]) {
			if (temp[end] == '(')
				open++;
			else if (temp[end] == ')')
				close++;
			else
				continue;
			if (open == close)
				break;
		}
		if (temp[end]) {
			end++;
			primary_key = xstrndup(temp, end);
			if (old_primary)
				xstrcat(query, " drop primary key,");
			xstrcat(correct_query, " drop primary key,");
			xstrfmtcat(query, " add %s,",  primary_key);
			xstrfmtcat(correct_query, " add %s,",  primary_key);

			xfree(primary_key);
		}
	}

	if ((temp = strstr(ending, "unique index"))) {
		int open = 0, close = 0;
		/* sizeof includes NULL, and end should start 1 back */
		int end = sizeof("unique index") - 2;
		char *udex_name = NULL, *name_marker = NULL;
		while (temp[end++]) {
			/*
			 * Extracts the index name, which is given explicitly
			 * or is the name of the first field included in the
			 * index.
			 * "unique index indexname (field1, field2)"
			 * "unique index (indexname, field2)"
			 * indexname is started by the first non '(' or ' '
			 *     after "unique index"
			 * indexname is terminated by '(' ')' ' ' or ','
			 */
			if (name_marker) {
				if (!udex_name && (temp[end] == '(' ||
						   temp[end] == ')' ||
						   temp[end] == ' ' ||
						   temp[end] == ','))
					udex_name = xstrndup(name_marker,
						temp + end - name_marker);
			} else if (temp[end] != '(' && temp[end] != ' ') {
				name_marker = temp + end;
			}

			/* find the end of the parenthetical expression */
			if (temp[end] == '(')
				open++;
			else if (temp[end] == ')')
				close++;
			else
				continue;
			if (open == close)
				break;
		}
		if (temp[end]) {
			end++;
			unique_index = xstrndup(temp, end);

			db_key = list_remove_first(keys_list, _find_db_key,
						   udex_name);
			if (db_key) {
				xstrfmtcat(query,
					   " drop index %s,", db_key->name);
				_destroy_db_key(db_key);
			} else {
				info("adding %s to table %s",
				     unique_index, table_name);
			}
			xstrfmtcat(correct_query, " drop index %s,", udex_name);
			xstrfmtcat(query, " add %s,", unique_index);
			xstrfmtcat(correct_query, " add %s,", unique_index);
			xfree(unique_index);
		}
		xfree(udex_name);
	}

	temp2 = ending;
	while ((temp = strstr(temp2, ", key "))) {
		int open = 0, close = 0, name_end = 0;
		int end = 5;
		char *new_key_name = NULL, *new_key = NULL;
		while (temp[end++]) {
			if (!name_end && (temp[end] == ' ')) {
				name_end = end;
				continue;
			} else if (temp[end] == '(') {
				open++;
				if (!name_end)
					name_end = end;
			} else if (temp[end] == ')')
				close++;
			else
				continue;
			if (open == close)
				break;
		}
		if (temp[end]) {
			end++;
			new_key_name = xstrndup(temp+6, name_end-6);
			new_key = xstrndup(temp+2, end-2); // skip ', '

			db_key = list_remove_first(keys_list, _find_db_key,
						   new_key_name);
			if (db_key) {
				xstrfmtcat(query,
					   " drop key %s,", db_key->name);
				_destroy_db_key(db_key);
			} else
				info("adding %s to table %s",
				     new_key, table_name);
			xstrfmtcat(correct_query,
				   " drop key %s,", new_key_name);

			xstrfmtcat(query, " add %s,",  new_key);
			xstrfmtcat(correct_query, " add %s,",  new_key);

			xfree(new_key);
			xfree(new_key_name);
		}
		temp2 = temp + end;
	}

	/* flush extra (old) keys */
	itr = list_iterator_create(keys_list);
	while ((db_key = list_next(itr))) {
		if (!db_key->non_unique) {
			info("dropping unique index %s from table %s",
			     db_key->name, table_name);
			xstrfmtcat(query, " drop index %s,", db_key->name);
		} else {
			info("dropping key %s from table %s",
			     db_key->name, table_name);
			xstrfmtcat(query, " drop key %s,", db_key->name);
		}
	}
	list_iterator_destroy(itr);

	FREE_NULL_LIST(keys_list);

	query[strlen(query)-1] = ';';
	correct_query[strlen(correct_query)-1] = ';';
	//info("%d query\n%s", __LINE__, query);

	/* see if table needs to be altered after db server upgrade */
	if (!adding && _alter_table_after_upgrade(mysql_conn, query)) {
		run_update = 3;
	} else if (!adding && !run_update) {
		/* see if we have already done this definition */
		char *quoted = slurm_add_slash_to_quotes(query);
		char *query2 = xstrdup_printf("select table_name from "
					      "%s where definition='%s'",
					      table_defs_table, quoted);
		MYSQL_RES *result = NULL;
		MYSQL_ROW row;

		xfree(quoted);
		run_update = 1;
		if ((result = mysql_db_query_ret(mysql_conn, query2, 0))) {
			if ((row = mysql_fetch_row(result)))
				run_update = 0;
			mysql_free_result(result);
		}
		xfree(query2);
		if (run_update) {
			run_update = 2;
			query2 = xstrdup_printf("select table_name from "
						"%s where table_name='%s'",
						table_defs_table, table_name);
			if ((result = mysql_db_query_ret(
				     mysql_conn, query2, 0))) {
				if ((row = mysql_fetch_row(result)))
					run_update = 1;
				mysql_free_result(result);
			}
			xfree(query2);
		}
	}

	/* if something has changed run the alter line */
	if (run_update || adding) {
		time_t now = time(NULL);
		char *query2 = NULL;
		char *quoted = NULL;

		if (run_update == 2)
			debug4("Table %s doesn't exist, adding", table_name);
		else if (run_update == 3)
			info("MariaDB >= 10.2.1 in use with a table from an earlier version or from MySQL. Updating table %s...",
			      table_name);
		else
			debug("Table %s has changed.  Updating...", table_name);

		debug2("query\n%s", query);
		if (mysql_db_query(mysql_conn, query)) {
			xfree(query);
			return SLURM_ERROR;
		}
		quoted = slurm_add_slash_to_quotes(correct_query);
		query2 = xstrdup_printf("insert into %s (creation_time, "
					"mod_time, table_name, definition) "
					"values (%ld, %ld, '%s', '%s') "
					"on duplicate key update "
					"definition='%s', mod_time=%ld;",
					table_defs_table, now, now,
					table_name, quoted,
					quoted, now);
		xfree(quoted);
		debug3("query\n%s", query2);
		if (mysql_db_query(mysql_conn, query2)) {
			xfree(query2);
			return SLURM_ERROR;
		}
		xfree(query2);
	}

	xfree(query);
	xfree(correct_query);
	query = xstrdup_printf("make table current %s", table_name);
	END_TIMER2(query);
	xfree(query);
	return SLURM_SUCCESS;
}

void _set_mysql_ssl_opts(MYSQL *db_conn, const char *options)
{
	char *tmp_opts, *token, *save_ptr = NULL;
	const char *key = NULL, *cert = NULL, *ca = NULL, *ca_path = NULL;
	const char *cipher = NULL;

	if (!options)
		return;

	tmp_opts = xstrdup(options);
	token = strtok_r(tmp_opts, ",", &save_ptr);
	while (token) {
		char *opt_str, *val_str = NULL;

		opt_str = strtok_r(token, "=", &val_str);

		if (!opt_str || !val_str) {
			error("Invalid storage option/val");
			goto next;
		} else if (!xstrcasecmp(opt_str, "SSL_CERT"))
			cert = val_str;
		else if (!xstrcasecmp(opt_str, "SSL_CA"))
			ca = val_str;
		else if (!xstrcasecmp(opt_str, "SSL_CAPATH"))
			ca_path = val_str;
		else if (!xstrcasecmp(opt_str, "SSL_KEY"))
			key = val_str;
		else if (!xstrcasecmp(opt_str, "SSL_CIPHER"))
			cipher = val_str;
		else {
			error("Invalid storage option '%s'", opt_str);
			goto next;
		}
next:
		token = strtok_r(NULL, ",", &save_ptr);
	}

	mysql_ssl_set(db_conn, key, cert, ca, ca_path, cipher);

	xfree(tmp_opts);
}

/* NOTE: Ensure that mysql_conn->lock is set on function entry */
static int _create_db(char *db_name, mysql_db_info_t *db_info)
{
	MYSQL *mysql_db = NULL;
	int rc = SLURM_ERROR;

	MYSQL *db_ptr = NULL;
	char *db_host = NULL;

	while (rc == SLURM_ERROR) {
		rc = SLURM_SUCCESS;
		if (!(mysql_db = mysql_init(mysql_db)))
			fatal("mysql_init failed: %s", mysql_error(mysql_db));

		_set_mysql_ssl_opts(mysql_db, db_info->params);

		db_host = db_info->host;
		db_ptr = mysql_real_connect(mysql_db,
					    db_host, db_info->user,
					    db_info->pass, NULL,
					    db_info->port, NULL, 0);

		if (!db_ptr && db_info->backup) {
			info("Connection failed to host = %s "
			     "user = %s port = %u",
			     db_host, db_info->user,
			     db_info->port);
			db_host = db_info->backup;
			db_ptr = mysql_real_connect(mysql_db, db_host,
						    db_info->user,
						    db_info->pass, NULL,
						    db_info->port, NULL, 0);
		}

		if (db_ptr) {
			char *create_line = NULL;
			xstrfmtcat(create_line, "create database %s", db_name);
			if (mysql_query(mysql_db, create_line)) {
				fatal("mysql_query failed: %d %s\n%s",
				      mysql_errno(mysql_db),
				      mysql_error(mysql_db), create_line);
			}
			xfree(create_line);
			if (mysql_thread_safe())
				mysql_thread_end();
			mysql_close(mysql_db);
		} else {
			info("Connection failed to host = %s "
			     "user = %s port = %u",
			     db_host, db_info->user,
			     db_info->port);
			error("mysql_real_connect failed: %d %s",
			      mysql_errno(mysql_db),
			      mysql_error(mysql_db));
			rc = SLURM_ERROR;
		}
		if (rc == SLURM_ERROR)
			sleep(3);
	}
	return rc;
}

extern mysql_conn_t *create_mysql_conn(int conn_num, bool rollback,
				       char *cluster_name)
{
	mysql_conn_t *mysql_conn = xmalloc(sizeof(mysql_conn_t));

	if (rollback)
		mysql_conn->flags |= DB_CONN_FLAG_ROLLBACK;
	mysql_conn->conn = conn_num;
	mysql_conn->cluster_name = xstrdup(cluster_name);
	mysql_conn->wsrep_trx_fragment_size_orig = NO_VAL64;
	slurm_mutex_init(&mysql_conn->lock);
	mysql_conn->update_list = list_create(slurmdb_destroy_update_object);

	return mysql_conn;
}

extern int destroy_mysql_conn(mysql_conn_t *mysql_conn)
{
	if (mysql_conn) {
		mysql_db_close_db_connection(mysql_conn);
		xfree(mysql_conn->pre_commit_query);
		xfree(mysql_conn->cluster_name);
		slurm_mutex_destroy(&mysql_conn->lock);
		FREE_NULL_LIST(mysql_conn->update_list);
		xfree(mysql_conn->wsrep_trx_fragment_unit_orig);
		xfree(mysql_conn);
	}

	return SLURM_SUCCESS;
}

extern mysql_db_info_t *create_mysql_db_info(slurm_mysql_plugin_type_t type)
{
	mysql_db_info_t *db_info = xmalloc(sizeof(mysql_db_info_t));

	switch (type) {
	case SLURM_MYSQL_PLUGIN_AS:
		db_info->port = slurm_conf.accounting_storage_port;
		db_info->host = xstrdup(slurm_conf.accounting_storage_host);
		db_info->backup =
			xstrdup(slurm_conf.accounting_storage_backup_host);
		db_info->user = xstrdup(slurmdbd_conf->storage_user);
		db_info->pass = xstrdup(slurm_conf.accounting_storage_pass);
		db_info->params = xstrdup(slurm_conf.accounting_storage_params);
		break;
	case SLURM_MYSQL_PLUGIN_JC:
		if (!slurm_conf.job_comp_port)
			slurm_conf.job_comp_port = DEFAULT_MYSQL_PORT;
		db_info->port = slurm_conf.job_comp_port;
		db_info->host = xstrdup(slurm_conf.job_comp_host);
		db_info->user = xstrdup(slurm_conf.job_comp_user);
		db_info->pass = xstrdup(slurm_conf.job_comp_pass);
		db_info->params = xstrdup(slurm_conf.accounting_storage_params);
		break;
	default:
		xfree(db_info);
		fatal("Unknown mysql_db_info %d", type);
	}
	return db_info;
}

extern int destroy_mysql_db_info(mysql_db_info_t *db_info)
{
	if (db_info) {
		xfree(db_info->backup);
		xfree(db_info->host);
		xfree(db_info->user);
		xfree(db_info->pass);
		xfree(db_info);
	}
	return SLURM_SUCCESS;
}

extern int mysql_db_get_db_connection(mysql_conn_t *mysql_conn, char *db_name,
				      mysql_db_info_t *db_info)
{
	int rc = SLURM_SUCCESS;
	bool storage_init = false;
	char *db_host = db_info->host;
	unsigned int my_timeout = 30;

	xassert(mysql_conn);

	slurm_mutex_lock(&mysql_conn->lock);

	if (!(mysql_conn->db_conn = mysql_init(mysql_conn->db_conn))) {
		slurm_mutex_unlock(&mysql_conn->lock);
		fatal("mysql_init failed: %s",
		      mysql_error(mysql_conn->db_conn));
	}

	/*
	 * If this ever changes you will need to alter
	 * src/common/slurmdbd_defs.c function _send_init_msg to
	 * handle a different timeout when polling for the
	 * response.
	 */
	mysql_options(mysql_conn->db_conn, MYSQL_OPT_CONNECT_TIMEOUT,
		      (char *)&my_timeout);

	_set_mysql_ssl_opts(mysql_conn->db_conn, db_info->params);

	while (!storage_init) {
		debug2("Attempting to connect to %s:%d", db_host,
		       db_info->port);
		if (!mysql_real_connect(mysql_conn->db_conn, db_host,
					db_info->user, db_info->pass,
					db_name, db_info->port, NULL,
					CLIENT_MULTI_STATEMENTS)) {
			const char *err_str = NULL;
			int err = mysql_errno(mysql_conn->db_conn);

			if (err == ER_BAD_DB_ERROR) {
				debug("Database %s not created.  Creating",
				      db_name);
				rc = _create_db(db_name, db_info);

				/*
				 * When using ca, cert and key the next
				 * connect will fail. Setting the options again
				 * fixes it.
				 */
				_set_mysql_ssl_opts(mysql_conn->db_conn,
						    db_info->params);
				continue;
			}

			err_str = mysql_error(mysql_conn->db_conn);

			if ((db_host == db_info->host) && db_info->backup) {
				debug2("mysql_real_connect failed: %d %s",
				       err, err_str);
				db_host = db_info->backup;
				continue;
			}

			error("mysql_real_connect failed: %d %s",
			      err, err_str);
			rc = ESLURM_DB_CONNECTION;
			mysql_close(mysql_conn->db_conn);
			mysql_conn->db_conn = NULL;
			break;
		}

		storage_init = true;
		if (mysql_conn->flags & DB_CONN_FLAG_ROLLBACK)
			mysql_autocommit(mysql_conn->db_conn, 0);
		rc = _mysql_query_internal(mysql_conn->db_conn,
					   "SET session sql_mode='ANSI_QUOTES,"
					   "NO_ENGINE_SUBSTITUTION';");
	}
	slurm_mutex_unlock(&mysql_conn->lock);
	errno = rc;
	return rc;
}

extern int mysql_db_close_db_connection(mysql_conn_t *mysql_conn)
{
	slurm_mutex_lock(&mysql_conn->lock);
	if (mysql_conn && mysql_conn->db_conn) {
		if (mysql_thread_safe())
			mysql_thread_end();
		mysql_close(mysql_conn->db_conn);
		mysql_conn->db_conn = NULL;
	}
	slurm_mutex_unlock(&mysql_conn->lock);
	return SLURM_SUCCESS;
}

extern int mysql_db_cleanup(void)
{
	debug3("starting mysql cleaning up");

#ifdef mysql_library_end
	mysql_library_end();
#else
	mysql_server_end();
#endif
	debug3("finished mysql cleaning up");
	return SLURM_SUCCESS;
}

extern int mysql_db_query(mysql_conn_t *mysql_conn, char *query)
{
	int rc = SLURM_SUCCESS;

	if (!mysql_conn || !mysql_conn->db_conn) {
		fatal("You haven't inited this storage yet.");
		return 0;	/* For CLANG false positive */
	}
	slurm_mutex_lock(&mysql_conn->lock);
	rc = _mysql_query_internal(mysql_conn->db_conn, query);
	slurm_mutex_unlock(&mysql_conn->lock);
	return rc;
}

/*
 * Executes a single delete sql query.
 * Returns the number of deleted rows, <0 for failure.
 */
extern int mysql_db_delete_affected_rows(mysql_conn_t *mysql_conn, char *query)
{
	int rc = SLURM_SUCCESS;

	if (!mysql_conn || !mysql_conn->db_conn) {
		fatal("You haven't inited this storage yet.");
		return 0;	/* For CLANG false positive */
	}
	slurm_mutex_lock(&mysql_conn->lock);
	if (!(rc = _mysql_query_internal(mysql_conn->db_conn, query)))
		rc = mysql_affected_rows(mysql_conn->db_conn);
	slurm_mutex_unlock(&mysql_conn->lock);
	return rc;
}

extern int mysql_db_ping(mysql_conn_t *mysql_conn)
{
	int rc;

	if (!mysql_conn->db_conn)
		return -1;

	/* clear out the old results so we don't get a 2014 error */
	slurm_mutex_lock(&mysql_conn->lock);
	_clear_results(mysql_conn->db_conn);
	rc = mysql_ping(mysql_conn->db_conn);
	/*
	 * Starting in MariaDB 10.2 many of the api commands started
	 * setting errno erroneously.
	 */
	if (!rc)
		errno = 0;
	slurm_mutex_unlock(&mysql_conn->lock);
	return rc;
}

extern int mysql_db_commit(mysql_conn_t *mysql_conn)
{
	int rc = SLURM_SUCCESS;

	if (!mysql_conn->db_conn)
		return SLURM_ERROR;

	slurm_mutex_lock(&mysql_conn->lock);
	/* clear out the old results so we don't get a 2014 error */
	_clear_results(mysql_conn->db_conn);
	if (mysql_commit(mysql_conn->db_conn)) {
		error("mysql_commit failed: %d %s",
		      mysql_errno(mysql_conn->db_conn),
		      mysql_error(mysql_conn->db_conn));
		errno = mysql_errno(mysql_conn->db_conn);
		rc = SLURM_ERROR;
	}
	slurm_mutex_unlock(&mysql_conn->lock);
	return rc;
}

extern int mysql_db_rollback(mysql_conn_t *mysql_conn)
{
	int rc = SLURM_SUCCESS;

	if (!mysql_conn->db_conn)
		return SLURM_ERROR;

	slurm_mutex_lock(&mysql_conn->lock);
	/* clear out the old results so we don't get a 2014 error */
	_clear_results(mysql_conn->db_conn);
	if (mysql_rollback(mysql_conn->db_conn)) {
		error("mysql_commit failed: %d %s",
		      mysql_errno(mysql_conn->db_conn),
		      mysql_error(mysql_conn->db_conn));
		errno = mysql_errno(mysql_conn->db_conn);
		rc = SLURM_ERROR;
	} else {
		/*
		 * Starting in MariaDB 10.2 many of the api commands started
		 * setting errno erroneously.
		 */
		errno = 0;
	}
	slurm_mutex_unlock(&mysql_conn->lock);
	return rc;

}

extern MYSQL_RES *mysql_db_query_ret(mysql_conn_t *mysql_conn,
				     char *query, bool last)
{
	MYSQL_RES *result = NULL;

	slurm_mutex_lock(&mysql_conn->lock);
	if (_mysql_query_internal(mysql_conn->db_conn, query) != SLURM_ERROR)  {
		if (mysql_errno(mysql_conn->db_conn) == ER_NO_SUCH_TABLE)
			goto fini;
		else if (last)
			result = _get_last_result(mysql_conn->db_conn);
		else
			result = _get_first_result(mysql_conn->db_conn);
		/*
		 * Starting in MariaDB 10.2 many of the api commands started
		 * setting errno erroneously.
		 */
		errno = 0;
		if (!result && mysql_field_count(mysql_conn->db_conn)) {
			/* should have returned data */
			error("We should have gotten a result: '%m' '%s'",
			      mysql_error(mysql_conn->db_conn));
		}
	}

fini:
	slurm_mutex_unlock(&mysql_conn->lock);
	return result;
}

extern int mysql_db_query_check_after(mysql_conn_t *mysql_conn, char *query)
{
	int rc = SLURM_SUCCESS;

	slurm_mutex_lock(&mysql_conn->lock);
	if ((rc = _mysql_query_internal(
		     mysql_conn->db_conn, query)) != SLURM_ERROR)
		rc = _clear_results(mysql_conn->db_conn);
	slurm_mutex_unlock(&mysql_conn->lock);
	return rc;
}

extern uint64_t mysql_db_insert_ret_id(mysql_conn_t *mysql_conn, char *query)
{
	uint64_t new_id = 0;

	slurm_mutex_lock(&mysql_conn->lock);
	if (_mysql_query_internal(mysql_conn->db_conn, query) != SLURM_ERROR)  {
		new_id = mysql_insert_id(mysql_conn->db_conn);
		if (!new_id) {
			/* should have new id */
			error("%s: We should have gotten a new id: %s",
			      __func__, mysql_error(mysql_conn->db_conn));
		}
	}
	slurm_mutex_unlock(&mysql_conn->lock);
	return new_id;

}

extern int mysql_db_create_table(mysql_conn_t *mysql_conn, char *table_name,
				 storage_field_t *fields, char *ending)
{
	char *query = NULL;
	int rc;
	storage_field_t *first_field = fields;

	if (!fields || !fields->name) {
		error("Not creating an empty table");
		return SLURM_ERROR;
	}

	/* We have an internal table called table_defs_table which
	 * contains the definition of each table in the database.  To
	 * speed things up we just check against that to see if
	 * anything has changed.
	 */
	query = xstrdup_printf("create table if not exists %s "
			       "(creation_time int unsigned not null, "
			       "mod_time int unsigned default 0 not null, "
			       "table_name text not null, "
			       "definition text not null, "
			       "primary key (table_name(50))) engine='innodb'",
			       table_defs_table);
	if (mysql_db_query(mysql_conn, query) == SLURM_ERROR) {
		xfree(query);
		return SLURM_ERROR;
	}
	xfree(query);

	query = xstrdup_printf("create table if not exists %s (`%s` %s",
			       table_name, fields->name, fields->options);
	fields++;

	while (fields && fields->name) {
		xstrfmtcat(query, ", `%s` %s", fields->name, fields->options);
		fields++;
	}
	xstrcat(query, ending);

	/* make sure we can do a rollback */
	xstrcat(query, " engine='innodb'");

	if (mysql_db_query(mysql_conn, query) == SLURM_ERROR) {
		xfree(query);
		return SLURM_ERROR;
	}
	xfree(query);

	rc = _mysql_make_table_current(
		mysql_conn, table_name, first_field, ending);
	return rc;
}

extern int mysql_db_get_var_str(mysql_conn_t *mysql_conn,
				const char *variable_name,
				char **value)
{
	MYSQL_ROW row = NULL;
	MYSQL_RES *result = NULL;
	char *query;

	query = xstrdup_printf("select @@%s;", variable_name);
	result = mysql_db_query_ret(mysql_conn, query, 0);
	if (!result) {
		error("%s: null result from query `%s`", __func__, query);
		xfree(query);
		return SLURM_ERROR;
	}

	if (mysql_num_rows(result) != 1) {
		error("%s: invalid results from query `%s`", __func__, query);
		xfree(query);
		mysql_free_result(result);
		return SLURM_ERROR;
	}

	xfree(query);

	row = mysql_fetch_row(result);
	*value = xstrdup(row[0]);

	mysql_free_result(result);

	return SLURM_SUCCESS;
}

extern int mysql_db_get_var_u64(mysql_conn_t *mysql_conn,
				const char *variable_name,
				uint64_t *value)
{
	char *err_check = NULL, *var_str = NULL;

	if (mysql_db_get_var_str(mysql_conn, variable_name, &var_str)) {
		return SLURM_ERROR;
	}

	*value = strtoull(var_str, &err_check, 10);

	if (*err_check) {
		error("%s: error parsing string to int `%s`",
		      __func__, var_str);
		xfree(var_str);
		return SLURM_ERROR;
	}
	xfree(var_str);

	return SLURM_SUCCESS;
}

extern void mysql_db_enable_streaming_replication(mysql_conn_t *mysql_conn)
{
	int rc = SLURM_SUCCESS;
	char *query;
	uint64_t wsrep_on, wsrep_max_ws_size, fragment_size;

	/* if this errors, assume wsrep_on doesn't exist, so must be disabled */
	if (mysql_db_get_var_u64(mysql_conn, "wsrep_on", &wsrep_on)) {
		wsrep_on = 0;
		if (errno == ER_UNKNOWN_SYSTEM_VARIABLE)
			error("The prior error message regarding an undefined 'wsrep_on' variable is innocuous.  MySQL and MariaDB < 10.1 do not have this variable and Slurm will operate normally without it.");
	}

	debug2("wsrep_on=%"PRIu64, wsrep_on);

	if (!wsrep_on)
		return;

	/*
	 * wsrep_max_ws_size represents the maximum write set size in bytes.
	 * The fragment cannot exceed this value.
	 */
	rc = mysql_db_get_var_u64(mysql_conn, "wsrep_max_ws_size",
			          &wsrep_max_ws_size);
	if (rc) {
		error("Failed to get wsrep_max_ws_size");
		return;
	}

	/*
	 * Save the initial wsrep settings so they can be restored later.
	 * If these were set previously, don't set them again.
	 *
	 * If these variables don't exist, streaming replication isn't supported
	 * so don't turn it on.
	 */
	if (!mysql_conn->wsrep_trx_fragment_unit_orig) {
		rc = mysql_db_get_var_str(
			mysql_conn,
			"wsrep_trx_fragment_unit",
			&mysql_conn->wsrep_trx_fragment_unit_orig);
		if (rc) {
			if (errno == ER_UNKNOWN_SYSTEM_VARIABLE)
				error("This version of galera does not support streaming replication.");
			error("Unable to fetch wsrep_trx_fragment_unit.");
			return;
		}
	}
	if (mysql_conn->wsrep_trx_fragment_size_orig == NO_VAL64) {
		rc = mysql_db_get_var_u64(
			mysql_conn,
			"wsrep_trx_fragment_size",
			&mysql_conn->wsrep_trx_fragment_size_orig);
		if (rc) {
			if (errno == ER_UNKNOWN_SYSTEM_VARIABLE)
				error("This version of galera does not support streaming replication.");
			error("Unable to fetch wsrep_trx_fragment_size.");
			return;
		}
	}

	/*
	 * Force the wsrep_trx_fragment_unit to bytes. The default may change
	 * in the future, or may have been set by the site, so don't rely on it
	 * being a specific value.
	 */
	query = xstrdup("SET @@SESSION.wsrep_trx_fragment_unit=\'bytes\';");
	rc = _mysql_query_internal(mysql_conn->db_conn, query);
	xfree(query);
	if (rc) {
		error("Unable to set wsrep_trx_fragment_unit.");
		return;
	}

	/*
	 * Set the fragment size to 128MiB, or wsrep_max_ws_size if it has been
	 * set below that. Simply setting it to the max size does not strictly
	 * result in the best performance.
	 */
	fragment_size = MIN(wsrep_max_ws_size, 134217700);
	query = xstrdup_printf("SET @@SESSION.wsrep_trx_fragment_size=%"PRIu64";",
			       fragment_size);
	rc = _mysql_query_internal(mysql_conn->db_conn, query);
	xfree(query);
	if (rc)
		error("Failed to set wsrep_trx_fragment_size");
	else
		debug2("set wsrep_trx_fragment_size=%"PRIu64" bytes",
		       fragment_size);

	return;
}

extern void mysql_db_restore_streaming_replication(mysql_conn_t *mysql_conn)
{
	int rc;
	char *query;

	/*
	 * Check if the connection has saved streaming replication settings.  If
	 * not there is nothing to restore.
	 */
	if (!mysql_conn->wsrep_trx_fragment_unit_orig &&
	    (mysql_conn->wsrep_trx_fragment_size_orig == NO_VAL64)) {
		debug2("no streaming replication settings to restore");
		return;
	}

	if (mysql_conn->wsrep_trx_fragment_unit_orig) {
		query = xstrdup_printf(
				"SET @@SESSION.wsrep_trx_fragment_unit=\'%s\';",
				mysql_conn->wsrep_trx_fragment_unit_orig);
		rc = _mysql_query_internal(mysql_conn->db_conn, query);
		xfree(query);
		if (rc) {
			error("Unable to restore wsrep_trx_fragment_unit.");
		} else {
			debug2("Restored wsrep_trx_fragment_unit=%s",
			       mysql_conn->wsrep_trx_fragment_unit_orig);
			xfree(mysql_conn->wsrep_trx_fragment_unit_orig);
		}
	}
	if (mysql_conn->wsrep_trx_fragment_size_orig != NO_VAL64) {
		query = xstrdup_printf(
				"SET @@SESSION.wsrep_trx_fragment_size=%"PRIu64";",
				mysql_conn->wsrep_trx_fragment_size_orig);
		rc = _mysql_query_internal(mysql_conn->db_conn, query);
		xfree(query);
		if (rc) {
			error("Unable to restore wsrep_trx_fragment_size.");
		} else {
			debug2("Restored wsrep_trx_fragment_size=%"PRIu64,
			       mysql_conn->wsrep_trx_fragment_size_orig);
			mysql_conn->wsrep_trx_fragment_size_orig = NO_VAL64;
		}
	}

	return;
}