repmgr: enable forced registration of a node with a downstream cascaded standby

dbutils.c

@@ -1628,6 +1628,89 @@ create_event_record(PGconn *conn, t_configuration_options *options, int node_id,
 }
 
 
+bool
+update_node_record(PGconn *conn, char *action, int node, char *type, int upstream_node, char *cluster_name, char *node_name, char *conninfo, int priority, char *slot_name, bool active)
+{
+	char		sqlquery[QUERY_STR_LEN];
+	char		upstream_node_id[MAXLEN];
+	char		slot_name_buf[MAXLEN];
+	PGresult   *res;
+
+	/* XXX this segment copied from create_node_record() */
+	if (upstream_node == NO_UPSTREAM_NODE)
+	{
+		/*
+		 * No explicit upstream node id provided for standby - attempt to
+		 * get primary node id
+		 */
+		if (strcmp(type, "standby") == 0)
+		{
+			int primary_node_id = get_master_node_id(conn, cluster_name);
+			maxlen_snprintf(upstream_node_id, "%i", primary_node_id);
+		}
+		else
+		{
+			maxlen_snprintf(upstream_node_id, "%s", "NULL");
+		}
+	}
+	else
+	{
+		maxlen_snprintf(upstream_node_id, "%i", upstream_node);
+	}
+
+	if (slot_name != NULL && slot_name[0])
+	{
+		maxlen_snprintf(slot_name_buf, "'%s'", slot_name);
+	}
+	else
+	{
+		maxlen_snprintf(slot_name_buf, "%s", "NULL");
+	}
+
+	/* XXX convert to placeholder query */
+	sqlquery_snprintf(sqlquery,
+					  "UPDATE %s.repl_nodes SET "
+					  "       type = '%s', "
+					  "       upstream_node_id = %s, "
+					  "       cluster = '%s', "
+					  "       name = '%s', "
+					  "       conninfo = '%s', "
+					  "       slot_name = %s, "
+					  "       priority = %i, "
+					  "       active = %s "
+					  " WHERE id = %i ",
+					  get_repmgr_schema_quoted(conn),
+					  type,
+					  upstream_node_id,
+					  cluster_name,
+					  node_name,
+					  conninfo,
+					  slot_name_buf,
+					  priority,
+					  active == true ? "TRUE" : "FALSE",
+					  node);
+
+	log_verbose(LOG_DEBUG, "update_node_record(): %s\n", sqlquery);
+
+	if (action != NULL)
+	{
+		log_verbose(LOG_DEBUG, "update_node_record(): action is \"%s\"\n", action);
+	}
+
+	res = PQexec(conn, sqlquery);
+	if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
+	{
+		log_err(_("Unable to update node record\n%s\n"),
+				PQerrorMessage(conn));
+		PQclear(res);
+		return false;
+	}
+
+	PQclear(res);
+
+	return true;
+}
+
 /*
  * Update node record following change of status
  * (e.g. inactive primary converted to standby)

dbutils.h

@@ -130,6 +130,7 @@ bool		create_node_record(PGconn *conn, char *action, int node, char *type, int u
 bool		delete_node_record(PGconn *conn, int node, char *action);
 int			get_node_record(PGconn *conn, char *cluster, int node_id, t_node_info *node_info);
 int			get_node_record_by_name(PGconn *conn, char *cluster, const char *node_name, t_node_info *node_info);
+bool        update_node_record(PGconn *conn, char *action, int node, char *type, int upstream_node, char *cluster_name, char *node_name, char *conninfo, int priority, char *slot_name, bool active);
 bool        update_node_record_status(PGconn *conn, char *cluster_name, int this_node_id, char *type, int upstream_node_id, bool active);
 bool        update_node_record_set_upstream(PGconn *conn, char *cluster_name, int this_node_id, int new_upstream_node_id);
 bool        create_event_record(PGconn *conn, t_configuration_options *options, int node_id, char *event, bool successful, char *details);

repmgr.c

@@ -2078,20 +2078,6 @@ do_standby_register(void)
 
 	/* Now register the standby */
 	log_info(_("registering the standby\n"));
-	if (runtime_options.force)
-	{
-		bool node_record_deleted = delete_node_record(master_conn,
-													  options.node,
-													  "standby register");
-
-		if (node_record_deleted == false)
-		{
-			PQfinish(master_conn);
-			if (PQstatus(conn) == CONNECTION_OK)
-				PQfinish(conn);
-			exit(ERR_BAD_CONFIG);
-		}
-	}
 
 	/*
 	 * Check that an active node with the same node_name doesn't exist already
@@ -2142,7 +2128,6 @@ do_standby_register(void)
 				exit(ERR_BAD_CONFIG);
 			}
 
-			/* */
 			log_notice(_("creating placeholder record for upstream node %i\n"),
 					   options.upstream_node);
 
@@ -2192,14 +2177,67 @@ do_standby_register(void)
 		else if (node_record.active == false)
 		{
 			/*
-			 * TODO:
+			 * upstream node is inactive and --force not supplied - refuse to register
-			 *   - emit warning if --force specified
-			 *   - else exit with error
 			 */
+			if (!runtime_options.force)
+			{
+				log_err(_("record for upstream node %i is marked as inactive\n"),
+							options.upstream_node);
+				log_hint(_("use option -F/--force to register a standby with an inactive upstream node\n"));
+				PQfinish(master_conn);
+				if (PQstatus(conn) == CONNECTION_OK)
+					PQfinish(conn);
+				exit(ERR_BAD_CONFIG);
+			}
+
+			/*
+			 * user is using the --force - notify about the potential footgun
+			 */
+			log_notice(_("registering node %i with inactive upstream node %i\n"),
+					   options.node,
+					   options.upstream_node);
 		}
 	}
 
 
+	/* Check if node record exists */
+
+	node_result = get_node_record(master_conn,
+								  options.cluster_name,
+								  options.node,
+								  &node_record);
+
+	if (node_result && !runtime_options.force)
+	{
+		log_err(_("node %i is already registered\n"),
+				options.node);
+		log_hint(_("use option -F/--force to overwrite an existing node record\n"));
+		PQfinish(master_conn);
+		if (PQstatus(conn) == CONNECTION_OK)
+			PQfinish(conn);
+		exit(ERR_BAD_CONFIG);
+	}
+
+	/*
+	 * node record exists - update it
+	 * (at this point we have already established that -F/--force is in use)
+	 */
+	if (node_result)
+	{
+		record_created = update_node_record(master_conn,
+											"standby register",
+											options.node,
+											"standby",
+											options.upstream_node,
+											options.cluster_name,
+											options.node_name,
+											options.conninfo,
+											options.priority,
+											repmgr_slot_name_ptr,
+											true);
+	}
+	else
+	{
 		record_created = create_node_record(master_conn,
 											"standby register",
 											options.node,
@@ -2211,15 +2249,19 @@ do_standby_register(void)
 											options.priority,
 											repmgr_slot_name_ptr,
 											true);
+	}
 
 	if (record_created == false)
 	{
-		if (!runtime_options.force)
+		/* XXX add event description */
-		{
+
-			log_hint(_("use option -F/--force to overwrite an existing node record\n"));
+		create_event_record(master_conn,
-		}
+							&options,
+							options.node,
+							"standby_register",
+							false,
+							NULL);
 
-		/* XXX log registration failure? */
 		PQfinish(master_conn);
 
 		if (PQstatus(conn) == CONNECTION_OK)