Initial database functions

Makefile.in

@@ -26,7 +26,7 @@ include Makefile.global
 
 $(info Building against PostgreSQL $(MAJORVERSION))
 
-REPMGR_CLIENT_OBJS = repmgr-client.o config.o log.o
+REPMGR_CLIENT_OBJS = repmgr-client.o config.o log.o strutil.o dbutils.o
 REPMGRD_OBJS = repmgrd.o
 
 repmgr4: $(REPMGR_CLIENT_OBJS)

config.c

@@ -182,8 +182,7 @@ parse_config(t_configuration_options *options)
 		log_warning(_("the following problems were found in the configuration file:"));
 		for (cell = config_warnings.head; cell; cell = cell->next)
 		{
-			fprintf(stderr, "  ");
+			fprintf(stderr, "  %s\n", cell->string);
-			log_warning("%s", cell->string);
 		}
 	}
 
@@ -327,6 +326,8 @@ _parse_config(t_configuration_options *options, ItemList *error_list, ItemList *
 			options->node_id = repmgr_atoi(value, name, error_list, 1);
 			node_id_found = true;
 		}
+		else if (strcmp(name, "node_name") == 0)
+			strncpy(options->node_name, value, MAXLEN);
 		else if (strcmp(name, "upstream_node_id") == 0)
 			options->upstream_node_id = repmgr_atoi(value, name, error_list, 1);
 		else if (strcmp(name, "conninfo") == 0)

dbutils.c

@@ -0,0 +1,187 @@
+/*
+ * dbutils.c - Database connection/management functions
+ *
+ * Copyright (c) 2ndQuadrant, 2010-2017
+ *
+ */
+
+#include <unistd.h>
+#include <time.h>
+#include <sys/time.h>
+
+#include "repmgr.h"
+
+
+#include "catalog/pg_control.h"
+
+static PGconn *_establish_db_connection(const char *conninfo,
+										const bool exit_on_error,
+										const bool log_notice,
+										const bool verbose_only);
+static bool _set_config(PGconn *conn, const char *config_param, const char *sqlquery);
+/* ==================== */
+/* Connection functions */
+/* ==================== */
+
+/*
+ * _establish_db_connection()
+ *
+ * Connect to a database using a conninfo string.
+ *
+ * NOTE: *do not* use this for replication connections; instead use:
+ *   establish_db_connection_by_params()
+ */
+
+static PGconn *
+_establish_db_connection(const char *conninfo, const bool exit_on_error, const bool log_notice, const bool verbose_only)
+{
+	PGconn	   *conn = NULL;
+	char		connection_string[MAXLEN];
+
+	strncpy(connection_string, conninfo, MAXLEN);
+
+	/* TODO: only set if not already present */
+	strcat(connection_string, " fallback_application_name='repmgr'");
+
+	log_debug(_("connecting to: '%s'"), connection_string);
+
+	conn = PQconnectdb(connection_string);
+
+	/* Check to see that the backend connection was successfully made */
+	if ((PQstatus(conn) != CONNECTION_OK))
+	{
+		bool emit_log = true;
+
+		if (verbose_only == true && verbose_logging == false)
+			emit_log = false;
+
+		if (emit_log)
+		{
+			if (log_notice)
+			{
+				log_notice(_("connection to database failed: %s"),
+						   PQerrorMessage(conn));
+			}
+			else
+			{
+				log_error(_("connection to database failed: %s"),
+						  PQerrorMessage(conn));
+			}
+		}
+
+		if (exit_on_error)
+		{
+			PQfinish(conn);
+			exit(ERR_DB_CON);
+		}
+	}
+
+	/*
+	 * set "synchronous_commit" to "local" in case synchronous replication is in use
+	 *
+	 * XXX set this explicitly before any write operations
+	 */
+
+	else if (set_config(conn, "synchronous_commit", "local") == false)
+	{
+		if (exit_on_error)
+		{
+			PQfinish(conn);
+			exit(ERR_DB_CON);
+		}
+	}
+
+	return conn;
+}
+
+
+/*
+ * Establish a database connection, optionally exit on error
+ */
+PGconn *
+establish_db_connection(const char *conninfo, const bool exit_on_error)
+{
+	return _establish_db_connection(conninfo, exit_on_error, false, false);
+}
+
+/* ========================== */
+/* GUC manipulation functions */
+/* ========================== */
+
+static bool
+_set_config(PGconn *conn, const char *config_param, const char *sqlquery)
+{
+	PGresult   *res;
+
+	res = PQexec(conn, sqlquery);
+
+	if (PQresultStatus(res) != PGRES_COMMAND_OK)
+	{
+		log_error("unable to set '%s': %s", config_param, PQerrorMessage(conn));
+		PQclear(res);
+		return false;
+	}
+
+	PQclear(res);
+
+	return true;
+}
+
+bool
+set_config(PGconn *conn, const char *config_param,  const char *config_value)
+{
+	char		sqlquery[MAX_QUERY_LEN];
+
+	sqlquery_snprintf(sqlquery,
+					  "SET %s TO '%s'",
+					  config_param,
+					  config_value);
+
+	log_verbose(LOG_DEBUG, "set_config():\n%s", sqlquery);
+
+	return _set_config(conn, config_param, sqlquery);
+}
+
+bool
+set_config_bool(PGconn *conn, const char *config_param, bool state)
+{
+	char		sqlquery[MAX_QUERY_LEN];
+
+	sqlquery_snprintf(sqlquery,
+					  "SET %s TO %s",
+					  config_param,
+					  state ? "TRUE" : "FALSE");
+
+	log_verbose(LOG_DEBUG, "set_config_bool():\n%s\n", sqlquery);
+
+	return _set_config(conn, config_param, sqlquery);
+}
+
+/* ============================ */
+/* Server information functions */
+/* ============================ */
+
+/*
+ * Return the server version number for the connection provided
+ */
+int
+get_server_version(PGconn *conn, char *server_version)
+{
+	PGresult   *res;
+	res = PQexec(conn,
+				 "SELECT pg_catalog.current_setting('server_version_num'), "
+				 "       pg_catalog.current_setting('server_version')");
+
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		log_error(_("unable to determine server version number:\n%s"),
+				  PQerrorMessage(conn));
+		PQclear(res);
+		return -1;
+	}
+
+	if (server_version != NULL)
+		strcpy(server_version, PQgetvalue(res, 0, 0));
+
+	return atoi(PQgetvalue(res, 0, 0));
+}

dbutils.h

@@ -0,0 +1,110 @@
+/*
+ * dbutils.h
+ *
+ * Copyright (c) 2ndQuadrant, 2010-2017
+ */
+
+
+
+#ifndef _REPMGR_DBUTILS_H_
+#define _REPMGR_DBUTILS_H_
+
+#include "access/xlogdefs.h"
+#include "pqexpbuffer.h"
+
+#include "config.h"
+#include "strutil.h"
+
+
+typedef enum {
+	UNKNOWN = 0,
+	MASTER,
+	STANDBY,
+	WITNESS,
+	BDR
+} t_server_type;
+
+/*
+ * Struct to store node information
+ */
+typedef struct s_node_info
+{
+	int			  node_id;
+	int			  upstream_node_id;
+	t_server_type type;
+	char		  name[MAXLEN];
+	char		  conninfo[MAXLEN];
+	char		  slot_name[MAXLEN];
+	int			  priority;
+	bool		  active;
+	bool		  is_ready;
+	bool		  is_visible;
+	XLogRecPtr	  xlog_location;
+}	t_node_info;
+
+
+#define T_NODE_INFO_INITIALIZER { \
+  NODE_NOT_FOUND, \
+  NO_UPSTREAM_NODE, \
+  UNKNOWN, \
+  "", \
+  "", \
+  "", \
+  DEFAULT_PRIORITY, \
+  true, \
+  false, \
+  false, \
+  InvalidXLogRecPtr \
+}
+
+
+typedef struct NodeInfoListCell
+{
+	struct NodeInfoListCell *next;
+	t_node_info *node_info;
+} NodeInfoListCell;
+
+typedef struct NodeInfoList
+{
+	NodeInfoListCell *head;
+	NodeInfoListCell *tail;
+} NodeInfoList;
+
+
+typedef struct s_event_info
+{
+	char		  *node_name;
+	char		  *conninfo_str;
+}   t_event_info;
+
+#define T_EVENT_INFO_INITIALIZER { \
+  NULL, \
+  NULL \
+}
+
+/*
+ * Struct to store replication slot information
+ */
+
+typedef struct s_replication_slot
+{
+	char slot_name[MAXLEN];
+    char slot_type[MAXLEN];
+	bool active;
+}   t_replication_slot;
+
+
+/* connection functions */
+PGconn *establish_db_connection(const char *conninfo,
+								const bool exit_on_error);
+
+
+/* GUC manipulation functions */
+bool		set_config(PGconn *conn, const char *config_param,  const char *config_value);
+bool		set_config_bool(PGconn *conn, const char *config_param, bool state);
+
+/* Server information functions */
+int			get_server_version(PGconn *conn, char *server_version);
+
+#endif
+

repmgr-client.c

@@ -26,6 +26,11 @@ t_configuration_options config_file_options = T_CONFIGURATION_OPTIONS_INITIALIZE
 ItemList	cli_errors = { NULL, NULL };
 ItemList	cli_warnings = { NULL, NULL };
 
+static bool	config_file_required = true;
+
+static char  repmgr_slot_name[MAXLEN] = "";
+static char *repmgr_slot_name_ptr = NULL;
+
 int
 main(int argc, char **argv)
 {
@@ -267,10 +272,89 @@ main(int argc, char **argv)
 	if (runtime_options.terse)
 		logger_set_terse();
 
+	/*
+	 * Node configuration information is not needed for all actions, with
+	 * STANDBY CLONE being the main exception.
+	 */
+	if (config_file_required)
+	{
+		/*
+		 * if a configuration file was provided, the configuration file parser
+		 * will already have errored out if no valid node_id found
+		 */
+		if (config_file_options.node_id == NODE_NOT_FOUND)
+		{
+			log_error(_("no node information was found - "
+						"please supply a configuration file"));
+			exit(ERR_BAD_CONFIG);
+		}
+	}
+
+	/*
+	 * Initialise slot name, if required (9.4 and later)
+	 *
+	 * NOTE: the slot name will be defined for each record, including
+	 * the master; the `slot_name` column in `repl_nodes` defines
+	 * the name of the slot, but does not imply a slot has been created.
+	 * The version check for 9.4 or later  is done in check_upstream_config()
+	 */
+	if (config_file_options.use_replication_slots)
+	{
+		maxlen_snprintf(repmgr_slot_name, "repmgr_slot_%i", config_file_options.node_id);
+		repmgr_slot_name_ptr = repmgr_slot_name;
+		log_verbose(LOG_DEBUG, "slot name initialised as: %s", repmgr_slot_name);
+	}
+
+	switch (action)
+	{
+		case MASTER_REGISTER:
+			do_master_register();
+			break;
+		default:
+			/* An action will have been determined by this point  */
+			break;
+	}
+
 	return SUCCESS;
 }
 
 
+
+static void
+exit_with_errors(void)
+{
+	fprintf(stderr, _("The following command line errors were encountered:\n"));
+
+	print_error_list(&cli_errors, LOG_ERR);
+
+	fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname());
+
+	exit(ERR_BAD_CONFIG);
+}
+
+
+static void
+print_error_list(ItemList *error_list, int log_level)
+{
+	ItemListCell *cell;
+
+	for (cell = error_list->head; cell; cell = cell->next)
+	{
+        fprintf(stderr, "  ");
+		switch(log_level)
+		{
+			/* Currently we only need errors and warnings */
+            case LOG_ERROR:
+				log_error("%s", cell->string);
+				break;
+			case LOG_WARNING:
+				log_warning("%s", cell->string);
+				break;
+		}
+	}
+}
+
+
 static void
 do_help(void)
 {
@@ -303,41 +387,66 @@ do_help(void)
 	printf(_("  --log-to-file                       log to file (or logging facility) defined in repmgr.conf\n"));
 	printf(_("  -t, --terse                         don't display hints and other non-critical output\n"));
 	printf(_("  -v, --verbose                       display additional log output (useful for debugging)\n"));
-	printf(_("\n"));
 
     puts("");
 }
 
+
 static void
-exit_with_errors(void)
+do_master_register(void)
 {
-	fprintf(stderr, _("The following command line errors were encountered:\n"));
+	PGconn	   *conn;
 
-	print_error_list(&cli_errors, LOG_ERR);
+	log_info(_("connecting to master database..."));
 
-	fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname());
+	// XXX if con fails, have this print offending conninfo!
+	conn = establish_db_connection(config_file_options.conninfo, true);
 
-	exit(ERR_BAD_CONFIG);
+	check_server_version(conn, "master", true, NULL);
 }
 
-
+/**
-static void
+ * check_server_version()
-print_error_list(ItemList *error_list, int log_level)
+ *
+ * Verify that the server is MIN_SUPPORTED_VERSION_NUM or later
+ *
+ * PGconn *conn:
+ *   the connection to check
+ *
+ * char *server_type:
+ *   either "master" or "standby"; used to format error message
+ *
+ * bool exit_on_error:
+ *   exit if reported server version is too low; optional to enable some callers
+ *   to perform additional cleanup
+ *
+ * char *server_version_string
+ *   passed to get_server_version(), which will place the human-readable
+ *   server version string there (e.g. "9.4.0")
+ */
+static int
+check_server_version(PGconn *conn, char *server_type, bool exit_on_error, char *server_version_string)
 {
-	ItemListCell *cell;
+	int			server_version_num = 0;
 
-	for (cell = error_list->head; cell; cell = cell->next)
+	server_version_num = get_server_version(conn, server_version_string);
+	if (server_version_num < MIN_SUPPORTED_VERSION_NUM)
 	{
-        fprintf(stderr, "  ");
+		if (server_version_num > 0)
-		switch(log_level)
+			log_error(_("%s requires %s to be PostgreSQL %s or later"),
+					  progname(),
+					  server_type,
+					  MIN_SUPPORTED_VERSION
+				);
+
+		if (exit_on_error == true)
 		{
-			/* Currently we only need errors and warnings */
+			PQfinish(conn);
-            case LOG_ERROR:
+			exit(ERR_BAD_CONFIG);
-				log_error("%s", cell->string);
-				break;
-			case LOG_WARNING:
-				log_warning("%s", cell->string);
-				break;
 		}
+
+		return -1;
 	}
+
+	return server_version_num;
 }

repmgr-client.h

@@ -131,8 +131,10 @@ typedef struct
 		"", false, false, false}
 
 static void do_help(void);
+static void do_master_register(void);
 
 static void exit_with_errors(void);
 static void print_error_list(ItemList *error_list, int log_level);
+static int	check_server_version(PGconn *conn, char *server_type, bool exit_on_error, char *server_version_string);
 
 #endif

repmgr.h

@@ -16,6 +16,8 @@
 #include "errcode.h"
 #include "strutil.h"
 #include "config.h"
+#include "dbutils.h"
+#include "log.h"
 
 #define MIN_SUPPORTED_VERSION		"9.3"
 #define MIN_SUPPORTED_VERSION_NUM	90300

strutil.c

@@ -0,0 +1,78 @@
+/*
+ * strutil.c
+ *
+ * Copyright (c) 2ndQuadrant, 2010-2017
+ */
+
+#include <stdarg.h>
+#include <stdio.h>
+#include <stdlib.h>
+
+#include "log.h"
+#include "strutil.h"
+
+static int
+xvsnprintf(char *str, size_t size, const char *format, va_list ap)
+__attribute__((format(PG_PRINTF_ATTRIBUTE, 3, 0)));
+
+static int
+xvsnprintf(char *str, size_t size, const char *format, va_list ap)
+{
+	int			retval;
+
+	retval = vsnprintf(str, size, format, ap);
+
+	if (retval >= (int) size)
+	{
+		log_error(_("Buffer of size not large enough to format entire string '%s'"),
+				str);
+		exit(ERR_STR_OVERFLOW);
+	}
+
+	return retval;
+}
+
+
+int
+sqlquery_snprintf(char *str, const char *format,...)
+{
+	va_list		arglist;
+	int			retval;
+
+	va_start(arglist, format);
+	retval = xvsnprintf(str, MAX_QUERY_LEN, format, arglist);
+	va_end(arglist);
+
+	return retval;
+}
+
+int
+maxlen_snprintf(char *str, const char *format,...)
+{
+	va_list		arglist;
+	int			retval;
+
+	va_start(arglist, format);
+	retval = xvsnprintf(str, MAXLEN, format, arglist);
+	va_end(arglist);
+
+	return retval;
+}
+
+
+/*
+ * Escape a string for use as a parameter in recovery.conf
+ * Caller must free returned value
+ */
+char *
+escape_recovery_conf_value(const char *src)
+{
+	char	   *result = escape_single_quotes_ascii(src);
+
+	if (!result)
+	{
+		fprintf(stderr, _("%s: out of memory\n"), progname());
+		exit(ERR_INTERNAL);
+	}
+	return result;
+}

strutil.h

@@ -7,6 +7,18 @@
 #define _STRUTIL_H_
 
 #define MAXLEN			1024
+#define MAX_QUERY_LEN	8192
+/* Why? http://stackoverflow.com/a/5459929/398670 */
+#define STR(x) CppAsString(x)
 
+#define MAXLEN_STR STR(MAXLEN)
+
+extern int
+sqlquery_snprintf(char *str, const char *format,...)
+__attribute__((format(PG_PRINTF_ATTRIBUTE, 2, 3)));
+
+extern int
+maxlen_snprintf(char *str, const char *format,...)
+__attribute__((format(PG_PRINTF_ATTRIBUTE, 2, 3)));
 
 #endif