pg_createsubscriber.c File Reference

#include "postgres_fe.h"
#include <sys/stat.h>
#include <sys/time.h>
#include <sys/wait.h>
#include "common/connect.h"
#include "common/controldata_utils.h"
#include "common/file_perm.h"
#include "common/file_utils.h"
#include "common/logging.h"
#include "common/pg_prng.h"
#include "common/restricted_token.h"
#include "datatype/timestamp.h"
#include "fe_utils/recovery_gen.h"
#include "fe_utils/simple_list.h"
#include "fe_utils/string_utils.h"
#include "fe_utils/version.h"
#include "getopt_long.h"

Include dependency graph for pg_createsubscriber.c:

Go to the source code of this file.

Data Structures
struct	CreateSubscriberOptions
struct	LogicalRepInfo
struct	LogicalRepInfos

Macros
#define	DEFAULT_SUB_PORT   "50432"
#define	OBJECTTYPE_PUBLICATIONS   0x0001
#define	PG_AUTOCONF_FILENAME   "postgresql.auto.conf"
#define	INCLUDED_CONF_FILE   "pg_createsubscriber.conf"
#define	INCLUDED_CONF_FILE_DISABLED   INCLUDED_CONF_FILE ".disabled"
#define	SERVER_LOG_FILE_NAME   "pg_createsubscriber_server.log"
#define	INTERNAL_LOG_FILE_NAME   "pg_createsubscriber_internal.log"
#define	WAIT_INTERVAL   1 /* 1 second */

Functions
static void	cleanup_objects_atexit (void)
static void	usage (void)
static char *	get_base_conninfo (const char *conninfo, char **dbname)
static char *	get_sub_conninfo (const struct CreateSubscriberOptions *opt)
static char *	get_exec_path (const char *argv0, const char *progname)
static void	check_data_directory (const char *datadir)
static char *	concat_conninfo_dbname (const char *conninfo, const char *dbname)
static struct LogicalRepInfo *	store_pub_sub_info (const struct CreateSubscriberOptions *opt, const char *pub_base_conninfo, const char *sub_base_conninfo)
static PGconn *	connect_database (const char *conninfo, bool exit_on_error)
static void	disconnect_database (PGconn *conn, bool exit_on_error)
static uint64	get_primary_sysid (const char *conninfo)
static uint64	get_standby_sysid (const char *datadir)
static void	modify_subscriber_sysid (const struct CreateSubscriberOptions *opt)
static bool	server_is_in_recovery (PGconn *conn)
static char *	generate_object_name (PGconn *conn)
static void	check_publisher (const struct LogicalRepInfo *dbinfo)
static char *	setup_publisher (struct LogicalRepInfo *dbinfo)
static void	check_subscriber (const struct LogicalRepInfo *dbinfo)
static void	setup_subscriber (struct LogicalRepInfo *dbinfo, const char *consistent_lsn)
static void	setup_recovery (const struct LogicalRepInfo *dbinfo, const char *datadir, const char *lsn)
static void	drop_primary_replication_slot (struct LogicalRepInfo *dbinfo, const char *slotname)
static void	drop_failover_replication_slots (struct LogicalRepInfo *dbinfo)
static char *	create_logical_replication_slot (PGconn *conn, struct LogicalRepInfo *dbinfo)
static void	drop_replication_slot (PGconn *conn, struct LogicalRepInfo *dbinfo, const char *slot_name)
static void	pg_ctl_status (const char *pg_ctl_cmd, int rc)
static void	start_standby_server (const struct CreateSubscriberOptions *opt, bool restricted_access, bool restrict_logical_worker)
static void	stop_standby_server (const char *datadir)
static void	wait_for_end_recovery (const char *conninfo, const struct CreateSubscriberOptions *opt)
static void	create_publication (PGconn *conn, struct LogicalRepInfo *dbinfo)
static bool	find_publication (PGconn *conn, const char *pubname, const char *dbname)
static void	drop_publication (PGconn *conn, const char *pubname, const char *dbname, bool *made_publication)
static void	check_and_drop_publications (PGconn *conn, struct LogicalRepInfo *dbinfo)
static void	create_subscription (PGconn *conn, const struct LogicalRepInfo *dbinfo)
static void	set_replication_progress (PGconn *conn, const struct LogicalRepInfo *dbinfo, const char *lsn)
static void	enable_subscription (PGconn *conn, const struct LogicalRepInfo *dbinfo)
static void	check_and_drop_existing_subscriptions (PGconn *conn, const struct LogicalRepInfo *dbinfo)
static void	drop_existing_subscription (PGconn *conn, const char *subname, const char *dbname)
static void	get_publisher_databases (struct CreateSubscriberOptions *opt, bool dbnamespecified)
static void	report_createsub_log (enum pg_log_level, enum pg_log_part, const char *pg_restrict fmt,...) pg_attribute_printf(3
static void static void	report_createsub_log_v (enum pg_log_level level, enum pg_log_part part, const char *pg_restrict fmt, va_list args) pg_attribute_printf(3
static void static void static pg_noreturn void	report_createsub_fatal (const char *pg_restrict fmt,...) pg_attribute_printf(1
static void static void static pg_noreturn void static void	internal_log_file_write (enum pg_log_level level, enum pg_log_part part, const char *pg_restrict fmt, va_list args) pg_attribute_printf(3
static void	appendConnStrItem (PQExpBuffer buf, const char *keyword, const char *val)
static FILE *	logfile_open (const char *filename, const char *mode)
static void	make_output_dirs (const char *log_basedir)
int	main (int argc, char **argv)

Variables
static const char *	progname
static char *	primary_slot_name = NULL
static bool	dry_run = false
static bool	success = false
static struct LogicalRepInfos	dbinfos
static int	num_dbs = 0
static int	num_pubs = 0
static int	num_subs = 0
static int	num_replslots = 0
static pg_prng_state	prng_state
static char *	pg_ctl_path = NULL
static char *	pg_resetwal_path = NULL
static FILE *	internal_log_file_fp = NULL
static char	logdir [MAXPGPATH]
static char *	subscriber_dir = NULL
static bool	recovery_ended = false
static bool	standby_running = false
static bool	recovery_params_set = false

Macro Definition Documentation

DEFAULT_SUB_PORT

#define DEFAULT_SUB_PORT   "50432"

Definition at line 35 of file pg_createsubscriber.c.

INCLUDED_CONF_FILE

#define INCLUDED_CONF_FILE   "pg_createsubscriber.conf"

Definition at line 50 of file pg_createsubscriber.c.

INCLUDED_CONF_FILE_DISABLED

#define INCLUDED_CONF_FILE_DISABLED   INCLUDED_CONF_FILE ".disabled"

Definition at line 51 of file pg_createsubscriber.c.

INTERNAL_LOG_FILE_NAME

#define INTERNAL_LOG_FILE_NAME   "pg_createsubscriber_internal.log"

Definition at line 54 of file pg_createsubscriber.c.

OBJECTTYPE_PUBLICATIONS

#define OBJECTTYPE_PUBLICATIONS   0x0001

Definition at line 36 of file pg_createsubscriber.c.

PG_AUTOCONF_FILENAME

#define PG_AUTOCONF_FILENAME   "postgresql.auto.conf"

Definition at line 49 of file pg_createsubscriber.c.

SERVER_LOG_FILE_NAME

#define SERVER_LOG_FILE_NAME   "pg_createsubscriber_server.log"

Definition at line 53 of file pg_createsubscriber.c.

WAIT_INTERVAL

#define WAIT_INTERVAL   1 /* 1 second */

Definition at line 167 of file pg_createsubscriber.c.

Function Documentation

appendConnStrItem()

static void appendConnStrItem ( PQExpBuffer  buf, const char *  keyword, const char *  val  )

static

Definition at line 397 of file pg_createsubscriber.c.

{
    if (buf->len > 0)
        appendPQExpBufferChar(buf, ' ');
    appendPQExpBufferStr(buf, keyword);
    appendPQExpBufferChar(buf, '=');
    appendConnStrVal(buf, val);
}

References appendConnStrVal(), appendPQExpBufferChar(), appendPQExpBufferStr(), buf, and val.

Referenced by concat_conninfo_dbname(), get_base_conninfo(), and get_sub_conninfo().

check_and_drop_existing_subscriptions()

static void check_and_drop_existing_subscriptions ( PGconn *  conn, const struct LogicalRepInfo *  dbinfo  )

static

Definition at line 1500 of file pg_createsubscriber.c.

{
    PQExpBuffer query = createPQExpBuffer();
    char       *dbname;
    PGresult   *res;
 
    Assert(conn != NULL);
 
    dbname = PQescapeLiteral(conn, dbinfo->dbname, strlen(dbinfo->dbname));
 
    appendPQExpBuffer(query,
                      "SELECT s.subname FROM pg_catalog.pg_subscription s "
                      "INNER JOIN pg_catalog.pg_database d ON (s.subdbid = d.oid) "
                      "WHERE d.datname = %s",
                      dbname);
    res = PQexec(conn, query->data);
 
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not obtain pre-existing subscriptions: %s",
                             PQresultErrorMessage(res));
        disconnect_database(conn, true);
    }
 
    for (int i = 0; i < PQntuples(res); i++)
        drop_existing_subscription(conn, PQgetvalue(res, i, 0),
                                   dbinfo->dbname);
 
    PQclear(res);
    destroyPQExpBuffer(query);
    PQfreemem(dbname);
}

References appendPQExpBuffer(), Assert, conn, createPQExpBuffer(), PQExpBufferData::data, LogicalRepInfo::dbname, dbname, destroyPQExpBuffer(), disconnect_database(), drop_existing_subscription(), fb(), i, PG_LOG_ERROR, PG_LOG_PRIMARY, PGRES_TUPLES_OK, PQclear, PQescapeLiteral(), PQexec(), PQfreemem(), PQgetvalue, PQntuples, PQresultErrorMessage, PQresultStatus, and report_createsub_log().

Referenced by setup_subscriber().

check_and_drop_publications()

static void check_and_drop_publications ( PGconn *  conn, struct LogicalRepInfo *  dbinfo  )

static

Definition at line 2167 of file pg_createsubscriber.c.

{
    PGresult   *res;
    bool        drop_all_pubs = dbinfos.objecttypes_to_clean & OBJECTTYPE_PUBLICATIONS;
 
    Assert(conn != NULL);
 
    if (drop_all_pubs)
    {
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "dropping all existing publications in database \"%s\"",
                             dbinfo->dbname);
 
        /* Fetch all publication names */
        res = PQexec(conn, "SELECT pubname FROM pg_catalog.pg_publication;");
        if (PQresultStatus(res) != PGRES_TUPLES_OK)
        {
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "could not obtain publication information: %s",
                                 PQresultErrorMessage(res));
            PQclear(res);
            disconnect_database(conn, true);
        }
 
        /* Drop each publication */
        for (int i = 0; i < PQntuples(res); i++)
            drop_publication(conn, PQgetvalue(res, i, 0), dbinfo->dbname,
                             &dbinfo->made_publication);
 
        PQclear(res);
    }
    else
    {
        /* Drop publication only if it was created by this tool */
        if (dbinfo->made_publication)
        {
            drop_publication(conn, dbinfo->pubname, dbinfo->dbname,
                             &dbinfo->made_publication);
        }
        else
        {
            if (dry_run)
                report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                                     "dry-run: would preserve existing publication \"%s\" in database \"%s\"",
                                     dbinfo->pubname, dbinfo->dbname);
            else
                report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                                     "preserve existing publication \"%s\" in database \"%s\"",
                                     dbinfo->pubname, dbinfo->dbname);
        }
    }
}

References Assert, conn, dbinfos, LogicalRepInfo::dbname, disconnect_database(), drop_publication(), dry_run, fb(), i, LogicalRepInfo::made_publication, OBJECTTYPE_PUBLICATIONS, LogicalRepInfos::objecttypes_to_clean, PG_LOG_ERROR, PG_LOG_INFO, PG_LOG_PRIMARY, PGRES_TUPLES_OK, PQclear, PQexec(), PQgetvalue, PQntuples, PQresultErrorMessage, PQresultStatus, LogicalRepInfo::pubname, and report_createsub_log().

Referenced by setup_subscriber().

check_data_directory()

static void check_data_directory ( const char *  datadir )

static

Definition at line 527 of file pg_createsubscriber.c.

{
    struct stat statbuf;
    uint32      major_version;
    char       *version_str;
 
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "checking if directory \"%s\" is a cluster data directory",
                         datadir);
 
    if (stat(datadir, &statbuf) != 0)
    {
        if (errno == ENOENT)
            report_createsub_fatal("data directory \"%s\" does not exist", datadir);
        else
            report_createsub_fatal("could not access directory \"%s\": %m", datadir);
    }
 
    /*
     * Retrieve the contents of this cluster's PG_VERSION.  We require
     * compatibility with the same major version as the one this tool is
     * compiled with.
     */
    major_version = GET_PG_MAJORVERSION_NUM(get_pg_version(datadir, &version_str));
    if (major_version != PG_MAJORVERSION_NUM)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "data directory is of wrong version");
        report_createsub_log(PG_LOG_ERROR, PG_LOG_DETAIL,
                             "File \"%s\" contains \"%s\", which is not compatible with this program's version \"%s\".",
                             "PG_VERSION", version_str, PG_MAJORVERSION);
        exit(1);
    }
}

References datadir, fb(), GET_PG_MAJORVERSION_NUM, get_pg_version(), PG_LOG_DETAIL, PG_LOG_ERROR, PG_LOG_INFO, PG_LOG_PRIMARY, report_createsub_fatal(), report_createsub_log(), and stat.

Referenced by main().

check_publisher()

static void check_publisher ( const struct LogicalRepInfo *  dbinfo )

static

Definition at line 1180 of file pg_createsubscriber.c.

{
    PGconn     *conn;
    PGresult   *res;
    bool        failed = false;
 
    char       *wal_level;
    int         max_repslots;
    int         cur_repslots;
    int         max_walsenders;
    int         cur_walsenders;
    int         max_prepared_transactions;
    char       *max_slot_wal_keep_size;
 
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "checking settings on publisher");
 
    conn = connect_database(dbinfo[0].pubconninfo, true);
 
    /*
     * If the primary server is in recovery (i.e. cascading replication),
     * objects (publication) cannot be created because it is read only.
     */
    if (server_is_in_recovery(conn))
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "primary server cannot be in recovery");
        disconnect_database(conn, true);
    }
 
    /*------------------------------------------------------------------------
     * Logical replication requires a few parameters to be set on publisher.
     * Since these parameters are not a requirement for physical replication,
     * we should check it to make sure it won't fail.
     *
     * - wal_level >= replica
     * - max_replication_slots >= current + number of dbs to be converted
     * - max_wal_senders >= current + number of dbs to be converted
     * - max_slot_wal_keep_size = -1 (to prevent deletion of required WAL files)
     * -----------------------------------------------------------------------
     */
    res = PQexec(conn,
                 "SELECT pg_catalog.current_setting('wal_level'),"
                 " pg_catalog.current_setting('max_replication_slots'),"
                 " (SELECT count(*) FROM pg_catalog.pg_replication_slots),"
                 " pg_catalog.current_setting('max_wal_senders'),"
                 " (SELECT count(*) FROM pg_catalog.pg_stat_activity WHERE backend_type = 'walsender'),"
                 " pg_catalog.current_setting('max_prepared_transactions'),"
                 " pg_catalog.current_setting('max_slot_wal_keep_size')");
 
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not obtain publisher settings: %s",
                             PQresultErrorMessage(res));
        disconnect_database(conn, true);
    }
 
    wal_level = pg_strdup(PQgetvalue(res, 0, 0));
    max_repslots = atoi(PQgetvalue(res, 0, 1));
    cur_repslots = atoi(PQgetvalue(res, 0, 2));
    max_walsenders = atoi(PQgetvalue(res, 0, 3));
    cur_walsenders = atoi(PQgetvalue(res, 0, 4));
    max_prepared_transactions = atoi(PQgetvalue(res, 0, 5));
    max_slot_wal_keep_size = pg_strdup(PQgetvalue(res, 0, 6));
 
    PQclear(res);
 
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "publisher: wal_level: %s", wal_level);
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "publisher: max_replication_slots: %d", max_repslots);
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "publisher: current replication slots: %d", cur_repslots);
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "publisher: max_wal_senders: %d", max_walsenders);
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "publisher: current wal senders: %d", cur_walsenders);
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "publisher: max_prepared_transactions: %d",
                         max_prepared_transactions);
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "publisher: max_slot_wal_keep_size: %s",
                         max_slot_wal_keep_size);
 
    disconnect_database(conn, false);
 
    if (strcmp(wal_level, "minimal") == 0)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "publisher requires \"wal_level\" >= \"replica\"");
        failed = true;
    }
 
    if (max_repslots - cur_repslots < num_dbs)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "publisher requires %d replication slots, but only %d remain",
                             num_dbs, max_repslots - cur_repslots);
        report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                             "Increase the configuration parameter \"%s\" to at least %d.",
                             "max_replication_slots", cur_repslots + num_dbs);
        failed = true;
    }
 
    if (max_walsenders - cur_walsenders < num_dbs)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "publisher requires %d WAL sender processes, but only %d remain",
                             num_dbs, max_walsenders - cur_walsenders);
        report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                             "Increase the configuration parameter \"%s\" to at least %d.",
                             "max_wal_senders", cur_walsenders + num_dbs);
        failed = true;
    }
 
    if (max_prepared_transactions != 0 && !dbinfos.two_phase)
    {
        report_createsub_log(PG_LOG_WARNING, PG_LOG_PRIMARY,
                             "two_phase option will not be enabled for replication slots");
        report_createsub_log(PG_LOG_WARNING, PG_LOG_DETAIL,
                             "Subscriptions will be created with the two_phase option disabled.  "
                             "Prepared transactions will be replicated at COMMIT PREPARED.");
        report_createsub_log(PG_LOG_WARNING, PG_LOG_HINT,
                             "You can use the command-line option --enable-two-phase to enable two_phase.");
    }
 
    /*
     * In dry-run mode, validate 'max_slot_wal_keep_size'. If this parameter
     * is set to a non-default value, it may cause replication failures due to
     * required WAL files being prematurely removed.
     */
    if (dry_run && (strcmp(max_slot_wal_keep_size, "-1") != 0))
    {
        report_createsub_log(PG_LOG_WARNING, PG_LOG_PRIMARY,
                             "required WAL could be removed from the publisher");
        report_createsub_log(PG_LOG_WARNING, PG_LOG_HINT,
                             "Set the configuration parameter \"%s\" to -1 to ensure that required WAL files are not prematurely removed.",
                             "max_slot_wal_keep_size");
    }
 
    pg_free(wal_level);
 
    if (failed)
        exit(1);
}

References conn, connect_database(), dbinfos, disconnect_database(), dry_run, fb(), num_dbs, pg_free(), PG_LOG_DEBUG, PG_LOG_DETAIL, PG_LOG_ERROR, PG_LOG_HINT, PG_LOG_INFO, PG_LOG_PRIMARY, PG_LOG_WARNING, pg_strdup(), PGRES_TUPLES_OK, PQclear, PQexec(), PQgetvalue, PQresultErrorMessage, PQresultStatus, report_createsub_log(), server_is_in_recovery(), LogicalRepInfos::two_phase, and wal_level.

Referenced by main().

check_subscriber()

static void check_subscriber ( const struct LogicalRepInfo *  dbinfo )

static

Definition at line 1339 of file pg_createsubscriber.c.

{
    PGconn     *conn;
    PGresult   *res;
    bool        failed = false;
 
    int         max_lrworkers;
    int         max_replorigins;
    int         max_wprocs;
 
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "checking settings on subscriber");
 
    conn = connect_database(dbinfo[0].subconninfo, true);
 
    /* The target server must be a standby */
    if (!server_is_in_recovery(conn))
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "target server must be a standby");
        disconnect_database(conn, true);
    }
 
    /*------------------------------------------------------------------------
     * Logical replication requires a few parameters to be set on subscriber.
     * Since these parameters are not a requirement for physical replication,
     * we should check it to make sure it won't fail.
     *
     * - max_active_replication_origins >= number of dbs to be converted
     * - max_logical_replication_workers >= number of dbs to be converted
     * - max_worker_processes >= 1 + number of dbs to be converted
     *------------------------------------------------------------------------
     */
    res = PQexec(conn,
                 "SELECT setting FROM pg_catalog.pg_settings WHERE name IN ("
                 "'max_logical_replication_workers', "
                 "'max_active_replication_origins', "
                 "'max_worker_processes', "
                 "'primary_slot_name') "
                 "ORDER BY name");
 
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not obtain subscriber settings: %s",
                             PQresultErrorMessage(res));
        disconnect_database(conn, true);
    }
 
    max_replorigins = atoi(PQgetvalue(res, 0, 0));
    max_lrworkers = atoi(PQgetvalue(res, 1, 0));
    max_wprocs = atoi(PQgetvalue(res, 2, 0));
    if (strcmp(PQgetvalue(res, 3, 0), "") != 0)
        primary_slot_name = pg_strdup(PQgetvalue(res, 3, 0));
 
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "subscriber: max_logical_replication_workers: %d",
                         max_lrworkers);
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "subscriber: max_active_replication_origins: %d", max_replorigins);
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "subscriber: max_worker_processes: %d", max_wprocs);
    if (primary_slot_name)
        report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                             "subscriber: primary_slot_name: %s", primary_slot_name);
 
    PQclear(res);
 
    disconnect_database(conn, false);
 
    if (max_replorigins < num_dbs)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "subscriber requires %d active replication origins, but only %d remain",
                             num_dbs, max_replorigins);
        report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                             "Increase the configuration parameter \"%s\" to at least %d.",
                             "max_active_replication_origins", num_dbs);
        failed = true;
    }
 
    if (max_lrworkers < num_dbs)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "subscriber requires %d logical replication workers, but only %d remain",
                             num_dbs, max_lrworkers);
        report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                             "Increase the configuration parameter \"%s\" to at least %d.",
                             "max_logical_replication_workers", num_dbs);
        failed = true;
    }
 
    if (max_wprocs < num_dbs + 1)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "subscriber requires %d worker processes, but only %d remain",
                             num_dbs + 1, max_wprocs);
        report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                             "Increase the configuration parameter \"%s\" to at least %d.",
                             "max_worker_processes", num_dbs + 1);
        failed = true;
    }
 
    if (failed)
        exit(1);
}

References conn, connect_database(), disconnect_database(), fb(), num_dbs, PG_LOG_DEBUG, PG_LOG_ERROR, PG_LOG_HINT, PG_LOG_INFO, PG_LOG_PRIMARY, pg_strdup(), PGRES_TUPLES_OK, PQclear, PQexec(), PQgetvalue, PQresultErrorMessage, PQresultStatus, primary_slot_name, report_createsub_log(), and server_is_in_recovery().

Referenced by main().

cleanup_objects_atexit()

static void cleanup_objects_atexit ( void  )

static

Definition at line 269 of file pg_createsubscriber.c.

{
    /* Rename the included configuration file, if necessary. */
    if (recovery_params_set)
    {
        char        conf_filename[MAXPGPATH];
        char        conf_filename_disabled[MAXPGPATH];
 
        snprintf(conf_filename, MAXPGPATH, "%s/%s", subscriber_dir,
                 INCLUDED_CONF_FILE);
        snprintf(conf_filename_disabled, MAXPGPATH, "%s/%s", subscriber_dir,
                 INCLUDED_CONF_FILE_DISABLED);
 
        if (durable_rename(conf_filename, conf_filename_disabled) != 0)
        {
            /* durable_rename() has already logged something. */
            report_createsub_log(PG_LOG_WARNING, PG_LOG_HINT,
                                 "A manual removal of the recovery parameters may be required.");
        }
    }
 
    if (success)
        return;
 
    /*
     * If the server is promoted, there is no way to use the current setup
     * again. Warn the user that a new replication setup should be done before
     * trying again.
     */
    if (recovery_ended)
    {
        report_createsub_log(PG_LOG_WARNING, PG_LOG_PRIMARY,
                             "failed after the end of recovery");
        report_createsub_log(PG_LOG_WARNING, PG_LOG_HINT,
                             "The target server cannot be used as a physical replica anymore.  "
                             "You must recreate the physical replica before continuing.");
    }
 
    for (int i = 0; i < num_dbs; i++)
    {
        struct LogicalRepInfo *dbinfo = &dbinfos.dbinfo[i];
 
        if (dbinfo->made_publication || dbinfo->made_replslot)
        {
            PGconn     *conn;
 
            conn = connect_database(dbinfo->pubconninfo, false);
            if (conn != NULL)
            {
                if (dbinfo->made_publication)
                    drop_publication(conn, dbinfo->pubname, dbinfo->dbname,
                                     &dbinfo->made_publication);
                if (dbinfo->made_replslot)
                    drop_replication_slot(conn, dbinfo, dbinfo->replslotname);
                disconnect_database(conn, false);
            }
            else
            {
                /*
                 * If a connection could not be established, inform the user
                 * that some objects were left on primary and should be
                 * removed before trying again.
                 */
                if (dbinfo->made_publication)
                {
                    report_createsub_log(PG_LOG_WARNING, PG_LOG_PRIMARY,
                                         "publication \"%s\" created in database \"%s\" on primary was left behind",
                                         dbinfo->pubname,
                                         dbinfo->dbname);
                    report_createsub_log(PG_LOG_WARNING, PG_LOG_HINT,
                                         "Drop this publication before trying again.");
                }
                if (dbinfo->made_replslot)
                {
                    report_createsub_log(PG_LOG_WARNING, PG_LOG_PRIMARY,
                                         "replication slot \"%s\" created in database \"%s\" on primary was left behind",
                                         dbinfo->replslotname,
                                         dbinfo->dbname);
                    report_createsub_log(PG_LOG_WARNING, PG_LOG_HINT,
                                         "Drop this replication slot soon to avoid retention of WAL files.");
                }
            }
        }
    }
 
    if (standby_running)
        stop_standby_server(subscriber_dir);
}

References conn, connect_database(), LogicalRepInfos::dbinfo, dbinfos, LogicalRepInfo::dbname, disconnect_database(), drop_publication(), drop_replication_slot(), durable_rename(), fb(), i, INCLUDED_CONF_FILE, INCLUDED_CONF_FILE_DISABLED, LogicalRepInfo::made_publication, LogicalRepInfo::made_replslot, MAXPGPATH, num_dbs, PG_LOG_HINT, PG_LOG_PRIMARY, PG_LOG_WARNING, LogicalRepInfo::pubconninfo, LogicalRepInfo::pubname, recovery_ended, recovery_params_set, LogicalRepInfo::replslotname, report_createsub_log(), snprintf, standby_running, stop_standby_server(), subscriber_dir, and success.

Referenced by main().

concat_conninfo_dbname()

static char * concat_conninfo_dbname ( const char *  conninfo, const char *  dbname  )

static

Definition at line 570 of file pg_createsubscriber.c.

{
    PQExpBuffer buf = createPQExpBuffer();
    char       *ret;
 
    Assert(conninfo != NULL);
 
    appendPQExpBufferStr(buf, conninfo);
    appendConnStrItem(buf, "dbname", dbname);
 
    ret = pg_strdup(buf->data);
    destroyPQExpBuffer(buf);
 
    return ret;
}

References appendConnStrItem(), appendPQExpBufferStr(), Assert, buf, createPQExpBuffer(), dbname, destroyPQExpBuffer(), fb(), and pg_strdup().

Referenced by get_publisher_databases(), and store_pub_sub_info().

connect_database()

static PGconn * connect_database ( const char *  conninfo, bool  exit_on_error  )

static

Definition at line 669 of file pg_createsubscriber.c.

{
    PGconn     *conn;
    PGresult   *res;
 
    conn = PQconnectdb(conninfo);
    if (PQstatus(conn) != CONNECTION_OK)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "connection to database failed: %s",
                             PQerrorMessage(conn));
        PQfinish(conn);
 
        if (exit_on_error)
            exit(1);
        return NULL;
    }
 
    /* Secure search_path */
    res = PQexec(conn, ALWAYS_SECURE_SEARCH_PATH_SQL);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not clear \"search_path\": %s",
                             PQresultErrorMessage(res));
        PQclear(res);
        PQfinish(conn);
 
        if (exit_on_error)
            exit(1);
        return NULL;
    }
    PQclear(res);
 
    return conn;
}

References ALWAYS_SECURE_SEARCH_PATH_SQL, conn, CONNECTION_OK, fb(), PG_LOG_ERROR, PG_LOG_PRIMARY, PGRES_TUPLES_OK, PQclear, PQconnectdb(), PQerrorMessage(), PQexec(), PQfinish(), PQresultErrorMessage, PQresultStatus, PQstatus(), and report_createsub_log().

Referenced by check_publisher(), check_subscriber(), cleanup_objects_atexit(), drop_failover_replication_slots(), drop_primary_replication_slot(), get_primary_sysid(), get_publisher_databases(), setup_publisher(), setup_recovery(), setup_subscriber(), and wait_for_end_recovery().

create_logical_replication_slot()

static char * create_logical_replication_slot ( PGconn *  conn, struct LogicalRepInfo *  dbinfo  )

static

Definition at line 1750 of file pg_createsubscriber.c.

{
    PQExpBuffer str = createPQExpBuffer();
    PGresult   *res = NULL;
    const char *slot_name = dbinfo->replslotname;
    char       *slot_name_esc;
    char       *lsn = NULL;
 
    Assert(conn != NULL);
 
    if (dry_run)
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "dry-run: would create the replication slot \"%s\" in database \"%s\" on publisher",
                             slot_name, dbinfo->dbname);
    else
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "creating the replication slot \"%s\" in database \"%s\" on publisher",
                             slot_name, dbinfo->dbname);
 
    slot_name_esc = PQescapeLiteral(conn, slot_name, strlen(slot_name));
 
    appendPQExpBuffer(str,
                      "SELECT lsn FROM pg_catalog.pg_create_logical_replication_slot(%s, 'pgoutput', false, %s, false)",
                      slot_name_esc,
                      dbinfos.two_phase ? "true" : "false");
 
    PQfreemem(slot_name_esc);
 
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "command is: %s", str->data);
 
    if (!dry_run)
    {
        res = PQexec(conn, str->data);
        if (PQresultStatus(res) != PGRES_TUPLES_OK)
        {
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "could not create replication slot \"%s\" in database \"%s\": %s",
                                 slot_name, dbinfo->dbname,
                                 PQresultErrorMessage(res));
            PQclear(res);
            destroyPQExpBuffer(str);
            return NULL;
        }
 
        lsn = pg_strdup(PQgetvalue(res, 0, 0));
        PQclear(res);
    }
 
    /* For cleanup purposes */
    dbinfo->made_replslot = true;
 
    destroyPQExpBuffer(str);
 
    return lsn;
}

References appendPQExpBuffer(), Assert, conn, createPQExpBuffer(), dbinfos, LogicalRepInfo::dbname, destroyPQExpBuffer(), dry_run, fb(), LogicalRepInfo::made_replslot, PG_LOG_DEBUG, PG_LOG_ERROR, PG_LOG_INFO, PG_LOG_PRIMARY, pg_strdup(), PGRES_TUPLES_OK, PQclear, PQescapeLiteral(), PQexec(), PQfreemem(), PQgetvalue, PQresultErrorMessage, PQresultStatus, LogicalRepInfo::replslotname, report_createsub_log(), str, and LogicalRepInfos::two_phase.

Referenced by setup_publisher().

create_publication()

static void create_publication ( PGconn *  conn, struct LogicalRepInfo *  dbinfo  )

static

Definition at line 2022 of file pg_createsubscriber.c.

{
    PQExpBuffer str = createPQExpBuffer();
    PGresult   *res;
    char       *ipubname_esc;
    char       *spubname_esc;
 
    Assert(conn != NULL);
 
    ipubname_esc = PQescapeIdentifier(conn, dbinfo->pubname, strlen(dbinfo->pubname));
    spubname_esc = PQescapeLiteral(conn, dbinfo->pubname, strlen(dbinfo->pubname));
 
    /* Check if the publication already exists */
    appendPQExpBuffer(str,
                      "SELECT 1 FROM pg_catalog.pg_publication "
                      "WHERE pubname = %s",
                      spubname_esc);
    res = PQexec(conn, str->data);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not obtain publication information: %s",
                             PQresultErrorMessage(res));
        disconnect_database(conn, true);
    }
 
    if (PQntuples(res) == 1)
    {
        /*
         * Unfortunately, if it reaches this code path, it will always fail
         * (unless you decide to change the existing publication name). That's
         * bad but it is very unlikely that the user will choose a name with
         * pg_createsubscriber_ prefix followed by the exact database oid and
         * a random number.
         */
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "publication \"%s\" already exists", dbinfo->pubname);
        report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                             "Consider renaming this publication before continuing.");
        disconnect_database(conn, true);
    }
 
    PQclear(res);
    resetPQExpBuffer(str);
 
    if (dry_run)
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "dry-run: would create publication \"%s\" in database \"%s\"",
                             dbinfo->pubname, dbinfo->dbname);
    else
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "creating publication \"%s\" in database \"%s\"",
                             dbinfo->pubname, dbinfo->dbname);
 
    appendPQExpBuffer(str, "CREATE PUBLICATION %s FOR ALL TABLES",
                      ipubname_esc);
 
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "command is: %s", str->data);
 
    if (!dry_run)
    {
        res = PQexec(conn, str->data);
        if (PQresultStatus(res) != PGRES_COMMAND_OK)
        {
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "could not create publication \"%s\" in database \"%s\": %s",
                                 dbinfo->pubname, dbinfo->dbname, PQresultErrorMessage(res));
            disconnect_database(conn, true);
        }
        PQclear(res);
    }
 
    /* For cleanup purposes */
    dbinfo->made_publication = true;
 
    PQfreemem(ipubname_esc);
    PQfreemem(spubname_esc);
    destroyPQExpBuffer(str);
}

References appendPQExpBuffer(), Assert, conn, createPQExpBuffer(), LogicalRepInfo::dbname, destroyPQExpBuffer(), disconnect_database(), dry_run, fb(), LogicalRepInfo::made_publication, PG_LOG_DEBUG, PG_LOG_ERROR, PG_LOG_HINT, PG_LOG_INFO, PG_LOG_PRIMARY, PGRES_COMMAND_OK, PGRES_TUPLES_OK, PQclear, PQescapeIdentifier(), PQescapeLiteral(), PQexec(), PQfreemem(), PQntuples, PQresultErrorMessage, PQresultStatus, LogicalRepInfo::pubname, report_createsub_log(), resetPQExpBuffer(), and str.

Referenced by setup_publisher().

create_subscription()

static void create_subscription ( PGconn *  conn, const struct LogicalRepInfo *  dbinfo  )

static

Definition at line 2232 of file pg_createsubscriber.c.

{
    PQExpBuffer str = createPQExpBuffer();
    PGresult   *res;
    char       *pubname_esc;
    char       *subname_esc;
    char       *pubconninfo_esc;
    char       *replslotname_esc;
 
    Assert(conn != NULL);
 
    pubname_esc = PQescapeIdentifier(conn, dbinfo->pubname, strlen(dbinfo->pubname));
    subname_esc = PQescapeIdentifier(conn, dbinfo->subname, strlen(dbinfo->subname));
    pubconninfo_esc = PQescapeLiteral(conn, dbinfo->pubconninfo, strlen(dbinfo->pubconninfo));
    replslotname_esc = PQescapeLiteral(conn, dbinfo->replslotname, strlen(dbinfo->replslotname));
 
    if (dry_run)
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "dry-run: would create subscription \"%s\" in database \"%s\"",
                             dbinfo->subname, dbinfo->dbname);
    else
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "creating subscription \"%s\" in database \"%s\"",
                             dbinfo->subname, dbinfo->dbname);
 
    appendPQExpBuffer(str,
                      "CREATE SUBSCRIPTION %s CONNECTION %s PUBLICATION %s "
                      "WITH (create_slot = false, enabled = false, "
                      "slot_name = %s, copy_data = false, two_phase = %s)",
                      subname_esc, pubconninfo_esc, pubname_esc, replslotname_esc,
                      dbinfos.two_phase ? "true" : "false");
 
    PQfreemem(pubname_esc);
    PQfreemem(subname_esc);
    PQfreemem(pubconninfo_esc);
    PQfreemem(replslotname_esc);
 
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "command is: %s", str->data);
 
    if (!dry_run)
    {
        res = PQexec(conn, str->data);
        if (PQresultStatus(res) != PGRES_COMMAND_OK)
        {
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "could not create subscription \"%s\" in database \"%s\": %s",
                                 dbinfo->subname, dbinfo->dbname, PQresultErrorMessage(res));
            disconnect_database(conn, true);
        }
        PQclear(res);
    }
 
    destroyPQExpBuffer(str);
}

References appendPQExpBuffer(), Assert, conn, createPQExpBuffer(), dbinfos, LogicalRepInfo::dbname, destroyPQExpBuffer(), disconnect_database(), dry_run, fb(), PG_LOG_DEBUG, PG_LOG_ERROR, PG_LOG_INFO, PG_LOG_PRIMARY, PGRES_COMMAND_OK, PQclear, PQescapeIdentifier(), PQescapeLiteral(), PQexec(), PQfreemem(), PQresultErrorMessage, PQresultStatus, LogicalRepInfo::pubconninfo, LogicalRepInfo::pubname, LogicalRepInfo::replslotname, report_createsub_log(), str, LogicalRepInfo::subname, and LogicalRepInfos::two_phase.

Referenced by setup_subscriber().

disconnect_database()

static void disconnect_database ( PGconn *  conn, bool  exit_on_error  )

static

Definition at line 711 of file pg_createsubscriber.c.

{
    Assert(conn != NULL);
 
    PQfinish(conn);
 
    if (exit_on_error)
        exit(1);
}

References Assert, conn, fb(), and PQfinish().

Referenced by check_and_drop_existing_subscriptions(), check_and_drop_publications(), check_publisher(), check_subscriber(), cleanup_objects_atexit(), create_publication(), create_subscription(), drop_existing_subscription(), drop_failover_replication_slots(), drop_primary_replication_slot(), enable_subscription(), find_publication(), generate_object_name(), get_primary_sysid(), get_publisher_databases(), server_is_in_recovery(), set_replication_progress(), setup_publisher(), setup_recovery(), setup_subscriber(), and wait_for_end_recovery().

drop_existing_subscription()

static void drop_existing_subscription ( PGconn *  conn, const char *  subname, const char *  dbname  )

static

Definition at line 1453 of file pg_createsubscriber.c.

{
    PQExpBuffer query = createPQExpBuffer();
    PGresult   *res;
 
    Assert(conn != NULL);
 
    /*
     * Construct a query string. These commands are allowed to be executed
     * within a transaction.
     */
    appendPQExpBuffer(query, "ALTER SUBSCRIPTION %s DISABLE;",
                      subname);
    appendPQExpBuffer(query, " ALTER SUBSCRIPTION %s SET (slot_name = NONE);",
                      subname);
    appendPQExpBuffer(query, " DROP SUBSCRIPTION %s;", subname);
 
    if (dry_run)
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "dry-run: would drop subscription \"%s\" in database \"%s\"",
                             subname, dbname);
    else
    {
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "dropping subscription \"%s\" in database \"%s\"",
                             subname, dbname);
 
        res = PQexec(conn, query->data);
 
        if (PQresultStatus(res) != PGRES_COMMAND_OK)
        {
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "could not drop subscription \"%s\": %s",
                                 subname, PQresultErrorMessage(res));
            disconnect_database(conn, true);
        }
 
        PQclear(res);
    }
 
    destroyPQExpBuffer(query);
}

References appendPQExpBuffer(), Assert, conn, createPQExpBuffer(), PQExpBufferData::data, dbname, destroyPQExpBuffer(), disconnect_database(), dry_run, fb(), PG_LOG_ERROR, PG_LOG_INFO, PG_LOG_PRIMARY, PGRES_COMMAND_OK, PQclear, PQexec(), PQresultErrorMessage, PQresultStatus, report_createsub_log(), and subname.

Referenced by check_and_drop_existing_subscriptions().

drop_failover_replication_slots()

static void drop_failover_replication_slots ( struct LogicalRepInfo *  dbinfo )

static

Definition at line 1704 of file pg_createsubscriber.c.

{
    PGconn     *conn;
    PGresult   *res;
 
    conn = connect_database(dbinfo[0].subconninfo, false);
    if (conn != NULL)
    {
        /* Get failover replication slot names */
        res = PQexec(conn,
                     "SELECT slot_name FROM pg_catalog.pg_replication_slots WHERE failover");
 
        if (PQresultStatus(res) == PGRES_TUPLES_OK)
        {
            /* Remove failover replication slots from subscriber */
            for (int i = 0; i < PQntuples(res); i++)
                drop_replication_slot(conn, &dbinfo[0], PQgetvalue(res, i, 0));
        }
        else
        {
            report_createsub_log(PG_LOG_WARNING, PG_LOG_PRIMARY,
                                 "could not obtain failover replication slot information: %s",
                                 PQresultErrorMessage(res));
            report_createsub_log(PG_LOG_WARNING, PG_LOG_HINT,
                                 "Drop the failover replication slots on subscriber soon to avoid retention of WAL files.");
        }
 
        PQclear(res);
        disconnect_database(conn, false);
    }
    else
    {
        report_createsub_log(PG_LOG_WARNING, PG_LOG_PRIMARY,
                             "could not drop failover replication slot");
        report_createsub_log(PG_LOG_WARNING, PG_LOG_HINT,
                             "Drop the failover replication slots on subscriber soon to avoid retention of WAL files.");
    }
}

References conn, connect_database(), disconnect_database(), drop_replication_slot(), fb(), i, PG_LOG_HINT, PG_LOG_PRIMARY, PG_LOG_WARNING, PGRES_TUPLES_OK, PQclear, PQexec(), PQgetvalue, PQntuples, PQresultErrorMessage, PQresultStatus, and report_createsub_log().

Referenced by main().

drop_primary_replication_slot()

static void drop_primary_replication_slot ( struct LogicalRepInfo *  dbinfo, const char *  slotname  )

static

Definition at line 1672 of file pg_createsubscriber.c.

{
    PGconn     *conn;
 
    /* Replication slot does not exist, do nothing */
    if (!primary_slot_name)
        return;
 
    conn = connect_database(dbinfo[0].pubconninfo, false);
    if (conn != NULL)
    {
        drop_replication_slot(conn, &dbinfo[0], slotname);
        disconnect_database(conn, false);
    }
    else
    {
        report_createsub_log(PG_LOG_WARNING, PG_LOG_PRIMARY,
                             "could not drop replication slot \"%s\" on primary",
                             slotname);
        report_createsub_log(PG_LOG_WARNING, PG_LOG_HINT,
                             "Drop this replication slot soon to avoid retention of WAL files.");
    }
}

References conn, connect_database(), disconnect_database(), drop_replication_slot(), fb(), PG_LOG_HINT, PG_LOG_PRIMARY, PG_LOG_WARNING, primary_slot_name, and report_createsub_log().

Referenced by main().

drop_publication()

static void drop_publication ( PGconn *  conn, const char *  pubname, const char *  dbname, bool *  made_publication  )

static

Definition at line 2107 of file pg_createsubscriber.c.

{
    PQExpBuffer str = createPQExpBuffer();
    PGresult   *res;
    char       *pubname_esc;
 
    Assert(conn != NULL);
 
    pubname_esc = PQescapeIdentifier(conn, pubname, strlen(pubname));
 
    if (dry_run)
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "dry-run: would drop publication \"%s\" in database \"%s\"",
                             pubname, dbname);
    else
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "dropping publication \"%s\" in database \"%s\"",
                             pubname, dbname);
 
    appendPQExpBuffer(str, "DROP PUBLICATION %s", pubname_esc);
 
    PQfreemem(pubname_esc);
 
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "command is: %s", str->data);
 
    if (!dry_run)
    {
        res = PQexec(conn, str->data);
        if (PQresultStatus(res) != PGRES_COMMAND_OK)
        {
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "could not drop publication \"%s\" in database \"%s\": %s",
                                 pubname, dbname, PQresultErrorMessage(res));
            *made_publication = false;  /* don't try again. */
 
            /*
             * Don't disconnect and exit here. This routine is used by primary
             * (cleanup publication / replication slot due to an error) and
             * subscriber (remove the replicated publications). In both cases,
             * it can continue and provide instructions for the user to remove
             * it later if cleanup fails.
             */
        }
        PQclear(res);
    }
 
    destroyPQExpBuffer(str);
}

References appendPQExpBuffer(), Assert, conn, createPQExpBuffer(), dbname, destroyPQExpBuffer(), dry_run, fb(), PG_LOG_DEBUG, PG_LOG_ERROR, PG_LOG_INFO, PG_LOG_PRIMARY, PGRES_COMMAND_OK, PQclear, PQescapeIdentifier(), PQexec(), PQfreemem(), PQresultErrorMessage, PQresultStatus, report_createsub_log(), and str.

Referenced by check_and_drop_publications(), and cleanup_objects_atexit().

drop_replication_slot()

static void drop_replication_slot ( PGconn *  conn, struct LogicalRepInfo *  dbinfo, const char *  slot_name  )

static

Definition at line 1808 of file pg_createsubscriber.c.

{
    PQExpBuffer str = createPQExpBuffer();
    char       *slot_name_esc;
    PGresult   *res;
 
    Assert(conn != NULL);
 
    if (dry_run)
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "dry-run: would drop the replication slot \"%s\" in database \"%s\"",
                             slot_name, dbinfo->dbname);
    else
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "dropping the replication slot \"%s\" in database \"%s\"",
                             slot_name, dbinfo->dbname);
 
    slot_name_esc = PQescapeLiteral(conn, slot_name, strlen(slot_name));
 
    appendPQExpBuffer(str, "SELECT pg_catalog.pg_drop_replication_slot(%s)", slot_name_esc);
 
    PQfreemem(slot_name_esc);
 
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "command is: %s", str->data);
 
    if (!dry_run)
    {
        res = PQexec(conn, str->data);
        if (PQresultStatus(res) != PGRES_TUPLES_OK)
        {
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "could not drop replication slot \"%s\" in database \"%s\": %s",
                                 slot_name, dbinfo->dbname, PQresultErrorMessage(res));
            dbinfo->made_replslot = false;  /* don't try again. */
        }
 
        PQclear(res);
    }
 
    destroyPQExpBuffer(str);
}

References appendPQExpBuffer(), Assert, conn, createPQExpBuffer(), LogicalRepInfo::dbname, destroyPQExpBuffer(), dry_run, fb(), LogicalRepInfo::made_replslot, PG_LOG_DEBUG, PG_LOG_ERROR, PG_LOG_INFO, PG_LOG_PRIMARY, PGRES_TUPLES_OK, PQclear, PQescapeLiteral(), PQexec(), PQfreemem(), PQresultErrorMessage, PQresultStatus, report_createsub_log(), and str.

Referenced by cleanup_objects_atexit(), drop_failover_replication_slots(), and drop_primary_replication_slot().

enable_subscription()

static void enable_subscription ( PGconn *  conn, const struct LogicalRepInfo *  dbinfo  )

static

Definition at line 2400 of file pg_createsubscriber.c.

{
    PQExpBuffer str = createPQExpBuffer();
    PGresult   *res;
    char       *subname;
 
    Assert(conn != NULL);
 
    subname = PQescapeIdentifier(conn, dbinfo->subname, strlen(dbinfo->subname));
 
    if (dry_run)
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "dry-run: would enable subscription \"%s\" in database \"%s\"",
                             dbinfo->subname, dbinfo->dbname);
    else
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "enabling subscription \"%s\" in database \"%s\"",
                             dbinfo->subname, dbinfo->dbname);
 
    appendPQExpBuffer(str, "ALTER SUBSCRIPTION %s ENABLE", subname);
 
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "command is: %s", str->data);
 
    if (!dry_run)
    {
        res = PQexec(conn, str->data);
        if (PQresultStatus(res) != PGRES_COMMAND_OK)
        {
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "could not enable subscription \"%s\": %s",
                                 dbinfo->subname, PQresultErrorMessage(res));
            disconnect_database(conn, true);
        }
 
        PQclear(res);
    }
 
    PQfreemem(subname);
    destroyPQExpBuffer(str);
}

References appendPQExpBuffer(), Assert, conn, createPQExpBuffer(), LogicalRepInfo::dbname, destroyPQExpBuffer(), disconnect_database(), dry_run, fb(), PG_LOG_DEBUG, PG_LOG_ERROR, PG_LOG_INFO, PG_LOG_PRIMARY, PGRES_COMMAND_OK, PQclear, PQescapeIdentifier(), PQexec(), PQfreemem(), PQresultErrorMessage, PQresultStatus, report_createsub_log(), str, LogicalRepInfo::subname, and subname.

Referenced by setup_subscriber().

find_publication()

static bool find_publication ( PGconn *  conn, const char *  pubname, const char *  dbname  )

static

Definition at line 931 of file pg_createsubscriber.c.

{
    PQExpBuffer str = createPQExpBuffer();
    PGresult   *res;
    bool        found = false;
    char       *pubname_esc = PQescapeLiteral(conn, pubname, strlen(pubname));
 
    appendPQExpBuffer(str,
                      "SELECT 1 FROM pg_catalog.pg_publication "
                      "WHERE pubname = %s",
                      pubname_esc);
    res = PQexec(conn, str->data);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not find publication \"%s\" in database \"%s\": %s",
                             pubname, dbname, PQerrorMessage(conn));
        disconnect_database(conn, true);
    }
 
    if (PQntuples(res) == 1)
        found = true;
 
    PQclear(res);
    PQfreemem(pubname_esc);
    destroyPQExpBuffer(str);
 
    return found;
}

References appendPQExpBuffer(), conn, createPQExpBuffer(), dbname, destroyPQExpBuffer(), disconnect_database(), fb(), PG_LOG_ERROR, PG_LOG_PRIMARY, PGRES_TUPLES_OK, PQclear, PQerrorMessage(), PQescapeLiteral(), PQexec(), PQfreemem(), PQntuples, PQresultStatus, report_createsub_log(), and str.

Referenced by setup_publisher().

generate_object_name()

static char * generate_object_name ( PGconn *  conn )

static

Definition at line 883 of file pg_createsubscriber.c.

{
    PGresult   *res;
    Oid         oid;
    uint32      rand;
    char       *objname;
 
    res = PQexec(conn,
                 "SELECT oid FROM pg_catalog.pg_database "
                 "WHERE datname = pg_catalog.current_database()");
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not obtain database OID: %s",
                             PQresultErrorMessage(res));
        disconnect_database(conn, true);
    }
 
    if (PQntuples(res) != 1)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not obtain database OID: got %d rows, expected %d row",
                             PQntuples(res), 1);
        disconnect_database(conn, true);
    }
 
    /* Database OID */
    oid = strtoul(PQgetvalue(res, 0, 0), NULL, 10);
 
    PQclear(res);
 
    /* Random unsigned integer */
    rand = pg_prng_uint32(&prng_state);
 
    /*
     * Build the object name. The name must not exceed NAMEDATALEN - 1. This
     * current schema uses a maximum of 40 characters (20 + 10 + 1 + 8 +
     * '\0').
     */
    objname = psprintf("pg_createsubscriber_%u_%x", oid, rand);
 
    return objname;
}

References conn, disconnect_database(), fb(), PG_LOG_ERROR, PG_LOG_PRIMARY, pg_prng_uint32(), PGRES_TUPLES_OK, PQclear, PQexec(), PQgetvalue, PQntuples, PQresultErrorMessage, PQresultStatus, prng_state, psprintf(), and report_createsub_log().

Referenced by setup_publisher().

get_base_conninfo()

static char * get_base_conninfo ( const char *  conninfo, char **  dbname  )

static

Definition at line 419 of file pg_createsubscriber.c.

{
    PQExpBuffer buf;
    PQconninfoOption *conn_opts;
    PQconninfoOption *conn_opt;
    char       *errmsg = NULL;
    char       *ret;
 
    conn_opts = PQconninfoParse(conninfo, &errmsg);
    if (conn_opts == NULL)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not parse connection string: %s", errmsg);
        PQfreemem(errmsg);
        return NULL;
    }
 
    buf = createPQExpBuffer();
    for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
    {
        if (conn_opt->val != NULL && conn_opt->val[0] != '\0')
        {
            if (strcmp(conn_opt->keyword, "dbname") == 0)
            {
                if (dbname)
                    *dbname = pg_strdup(conn_opt->val);
                continue;
            }
            appendConnStrItem(buf, conn_opt->keyword, conn_opt->val);
        }
    }
 
    ret = pg_strdup(buf->data);
 
    destroyPQExpBuffer(buf);
    PQconninfoFree(conn_opts);
 
    return ret;
}

References appendConnStrItem(), buf, createPQExpBuffer(), dbname, destroyPQExpBuffer(), errmsg, fb(), PG_LOG_ERROR, PG_LOG_PRIMARY, pg_strdup(), PQconninfoFree(), PQconninfoParse(), PQfreemem(), and report_createsub_log().

Referenced by main().

get_exec_path()

static char * get_exec_path ( const char *  argv0, const char *  progname  )

static

Definition at line 490 of file pg_createsubscriber.c.

{
    char       *versionstr;
    char       *exec_path;
    int         ret;
 
    versionstr = psprintf("%s (PostgreSQL) %s\n", progname, PG_VERSION);
    exec_path = pg_malloc(MAXPGPATH);
    ret = find_other_exec(argv0, progname, versionstr, exec_path);
 
    if (ret < 0)
    {
        char        full_path[MAXPGPATH];
 
        if (find_my_exec(argv0, full_path) < 0)
            strlcpy(full_path, progname, sizeof(full_path));
 
        if (ret == -1)
            report_createsub_fatal("program \"%s\" is needed by %s but was not found in the same directory as \"%s\"",
                                   progname, "pg_createsubscriber", full_path);
        else
            report_createsub_fatal("program \"%s\" was found by \"%s\" but was not the same version as %s",
                                   progname, full_path, "pg_createsubscriber");
    }
 
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "%s path is:  %s", progname, exec_path);
 
    return exec_path;
}

References argv0, exec_path, fb(), find_my_exec(), find_other_exec(), MAXPGPATH, PG_LOG_DEBUG, PG_LOG_PRIMARY, pg_malloc(), progname, psprintf(), report_createsub_fatal(), report_createsub_log(), and strlcpy().

Referenced by main().

get_primary_sysid()

static uint64 get_primary_sysid ( const char *  conninfo )

static

Definition at line 726 of file pg_createsubscriber.c.

{
    PGconn     *conn;
    PGresult   *res;
    uint64      sysid;
 
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "getting system identifier from publisher");
 
    conn = connect_database(conninfo, true);
 
    res = PQexec(conn, "SELECT system_identifier FROM pg_catalog.pg_control_system()");
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not get system identifier: %s",
                             PQresultErrorMessage(res));
        disconnect_database(conn, true);
    }
    if (PQntuples(res) != 1)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not get system identifier: got %d rows, expected %d row",
                             PQntuples(res), 1);
        disconnect_database(conn, true);
    }
 
    sysid = strtou64(PQgetvalue(res, 0, 0), NULL, 10);
 
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "system identifier is %" PRIu64 " on publisher", sysid);
 
    PQclear(res);
    disconnect_database(conn, false);
 
    return sysid;
}

References conn, connect_database(), disconnect_database(), fb(), PG_LOG_ERROR, PG_LOG_INFO, PG_LOG_PRIMARY, PGRES_TUPLES_OK, PQclear, PQexec(), PQgetvalue, PQntuples, PQresultErrorMessage, PQresultStatus, and report_createsub_log().

Referenced by main().

get_publisher_databases()

static void get_publisher_databases ( struct CreateSubscriberOptions *  opt, bool  dbnamespecified  )

static

Definition at line 2448 of file pg_createsubscriber.c.

{
    PGconn     *conn;
    PGresult   *res;
 
    /* If a database name was specified, just connect to it. */
    if (dbnamespecified)
        conn = connect_database(opt->pub_conninfo_str, true);
    else
    {
        /* Otherwise, try postgres first and then template1. */
        char       *conninfo;
 
        conninfo = concat_conninfo_dbname(opt->pub_conninfo_str, "postgres");
        conn = connect_database(conninfo, false);
        pg_free(conninfo);
        if (!conn)
        {
            conninfo = concat_conninfo_dbname(opt->pub_conninfo_str, "template1");
            conn = connect_database(conninfo, true);
            pg_free(conninfo);
        }
    }
 
    res = PQexec(conn, "SELECT datname FROM pg_database WHERE datistemplate = false AND datallowconn AND datconnlimit <> -2 ORDER BY 1");
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not obtain a list of databases: %s",
                             PQresultErrorMessage(res));
        PQclear(res);
        disconnect_database(conn, true);
    }
 
    for (int i = 0; i < PQntuples(res); i++)
    {
        const char *dbname = PQgetvalue(res, i, 0);
 
        simple_string_list_append(&opt->database_names, dbname);
 
        /* Increment num_dbs to reflect multiple --database options */
        num_dbs++;
    }
 
    PQclear(res);
    disconnect_database(conn, false);
}

References concat_conninfo_dbname(), conn, connect_database(), CreateSubscriberOptions::database_names, dbname, disconnect_database(), fb(), i, num_dbs, pg_free(), PG_LOG_ERROR, PG_LOG_PRIMARY, PGRES_TUPLES_OK, PQclear, PQexec(), PQgetvalue, PQntuples, PQresultErrorMessage, PQresultStatus, CreateSubscriberOptions::pub_conninfo_str, report_createsub_log(), and simple_string_list_append().

Referenced by main().

get_standby_sysid()

static uint64 get_standby_sysid ( const char *  datadir )

static

Definition at line 770 of file pg_createsubscriber.c.

{
    ControlFileData *cf;
    bool        crc_ok;
    uint64      sysid;
 
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "getting system identifier from subscriber");
 
    cf = get_controlfile(datadir, &crc_ok);
    if (!crc_ok)
        report_createsub_fatal("control file appears to be corrupt");
 
    sysid = cf->system_identifier;
 
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "system identifier is %" PRIu64 " on subscriber", sysid);
 
    pg_free(cf);
 
    return sysid;
}

References datadir, fb(), get_controlfile(), pg_free(), PG_LOG_INFO, PG_LOG_PRIMARY, report_createsub_fatal(), and report_createsub_log().

Referenced by main().

get_sub_conninfo()

static char * get_sub_conninfo ( const struct CreateSubscriberOptions *  opt )

static

Definition at line 464 of file pg_createsubscriber.c.

{
    PQExpBuffer buf = createPQExpBuffer();
    char       *ret;
 
    appendConnStrItem(buf, "port", opt->sub_port);
#if !defined(WIN32)
    appendConnStrItem(buf, "host", opt->socket_dir);
#endif
    if (opt->sub_username != NULL)
        appendConnStrItem(buf, "user", opt->sub_username);
    appendConnStrItem(buf, "fallback_application_name", progname);
 
    ret = pg_strdup(buf->data);
 
    destroyPQExpBuffer(buf);
 
    return ret;
}

References appendConnStrItem(), buf, createPQExpBuffer(), destroyPQExpBuffer(), fb(), pg_strdup(), progname, CreateSubscriberOptions::socket_dir, CreateSubscriberOptions::sub_port, and CreateSubscriberOptions::sub_username.

Referenced by main().

internal_log_file_write()

static void internal_log_file_write ( enum pg_log_level  level, enum pg_log_part  part, const char *pg_restrict  fmt, va_list  args  )

static

Definition at line 1082 of file pg_createsubscriber.c.

{
    Assert(internal_log_file_fp);
 
    /* Do nothing if log level is too low. */
    if (level < __pg_log_level)
        return;
 
    /* Add prefix based on the log part and log level */
    switch (part)
    {
        case PG_LOG_PRIMARY:
            switch (level)
            {
                case PG_LOG_ERROR:
                    fprintf(internal_log_file_fp, _("error: "));
                    break;
                case PG_LOG_WARNING:
                    fprintf(internal_log_file_fp, _("warning: "));
                    break;
                default:
                    break;
            }
            break;
        case PG_LOG_DETAIL:
            fprintf(internal_log_file_fp, _("detail: "));
            break;
        case PG_LOG_HINT:
            fprintf(internal_log_file_fp, _("hint: "));
            break;
    }
 
    vfprintf(internal_log_file_fp, _(fmt), args);
 
    fprintf(internal_log_file_fp, "\n");
    fflush(internal_log_file_fp);
}

References _, __pg_log_level, Assert, fb(), fprintf, internal_log_file_fp, PG_LOG_DETAIL, PG_LOG_ERROR, PG_LOG_HINT, PG_LOG_PRIMARY, PG_LOG_WARNING, and vfprintf.

Referenced by report_createsub_log_v().

logfile_open()

static FILE * logfile_open ( const char *  filename, const char *  mode  )

static

Definition at line 1125 of file pg_createsubscriber.c.

{
    FILE       *fh;
 
    fh = fopen(filename, mode);
 
    if (!fh)
        report_createsub_fatal("could not open log file \"%s\": %m",
                               filename);
 
    return fh;
}

References fb(), filename, mode, and report_createsub_fatal().

Referenced by main().

main()

int main	(	int	argc,
		char **	argv
	)

Definition at line 2498 of file pg_createsubscriber.c.

{
    static struct option long_options[] =
    {
        {"all", no_argument, NULL, 'a'},
        {"database", required_argument, NULL, 'd'},
        {"pgdata", required_argument, NULL, 'D'},
        {"logdir", required_argument, NULL, 'l'},
        {"dry-run", no_argument, NULL, 'n'},
        {"subscriber-port", required_argument, NULL, 'p'},
        {"publisher-server", required_argument, NULL, 'P'},
        {"socketdir", required_argument, NULL, 's'},
        {"recovery-timeout", required_argument, NULL, 't'},
        {"enable-two-phase", no_argument, NULL, 'T'},
        {"subscriber-username", required_argument, NULL, 'U'},
        {"verbose", no_argument, NULL, 'v'},
        {"version", no_argument, NULL, 'V'},
        {"help", no_argument, NULL, '?'},
        {"config-file", required_argument, NULL, 1},
        {"publication", required_argument, NULL, 2},
        {"replication-slot", required_argument, NULL, 3},
        {"subscription", required_argument, NULL, 4},
        {"clean", required_argument, NULL, 5},
        {NULL, 0, NULL, 0}
    };
 
    struct CreateSubscriberOptions opt = {0};
 
    int         c;
    int         option_index;
 
    char       *pub_base_conninfo;
    char       *sub_base_conninfo;
    char       *dbname_conninfo = NULL;
 
    uint64      pub_sysid;
    uint64      sub_sysid;
    struct stat statbuf;
 
    char       *consistent_lsn;
 
    char        pidfile[MAXPGPATH];
 
    pg_logging_init(argv[0]);
    pg_logging_set_level(PG_LOG_WARNING);
    progname = get_progname(argv[0]);
    set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_basebackup"));
 
    if (argc > 1)
    {
        if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
        {
            usage();
            exit(0);
        }
        else if (strcmp(argv[1], "-V") == 0
                 || strcmp(argv[1], "--version") == 0)
        {
            puts("pg_createsubscriber (PostgreSQL) " PG_VERSION);
            exit(0);
        }
    }
 
    /* Default settings */
    subscriber_dir = NULL;
    opt.config_file = NULL;
    opt.log_dir = NULL;
    opt.pub_conninfo_str = NULL;
    opt.socket_dir = NULL;
    opt.sub_port = DEFAULT_SUB_PORT;
    opt.sub_username = NULL;
    opt.two_phase = false;
    opt.database_names = (SimpleStringList)
    {
        0
    };
    opt.recovery_timeout = 0;
    opt.all_dbs = false;
 
    /*
     * Don't allow it to be run as root. It uses pg_ctl which does not allow
     * it either.
     */
#ifndef WIN32
    if (geteuid() == 0)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "cannot be executed by \"root\"");
        report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                             "You must run %s as the PostgreSQL superuser.",
                             progname);
        exit(1);
    }
#endif
 
    get_restricted_token();
 
    while ((c = getopt_long(argc, argv, "ad:D:l:np:P:s:t:TU:v",
                            long_options, &option_index)) != -1)
    {
        switch (c)
        {
            case 'a':
                opt.all_dbs = true;
                break;
            case 'd':
                if (!simple_string_list_member(&opt.database_names, optarg))
                {
                    simple_string_list_append(&opt.database_names, optarg);
                    num_dbs++;
                }
                else
                    report_createsub_fatal("database \"%s\" specified more than once for -d/--database", optarg);
                break;
            case 'D':
                subscriber_dir = pg_strdup(optarg);
                canonicalize_path(subscriber_dir);
                break;
            case 'l':
                opt.log_dir = pg_strdup(optarg);
                canonicalize_path(opt.log_dir);
                break;
            case 'n':
                dry_run = true;
                break;
            case 'p':
                opt.sub_port = pg_strdup(optarg);
                break;
            case 'P':
                opt.pub_conninfo_str = pg_strdup(optarg);
                break;
            case 's':
                opt.socket_dir = pg_strdup(optarg);
                canonicalize_path(opt.socket_dir);
                break;
            case 't':
                opt.recovery_timeout = atoi(optarg);
                break;
            case 'T':
                opt.two_phase = true;
                break;
            case 'U':
                opt.sub_username = pg_strdup(optarg);
                break;
            case 'v':
                pg_logging_increase_verbosity();
                break;
            case 1:
                opt.config_file = pg_strdup(optarg);
                break;
            case 2:
                if (!simple_string_list_member(&opt.pub_names, optarg))
                {
                    simple_string_list_append(&opt.pub_names, optarg);
                    num_pubs++;
                }
                else
                    report_createsub_fatal("publication \"%s\" specified more than once for --publication", optarg);
                break;
            case 3:
                if (!simple_string_list_member(&opt.replslot_names, optarg))
                {
                    simple_string_list_append(&opt.replslot_names, optarg);
                    num_replslots++;
                }
                else
                    report_createsub_fatal("replication slot \"%s\" specified more than once for --replication-slot", optarg);
                break;
            case 4:
                if (!simple_string_list_member(&opt.sub_names, optarg))
                {
                    simple_string_list_append(&opt.sub_names, optarg);
                    num_subs++;
                }
                else
                    report_createsub_fatal("subscription \"%s\" specified more than once for --subscription", optarg);
                break;
            case 5:
                if (!simple_string_list_member(&opt.objecttypes_to_clean, optarg))
                    simple_string_list_append(&opt.objecttypes_to_clean, optarg);
                else
                    report_createsub_fatal("object type \"%s\" specified more than once for --clean", optarg);
                break;
            default:
                /* getopt_long already emitted a complaint */
                report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                                     "Try \"%s --help\" for more information.",
                                     progname);
                exit(1);
        }
    }
 
    /* Validate that --all is not used with incompatible options */
    if (opt.all_dbs)
    {
        char       *bad_switch = NULL;
 
        if (num_dbs > 0)
            bad_switch = "--database";
        else if (num_pubs > 0)
            bad_switch = "--publication";
        else if (num_replslots > 0)
            bad_switch = "--replication-slot";
        else if (num_subs > 0)
            bad_switch = "--subscription";
 
        if (bad_switch)
        {
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "options %s and %s cannot be used together",
                                 bad_switch, "-a/--all");
            report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                                 "Try \"%s --help\" for more information.",
                                 progname);
            exit(1);
        }
    }
 
    /* Any non-option arguments? */
    if (optind < argc)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "too many command-line arguments (first is \"%s\")",
                             argv[optind]);
        report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                             "Try \"%s --help\" for more information.", progname);
        exit(1);
    }
 
    /* Required arguments */
    if (subscriber_dir == NULL)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "no subscriber data directory specified");
        report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                             "Try \"%s --help\" for more information.", progname);
        exit(1);
    }
 
    /* If socket directory is not provided, use the current directory */
    if (opt.socket_dir == NULL)
    {
        char        cwd[MAXPGPATH];
 
        if (!getcwd(cwd, MAXPGPATH))
            report_createsub_fatal("could not determine current directory");
        opt.socket_dir = pg_strdup(cwd);
        canonicalize_path(opt.socket_dir);
    }
 
    /*
     * Parse connection string. Build a base connection string that might be
     * reused by multiple databases.
     */
    if (opt.pub_conninfo_str == NULL)
    {
        /*
         * TODO use primary_conninfo (if available) from subscriber and
         * extract publisher connection string. Assume that there are
         * identical entries for physical and logical replication. If there is
         * not, we would fail anyway.
         */
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "no publisher connection string specified");
        report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                             "Try \"%s --help\" for more information.", progname);
        exit(1);
    }
 
    if (opt.log_dir != NULL)
    {
        char       *internal_log_file;
 
        umask(PG_MODE_MASK_OWNER);
 
        /*
         * Set mask based on PGDATA permissions, needed for the creation of
         * the output directories with correct permissions, similar with
         * pg_ctl and pg_upgrade.
         *
         * Don't error here if the data directory cannot be stat'd. Upcoming
         * checks for the data directory would raise the fatal error later.
         */
        if (GetDataDirectoryCreatePerm(subscriber_dir))
            umask(pg_mode_mask);
 
        make_output_dirs(opt.log_dir);
        internal_log_file = psprintf("%s/%s", logdir, INTERNAL_LOG_FILE_NAME);
 
        /* logfile_open() will exit if there is an error */
        internal_log_file_fp = logfile_open(internal_log_file, "a");
        pg_free(internal_log_file);
    }
 
    if (dry_run)
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "Executing in dry-run mode.\n"
                             "The target directory will not be modified.");
 
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "validating publisher connection string");
    pub_base_conninfo = get_base_conninfo(opt.pub_conninfo_str,
                                          &dbname_conninfo);
    if (pub_base_conninfo == NULL)
        exit(1);
 
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "validating subscriber connection string");
    sub_base_conninfo = get_sub_conninfo(&opt);
 
    /*
     * Fetch all databases from the source (publisher) and treat them as if
     * the user specified has multiple --database options, one for each source
     * database.
     */
    if (opt.all_dbs)
    {
        bool        dbnamespecified = (dbname_conninfo != NULL);
 
        get_publisher_databases(&opt, dbnamespecified);
    }
 
    if (opt.database_names.head == NULL)
    {
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "no database was specified");
 
        /*
         * Try to obtain the dbname from the publisher conninfo. If dbname
         * parameter is not available, error out.
         */
        if (dbname_conninfo)
        {
            simple_string_list_append(&opt.database_names, dbname_conninfo);
            num_dbs++;
 
            report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                                 "database name \"%s\" was extracted from the publisher connection string",
                                 dbname_conninfo);
        }
        else
        {
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "no database name specified");
            report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                                 "Try \"%s --help\" for more information.",
                                 progname);
            exit(1);
        }
    }
 
    /* Number of object names must match number of databases */
    if (num_pubs > 0 && num_pubs != num_dbs)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "wrong number of publication names specified");
        report_createsub_log(PG_LOG_ERROR, PG_LOG_DETAIL,
                             "The number of specified publication names (%d) must match the number of specified database names (%d).",
                             num_pubs, num_dbs);
        exit(1);
    }
    if (num_subs > 0 && num_subs != num_dbs)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "wrong number of subscription names specified");
        report_createsub_log(PG_LOG_ERROR, PG_LOG_DETAIL,
                             "The number of specified subscription names (%d) must match the number of specified database names (%d).",
                             num_subs, num_dbs);
        exit(1);
    }
    if (num_replslots > 0 && num_replslots != num_dbs)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "wrong number of replication slot names specified");
        report_createsub_log(PG_LOG_ERROR, PG_LOG_DETAIL,
                             "The number of specified replication slot names (%d) must match the number of specified database names (%d).",
                             num_replslots, num_dbs);
        exit(1);
    }
 
    /* Verify the object types specified for removal from the subscriber */
    for (SimpleStringListCell *cell = opt.objecttypes_to_clean.head; cell; cell = cell->next)
    {
        if (pg_strcasecmp(cell->val, "publications") == 0)
            dbinfos.objecttypes_to_clean |= OBJECTTYPE_PUBLICATIONS;
        else
        {
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "invalid object type \"%s\" specified for %s",
                                 cell->val, "--clean");
            report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                                 "The valid value is: \"%s\"", "publications");
            exit(1);
        }
    }
 
    /* Get the absolute path of pg_ctl and pg_resetwal on the subscriber */
    pg_ctl_path = get_exec_path(argv[0], "pg_ctl");
    pg_resetwal_path = get_exec_path(argv[0], "pg_resetwal");
 
    /* Rudimentary check for a data directory */
    check_data_directory(subscriber_dir);
 
    dbinfos.two_phase = opt.two_phase;
 
    /*
     * Store database information for publisher and subscriber. It should be
     * called before atexit() because its return is used in the
     * cleanup_objects_atexit().
     */
    dbinfos.dbinfo = store_pub_sub_info(&opt, pub_base_conninfo, sub_base_conninfo);
 
    /* Register a function to clean up objects in case of failure */
    atexit(cleanup_objects_atexit);
 
    /*
     * Check if the subscriber data directory has the same system identifier
     * than the publisher data directory.
     */
    pub_sysid = get_primary_sysid(dbinfos.dbinfo[0].pubconninfo);
    sub_sysid = get_standby_sysid(subscriber_dir);
    if (pub_sysid != sub_sysid)
        report_createsub_fatal("subscriber data directory is not a copy of the source database cluster");
 
    /* Subscriber PID file */
    snprintf(pidfile, MAXPGPATH, "%s/postmaster.pid", subscriber_dir);
 
    /*
     * The standby server must not be running. If the server is started under
     * service manager and pg_createsubscriber stops it, the service manager
     * might react to this action and start the server again. Therefore,
     * refuse to proceed if the server is running to avoid possible failures.
     */
    if (stat(pidfile, &statbuf) == 0)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "standby server is running");
        report_createsub_log(PG_LOG_ERROR, PG_LOG_HINT,
                             "Stop the standby server and try again.");
        exit(1);
    }
 
    /*
     * Start a short-lived standby server with temporary parameters (provided
     * by command-line options). The goal is to avoid connections during the
     * transformation steps.
     */
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "starting the standby server with command-line options");
    start_standby_server(&opt, true, false);
 
    /* Check if the standby server is ready for logical replication */
    check_subscriber(dbinfos.dbinfo);
 
    /* Check if the primary server is ready for logical replication */
    check_publisher(dbinfos.dbinfo);
 
    /*
     * Stop the target server. The recovery process requires that the server
     * reaches a consistent state before targeting the recovery stop point.
     * Make sure a consistent state is reached (stop the target server
     * guarantees it) *before* creating the replication slots in
     * setup_publisher().
     */
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "stopping the subscriber");
    stop_standby_server(subscriber_dir);
 
    /* Create the required objects for each database on publisher */
    consistent_lsn = setup_publisher(dbinfos.dbinfo);
 
    /* Write the required recovery parameters */
    setup_recovery(dbinfos.dbinfo, subscriber_dir, consistent_lsn);
 
    /*
     * Start subscriber so the recovery parameters will take effect. Wait
     * until accepting connections. We don't want to start logical replication
     * during setup.
     */
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "starting the subscriber");
    start_standby_server(&opt, true, true);
 
    /* Waiting the subscriber to be promoted */
    wait_for_end_recovery(dbinfos.dbinfo[0].subconninfo, &opt);
 
    /*
     * Create the subscription for each database on subscriber. It does not
     * enable it immediately because it needs to adjust the replication start
     * point to the LSN reported by setup_publisher().  It also cleans up
     * publications created by this tool and replication to the standby.
     */
    setup_subscriber(dbinfos.dbinfo, consistent_lsn);
 
    /* Remove primary_slot_name if it exists on primary */
    drop_primary_replication_slot(dbinfos.dbinfo, primary_slot_name);
 
    /* Remove failover replication slots if they exist on subscriber */
    drop_failover_replication_slots(dbinfos.dbinfo);
 
    /* Stop the subscriber */
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "stopping the subscriber");
    stop_standby_server(subscriber_dir);
 
    /* Change system identifier from subscriber */
    modify_subscriber_sysid(&opt);
 
    success = true;
 
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "Done!");
 
    return 0;
}

References CreateSubscriberOptions::all_dbs, canonicalize_path(), check_data_directory(), check_publisher(), check_subscriber(), cleanup_objects_atexit(), CreateSubscriberOptions::config_file, CreateSubscriberOptions::database_names, LogicalRepInfos::dbinfo, dbinfos, DEFAULT_SUB_PORT, drop_failover_replication_slots(), drop_primary_replication_slot(), dry_run, fb(), get_base_conninfo(), get_exec_path(), get_primary_sysid(), get_progname(), get_publisher_databases(), get_restricted_token(), get_standby_sysid(), get_sub_conninfo(), GetDataDirectoryCreatePerm(), getopt_long(), SimpleStringList::head, internal_log_file_fp, INTERNAL_LOG_FILE_NAME, CreateSubscriberOptions::log_dir, logdir, logfile_open(), make_output_dirs(), MAXPGPATH, modify_subscriber_sysid(), no_argument, num_dbs, num_pubs, num_replslots, num_subs, OBJECTTYPE_PUBLICATIONS, CreateSubscriberOptions::objecttypes_to_clean, LogicalRepInfos::objecttypes_to_clean, optarg, optind, pg_ctl_path, pg_free(), PG_LOG_DETAIL, PG_LOG_ERROR, PG_LOG_HINT, PG_LOG_INFO, PG_LOG_PRIMARY, PG_LOG_WARNING, pg_logging_increase_verbosity(), pg_logging_init(), pg_logging_set_level(), pg_mode_mask, PG_MODE_MASK_OWNER, pg_resetwal_path, pg_strcasecmp(), pg_strdup(), PG_TEXTDOMAIN, primary_slot_name, progname, psprintf(), CreateSubscriberOptions::pub_conninfo_str, CreateSubscriberOptions::pub_names, LogicalRepInfo::pubconninfo, CreateSubscriberOptions::recovery_timeout, CreateSubscriberOptions::replslot_names, report_createsub_fatal(), report_createsub_log(), required_argument, set_pglocale_pgservice(), setup_publisher(), setup_recovery(), setup_subscriber(), simple_string_list_append(), simple_string_list_member(), snprintf, CreateSubscriberOptions::socket_dir, start_standby_server(), stat, stop_standby_server(), store_pub_sub_info(), CreateSubscriberOptions::sub_names, CreateSubscriberOptions::sub_port, CreateSubscriberOptions::sub_username, LogicalRepInfo::subconninfo, subscriber_dir, success, CreateSubscriberOptions::two_phase, LogicalRepInfos::two_phase, usage(), and wait_for_end_recovery().

make_output_dirs()

static void make_output_dirs ( const char *  log_basedir )

static

Definition at line 1139 of file pg_createsubscriber.c.

{
    char        timestamp[128];
    struct timeval tval;
    time_t      now;
    struct tm   tmbuf;
    int         len;
 
    /* Generate timestamp */
    gettimeofday(&tval, NULL);
    now = tval.tv_sec;
 
    strftime(timestamp, sizeof(timestamp), "%Y%m%dT%H%M%S",
             localtime_r(&now, &tmbuf));
 
    /* Append milliseconds */
    snprintf(timestamp + strlen(timestamp),
             sizeof(timestamp) - strlen(timestamp), ".%03u",
             (unsigned int) (tval.tv_usec / 1000));
 
    /* Build timestamp directory path */
    len = snprintf(logdir, MAXPGPATH, "%s/%s", log_basedir, timestamp);
 
    if (len >= MAXPGPATH)
        report_createsub_fatal("directory path for log files is too long");
 
    /* Create base directory (ignore if exists) */
    if (mkdir(log_basedir, pg_dir_create_mode) < 0 && errno != EEXIST)
        report_createsub_fatal("could not create directory \"%s\": %m", log_basedir);
 
    /* Create a timestamp-named subdirectory under the base directory */
    if (mkdir(logdir, pg_dir_create_mode) < 0)
        report_createsub_fatal("could not create directory \"%s\": %m", logdir);
}

References fb(), gettimeofday(), len, logdir, MAXPGPATH, mkdir, now(), pg_dir_create_mode, report_createsub_fatal(), snprintf, and tm.

Referenced by main().

modify_subscriber_sysid()

static void modify_subscriber_sysid ( const struct CreateSubscriberOptions *  opt )

static

Definition at line 799 of file pg_createsubscriber.c.

{
    ControlFileData *cf;
    bool        crc_ok;
    struct timeval tv;
 
    char       *out_file;
    char       *cmd_str;
 
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "modifying system identifier of subscriber");
 
    cf = get_controlfile(subscriber_dir, &crc_ok);
    if (!crc_ok)
        report_createsub_fatal("control file appears to be corrupt");
 
    /*
     * Select a new system identifier.
     *
     * XXX this code was extracted from BootStrapXLOG().
     */
    gettimeofday(&tv, NULL);
    cf->system_identifier = ((uint64) tv.tv_sec) << 32;
    cf->system_identifier |= ((uint64) tv.tv_usec) << 12;
    cf->system_identifier |= getpid() & 0xFFF;
 
    if (dry_run)
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "dry-run: would set system identifier to %" PRIu64 " on subscriber",
                             cf->system_identifier);
    else
    {
        update_controlfile(subscriber_dir, cf, true);
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "system identifier is %" PRIu64 " on subscriber",
                             cf->system_identifier);
    }
 
    if (dry_run)
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "dry-run: would run pg_resetwal on the subscriber");
    else
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "running pg_resetwal on the subscriber");
 
    /*
     * Redirecting the output to the logfile if specified. Since the output
     * would be very short, around one line, we do not provide a separate file
     * for it; it's done as a part of the server log.
     */
    if (opt->log_dir)
        out_file = psprintf("%s/%s", logdir, SERVER_LOG_FILE_NAME);
    else
        out_file = DEVNULL;
 
    cmd_str = psprintf("\"%s\" -D \"%s\" >> \"%s\"", pg_resetwal_path,
                       subscriber_dir, out_file);
    if (opt->log_dir)
        pg_free(out_file);
 
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "pg_resetwal command is: %s", cmd_str);
 
    if (!dry_run)
    {
        int         rc = system(cmd_str);
 
        if (rc == 0)
            report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                                 "successfully reset WAL on the subscriber");
        else
            report_createsub_fatal("could not reset WAL on subscriber: %s", wait_result_to_str(rc));
    }
 
    pg_free(cf);
    pg_free(cmd_str);
}

References DEVNULL, dry_run, fb(), get_controlfile(), gettimeofday(), CreateSubscriberOptions::log_dir, logdir, pg_free(), PG_LOG_DEBUG, PG_LOG_INFO, PG_LOG_PRIMARY, pg_resetwal_path, psprintf(), report_createsub_fatal(), report_createsub_log(), SERVER_LOG_FILE_NAME, subscriber_dir, update_controlfile(), and wait_result_to_str().

Referenced by main().

pg_ctl_status()

static void pg_ctl_status ( const char *  pg_ctl_cmd, int  rc  )

static

Definition at line 1856 of file pg_createsubscriber.c.

{
    if (rc != 0)
    {
        if (WIFEXITED(rc))
        {
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "pg_ctl failed with exit code %d",
                                 WEXITSTATUS(rc));
        }
        else if (WIFSIGNALED(rc))
        {
#if defined(WIN32)
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "pg_ctl was terminated by exception 0x%X",
                                 WTERMSIG(rc));
            report_createsub_log(PG_LOG_ERROR, PG_LOG_DETAIL,
                                 "See C include file \"ntstatus.h\" for a description of the hexadecimal value.");
#else
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "pg_ctl was terminated by signal %d: %s",
                                 WTERMSIG(rc), pg_strsignal(WTERMSIG(rc)));
#endif
        }
        else
        {
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "pg_ctl exited with unrecognized status %d", rc);
        }
 
        report_createsub_log(PG_LOG_ERROR, PG_LOG_DETAIL,
                             "The failed command was: %s", pg_ctl_cmd);
        exit(1);
    }
}

References fb(), PG_LOG_DETAIL, PG_LOG_ERROR, PG_LOG_PRIMARY, pg_strsignal(), report_createsub_log(), WEXITSTATUS, WIFEXITED, WIFSIGNALED, and WTERMSIG.

Referenced by start_standby_server(), and stop_standby_server().

report_createsub_fatal()

static void report_createsub_fatal ( const char *pg_restrict  fmt,   ...  )

static

Definition at line 242 of file pg_createsubscriber.c.

{
    va_list     args;
 
    va_start(args, fmt);
 
    report_createsub_log_v(PG_LOG_ERROR, PG_LOG_PRIMARY, fmt, args);
 
    va_end(args);
 
    exit(1);
}

References fb(), PG_LOG_ERROR, PG_LOG_PRIMARY, and report_createsub_log_v().

Referenced by check_data_directory(), get_exec_path(), get_standby_sysid(), logfile_open(), main(), make_output_dirs(), modify_subscriber_sysid(), setup_recovery(), and wait_for_end_recovery().

report_createsub_log()

static void report_createsub_log ( enum pg_log_level  level, enum pg_log_part  part, const char *pg_restrict  fmt,   ...  )

static

Definition at line 226 of file pg_createsubscriber.c.

{
    va_list     args;
 
    va_start(args, fmt);
 
    report_createsub_log_v(level, part, fmt, args);
 
    va_end(args);
}

References fb(), and report_createsub_log_v().

Referenced by check_and_drop_existing_subscriptions(), check_and_drop_publications(), check_data_directory(), check_publisher(), check_subscriber(), cleanup_objects_atexit(), connect_database(), create_logical_replication_slot(), create_publication(), create_subscription(), drop_existing_subscription(), drop_failover_replication_slots(), drop_primary_replication_slot(), drop_publication(), drop_replication_slot(), enable_subscription(), find_publication(), generate_object_name(), get_base_conninfo(), get_exec_path(), get_primary_sysid(), get_publisher_databases(), get_standby_sysid(), main(), modify_subscriber_sysid(), pg_ctl_status(), server_is_in_recovery(), set_replication_progress(), setup_publisher(), setup_recovery(), start_standby_server(), stop_standby_server(), store_pub_sub_info(), and wait_for_end_recovery().

report_createsub_log_v()

static void report_createsub_log_v ( enum pg_log_level  level, enum pg_log_part  part, const char *pg_restrict  fmt, va_list  args  )

static

Definition at line 206 of file pg_createsubscriber.c.

{
    int         save_errno = errno;
 
    if (internal_log_file_fp != NULL)
    {
        /* Output to both stderr and the log file */
        va_list     arg_cpy;
 
        va_copy(arg_cpy, args);
        internal_log_file_write(level, part, fmt, arg_cpy);
        va_end(arg_cpy);
        /* Restore errno in case internal_log_file_write changed it */
        errno = save_errno;
    }
    pg_log_generic_v(level, part, fmt, args);
}

References fb(), internal_log_file_fp, internal_log_file_write(), and pg_log_generic_v().

Referenced by report_createsub_fatal(), and report_createsub_log().

server_is_in_recovery()

static bool server_is_in_recovery ( PGconn *  conn )

static

Definition at line 1058 of file pg_createsubscriber.c.

{
    PGresult   *res;
    int         ret;
 
    res = PQexec(conn, "SELECT pg_catalog.pg_is_in_recovery()");
 
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not obtain recovery progress: %s",
                             PQresultErrorMessage(res));
        disconnect_database(conn, true);
    }
 
 
    ret = strcmp("t", PQgetvalue(res, 0, 0));
 
    PQclear(res);
 
    return ret == 0;
}

References conn, disconnect_database(), fb(), PG_LOG_ERROR, PG_LOG_PRIMARY, PGRES_TUPLES_OK, PQclear, PQexec(), PQgetvalue, PQresultErrorMessage, PQresultStatus, and report_createsub_log().

Referenced by check_publisher(), check_subscriber(), and wait_for_end_recovery().

set_replication_progress()

static void set_replication_progress ( PGconn *  conn, const struct LogicalRepInfo *  dbinfo, const char *  lsn  )

static

Definition at line 2299 of file pg_createsubscriber.c.

{
    PQExpBuffer str = createPQExpBuffer();
    PGresult   *res;
    Oid         suboid;
    char       *subname;
    char       *dbname;
    char       *originname;
    char       *lsnstr;
 
    Assert(conn != NULL);
 
    subname = PQescapeLiteral(conn, dbinfo->subname, strlen(dbinfo->subname));
    dbname = PQescapeLiteral(conn, dbinfo->dbname, strlen(dbinfo->dbname));
 
    appendPQExpBuffer(str,
                      "SELECT s.oid FROM pg_catalog.pg_subscription s "
                      "INNER JOIN pg_catalog.pg_database d ON (s.subdbid = d.oid) "
                      "WHERE s.subname = %s AND d.datname = %s",
                      subname, dbname);
 
    res = PQexec(conn, str->data);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not obtain subscription OID: %s",
                             PQresultErrorMessage(res));
        disconnect_database(conn, true);
    }
 
    if (PQntuples(res) != 1 && !dry_run)
    {
        report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                             "could not obtain subscription OID: got %d rows, expected %d row",
                             PQntuples(res), 1);
        disconnect_database(conn, true);
    }
 
    if (dry_run)
    {
        suboid = InvalidOid;
        lsnstr = psprintf("%X/%08X", LSN_FORMAT_ARGS((XLogRecPtr) InvalidXLogRecPtr));
    }
    else
    {
        suboid = strtoul(PQgetvalue(res, 0, 0), NULL, 10);
        lsnstr = psprintf("%s", lsn);
    }
 
    PQclear(res);
 
    /*
     * The origin name is defined as pg_%u. %u is the subscription OID. See
     * ApplyWorkerMain().
     */
    originname = psprintf("pg_%u", suboid);
 
    if (dry_run)
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "dry-run: would set the replication progress (node name \"%s\", LSN %s) in database \"%s\"",
                             originname, lsnstr, dbinfo->dbname);
    else
        report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                             "setting the replication progress (node name \"%s\", LSN %s) in database \"%s\"",
                             originname, lsnstr, dbinfo->dbname);
 
    resetPQExpBuffer(str);
    appendPQExpBuffer(str,
                      "SELECT pg_catalog.pg_replication_origin_advance('%s', '%s')",
                      originname, lsnstr);
 
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "command is: %s", str->data);
 
    if (!dry_run)
    {
        res = PQexec(conn, str->data);
        if (PQresultStatus(res) != PGRES_TUPLES_OK)
        {
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "could not set replication progress for subscription \"%s\": %s",
                                 dbinfo->subname, PQresultErrorMessage(res));
            disconnect_database(conn, true);
        }
        PQclear(res);
    }
 
    PQfreemem(subname);
    PQfreemem(dbname);
    pg_free(originname);
    pg_free(lsnstr);
    destroyPQExpBuffer(str);
}

References appendPQExpBuffer(), Assert, conn, createPQExpBuffer(), LogicalRepInfo::dbname, dbname, destroyPQExpBuffer(), disconnect_database(), dry_run, fb(), InvalidOid, InvalidXLogRecPtr, LSN_FORMAT_ARGS, pg_free(), PG_LOG_DEBUG, PG_LOG_ERROR, PG_LOG_INFO, PG_LOG_PRIMARY, PGRES_TUPLES_OK, PQclear, PQescapeLiteral(), PQexec(), PQfreemem(), PQgetvalue, PQntuples, PQresultErrorMessage, PQresultStatus, psprintf(), report_createsub_log(), resetPQExpBuffer(), str, LogicalRepInfo::subname, and subname.

Referenced by setup_subscriber().

setup_publisher()

static char * setup_publisher ( struct LogicalRepInfo *  dbinfo )

static

Definition at line 968 of file pg_createsubscriber.c.

{
    char       *lsn = NULL;
 
    pg_prng_seed(&prng_state, (uint64) (getpid() ^ time(NULL)));
 
    for (int i = 0; i < num_dbs; i++)
    {
        PGconn     *conn;
        char       *genname = NULL;
 
        conn = connect_database(dbinfo[i].pubconninfo, true);
 
        /*
         * If an object name was not specified as command-line options, assign
         * a generated object name. The replication slot has a different rule.
         * The subscription name is assigned to the replication slot name if
         * no replication slot is specified. It follows the same rule as
         * CREATE SUBSCRIPTION.
         */
        if (num_pubs == 0 || num_subs == 0 || num_replslots == 0)
            genname = generate_object_name(conn);
        if (num_pubs == 0)
            dbinfo[i].pubname = pg_strdup(genname);
        if (num_subs == 0)
            dbinfo[i].subname = pg_strdup(genname);
        if (num_replslots == 0)
            dbinfo[i].replslotname = pg_strdup(dbinfo[i].subname);
 
        if (find_publication(conn, dbinfo[i].pubname, dbinfo[i].dbname))
        {
            /* Reuse existing publication on publisher. */
            report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                                 "use existing publication \"%s\" in database \"%s\"",
                                 dbinfo[i].pubname, dbinfo[i].dbname);
            /* Don't remove pre-existing publication if an error occurs. */
            dbinfo[i].made_publication = false;
        }
        else
        {
            /*
             * Create publication on publisher. This step should be executed
             * *before* promoting the subscriber to avoid any transactions
             * between consistent LSN and the new publication rows (such
             * transactions wouldn't see the new publication rows resulting in
             * an error).
             */
            create_publication(conn, &dbinfo[i]);
        }
 
        /* Create replication slot on publisher */
        if (lsn)
            pg_free(lsn);
        lsn = create_logical_replication_slot(conn, &dbinfo[i]);
        if (lsn == NULL && !dry_run)
            exit(1);
 
        /*
         * Since we are using the LSN returned by the last replication slot as
         * recovery_target_lsn, this LSN is ahead of the current WAL position
         * and the recovery waits until the publisher writes a WAL record to
         * reach the target and ends the recovery. On idle systems, this wait
         * time is unpredictable and could lead to failure in promoting the
         * subscriber. To avoid that, insert a harmless WAL record.
         */
        if (i == num_dbs - 1 && !dry_run)
        {
            PGresult   *res;
 
            res = PQexec(conn, "SELECT pg_log_standby_snapshot()");
            if (PQresultStatus(res) != PGRES_TUPLES_OK)
            {
                report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                     "could not write an additional WAL record: %s",
                                     PQresultErrorMessage(res));
                disconnect_database(conn, true);
            }
            PQclear(res);
        }
 
        disconnect_database(conn, false);
    }
 
    return lsn;
}

References conn, connect_database(), create_logical_replication_slot(), create_publication(), dbname, disconnect_database(), dry_run, fb(), find_publication(), generate_object_name(), i, LogicalRepInfo::made_publication, num_dbs, num_pubs, num_replslots, num_subs, pg_free(), PG_LOG_ERROR, PG_LOG_INFO, PG_LOG_PRIMARY, pg_prng_seed(), pg_strdup(), PGRES_TUPLES_OK, PQclear, PQexec(), PQresultErrorMessage, PQresultStatus, prng_state, LogicalRepInfo::pubname, LogicalRepInfo::replslotname, report_createsub_log(), LogicalRepInfo::subname, and subname.

Referenced by main().

setup_recovery()

static void setup_recovery ( const struct LogicalRepInfo *  dbinfo, const char *  datadir, const char *  lsn  )

static

Definition at line 1577 of file pg_createsubscriber.c.

{
    PGconn     *conn;
    PQExpBuffer recoveryconfcontents;
 
    /*
     * Despite of the recovery parameters will be written to the subscriber,
     * use a publisher connection. The primary_conninfo is generated using the
     * connection settings.
     */
    conn = connect_database(dbinfo[0].pubconninfo, true);
 
    /*
     * Write recovery parameters.
     *
     * The subscriber is not running yet. In dry run mode, the recovery
     * parameters *won't* be written. An invalid LSN is used for printing
     * purposes. Additional recovery parameters are added here. It avoids
     * unexpected behavior such as end of recovery as soon as a consistent
     * state is reached (recovery_target) and failure due to multiple recovery
     * targets (name, time, xid, LSN).
     */
    recoveryconfcontents = GenerateRecoveryConfig(conn, NULL, NULL);
    appendPQExpBufferStr(recoveryconfcontents, "recovery_target = ''\n");
    appendPQExpBufferStr(recoveryconfcontents,
                         "recovery_target_timeline = 'latest'\n");
 
    /*
     * Set recovery_target_inclusive = false to avoid reapplying the
     * transaction committed at 'lsn' after subscription is enabled. This is
     * because the provided 'lsn' is also used as the replication start point
     * for the subscription. So, the server can send the transaction committed
     * at that 'lsn' after replication is started which can lead to applying
     * the same transaction twice if we keep recovery_target_inclusive = true.
     */
    appendPQExpBufferStr(recoveryconfcontents,
                         "recovery_target_inclusive = false\n");
    appendPQExpBufferStr(recoveryconfcontents,
                         "recovery_target_action = promote\n");
    appendPQExpBufferStr(recoveryconfcontents, "recovery_target_name = ''\n");
    appendPQExpBufferStr(recoveryconfcontents, "recovery_target_time = ''\n");
    appendPQExpBufferStr(recoveryconfcontents, "recovery_target_xid = ''\n");
 
    if (dry_run)
    {
        appendPQExpBufferStr(recoveryconfcontents, "# dry run mode\n");
        appendPQExpBuffer(recoveryconfcontents,
                          "recovery_target_lsn = '%X/%08X'\n",
                          LSN_FORMAT_ARGS((XLogRecPtr) InvalidXLogRecPtr));
    }
    else
    {
        appendPQExpBuffer(recoveryconfcontents, "recovery_target_lsn = '%s'\n",
                          lsn);
    }
 
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "recovery parameters:\n%s", recoveryconfcontents->data);
 
    if (!dry_run)
    {
        char        conf_filename[MAXPGPATH];
        FILE       *fd;
 
        /* Write the recovery parameters to INCLUDED_CONF_FILE */
        snprintf(conf_filename, MAXPGPATH, "%s/%s", datadir,
                 INCLUDED_CONF_FILE);
        fd = fopen(conf_filename, "w");
        if (fd == NULL)
            report_createsub_fatal("could not open file \"%s\": %m", conf_filename);
 
        if (fwrite(recoveryconfcontents->data, recoveryconfcontents->len, 1, fd) != 1)
            report_createsub_fatal("could not write to file \"%s\": %m", conf_filename);
 
        fclose(fd);
        recovery_params_set = true;
 
        /* Include conditionally the recovery parameters. */
        resetPQExpBuffer(recoveryconfcontents);
        appendPQExpBufferStr(recoveryconfcontents,
                             "include_if_exists '" INCLUDED_CONF_FILE "'\n");
        WriteRecoveryConfig(conn, datadir, recoveryconfcontents);
    }
 
    disconnect_database(conn, false);
}

References appendPQExpBuffer(), appendPQExpBufferStr(), conn, connect_database(), PQExpBufferData::data, datadir, disconnect_database(), dry_run, fb(), fd(), GenerateRecoveryConfig(), INCLUDED_CONF_FILE, InvalidXLogRecPtr, PQExpBufferData::len, LSN_FORMAT_ARGS, MAXPGPATH, PG_LOG_DEBUG, PG_LOG_PRIMARY, recovery_params_set, recoveryconfcontents, report_createsub_fatal(), report_createsub_log(), resetPQExpBuffer(), snprintf, and WriteRecoveryConfig().

Referenced by main().

setup_subscriber()

static void setup_subscriber ( struct LogicalRepInfo *  dbinfo, const char *  consistent_lsn  )

static

Definition at line 1541 of file pg_createsubscriber.c.

{
    for (int i = 0; i < num_dbs; i++)
    {
        PGconn     *conn;
 
        /* Connect to subscriber. */
        conn = connect_database(dbinfo[i].subconninfo, true);
 
        /*
         * We don't need the pre-existing subscriptions on the newly formed
         * subscriber. They can connect to other publisher nodes and either
         * get some unwarranted data or can lead to ERRORs in connecting to
         * such nodes.
         */
        check_and_drop_existing_subscriptions(conn, &dbinfo[i]);
 
        /* Check and drop the required publications in the given database. */
        check_and_drop_publications(conn, &dbinfo[i]);
 
        create_subscription(conn, &dbinfo[i]);
 
        /* Set the replication progress to the correct LSN */
        set_replication_progress(conn, &dbinfo[i], consistent_lsn);
 
        /* Enable subscription */
        enable_subscription(conn, &dbinfo[i]);
 
        disconnect_database(conn, false);
    }
}

References check_and_drop_existing_subscriptions(), check_and_drop_publications(), conn, connect_database(), create_subscription(), disconnect_database(), enable_subscription(), fb(), i, num_dbs, and set_replication_progress().

Referenced by main().

start_standby_server()

static void start_standby_server ( const struct CreateSubscriberOptions *  opt, bool  restricted_access, bool  restrict_logical_worker  )

static

Definition at line 1893 of file pg_createsubscriber.c.

{
    PQExpBuffer pg_ctl_cmd = createPQExpBuffer();
    int         rc;
 
    appendPQExpBuffer(pg_ctl_cmd, "\"%s\" start -D ", pg_ctl_path);
    appendShellString(pg_ctl_cmd, subscriber_dir);
    appendPQExpBufferStr(pg_ctl_cmd, " -s -o \"-c sync_replication_slots=off\"");
 
    /* Prevent unintended slot invalidation */
    appendPQExpBufferStr(pg_ctl_cmd, " -o \"-c idle_replication_slot_timeout=0\"");
 
    if (restricted_access)
    {
        appendPQExpBuffer(pg_ctl_cmd, " -o \"-p %s\"", opt->sub_port);
#if !defined(WIN32)
 
        /*
         * An empty listen_addresses list means the server does not listen on
         * any IP interfaces; only Unix-domain sockets can be used to connect
         * to the server. Prevent external connections to minimize the chance
         * of failure.
         */
        appendPQExpBufferStr(pg_ctl_cmd, " -o \"-c listen_addresses='' -c unix_socket_permissions=0700");
        if (opt->socket_dir)
            appendPQExpBuffer(pg_ctl_cmd, " -c unix_socket_directories='%s'",
                              opt->socket_dir);
        appendPQExpBufferChar(pg_ctl_cmd, '"');
#endif
    }
    if (opt->config_file != NULL)
        appendPQExpBuffer(pg_ctl_cmd, " -o \"-c config_file=%s\"",
                          opt->config_file);
 
    /* Suppress to start logical replication if requested */
    if (restrict_logical_worker)
        appendPQExpBufferStr(pg_ctl_cmd, " -o \"-c max_logical_replication_workers=0\"");
 
    if (opt->log_dir)
        appendPQExpBuffer(pg_ctl_cmd, " -l \"%s/%s\"", logdir, SERVER_LOG_FILE_NAME);
 
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "pg_ctl command is: %s", pg_ctl_cmd->data);
    rc = system(pg_ctl_cmd->data);
    pg_ctl_status(pg_ctl_cmd->data, rc);
    standby_running = true;
    destroyPQExpBuffer(pg_ctl_cmd);
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "server was started");
}

References appendPQExpBuffer(), appendPQExpBufferChar(), appendPQExpBufferStr(), appendShellString(), CreateSubscriberOptions::config_file, createPQExpBuffer(), destroyPQExpBuffer(), fb(), CreateSubscriberOptions::log_dir, logdir, pg_ctl_path, pg_ctl_status(), PG_LOG_DEBUG, PG_LOG_INFO, PG_LOG_PRIMARY, report_createsub_log(), SERVER_LOG_FILE_NAME, CreateSubscriberOptions::socket_dir, standby_running, CreateSubscriberOptions::sub_port, and subscriber_dir.

Referenced by main().

stop_standby_server()

static void stop_standby_server ( const char *  datadir )

static

Definition at line 1946 of file pg_createsubscriber.c.

{
    char       *pg_ctl_cmd;
    int         rc;
 
    pg_ctl_cmd = psprintf("\"%s\" stop -D \"%s\" -s", pg_ctl_path,
                          datadir);
    report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                         "pg_ctl command is: %s", pg_ctl_cmd);
    rc = system(pg_ctl_cmd);
    pg_ctl_status(pg_ctl_cmd, rc);
    standby_running = false;
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "server was stopped");
}

References datadir, fb(), pg_ctl_path, pg_ctl_status(), PG_LOG_DEBUG, PG_LOG_INFO, PG_LOG_PRIMARY, psprintf(), report_createsub_log(), and standby_running.

Referenced by cleanup_objects_atexit(), main(), and wait_for_end_recovery().

store_pub_sub_info()

static struct LogicalRepInfo * store_pub_sub_info ( const struct CreateSubscriberOptions *  opt, const char *  pub_base_conninfo, const char *  sub_base_conninfo  )

static

Definition at line 594 of file pg_createsubscriber.c.

{
    struct LogicalRepInfo *dbinfo;
    SimpleStringListCell *pubcell = NULL;
    SimpleStringListCell *subcell = NULL;
    SimpleStringListCell *replslotcell = NULL;
    int         i = 0;
 
    dbinfo = pg_malloc_array(struct LogicalRepInfo, num_dbs);
 
    if (num_pubs > 0)
        pubcell = opt->pub_names.head;
    if (num_subs > 0)
        subcell = opt->sub_names.head;
    if (num_replslots > 0)
        replslotcell = opt->replslot_names.head;
 
    for (SimpleStringListCell *cell = opt->database_names.head; cell; cell = cell->next)
    {
        char       *conninfo;
 
        /* Fill publisher attributes */
        conninfo = concat_conninfo_dbname(pub_base_conninfo, cell->val);
        dbinfo[i].pubconninfo = conninfo;
        dbinfo[i].dbname = cell->val;
        if (num_pubs > 0)
            dbinfo[i].pubname = pubcell->val;
        else
            dbinfo[i].pubname = NULL;
        if (num_replslots > 0)
            dbinfo[i].replslotname = replslotcell->val;
        else
            dbinfo[i].replslotname = NULL;
        dbinfo[i].made_replslot = false;
        dbinfo[i].made_publication = false;
        /* Fill subscriber attributes */
        conninfo = concat_conninfo_dbname(sub_base_conninfo, cell->val);
        dbinfo[i].subconninfo = conninfo;
        if (num_subs > 0)
            dbinfo[i].subname = subcell->val;
        else
            dbinfo[i].subname = NULL;
        /* Other fields will be filled later */
 
        report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                             "publisher(%d): publication: %s ; replication slot: %s ; connection string: %s", i,
                             dbinfo[i].pubname ? dbinfo[i].pubname : "(auto)",
                             dbinfo[i].replslotname ? dbinfo[i].replslotname : "(auto)",
                             dbinfo[i].pubconninfo);
        report_createsub_log(PG_LOG_DEBUG, PG_LOG_PRIMARY,
                             "subscriber(%d): subscription: %s ; connection string: %s, two_phase: %s", i,
                             dbinfo[i].subname ? dbinfo[i].subname : "(auto)",
                             dbinfo[i].subconninfo,
                             dbinfos.two_phase ? "true" : "false");
 
        if (num_pubs > 0)
            pubcell = pubcell->next;
        if (num_subs > 0)
            subcell = subcell->next;
        if (num_replslots > 0)
            replslotcell = replslotcell->next;
 
        i++;
    }
 
    return dbinfo;
}

References concat_conninfo_dbname(), CreateSubscriberOptions::database_names, dbinfos, LogicalRepInfo::dbname, fb(), SimpleStringList::head, i, LogicalRepInfo::made_publication, LogicalRepInfo::made_replslot, num_dbs, num_pubs, num_replslots, num_subs, PG_LOG_DEBUG, PG_LOG_PRIMARY, pg_malloc_array, CreateSubscriberOptions::pub_names, LogicalRepInfo::pubconninfo, LogicalRepInfo::pubname, CreateSubscriberOptions::replslot_names, LogicalRepInfo::replslotname, report_createsub_log(), CreateSubscriberOptions::sub_names, LogicalRepInfo::subconninfo, LogicalRepInfo::subname, subname, and LogicalRepInfos::two_phase.

Referenced by main().

usage()

static void usage ( void  )

static

Definition at line 359 of file pg_createsubscriber.c.

{
    printf(_("%s creates a new logical replica from a standby server.\n\n"),
           progname);
    printf(_("Usage:\n"));
    printf(_("  %s [OPTION]...\n"), progname);
    printf(_("\nOptions:\n"));
    printf(_("  -a, --all                       create subscriptions for all databases except template\n"
             "                                  databases and databases that don't allow connections\n"));
    printf(_("  -d, --database=DBNAME           database in which to create a subscription\n"));
    printf(_("  -D, --pgdata=DATADIR            location for the subscriber data directory\n"));
    printf(_("  -l, --logdir=LOGDIR             location for the log directory\n"));
    printf(_("  -n, --dry-run                   dry run, just show what would be done\n"));
    printf(_("  -p, --subscriber-port=PORT      subscriber port number (default %s)\n"), DEFAULT_SUB_PORT);
    printf(_("  -P, --publisher-server=CONNSTR  publisher connection string\n"));
    printf(_("  -s, --socketdir=DIR             socket directory to use (default current dir.)\n"));
    printf(_("  -t, --recovery-timeout=SECS     seconds to wait for recovery to end\n"));
    printf(_("  -T, --enable-two-phase          enable two-phase commit for all subscriptions\n"));
    printf(_("  -U, --subscriber-username=NAME  user name for subscriber connection\n"));
    printf(_("  -v, --verbose                   output verbose messages\n"));
    printf(_("      --clean=OBJECTTYPE          drop all objects of the specified type from specified\n"
             "                                  databases on the subscriber; accepts: \"%s\"\n"), "publications");
    printf(_("      --config-file=FILENAME      use specified main server configuration\n"
             "                                  file when running target cluster\n"));
    printf(_("      --publication=NAME          publication name\n"));
    printf(_("      --replication-slot=NAME     replication slot name\n"));
    printf(_("      --subscription=NAME         subscription name\n"));
    printf(_("  -V, --version                   output version information, then exit\n"));
    printf(_("  -?, --help                      show this help, then exit\n"));
    printf(_("\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
    printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
}

References _, DEFAULT_SUB_PORT, fb(), printf, and progname.

Referenced by main().

wait_for_end_recovery()

static void wait_for_end_recovery ( const char *  conninfo, const struct CreateSubscriberOptions *  opt  )

static

Definition at line 1972 of file pg_createsubscriber.c.

{
    PGconn     *conn;
    bool        ready = false;
    int         timer = 0;
 
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "waiting for the target server to reach the consistent state");
 
    conn = connect_database(conninfo, true);
 
    for (;;)
    {
        /* Did the recovery process finish? We're done if so. */
        if (dry_run || !server_is_in_recovery(conn))
        {
            ready = true;
            recovery_ended = true;
            break;
        }
 
        /* Bail out after recovery_timeout seconds if this option is set */
        if (opt->recovery_timeout > 0 && timer >= opt->recovery_timeout)
        {
            stop_standby_server(subscriber_dir);
            report_createsub_log(PG_LOG_ERROR, PG_LOG_PRIMARY,
                                 "recovery timed out");
            disconnect_database(conn, true);
        }
 
        /* Keep waiting */
        pg_usleep(WAIT_INTERVAL * USECS_PER_SEC);
        timer += WAIT_INTERVAL;
    }
 
    disconnect_database(conn, false);
 
    if (!ready)
        report_createsub_fatal("server did not end recovery");
 
    report_createsub_log(PG_LOG_INFO, PG_LOG_PRIMARY,
                         "target server reached the consistent state");
    report_createsub_log(PG_LOG_INFO, PG_LOG_HINT,
                         "If pg_createsubscriber fails after this point, you must recreate the physical replica before continuing.");
}

References conn, connect_database(), disconnect_database(), dry_run, fb(), PG_LOG_ERROR, PG_LOG_HINT, PG_LOG_INFO, PG_LOG_PRIMARY, pg_usleep(), recovery_ended, CreateSubscriberOptions::recovery_timeout, report_createsub_fatal(), report_createsub_log(), server_is_in_recovery(), stop_standby_server(), subscriber_dir, USECS_PER_SEC, and WAIT_INTERVAL.

Referenced by main().

Variable Documentation

dbinfos

struct LogicalRepInfos dbinfos

static

Definition at line 176 of file pg_createsubscriber.c.

Referenced by check_and_drop_publications(), check_publisher(), cleanup_objects_atexit(), create_logical_replication_slot(), create_subscription(), get_db_infos(), main(), and store_pub_sub_info().

dry_run

bool dry_run = false

static

Definition at line 172 of file pg_createsubscriber.c.

Referenced by check_and_drop_publications(), check_publisher(), copy_file(), create_logical_replication_slot(), create_publication(), create_subscription(), create_target_dir(), create_target_symlink(), debug_reconstruction(), drop_existing_subscription(), drop_publication(), drop_replication_slot(), enable_subscription(), main(), modify_subscriber_sysid(), open_target_file(), reconstruct_from_incremental_file(), remove_target_dir(), remove_target_file(), remove_target_symlink(), set_replication_progress(), setup_publisher(), setup_recovery(), sync_target_dir(), truncate_target_file(), wait_for_end_recovery(), write_reconstructed_file(), and write_target_range().

internal_log_file_fp

FILE* internal_log_file_fp = NULL

static

Definition at line 187 of file pg_createsubscriber.c.

Referenced by internal_log_file_write(), main(), and report_createsub_log_v().

logdir

char logdir[MAXPGPATH]

static

Definition at line 188 of file pg_createsubscriber.c.

Referenced by main(), make_output_dirs(), modify_subscriber_sysid(), and start_standby_server().

num_dbs

int num_dbs = 0

static

Definition at line 177 of file pg_createsubscriber.c.

Referenced by check_publisher(), check_subscriber(), cleanup_objects_atexit(), get_publisher_databases(), main(), setup_publisher(), setup_subscriber(), and store_pub_sub_info().

num_pubs

int num_pubs = 0

static

Definition at line 178 of file pg_createsubscriber.c.

Referenced by main(), setup_publisher(), and store_pub_sub_info().

num_replslots

int num_replslots = 0

static

Definition at line 180 of file pg_createsubscriber.c.

Referenced by main(), setup_publisher(), and store_pub_sub_info().

num_subs

int num_subs = 0

static

Definition at line 179 of file pg_createsubscriber.c.

Referenced by main(), setup_publisher(), and store_pub_sub_info().

pg_ctl_path

char* pg_ctl_path = NULL

static

Definition at line 184 of file pg_createsubscriber.c.

Referenced by main(), main(), start_standby_server(), and stop_standby_server().

pg_resetwal_path

char* pg_resetwal_path = NULL

static

Definition at line 185 of file pg_createsubscriber.c.

Referenced by main(), and modify_subscriber_sysid().

primary_slot_name

char* primary_slot_name = NULL

static

Definition at line 171 of file pg_createsubscriber.c.

Referenced by check_subscriber(), drop_primary_replication_slot(), and main().

prng_state

pg_prng_state prng_state

static

Definition at line 182 of file pg_createsubscriber.c.

Referenced by generate_object_name(), and setup_publisher().

progname

const char* progname

static

Definition at line 169 of file pg_createsubscriber.c.

Referenced by get_exec_path(), get_sub_conninfo(), main(), and usage().

recovery_ended

bool recovery_ended = false

static

Definition at line 195 of file pg_createsubscriber.c.

Referenced by cleanup_objects_atexit(), and wait_for_end_recovery().

recovery_params_set

bool recovery_params_set = false

static

Definition at line 197 of file pg_createsubscriber.c.

Referenced by cleanup_objects_atexit(), and setup_recovery().

standby_running

bool standby_running = false

static

Definition at line 196 of file pg_createsubscriber.c.

Referenced by cleanup_objects_atexit(), start_standby_server(), and stop_standby_server().

subscriber_dir

char* subscriber_dir = NULL

static

Definition at line 193 of file pg_createsubscriber.c.

Referenced by cleanup_objects_atexit(), main(), modify_subscriber_sysid(), start_standby_server(), and wait_for_end_recovery().

success

bool success = false

static

Definition at line 174 of file pg_createsubscriber.c.

Referenced by cleanup_objects_atexit(), and main().