31#define DEFAULT_SUB_PORT "50432"
81 const char *pub_base_conninfo,
82 const char *sub_base_conninfo);
94 const char *consistent_lsn);
98 const char *slotname);
103 const char *slot_name);
106 bool restricted_access,
107 bool restrict_logical_worker);
122#define USEC_PER_SEC 1000000
123#define WAIT_INTERVAL 1
182 "You must recreate the physical replica before continuing.");
211 pg_log_warning(
"publication \"%s\" created in database \"%s\" on primary was left behind",
218 pg_log_warning(
"replication slot \"%s\" created in database \"%s\" on primary was left behind",
234 printf(
_(
"%s creates a new logical replica from a standby server.\n\n"),
239 printf(
_(
" -d, --database=DBNAME database in which to create a subscription\n"));
240 printf(
_(
" -D, --pgdata=DATADIR location for the subscriber data directory\n"));
241 printf(
_(
" -n, --dry-run dry run, just show what would be done\n"));
243 printf(
_(
" -P, --publisher-server=CONNSTR publisher connection string\n"));
244 printf(
_(
" -s, --socketdir=DIR socket directory to use (default current dir.)\n"));
245 printf(
_(
" -t, --recovery-timeout=SECS seconds to wait for recovery to end\n"));
246 printf(
_(
" -T, --enable-two-phase enable two-phase commit for all subscriptions\n"));
247 printf(
_(
" -U, --subscriber-username=NAME user name for subscriber connection\n"));
248 printf(
_(
" -v, --verbose output verbose messages\n"));
249 printf(
_(
" --config-file=FILENAME use specified main server configuration\n"
250 " file when running target cluster\n"));
251 printf(
_(
" --publication=NAME publication name\n"));
252 printf(
_(
" --replication-slot=NAME replication slot name\n"));
253 printf(
_(
" --subscription=NAME subscription name\n"));
254 printf(
_(
" -V, --version output version information, then exit\n"));
255 printf(
_(
" -?, --help show this help, then exit\n"));
256 printf(
_(
"\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
257 printf(
_(
"%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
296 if (conn_opts == NULL)
304 for (conn_opt = conn_opts; conn_opt->
keyword != NULL; conn_opt++)
306 if (conn_opt->
val != NULL && conn_opt->
val[0] !=
'\0')
308 if (strcmp(conn_opt->
keyword,
"dbname") == 0)
375 pg_fatal(
"program \"%s\" is needed by %s but was not found in the same directory as \"%s\"",
376 progname,
"pg_createsubscriber", full_path);
378 pg_fatal(
"program \"%s\" was found by \"%s\" but was not the same version as %s",
379 progname, full_path,
"pg_createsubscriber");
398 pg_log_info(
"checking if directory \"%s\" is a cluster data directory",
410 if (
stat(versionfile, &statbuf) != 0 && errno == ENOENT)
412 pg_fatal(
"directory \"%s\" is not a database cluster directory",
450 const char *pub_base_conninfo,
451 const char *sub_base_conninfo)
495 pg_log_debug(
"publisher(%d): publication: %s ; replication slot: %s ; connection string: %s",
i,
499 pg_log_debug(
"subscriber(%d): subscription: %s ; connection string: %s, two_phase: %s",
i,
505 pubcell = pubcell->
next;
507 subcell = subcell->
next;
509 replslotcell = replslotcell->
next;
583 pg_log_info(
"getting system identifier from publisher");
587 res =
PQexec(
conn,
"SELECT system_identifier FROM pg_catalog.pg_control_system()");
596 pg_log_error(
"could not get system identifier: got %d rows, expected %d row",
601 sysid = strtou64(
PQgetvalue(res, 0, 0), NULL, 10);
603 pg_log_info(
"system identifier is %llu on publisher",
604 (
unsigned long long) sysid);
624 pg_log_info(
"getting system identifier from subscriber");
628 pg_fatal(
"control file appears to be corrupt");
632 pg_log_info(
"system identifier is %llu on subscriber",
633 (
unsigned long long) sysid);
654 pg_log_info(
"modifying system identifier of subscriber");
658 pg_fatal(
"control file appears to be corrupt");
673 pg_log_info(
"system identifier is %llu on subscriber",
676 pg_log_info(
"running pg_resetwal on the subscriber");
685 int rc = system(cmd_str);
688 pg_log_info(
"subscriber successfully changed the system identifier");
710 "SELECT oid FROM pg_catalog.pg_database "
711 "WHERE datname = pg_catalog.current_database()");
721 pg_log_error(
"could not obtain database OID: got %d rows, expected %d row",
727 oid = strtoul(
PQgetvalue(res, 0, 0), NULL, 10);
739 objname =
psprintf(
"pg_createsubscriber_%u_%x", oid, rand);
760 char *genname = NULL;
793 pg_log_info(
"create replication slot \"%s\" on publisher",
794 dbinfo[
i].replslotname);
810 res =
PQexec(
conn,
"SELECT pg_log_standby_snapshot()");
813 pg_log_error(
"could not write an additional WAL record: %s",
835 res =
PQexec(
conn,
"SELECT pg_catalog.pg_is_in_recovery()");
869 int max_prepared_transactions;
870 char *max_slot_wal_keep_size;
898 "SELECT pg_catalog.current_setting('wal_level'),"
899 " pg_catalog.current_setting('max_replication_slots'),"
900 " (SELECT count(*) FROM pg_catalog.pg_replication_slots),"
901 " pg_catalog.current_setting('max_wal_senders'),"
902 " (SELECT count(*) FROM pg_catalog.pg_stat_activity WHERE backend_type = 'walsender'),"
903 " pg_catalog.current_setting('max_prepared_transactions'),"
904 " pg_catalog.current_setting('max_slot_wal_keep_size')");
918 max_prepared_transactions = atoi(
PQgetvalue(res, 0, 5));
924 pg_log_debug(
"publisher: max_replication_slots: %d", max_repslots);
925 pg_log_debug(
"publisher: current replication slots: %d", cur_repslots);
926 pg_log_debug(
"publisher: max_wal_senders: %d", max_walsenders);
927 pg_log_debug(
"publisher: current wal senders: %d", cur_walsenders);
928 pg_log_debug(
"publisher: max_prepared_transactions: %d",
929 max_prepared_transactions);
931 max_slot_wal_keep_size);
937 pg_log_error(
"publisher requires \"wal_level\" >= \"logical\"");
941 if (max_repslots - cur_repslots <
num_dbs)
943 pg_log_error(
"publisher requires %d replication slots, but only %d remain",
944 num_dbs, max_repslots - cur_repslots);
946 "max_replication_slots", cur_repslots +
num_dbs);
950 if (max_walsenders - cur_walsenders <
num_dbs)
952 pg_log_error(
"publisher requires %d WAL sender processes, but only %d remain",
953 num_dbs, max_walsenders - cur_walsenders);
955 "max_wal_senders", cur_walsenders +
num_dbs);
961 pg_log_warning(
"two_phase option will not be enabled for replication slots");
963 "Prepared transactions will be replicated at COMMIT PREPARED.");
972 if (
dry_run && (strcmp(max_slot_wal_keep_size,
"-1") != 0))
974 pg_log_warning(
"required WAL could be removed from the publisher");
975 pg_log_warning_hint(
"Set the configuration parameter \"%s\" to -1 to ensure that required WAL files are not prematurely removed.",
976 "max_slot_wal_keep_size");
1001 bool failed =
false;
1029 "SELECT setting FROM pg_catalog.pg_settings WHERE name IN ("
1030 "'max_logical_replication_workers', "
1031 "'max_replication_slots', "
1032 "'max_worker_processes', "
1033 "'primary_slot_name') "
1038 pg_log_error(
"could not obtain subscriber settings: %s",
1049 pg_log_debug(
"subscriber: max_logical_replication_workers: %d",
1051 pg_log_debug(
"subscriber: max_replication_slots: %d", max_repslots);
1052 pg_log_debug(
"subscriber: max_worker_processes: %d", max_wprocs);
1062 pg_log_error(
"subscriber requires %d replication slots, but only %d remain",
1064 pg_log_error_hint(
"Increase the configuration parameter \"%s\" to at least %d.",
1065 "max_replication_slots",
num_dbs);
1071 pg_log_error(
"subscriber requires %d logical replication workers, but only %d remain",
1073 pg_log_error_hint(
"Increase the configuration parameter \"%s\" to at least %d.",
1074 "max_logical_replication_workers",
num_dbs);
1080 pg_log_error(
"subscriber requires %d worker processes, but only %d remain",
1082 pg_log_error_hint(
"Increase the configuration parameter \"%s\" to at least %d.",
1083 "max_worker_processes",
num_dbs + 1);
1115 pg_log_info(
"dropping subscription \"%s\" in database \"%s\"",
1151 "SELECT s.subname FROM pg_catalog.pg_subscription s "
1152 "INNER JOIN pg_catalog.pg_database d ON (s.subdbid = d.oid) "
1153 "WHERE d.datname = %s",
1159 pg_log_error(
"could not obtain pre-existing subscriptions: %s",
1244 "recovery_target_timeline = 'latest'\n");
1246 "recovery_target_inclusive = true\n");
1248 "recovery_target_action = promote\n");
1257 "recovery_target_lsn = '%X/%X'\n",
1295 pg_log_warning(
"could not drop replication slot \"%s\" on primary",
1319 "SELECT slot_name FROM pg_catalog.pg_replication_slots WHERE failover");
1329 pg_log_warning(
"could not obtain failover replication slot information: %s",
1331 pg_log_warning_hint(
"Drop the failover replication slots on subscriber soon to avoid retention of WAL files.");
1340 pg_log_warning_hint(
"Drop the failover replication slots on subscriber soon to avoid retention of WAL files.");
1356 char *slot_name_esc;
1361 pg_log_info(
"creating the replication slot \"%s\" in database \"%s\"",
1362 slot_name, dbinfo->
dbname);
1367 "SELECT lsn FROM pg_catalog.pg_create_logical_replication_slot(%s, 'pgoutput', false, %s, false)",
1380 pg_log_error(
"could not create replication slot \"%s\" in database \"%s\": %s",
1381 slot_name, dbinfo->
dbname,
1402 const char *slot_name)
1405 char *slot_name_esc;
1410 pg_log_info(
"dropping the replication slot \"%s\" in database \"%s\"",
1411 slot_name, dbinfo->
dbname);
1426 pg_log_error(
"could not drop replication slot \"%s\" in database \"%s\": %s",
1452 pg_log_error(
"pg_ctl was terminated by exception 0x%X",
1454 pg_log_error_detail(
"See C include file \"ntstatus.h\" for a description of the hexadecimal value.");
1462 pg_log_error(
"pg_ctl exited with unrecognized status %d", rc);
1472 bool restrict_logical_worker)
1484 if (restricted_access)
1495 appendPQExpBufferStr(pg_ctl_cmd,
" -o \"-c listen_addresses='' -c unix_socket_permissions=0700");
1507 if (restrict_logical_worker)
1511 rc = system(pg_ctl_cmd->
data);
1527 rc = system(pg_ctl_cmd);
1549 pg_log_info(
"waiting for the target server to reach the consistent state");
1585 pg_fatal(
"server did not end recovery");
1587 pg_log_info(
"target server reached the consistent state");
1588 pg_log_info_hint(
"If pg_createsubscriber fails after this point, you must recreate the physical replica before continuing.");
1609 "SELECT 1 FROM pg_catalog.pg_publication "
1610 "WHERE pubname = %s",
1615 pg_log_error(
"could not obtain publication information: %s",
1637 pg_log_info(
"creating publication \"%s\" in database \"%s\"",
1650 pg_log_error(
"could not create publication \"%s\" in database \"%s\": %s",
1679 pg_log_info(
"dropping publication \"%s\" in database \"%s\"",
1693 pg_log_error(
"could not drop publication \"%s\" in database \"%s\": %s",
1729 char *pubconninfo_esc;
1730 char *replslotname_esc;
1739 pg_log_info(
"creating subscription \"%s\" in database \"%s\"",
1743 "CREATE SUBSCRIPTION %s CONNECTION %s PUBLICATION %s "
1744 "WITH (create_slot = false, enabled = false, "
1745 "slot_name = %s, copy_data = false, two_phase = %s)",
1746 subname_esc, pubconninfo_esc, pubname_esc, replslotname_esc,
1761 pg_log_error(
"could not create subscription \"%s\" in database \"%s\": %s",
1798 "SELECT s.oid FROM pg_catalog.pg_subscription s "
1799 "INNER JOIN pg_catalog.pg_database d ON (s.subdbid = d.oid) "
1800 "WHERE s.subname = %s AND d.datname = %s",
1813 pg_log_error(
"could not obtain subscription OID: got %d rows, expected %d row",
1825 suboid = strtoul(
PQgetvalue(res, 0, 0), NULL, 10);
1835 originname =
psprintf(
"pg_%u", suboid);
1837 pg_log_info(
"setting the replication progress (node name \"%s\", LSN %s) in database \"%s\"",
1838 originname, lsnstr, dbinfo->
dbname);
1842 "SELECT pg_catalog.pg_replication_origin_advance('%s', '%s')",
1843 originname, lsnstr);
1852 pg_log_error(
"could not set replication progress for subscription \"%s\": %s",
1883 pg_log_info(
"enabling subscription \"%s\" in database \"%s\"",
1895 pg_log_error(
"could not enable subscription \"%s\": %s",
1910 static struct option long_options[] =
1936 char *pub_base_conninfo;
1937 char *sub_base_conninfo;
1938 char *dbname_conninfo = NULL;
1942 struct stat statbuf;
1944 char *consistent_lsn;
1955 if (strcmp(argv[1],
"--help") == 0 || strcmp(argv[1],
"-?") == 0)
1960 else if (strcmp(argv[1],
"-V") == 0
1961 || strcmp(argv[1],
"--version") == 0)
1963 puts(
"pg_createsubscriber (PostgreSQL) " PG_VERSION);
1998 while ((
c =
getopt_long(argc, argv,
"d:D:np:P:s:t:TU:v",
1999 long_options, &option_index)) != -1)
2093 pg_log_error(
"too many command-line arguments (first is \"%s\")",
2102 pg_log_error(
"no subscriber data directory specified");
2113 pg_fatal(
"could not determine current directory");
2130 pg_log_error(
"no publisher connection string specified");
2134 pg_log_info(
"validating publisher connection string");
2137 if (pub_base_conninfo == NULL)
2140 pg_log_info(
"validating subscriber connection string");
2152 if (dbname_conninfo)
2157 pg_log_info(
"database name \"%s\" was extracted from the publisher connection string",
2172 pg_log_error(
"wrong number of publication names specified");
2173 pg_log_error_detail(
"The number of specified publication names (%d) must match the number of specified database names (%d).",
2179 pg_log_error(
"wrong number of subscription names specified");
2180 pg_log_error_detail(
"The number of specified subscription names (%d) must match the number of specified database names (%d).",
2186 pg_log_error(
"wrong number of replication slot names specified");
2187 pg_log_error_detail(
"The number of specified replication slot names (%d) must match the number of specified database names (%d).",
2217 if (pub_sysid != sub_sysid)
2218 pg_fatal(
"subscriber data directory is not a copy of the source database cluster");
2229 if (
stat(pidfile, &statbuf) == 0)
2241 pg_log_info(
"starting the standby server with command-line options");
#define PG_TEXTDOMAIN(domain)
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)
int errmsg(const char *fmt,...)
PGconn * PQconnectdb(const char *conninfo)
void PQconninfoFree(PQconninfoOption *connOptions)
PQconninfoOption * PQconninfoParse(const char *conninfo, char **errmsg)
ConnStatusType PQstatus(const PGconn *conn)
void PQfinish(PGconn *conn)
char * PQerrorMessage(const PGconn *conn)
void PQfreemem(void *ptr)
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
ExecStatusType PQresultStatus(const PGresult *res)
void PQclear(PGresult *res)
int PQntuples(const PGresult *res)
char * PQresultErrorMessage(const PGresult *res)
char * PQescapeLiteral(PGconn *conn, const char *str, size_t len)
PGresult * PQexec(PGconn *conn, const char *query)
char * PQescapeIdentifier(PGconn *conn, const char *str, size_t len)
void * pg_malloc(size_t size)
char * pg_strdup(const char *in)
#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
Assert(PointerIsAligned(start, uint64))
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 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 struct LogicalRepInfos dbinfos
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(...)
const char * pg_strsignal(int signum)
void canonicalize_path(char *path)
const char * get_progname(const char *argv0)
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
struct LogicalRepInfo * dbinfo
char val[FLEXIBLE_ARRAY_MEMBER]
struct SimpleStringListCell * next
SimpleStringListCell * head
char * wait_result_to_str(int exitstatus)
int gettimeofday(struct timeval *tp, void *tzp)
#define LSN_FORMAT_ARGS(lsn)
#define InvalidXLogRecPtr