From 3c2b8e57922641621d6de9155a40c103dbc41b60 Mon Sep 17 00:00:00 2001
From: Ian Barwick <ian@2ndquadrant.com>
Date: Thu, 1 Mar 2018 15:42:25 +0900
Subject: [PATCH] "standby clone": remove restriction on replication slots in
 Barman mode

While it's preferable to avoid standby replication slots if Barman is in
use, there's no technical reason to prevent this.

Implements GitHub #379.
---
 repmgr-action-standby.c | 100 +++++++++++++++++++++++++++++++++++++---
 repmgr-client.c         |  18 --------
 2 files changed, 94 insertions(+), 24 deletions(-)

diff --git a/repmgr-action-standby.c b/repmgr-action-standby.c
index 743633de..7b7836e5 100644
--- a/repmgr-action-standby.c
+++ b/repmgr-action-standby.c
@@ -554,13 +554,17 @@ do_standby_clone(void)
 		}
 	}
 
-	if (mode == pg_basebackup)
+	switch (mode)
 	{
-		r = run_basebackup(&node_record);
-	}
-	else
-	{
-		r = run_file_backup(&node_record);
+		case pg_basebackup:
+			r = run_basebackup(&node_record);
+			break;
+		case barman:
+			r = run_file_backup(&node_record);
+			break;
+		default:
+			/* should never reach here */
+			log_error(_("unknown clone mode"));
 	}
 
 
@@ -5250,6 +5254,90 @@ stop_backup:
 		rmtree(local_repmgr_tmp_directory, true);
 	}
 
+
+	/*
+	 * if replication slots in use, create replication slot
+	 */
+	if (r == SUCCESS)
+	{
+		if (config_file_options.use_replication_slots == true)
+		{
+			bool slot_warning = false;
+			if (runtime_options.no_upstream_connection == true)
+			{
+				slot_warning = true;
+			}
+			else
+			{
+				t_node_info upstream_node_record = T_NODE_INFO_INITIALIZER;
+				t_replication_slot slot_info = T_REPLICATION_SLOT_INITIALIZER;
+				RecordStatus record_status = RECORD_NOT_FOUND;
+				PGconn	   *upstream_conn = NULL;
+
+				record_status = get_node_record(source_conn, upstream_node_id, &upstream_node_record);
+
+				if (record_status != RECORD_FOUND)
+				{
+					log_error(_("unable to retrieve node record for upstream node %i"), upstream_node_id);
+					slot_warning = true;
+				}
+				else
+				{
+					upstream_conn = establish_db_connection(upstream_node_record.conninfo, false);
+					if (PQstatus(upstream_conn) != CONNECTION_OK)
+					{
+						log_error(_("unable to connect to upstream node %i to create a replication slot"), upstream_node_id);
+						slot_warning = true;
+					}
+					else
+					{
+						record_status = get_slot_record(upstream_conn, node_record->slot_name, &slot_info);
+
+						if (record_status == RECORD_FOUND)
+						{
+							log_verbose(LOG_INFO,
+										_("replication slot \"%s\" aleady exists on upstream node %i"),
+										node_record->slot_name,
+										upstream_node_id);
+						}
+						else
+						{
+							PQExpBufferData errmsg;
+
+							initPQExpBuffer(&errmsg);
+
+							if (create_replication_slot(upstream_conn, node_record->slot_name, source_server_version_num, &errmsg) == false)
+							{
+								log_error(_("unable to create replication slot on upstream node %i"), upstream_node_id);
+								log_detail("%s", errmsg.data);
+								slot_warning = true;
+							}
+							else
+							{
+								log_notice(_("replication slot \"%s\" created on upstream node \"%s\" (ID: %i)"),
+										   node_record->slot_name,
+										   upstream_node_record.node_name,
+										   upstream_node_id );
+							}
+							termPQExpBuffer(&errmsg);
+						}
+
+						PQfinish(upstream_conn);
+					}
+				}
+			}
+
+
+			if (slot_warning == true)
+			{
+				log_warning(_("\"use_replication_slots\" specified but a replication slot could not be created"));
+				log_hint(_("ensure a replication slot called \"%s\" is created on the upstream node (ID: %i)"),
+						 node_record->slot_name,
+						 upstream_node_id);
+			}
+		}
+	}
+
 	return r;
 }
 
diff --git a/repmgr-client.c b/repmgr-client.c
index 139d37aa..be5ab9e7 100644
--- a/repmgr-client.c
+++ b/repmgr-client.c
@@ -998,24 +998,6 @@ main(int argc, char **argv)
 	}
 
 
-
-	/* check for conflicts between runtime options and configuration file */
-	/* ================================================================== */
-
-	if (action == STANDBY_CLONE)
-	{
-		standy_clone_mode mode = get_standby_clone_mode();
-
-		if (mode == barman && runtime_options.without_barman == false
-			&& config_file_options.use_replication_slots == true)
-		{
-			log_error(_("STANDBY CLONE in Barman mode is incompatible with configuration option \"use_replication_slots\""));
-			log_hint(_("set \"use_replication_slots\" to \"no\" in repmgr.conf, or use --without-barman to clone directly from the upstream server"));
-			exit(ERR_BAD_CONFIG);
-		}
-	}
-
-
 	/*
 	 * Check for configuration file items which can be overriden by runtime
 	 * options