20 #include "catalog/pg_authid_d.h"
32 #define DEFAULT_SUB_PORT "50432"
70 const char *pub_base_conninfo,
71 const char *sub_base_conninfo);
83 const char *consistent_lsn);
87 const char *slotname);
92 const char *slot_name);
95 bool restricted_access,
96 bool restrict_logical_worker);
111 #define USEC_PER_SEC 1000000
112 #define WAIT_INTERVAL 1
171 "You must recreate the physical replica before continuing.");
198 pg_log_warning(
"publication \"%s\" in database \"%s\" on primary might be left behind",
204 pg_log_warning(
"replication slot \"%s\" in database \"%s\" on primary might be left behind",
219 printf(
_(
"%s creates a new logical replica from a standby server.\n\n"),
224 printf(
_(
" -d, --database=DBNAME database to create a subscription\n"));
225 printf(
_(
" -D, --pgdata=DATADIR location for the subscriber data directory\n"));
226 printf(
_(
" -n, --dry-run dry run, just show what would be done\n"));
228 printf(
_(
" -P, --publisher-server=CONNSTR publisher connection string\n"));
229 printf(
_(
" -s, --socket-directory=DIR socket directory to use (default current directory)\n"));
230 printf(
_(
" -t, --recovery-timeout=SECS seconds to wait for recovery to end\n"));
231 printf(
_(
" -U, --subscriber-username=NAME subscriber username\n"));
232 printf(
_(
" -v, --verbose output verbose messages\n"));
233 printf(
_(
" --config-file=FILENAME use specified main server configuration\n"
234 " file when running target cluster\n"));
235 printf(
_(
" --publication=NAME publication name\n"));
236 printf(
_(
" --replication-slot=NAME replication slot name\n"));
237 printf(
_(
" --subscription=NAME subscription name\n"));
238 printf(
_(
" -V, --version output version information, then exit\n"));
239 printf(
_(
" -?, --help show this help, then exit\n"));
240 printf(
_(
"\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
241 printf(
_(
"%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
280 if (conn_opts == NULL)
288 for (conn_opt = conn_opts; conn_opt->
keyword != NULL; conn_opt++)
290 if (conn_opt->
val != NULL && conn_opt->
val[0] !=
'\0')
292 if (strcmp(conn_opt->
keyword,
"dbname") == 0)
359 pg_fatal(
"program \"%s\" is needed by %s but was not found in the same directory as \"%s\"",
360 progname,
"pg_createsubscriber", full_path);
362 pg_fatal(
"program \"%s\" was found by \"%s\" but was not the same version as %s",
363 progname, full_path,
"pg_createsubscriber");
382 pg_log_info(
"checking if directory \"%s\" is a cluster data directory",
394 if (
stat(versionfile, &statbuf) != 0 && errno == ENOENT)
396 pg_fatal(
"directory \"%s\" is not a database cluster directory",
434 const char *pub_base_conninfo,
435 const char *sub_base_conninfo)
479 pg_log_debug(
"publisher(%d): publication: %s ; replication slot: %s ; connection string: %s",
i,
483 pg_log_debug(
"subscriber(%d): subscription: %s ; connection string: %s",
i,
488 pubcell = pubcell->
next;
490 subcell = subcell->
next;
492 replslotcell = replslotcell->
next;
566 pg_log_info(
"getting system identifier from publisher");
570 res =
PQexec(
conn,
"SELECT system_identifier FROM pg_catalog.pg_control_system()");
579 pg_log_error(
"could not get system identifier: got %d rows, expected %d row",
586 pg_log_info(
"system identifier is %llu on publisher",
587 (
unsigned long long) sysid);
607 pg_log_info(
"getting system identifier from subscriber");
611 pg_fatal(
"control file appears to be corrupt");
615 pg_log_info(
"system identifier is %llu on subscriber",
616 (
unsigned long long) sysid);
637 pg_log_info(
"modifying system identifier of subscriber");
641 pg_fatal(
"control file appears to be corrupt");
656 pg_log_info(
"system identifier is %llu on subscriber",
659 pg_log_info(
"running pg_resetwal on the subscriber");
668 int rc = system(cmd_str);
671 pg_log_info(
"subscriber successfully changed the system identifier");
673 pg_fatal(
"subscriber failed to change system identifier: exit code: %d", rc);
693 "SELECT oid FROM pg_catalog.pg_database "
694 "WHERE datname = pg_catalog.current_database()");
704 pg_log_error(
"could not obtain database OID: got %d rows, expected %d row",
722 objname =
psprintf(
"pg_createsubscriber_%u_%x", oid, rand);
743 char *genname = NULL;
776 pg_log_info(
"create replication slot \"%s\" on publisher",
830 int max_prepared_transactions;
857 "SELECT pg_catalog.current_setting('wal_level'),"
858 " pg_catalog.current_setting('max_replication_slots'),"
859 " (SELECT count(*) FROM pg_catalog.pg_replication_slots),"
860 " pg_catalog.current_setting('max_wal_senders'),"
861 " (SELECT count(*) FROM pg_catalog.pg_stat_activity WHERE backend_type = 'walsender'),"
862 " pg_catalog.current_setting('max_prepared_transactions')");
881 pg_log_debug(
"publisher: max_replication_slots: %d", max_repslots);
882 pg_log_debug(
"publisher: current replication slots: %d", cur_repslots);
883 pg_log_debug(
"publisher: max_wal_senders: %d", max_walsenders);
884 pg_log_debug(
"publisher: current wal senders: %d", cur_walsenders);
885 pg_log_debug(
"publisher: max_prepared_transactions: %d",
886 max_prepared_transactions);
892 pg_log_error(
"publisher requires wal_level >= \"logical\"");
896 if (max_repslots - cur_repslots <
num_dbs)
898 pg_log_error(
"publisher requires %d replication slots, but only %d remain",
899 num_dbs, max_repslots - cur_repslots);
901 "max_replication_slots", cur_repslots +
num_dbs);
905 if (max_walsenders - cur_walsenders <
num_dbs)
907 pg_log_error(
"publisher requires %d wal sender processes, but only %d remain",
908 num_dbs, max_walsenders - cur_walsenders);
910 "max_wal_senders", cur_walsenders +
num_dbs);
914 if (max_prepared_transactions != 0)
918 "Prepared transactions will be replicated at COMMIT PREPARED.");
971 "SELECT setting FROM pg_catalog.pg_settings WHERE name IN ("
972 "'max_logical_replication_workers', "
973 "'max_replication_slots', "
974 "'max_worker_processes', "
975 "'primary_slot_name') "
980 pg_log_error(
"could not obtain subscriber settings: %s",
991 pg_log_debug(
"subscriber: max_logical_replication_workers: %d",
993 pg_log_debug(
"subscriber: max_replication_slots: %d", max_repslots);
994 pg_log_debug(
"subscriber: max_worker_processes: %d", max_wprocs);
1004 pg_log_error(
"subscriber requires %d replication slots, but only %d remain",
1006 pg_log_error_hint(
"Increase the configuration parameter \"%s\" to at least %d.",
1007 "max_replication_slots",
num_dbs);
1013 pg_log_error(
"subscriber requires %d logical replication workers, but only %d remain",
1015 pg_log_error_hint(
"Increase the configuration parameter \"%s\" to at least %d.",
1016 "max_logical_replication_workers",
num_dbs);
1022 pg_log_error(
"subscriber requires %d worker processes, but only %d remain",
1024 pg_log_error_hint(
"Increase the configuration parameter \"%s\" to at least %d.",
1025 "max_worker_processes",
num_dbs + 1);
1057 pg_log_info(
"dropping subscription \"%s\" in database \"%s\"",
1066 pg_log_error(
"could not drop a subscription \"%s\" settings: %s",
1093 "SELECT s.subname FROM pg_catalog.pg_subscription s "
1094 "INNER JOIN pg_catalog.pg_database d ON (s.subdbid = d.oid) "
1095 "WHERE d.datname = %s",
1101 pg_log_error(
"could not obtain pre-existing subscriptions: %s",
1185 "recovery_target_timeline = 'latest'\n");
1187 "recovery_target_inclusive = true\n");
1189 "recovery_target_action = promote\n");
1198 "recovery_target_lsn = '%X/%X'\n",
1236 pg_log_warning(
"could not drop replication slot \"%s\" on primary",
1260 "SELECT slot_name FROM pg_catalog.pg_replication_slots WHERE failover");
1270 pg_log_warning(
"could not obtain failover replication slot information: %s",
1272 pg_log_warning_hint(
"Drop the failover replication slots on subscriber soon to avoid retention of WAL files.");
1281 pg_log_warning_hint(
"Drop the failover replication slots on subscriber soon to avoid retention of WAL files.");
1297 char *slot_name_esc;
1302 pg_log_info(
"creating the replication slot \"%s\" in database \"%s\"",
1308 "SELECT lsn FROM pg_catalog.pg_create_logical_replication_slot(%s, 'pgoutput', false, false, false)",
1320 pg_log_error(
"could not create replication slot \"%s\" in database \"%s\": %s",
1342 const char *slot_name)
1345 char *slot_name_esc;
1350 pg_log_info(
"dropping the replication slot \"%s\" in database \"%s\"",
1366 pg_log_error(
"could not drop replication slot \"%s\" in database \"%s\": %s",
1392 pg_log_error(
"pg_ctl was terminated by exception 0x%X",
1394 pg_log_error_detail(
"See C include file \"ntstatus.h\" for a description of the hexadecimal value.");
1402 pg_log_error(
"pg_ctl exited with unrecognized status %d", rc);
1412 bool restrict_logical_worker)
1420 if (restricted_access)
1431 appendPQExpBufferStr(pg_ctl_cmd,
" -o \"-c listen_addresses='' -c unix_socket_permissions=0700");
1443 if (restrict_logical_worker)
1447 rc = system(pg_ctl_cmd->
data);
1463 rc = system(pg_ctl_cmd);
1485 pg_log_info(
"waiting for the target server to reach the consistent state");
1521 pg_fatal(
"server did not end recovery");
1523 pg_log_info(
"target server reached the consistent state");
1524 pg_log_info_hint(
"If pg_createsubscriber fails after this point, you must recreate the physical replica before continuing.");
1545 "SELECT 1 FROM pg_catalog.pg_publication "
1546 "WHERE pubname = %s",
1551 pg_log_error(
"could not obtain publication information: %s",
1573 pg_log_info(
"creating publication \"%s\" in database \"%s\"",
1586 pg_log_error(
"could not create publication \"%s\" in database \"%s\": %s",
1615 pg_log_info(
"dropping publication \"%s\" in database \"%s\"",
1629 pg_log_error(
"could not drop publication \"%s\" in database \"%s\": %s",
1665 char *pubconninfo_esc;
1666 char *replslotname_esc;
1675 pg_log_info(
"creating subscription \"%s\" in database \"%s\"",
1679 "CREATE SUBSCRIPTION %s CONNECTION %s PUBLICATION %s "
1680 "WITH (create_slot = false, enabled = false, "
1681 "slot_name = %s, copy_data = false)",
1682 subname_esc, pubconninfo_esc, pubname_esc, replslotname_esc);
1696 pg_log_error(
"could not create subscription \"%s\" in database \"%s\": %s",
1733 "SELECT s.oid FROM pg_catalog.pg_subscription s "
1734 "INNER JOIN pg_catalog.pg_database d ON (s.subdbid = d.oid) "
1735 "WHERE s.subname = %s AND d.datname = %s",
1748 pg_log_error(
"could not obtain subscription OID: got %d rows, expected %d row",
1770 originname =
psprintf(
"pg_%u", suboid);
1772 pg_log_info(
"setting the replication progress (node name \"%s\" ; LSN %s) in database \"%s\"",
1777 "SELECT pg_catalog.pg_replication_origin_advance('%s', '%s')",
1778 originname, lsnstr);
1787 pg_log_error(
"could not set replication progress for the subscription \"%s\": %s",
1818 pg_log_info(
"enabling subscription \"%s\" in database \"%s\"",
1830 pg_log_error(
"could not enable subscription \"%s\": %s",
1845 static struct option long_options[] =
1870 char *pub_base_conninfo;
1871 char *sub_base_conninfo;
1872 char *dbname_conninfo = NULL;
1876 struct stat statbuf;
1878 char *consistent_lsn;
1889 if (strcmp(argv[1],
"--help") == 0 || strcmp(argv[1],
"-?") == 0)
1894 else if (strcmp(argv[1],
"-V") == 0
1895 || strcmp(argv[1],
"--version") == 0)
1897 puts(
"pg_createsubscriber (PostgreSQL) " PG_VERSION);
1931 while ((
c =
getopt_long(argc, argv,
"d:D:np:P:s:t:U:v",
1932 long_options, &option_index)) != -1)
2023 pg_log_error(
"too many command-line arguments (first is \"%s\")",
2032 pg_log_error(
"no subscriber data directory specified");
2043 pg_fatal(
"could not determine current directory");
2060 pg_log_error(
"no publisher connection string specified");
2064 pg_log_info(
"validating connection string on publisher");
2067 if (pub_base_conninfo == NULL)
2070 pg_log_info(
"validating connection string on subscriber");
2082 if (dbname_conninfo)
2087 pg_log_info(
"database \"%s\" was extracted from the publisher connection string",
2103 pg_log_error_hint(
"Number of publication names (%d) must match number of database names (%d).",
2110 pg_log_error_hint(
"Number of subscription names (%d) must match number of database names (%d).",
2116 pg_log_error(
"wrong number of replication slot names");
2117 pg_log_error_hint(
"Number of replication slot names (%d) must match number of database names (%d).",
2145 if (pub_sysid != sub_sysid)
2146 pg_fatal(
"subscriber data directory is not a copy of the source database cluster");
2157 if (
stat(pidfile, &statbuf) == 0)
2169 pg_log_info(
"starting the standby with command-line options");
#define Assert(condition)
#define PG_TEXTDOMAIN(domain)
#define strtou64(str, endptr, base)
int find_my_exec(const char *argv0, char *retpath)
void set_pglocale_pgservice(const char *argv0, const char *app)
int find_other_exec(const char *argv0, const char *target, const char *versionstr, char *retpath)
#define ALWAYS_SECURE_SEARCH_PATH_SQL
void update_controlfile(const char *DataDir, ControlFileData *ControlFile, bool do_sync)
ControlFileData * get_controlfile(const char *DataDir, bool *crc_ok_p)
static void PGresult * res
int errmsg(const char *fmt,...)
PQconninfoOption * PQconninfoParse(const char *conninfo, char **errmsg)
void PQconninfoFree(PQconninfoOption *connOptions)
char * PQerrorMessage(const PGconn *conn)
ConnStatusType PQstatus(const PGconn *conn)
void PQfinish(PGconn *conn)
PGconn * PQconnectdb(const char *conninfo)
void PQfreemem(void *ptr)
char * PQescapeIdentifier(PGconn *conn, const char *str, size_t len)
ExecStatusType PQresultStatus(const PGresult *res)
char * PQresultErrorMessage(const PGresult *res)
int PQntuples(const PGresult *res)
char * PQescapeLiteral(PGconn *conn, const char *str, size_t len)
PGresult * PQexec(PGconn *conn, const char *query)
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
char * pg_strdup(const char *in)
void * pg_malloc(size_t size)
#define pg_malloc_array(type, count)
int getopt_long(int argc, char *const argv[], const char *optstring, const struct option *longopts, int *longindex)
#define required_argument
void pg_logging_increase_verbosity(void)
void pg_logging_init(const char *argv0)
void pg_logging_set_level(enum pg_log_level new_level)
#define pg_log_error(...)
#define pg_log_error_hint(...)
#define pg_log_warning_hint(...)
#define pg_log_info_hint(...)
#define pg_log_warning_detail(...)
#define pg_log_error_detail(...)
#define pg_log_debug(...)
static PQExpBuffer recoveryconfcontents
static void pg_ctl_status(const char *pg_ctl_cmd, int rc)
static void create_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo)
static struct LogicalRepInfo * store_pub_sub_info(const struct CreateSubscriberOptions *opt, const char *pub_base_conninfo, const char *sub_base_conninfo)
static struct LogicalRepInfo * dbinfo
static void drop_publication(PGconn *conn, struct LogicalRepInfo *dbinfo)
static void modify_subscriber_sysid(const struct CreateSubscriberOptions *opt)
static void stop_standby_server(const char *datadir)
static char * pg_ctl_path
static bool server_is_in_recovery(PGconn *conn)
static char * get_exec_path(const char *argv0, const char *progname)
static void appendConnStrItem(PQExpBuffer buf, const char *keyword, const char *val)
static void check_publisher(const struct LogicalRepInfo *dbinfo)
static char * subscriber_dir
static char * create_logical_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo)
int main(int argc, char **argv)
static char * primary_slot_name
static void cleanup_objects_atexit(void)
static void check_subscriber(const struct LogicalRepInfo *dbinfo)
static pg_prng_state prng_state
static char * concat_conninfo_dbname(const char *conninfo, const char *dbname)
static void check_data_directory(const char *datadir)
static char * setup_publisher(struct LogicalRepInfo *dbinfo)
@ POSTMASTER_STILL_STARTING
static void setup_subscriber(struct LogicalRepInfo *dbinfo, const char *consistent_lsn)
static char * get_base_conninfo(const char *conninfo, char **dbname)
static uint64 get_standby_sysid(const char *datadir)
static bool recovery_ended
static void start_standby_server(const struct CreateSubscriberOptions *opt, bool restricted_access, bool restrict_logical_worker)
static void drop_existing_subscriptions(PGconn *conn, const char *subname, const char *dbname)
static char * get_sub_conninfo(const struct CreateSubscriberOptions *opt)
static void disconnect_database(PGconn *conn, bool exit_on_error)
static void check_and_drop_existing_subscriptions(PGconn *conn, const struct LogicalRepInfo *dbinfo)
static void drop_primary_replication_slot(struct LogicalRepInfo *dbinfo, const char *slotname)
static void drop_failover_replication_slots(struct LogicalRepInfo *dbinfo)
static void wait_for_end_recovery(const char *conninfo, const struct CreateSubscriberOptions *opt)
static void create_publication(PGconn *conn, struct LogicalRepInfo *dbinfo)
static void set_replication_progress(PGconn *conn, const struct LogicalRepInfo *dbinfo, const char *lsn)
static char * pg_resetwal_path
static char * generate_object_name(PGconn *conn)
static const char * progname
static void drop_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo, const char *slot_name)
static bool standby_running
static uint64 get_primary_sysid(const char *conninfo)
static void setup_recovery(const struct LogicalRepInfo *dbinfo, const char *datadir, const char *lsn)
static void enable_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo)
static PGconn * connect_database(const char *conninfo, bool exit_on_error)
PGDLLIMPORT char * optarg
uint32 pg_prng_uint32(pg_prng_state *state)
void pg_prng_seed(pg_prng_state *state, uint64 seed)
#define pg_log_warning(...)
void canonicalize_path(char *path)
const char * get_progname(const char *argv0)
const char * pg_strsignal(int signum)
size_t strlcpy(char *dst, const char *src, size_t siz)
PQExpBuffer createPQExpBuffer(void)
void resetPQExpBuffer(PQExpBuffer str)
void appendPQExpBuffer(PQExpBuffer str, const char *fmt,...)
void destroyPQExpBuffer(PQExpBuffer str)
void appendPQExpBufferChar(PQExpBuffer str, char ch)
void appendPQExpBufferStr(PQExpBuffer str, const char *data)
char * psprintf(const char *fmt,...)
void WriteRecoveryConfig(PGconn *pgconn, const char *target_dir, PQExpBuffer contents)
PQExpBuffer GenerateRecoveryConfig(PGconn *pgconn, const char *replication_slot, char *dbname)
void get_restricted_token(void)
void pg_usleep(long microsec)
bool simple_string_list_member(SimpleStringList *list, const char *val)
void simple_string_list_append(SimpleStringList *list, const char *val)
struct SimpleStringList SimpleStringList
void appendShellString(PQExpBuffer buf, const char *str)
void appendConnStrVal(PQExpBuffer buf, const char *str)
const char * sub_username
SimpleStringList database_names
SimpleStringList sub_names
SimpleStringList replslot_names
SimpleStringList pub_names
char val[FLEXIBLE_ARRAY_MEMBER]
struct SimpleStringListCell * next
SimpleStringListCell * head
int gettimeofday(struct timeval *tp, void *tzp)
#define LSN_FORMAT_ARGS(lsn)
#define InvalidXLogRecPtr