"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

@@ -554,13 +554,17 @@ do_standby_clone(void)
 		}
 	}
 
-	if (mode == pg_basebackup)
+	switch (mode)
 	{
-		r = run_basebackup(&node_record);
+		case pg_basebackup:
-	}
+			r = run_basebackup(&node_record);
-	else
+			break;
-	{
+		case barman:
-		r = run_file_backup(&node_record);
+			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;
 }
 

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