Add "repmgr standby follow --upstream-node-id"

In an automatic failover situation, after a standby has been promoted
there's a risk the original primary may become available again before
"standby follow" is issued on another standby node, in which case "standby
follow" will reconnect to the original primary.

As the standby's repmgrd will have received a notification from the new
primary, it will know the primary's ID and can therefore explicitly
direct "standby follow" to follow that primary.

repmgr-action-standby.c

@@ -1408,6 +1408,7 @@ do_standby_follow(void)
 	PGconn	   *primary_conn = NULL;
 	int			primary_id = UNKNOWN_NODE_ID;
 	t_node_info primary_node_record = T_NODE_INFO_INITIALIZER;
+	RecordStatus	   record_status = RECORD_NOT_FOUND;
 
 	int	    	timer = 0;
 
@@ -1428,24 +1429,52 @@ do_standby_follow(void)
 	/* check this is a standby */
 	check_recovery_type(local_conn);
 
-	/*
+	if (runtime_options.upstream_node_id != NO_UPSTREAM_NODE)
-	 * Attempt to connect to primary.
-	 *
-	 * If --wait provided, loop for up `primary_follow_timeout`
-	 * seconds before giving up
-	 */
-
-	for (timer = 0; timer < config_file_options.primary_follow_timeout; timer++)
 	{
-		primary_conn = get_primary_connection_quiet(local_conn,
+		// XXX check not self!
-													&primary_id,
-													NULL);
 
-		if (PQstatus(primary_conn) == CONNECTION_OK || runtime_options.wait == false)
+		record_status = get_node_record(local_conn, runtime_options.upstream_node_id, &primary_node_record);
+
+		if (record_status != RECORD_FOUND)
 		{
-			break;
+			log_error(_("unable to find record for specified upstream node %i"),
+					  runtime_options.upstream_node_id);
+			PQfinish(local_conn);
+			exit(ERR_BAD_CONFIG);
+		}
+
+		for (timer = 0; timer < config_file_options.primary_follow_timeout; timer++)
+		{
+			primary_conn = establish_db_connection(config_file_options.conninfo, true);
+
+			if (PQstatus(primary_conn) == CONNECTION_OK || runtime_options.wait == false)
+			{
+				primary_id = runtime_options.upstream_node_id;
+				break;
+			}
+			sleep(1);
+		}
+	}
+	else
+	{
+		/*
+		 * Attempt to connect to primary.
+		 *
+		 * If --wait provided, loop for up `primary_follow_timeout`
+		 * seconds before giving up
+		 */
+
+		for (timer = 0; timer < config_file_options.primary_follow_timeout; timer++)
+		{
+			primary_conn = get_primary_connection_quiet(local_conn,
+														&primary_id,
+														NULL);
+			if (PQstatus(primary_conn) == CONNECTION_OK || runtime_options.wait == false)
+			{
+				break;
+			}
+			sleep(1);
 		}
-		sleep(1);
 	}
 
 	PQfinish(local_conn);
@@ -1457,10 +1486,19 @@ do_standby_follow(void)
 		exit(ERR_BAD_CONFIG);
 	}
 
-	get_node_record(primary_conn, primary_id, &primary_node_record);
+	record_status = get_node_record(primary_conn, primary_id, &primary_node_record);
 
+	if (record_status != RECORD_FOUND)
+	{
+		log_error(_("unable to find record for new upstream node %i"),
+				  runtime_options.upstream_node_id);
+		PQfinish(primary_conn);
+		exit(ERR_BAD_CONFIG);
+	}
+
+
+	// XXX check this is not current upstream anyway
 	/* check replication connection */
-
 	initialize_conninfo_params(&repl_conninfo, false);
 
 	conn_to_param_list(primary_conn, &repl_conninfo);

repmgr-client-global.h

@@ -65,9 +65,11 @@ typedef struct
 	char		recovery_min_apply_delay[MAXLEN];
 	char		replication_user[MAXLEN];
 	char		upstream_conninfo[MAXLEN];
-	int			upstream_node_id;
 	bool		without_barman;
 
+	/* "standby clone"/"standby follow" options */
+	int			upstream_node_id;
+
 	/* "standby register" options */
 	bool		wait_register_sync;
 	int			wait_register_sync_seconds;
@@ -123,7 +125,9 @@ typedef struct
 		UNKNOWN_NODE_ID, "", "", \
 		/* "standby clone" options */ \
 		false, CONFIG_FILE_SAMEPATH, false, false, false, "", "", "", \
-		NO_UPSTREAM_NODE, false,  \
+		false,  \
+		/* "standby clone"/"standby follow" options */ \
+		NO_UPSTREAM_NODE, \
 		/* "standby register" options */ \
 		false, 0, \
 		/* "standby switchover" options */ \

repmgr-client.c

@@ -1355,6 +1355,7 @@ check_cli_parameters(const int action)
 		{
 			case STANDBY_CLONE:
 			case STANDBY_REGISTER:
+			case STANDBY_FOLLOW:
 				break;
 			default:
 				item_list_append_format(&cli_warnings,