6 Star 3 Fork 5

Gitee 极速下载/repmgr

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
文件
此仓库是为了提升国内下载速度的镜像仓库,每日同步一次。 原始仓库: https://github.com/2ndQuadrant/repmgr/
克隆/下载
repmgrd.c 25.75 KB
一键复制 编辑 原始数据 按行查看 历史
Ian Barwick 提交于 2022-05-11 13:57 +08:00 . repmgr: fix error message
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090
/*
* repmgrd.c - Replication manager daemon
*
* Copyright (c) EnterpriseDB Corporation, 2010-2021
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <stdio.h>
#include <signal.h>
#include <stdlib.h>
#include <sys/stat.h>
#include "repmgr.h"
#include "repmgrd.h"
#include "repmgrd-physical.h"
#include "configfile.h"
#include "voting.h"
#define OPT_HELP 1
static char *config_file = NULL;
static bool verbose = false;
char pid_file[MAXPGPATH];
static bool daemonize = true;
static bool show_pid_file = false;
static bool no_pid_file = false;
t_node_info local_node_info = T_NODE_INFO_INITIALIZER;
PGconn *local_conn = NULL;
/* Collate command line errors here for friendlier reporting */
static ItemList cli_errors = {NULL, NULL};
bool startup_event_logged = false;
MonitoringState monitoring_state = MS_NORMAL;
instr_time degraded_monitoring_start;
/*
* Record receipt of SIGHUP; will cause configuration file to be reread
* at the appropriate point in the main loop.
*/
volatile sig_atomic_t got_SIGHUP = false;
static void show_help(void);
static void show_usage(void);
static void daemonize_process(void);
static void check_and_create_pid_file(const char *pid_file);
static void start_monitoring(void);
#ifndef WIN32
static void setup_event_handlers(void);
static void handle_sighup(SIGNAL_ARGS);
#endif
int calculate_elapsed(instr_time start_time);
void update_registration(PGconn *conn);
void terminate(int retval);
int
main(int argc, char **argv)
{
int optindex;
int c;
char cli_log_level[MAXLEN] = "";
bool cli_monitoring_history = false;
RecordStatus record_status;
ExtensionStatus extension_status = REPMGR_UNKNOWN;
t_extension_versions extversions = T_EXTENSION_VERSIONS_INITIALIZER;
FILE *fd;
static struct option long_options[] =
{
/* general options */
{"help", no_argument, NULL, OPT_HELP},
{"version", no_argument, NULL, 'V'},
/* configuration options */
{"config-file", required_argument, NULL, 'f'},
/* daemon options */
{"daemonize-short", optional_argument, NULL, 'd'},
{"daemonize", optional_argument, NULL, OPT_DAEMONIZE},
{"pid-file", required_argument, NULL, 'p'},
{"show-pid-file", no_argument, NULL, 's'},
{"no-pid-file", no_argument, NULL, OPT_NO_PID_FILE},
/* logging options */
{"log-level", required_argument, NULL, 'L'},
{"verbose", no_argument, NULL, 'v'},
/* legacy options */
{"monitoring-history", no_argument, NULL, 'm'},
{NULL, 0, NULL, 0}
};
set_progname(argv[0]);
/* Disallow running as root */
if (geteuid() == 0)
{
fprintf(stderr,
_("%s: cannot be run as root\n"
"Please log in (using, e.g., \"su\") as the "
"(unprivileged) user that owns "
"the data directory.\n"
),
progname());
exit(1);
}
srand(time(NULL));
memset(pid_file, 0, MAXPGPATH);
while ((c = getopt_long(argc, argv, "?Vf:L:vdp:sm", long_options, &optindex)) != -1)
{
switch (c)
{
/* general options */
case '?':
/* Actual help option given */
if (strcmp(argv[optind - 1], "-?") == 0)
{
show_help();
exit(SUCCESS);
}
/* unknown option reported by getopt */
goto unknown_option;
break;
case OPT_HELP:
show_help();
exit(SUCCESS);
case 'V':
/*
* in contrast to repmgr3 and earlier, we only display the
* repmgr version as it's not specific to a particular
* PostgreSQL version
*/
printf("%s %s\n", progname(), REPMGR_VERSION);
exit(SUCCESS);
/* configuration options */
case 'f':
config_file = optarg;
break;
/* daemon options */
case 'd':
daemonize = true;
break;
case OPT_DAEMONIZE:
daemonize = parse_bool(optarg, "-d/--daemonize", &cli_errors);
break;
case 'p':
strncpy(pid_file, optarg, MAXPGPATH);
break;
case 's':
show_pid_file = true;
break;
case OPT_NO_PID_FILE:
no_pid_file = true;
break;
/* logging options */
/* -L/--log-level */
case 'L':
{
int detected_cli_log_level = detect_log_level(optarg);
if (detected_cli_log_level != -1)
{
strncpy(cli_log_level, optarg, MAXLEN);
}
else
{
PQExpBufferData invalid_log_level;
initPQExpBuffer(&invalid_log_level);
appendPQExpBuffer(&invalid_log_level,
_("invalid log level \"%s\" provided"),
optarg);
item_list_append(&cli_errors, invalid_log_level.data);
termPQExpBuffer(&invalid_log_level);
}
break;
}
case 'v':
verbose = true;
break;
/* legacy options */
case 'm':
cli_monitoring_history = true;
break;
default:
unknown_option:
show_usage();
exit(ERR_BAD_CONFIG);
}
}
/* Exit here already if errors in command line options found */
if (cli_errors.head != NULL)
{
exit_with_cli_errors(&cli_errors, NULL);
}
startup_event_logged = false;
/*
* Tell the logger we're a daemon - this will ensure any output logged
* before the logger is initialized will be formatted correctly
*/
logger_output_mode = OM_DAEMON;
/*
* Parse the configuration file, if provided (if no configuration file was
* provided, an attempt will be made to find one in one of the default
* locations). If no configuration file is available, or it can't be parsed
* parse_config() will abort anyway, with an appropriate message.
*/
load_config(config_file, verbose, false, argv[0]);
/* Determine pid file location, unless --no-pid-file supplied */
if (no_pid_file == false)
{
if (config_file_options.repmgrd_pid_file[0] != '\0')
{
if (pid_file[0] != '\0')
{
log_warning(_("\"repmgrd_pid_file\" will be overridden by --pid-file"));
}
else
{
strncpy(pid_file, config_file_options.repmgrd_pid_file, MAXPGPATH);
}
}
/* no pid file provided - determine location */
if (pid_file[0] == '\0')
{
/* packagers: if feasible, patch PID file path into "package_pid_file" */
char package_pid_file[MAXPGPATH] = "";
if (package_pid_file[0] != '\0')
{
maxpath_snprintf(pid_file, "%s", package_pid_file);
}
else
{
const char *tmpdir = getenv("TMPDIR");
if (!tmpdir)
tmpdir = "/tmp";
maxpath_snprintf(pid_file, "%s/repmgrd.pid", tmpdir);
}
}
}
else
{
/* --no-pid-file supplied - overwrite any value provided with --pid-file ... */
memset(pid_file, 0, MAXPGPATH);
}
/* If --show-pid-file supplied, output the location (if set) and exit */
if (show_pid_file == true)
{
printf("%s\n", pid_file);
exit(SUCCESS);
}
/* Some configuration file items can be overridden by command line options */
/*
* Command-line parameter -L/--log-level overrides any setting in config
* file
*/
if (*cli_log_level != '\0')
{
strncpy(config_file_options.log_level, cli_log_level, MAXLEN);
}
/*
* -m/--monitoring-history, if provided, will override repmgr.conf's
* monitoring_history; this is for backwards compatibility as it's
* possible this may be baked into various startup scripts.
*/
if (cli_monitoring_history == true)
{
config_file_options.monitoring_history = true;
}
fd = freopen("/dev/null", "r", stdin);
if (fd == NULL)
{
fprintf(stderr, "error reopening stdin to \"/dev/null\":\n %s\n",
strerror(errno));
}
fd = freopen("/dev/null", "w", stdout);
if (fd == NULL)
{
fprintf(stderr, "error reopening stdout to \"/dev/null\":\n %s\n",
strerror(errno));
}
logger_init(&config_file_options, progname());
log_notice(_("repmgrd (%s %s) starting up"), progname(), REPMGR_VERSION);
if (verbose)
logger_set_verbose();
if (log_type == REPMGR_SYSLOG)
{
fd = freopen("/dev/null", "w", stderr);
if (fd == NULL)
{
fprintf(stderr, "error reopening stderr to \"/dev/null\":\n %s\n",
strerror(errno));
}
}
log_info(_("connecting to database \"%s\""),
config_file_options.conninfo);
/* abort if local node not available at startup */
local_conn = establish_db_connection(config_file_options.conninfo, true);
/*
* sanity checks
*
* Note: previous repmgr versions checked the PostgreSQL version at this
* point, but we'll skip that and assume the presence of a node record
* means we're dealing with a supported installation.
*
* The absence of a node record will also indicate that either the node or
* repmgr has not been properly configured.
*/
/* warn about any settings which might not be relevant for the current PostgreSQL version */
if (config_file_options.standby_disconnect_on_failover == true && PQserverVersion(local_conn) < 90500)
{
log_warning(_("\"standby_disconnect_on_failover\" specified, but not available for this PostgreSQL version"));
/* TODO: format server version */
log_detail(_("available from PostgreSQL 9.5, this PostgreSQL version is %i"), PQserverVersion(local_conn));
}
/* Check "repmgr" the extension is installed */
extension_status = get_repmgr_extension_status(local_conn, &extversions);
if (extension_status == REPMGR_INSTALLED)
{
/*
* extension is the latest available according to "pg_available_extensions" -
* - does our (major) version match that?
*/
log_verbose(LOG_DEBUG, "expected extension version: %i; extension version: %i",
REPMGR_EXTENSION_NUM, extversions.installed_version_num);
if ((REPMGR_EXTENSION_NUM/100) < (extversions.installed_version_num / 100))
{
log_error(_("this \"repmgr\" version is older than the installed \"repmgr\" extension version"));
log_detail(_("\"repmgr\" version %s providing extension version %s is installed but extension is version %s"),
REPMGR_VERSION,
REPMGR_EXTENSION_VERSION,
extversions.installed_version);
log_hint(_("update the repmgr binaries to match the installed extension version"));
close_connection(&local_conn);
exit(ERR_BAD_CONFIG);
}
if ((REPMGR_EXTENSION_NUM/100) > (extversions.installed_version_num / 100))
{
log_error(_("this \"repmgr\" version is newer than the installed \"repmgr\" extension version"));
log_detail(_("\"repmgr\" version %s providing extension version %s is installed but extension is version %s"),
REPMGR_VERSION,
REPMGR_EXTENSION_VERSION,
extversions.installed_version);
log_hint(_("update the installed extension version by executing \"ALTER EXTENSION repmgr UPDATE\" in the repmgr database"));
close_connection(&local_conn);
exit(ERR_BAD_CONFIG);
}
}
else
{
/* this is unlikely to happen */
if (extension_status == REPMGR_UNKNOWN)
{
log_error(_("unable to determine status of \"repmgr\" extension"));
log_detail("\n%s", PQerrorMessage(local_conn));
close_connection(&local_conn);
exit(ERR_DB_QUERY);
}
if (extension_status == REPMGR_OLD_VERSION_INSTALLED)
{
log_error(_("an older version of the \"repmgr\" extension is installed"));
log_detail(_("extension version %s is installed but newer version %s is available"),
extversions.installed_version,
extversions.default_version);
log_hint(_("verify the repmgr installation is updated properly before continuing"));
}
else
{
log_error(_("repmgr extension not found on this node"));
if (extension_status == REPMGR_AVAILABLE)
{
log_detail(_("repmgr extension is available but not installed in database \"%s\""),
PQdb(local_conn));
}
else if (extension_status == REPMGR_UNAVAILABLE)
{
log_detail(_("repmgr extension is not available on this node"));
}
log_hint(_("check that this node is part of a repmgr cluster"));
}
close_connection(&local_conn);
exit(ERR_BAD_CONFIG);
}
/* Retrieve record for this node from the local database */
record_status = get_node_record(local_conn, config_file_options.node_id, &local_node_info);
/*
* Terminate if we can't find the local node record. This is a
* "fix-the-config" situation, not a lot else we can do.
*/
if (record_status != RECORD_FOUND)
{
log_error(_("no metadata record found for this node - terminating"));
switch (config_file_options.replication_type)
{
case REPLICATION_TYPE_PHYSICAL:
log_hint(_("check that 'repmgr (primary|standby|witness) register' was executed for this node"));
break;
}
close_connection(&local_conn);
terminate(ERR_BAD_CONFIG);
}
repmgrd_set_local_node_id(local_conn, config_file_options.node_id);
{
/*
* sanity-check that the shared library is loaded and shared memory
* can be written by attempting to retrieve the previously stored node_id
*/
int stored_local_node_id = UNKNOWN_NODE_ID;
stored_local_node_id = repmgrd_get_local_node_id(local_conn);
if (stored_local_node_id == UNKNOWN_NODE_ID)
{
log_error(_("unable to write to shared memory"));
log_hint(_("ensure \"shared_preload_libraries\" includes \"repmgr\""));
close_connection(&local_conn);
terminate(ERR_BAD_CONFIG);
}
}
if (config_file_options.replication_type == REPLICATION_TYPE_PHYSICAL)
{
log_debug("node id is %i, upstream node id is %i",
local_node_info.node_id,
local_node_info.upstream_node_id);
do_physical_node_check(local_conn);
}
if (daemonize == true)
{
daemonize_process();
}
if (pid_file[0] != '\0')
{
check_and_create_pid_file(pid_file);
}
repmgrd_set_pid(local_conn, getpid(), pid_file);
#ifndef WIN32
setup_event_handlers();
#endif
start_monitoring();
logger_shutdown();
return SUCCESS;
}
static void
start_monitoring(void)
{
log_notice(_("starting monitoring of node \"%s\" (ID: %i)"),
local_node_info.node_name,
local_node_info.node_id);
log_info(_("\"connection_check_type\" set to \"%s\""), print_connection_check_type(config_file_options.connection_check_type));
while (true)
{
switch (local_node_info.type)
{
case PRIMARY:
monitor_streaming_primary();
break;
case STANDBY:
monitor_streaming_standby();
break;
case WITNESS:
monitor_streaming_witness();
break;
case UNKNOWN:
/* should never happen */
break;
}
}
}
void
update_registration(PGconn *conn)
{
bool success = update_node_record_conn_priority(local_conn,
&config_file_options);
/* check values have actually changed */
if (success == false)
{
PQExpBufferData errmsg;
initPQExpBuffer(&errmsg);
appendPQExpBuffer(&errmsg,
_("unable to update local node record:\n %s"),
PQerrorMessage(conn));
create_event_record(conn,
&config_file_options,
config_file_options.node_id,
"repmgrd_config_reload",
false,
errmsg.data);
termPQExpBuffer(&errmsg);
}
return;
}
static void
daemonize_process(void)
{
char *ptr,
path[MAXPGPATH];
pid_t pid = fork();
int ret;
switch (pid)
{
case -1:
log_error(_("error in fork()"));
log_detail("%s", strerror(errno));
exit(ERR_SYS_FAILURE);
break;
case 0:
/* create independent session ID */
pid = setsid();
if (pid == (pid_t) -1)
{
log_error(_("error executing setsid()"));
log_detail("%s", strerror(errno));
exit(ERR_SYS_FAILURE);
}
/* ensure that we are no longer able to open a terminal */
pid = fork();
/* error case */
if (pid == -1)
{
log_error(_("error executing fork()"));
log_detail("%s", strerror(errno));
exit(ERR_SYS_FAILURE);
}
/* parent process */
if (pid != 0)
{
exit(0);
}
/* child process */
memset(path, 0, MAXPGPATH);
for (ptr = config_file_path + strlen(config_file_path); ptr > config_file_path; --ptr)
{
if (*ptr == '/')
{
strncpy(path, config_file_path, ptr - config_file_path);
}
}
if (*path == '\0')
{
*path = '/';
}
log_debug("dir now %s", path);
ret = chdir(path);
if (ret != 0)
{
log_error(_("error changing directory to \"%s\":\n %s"), path,
strerror(errno));
}
break;
default: /* parent process */
exit(0);
}
}
static void
check_and_create_pid_file(const char *pid_file)
{
struct stat st;
FILE *fd;
char buff[MAXLEN];
pid_t pid;
size_t nread;
if (stat(pid_file, &st) != -1)
{
memset(buff, 0, MAXLEN);
fd = fopen(pid_file, "r");
if (fd == NULL)
{
log_error(_("PID file \"%s\" exists but could not opened for reading"), pid_file);
log_hint(_("if repmgrd is no longer alive, remove the file and restart repmgrd"));
exit(ERR_BAD_PIDFILE);
}
nread = fread(buff, MAXLEN - 1, 1, fd);
if (nread == 0 && ferror(fd))
{
log_error(_("error reading PID file \"%s\", aborting"), pid_file);
exit(ERR_BAD_PIDFILE);
}
fclose(fd);
pid = atoi(buff);
if (pid != 0)
{
if (kill(pid, 0) != -1)
{
log_error(_("PID file \"%s\" exists and seems to contain a valid PID"), pid_file);
log_hint(_("if repmgrd is no longer alive, remove the file and restart repmgrd"));
exit(ERR_BAD_PIDFILE);
}
}
}
fd = fopen(pid_file, "w");
if (fd == NULL)
{
log_error(_("could not open PID file %s"), pid_file);
exit(ERR_BAD_CONFIG);
}
fprintf(fd, "%d", getpid());
fclose(fd);
}
#ifndef WIN32
/* SIGHUP: set flag to re-read config file at next convenient time */
static void
handle_sighup(SIGNAL_ARGS)
{
got_SIGHUP = true;
}
static void
setup_event_handlers(void)
{
pqsignal(SIGHUP, handle_sighup);
/*
* we want to be able to write a "repmgrd_shutdown" event, so delegate
* signal handling to the respective replication type handler, as it
* will know best which database connection to use
*/
switch (config_file_options.replication_type)
{
case REPLICATION_TYPE_PHYSICAL:
pqsignal(SIGINT, handle_sigint_physical);
pqsignal(SIGTERM, handle_sigint_physical);
break;
}
}
#endif
void
show_usage(void)
{
fprintf(stderr, _("%s: replication management daemon for PostgreSQL\n"), progname());
fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname());
}
void
show_help(void)
{
printf(_("%s: replication management daemon for PostgreSQL\n"), progname());
puts("");
printf(_("%s monitors a cluster of servers and optionally performs failover.\n"), progname());
puts("");
printf(_("Usage:\n"));
printf(_(" %s [OPTIONS]\n"), progname());
printf(_("\n"));
printf(_("Options:\n"));
puts("");
printf(_("General options:\n"));
printf(_(" -?, --help show this help, then exit\n"));
printf(_(" -V, --version output version information, then exit\n"));
puts("");
printf(_("General configuration options:\n"));
printf(_(" -v, --verbose output verbose activity information\n"));
printf(_(" -f, --config-file=PATH path to the configuration file\n"));
puts("");
printf(_("Daemon configuration options:\n"));
printf(_(" -d\n"));
printf(_(" --daemonize[=true/false]\n"));
printf(_(" detach process from foreground (default: true)\n"));
printf(_(" -p, --pid-file=PATH use the specified PID file\n"));
printf(_(" -s, --show-pid-file show PID file which would be used by the current configuration\n"));
printf(_(" --no-pid-file don't write a PID file\n"));
puts("");
}
bool
check_upstream_connection(PGconn **conn, const char *conninfo, PGconn **paired_conn)
{
/* Check the connection status twice in case it changes after reset */
bool twice = false;
log_debug("connection check type is \"%s\"",
print_connection_check_type(config_file_options.connection_check_type));
/*
* For the check types which do not involve using the existing database
* connection, we'll perform the actual check, then as an additional
* safeguard verify that the connection is still valid (as it might have
* gone away during a brief outage between checks).
*/
if (config_file_options.connection_check_type != CHECK_QUERY)
{
bool success = true;
if (config_file_options.connection_check_type == CHECK_PING)
{
success = is_server_available(conninfo);
}
else if (config_file_options.connection_check_type == CHECK_CONNECTION)
{
/*
* This connection is thrown away, and we never execute a query on it.
*/
PGconn *test_conn = PQconnectdb(conninfo);
log_debug("check_upstream_connection(): attempting to connect to \"%s\"", conninfo);
if (PQstatus(test_conn) != CONNECTION_OK)
{
log_warning(_("unable to connect to \"%s\""), conninfo);
log_detail("\n%s", PQerrorMessage(test_conn));
success = false;
}
PQfinish(test_conn);
}
if (success == false)
return false;
if (PQstatus(*conn) == CONNECTION_OK)
return true;
/*
* Checks have succeeded, but the open connection to the primary has gone away,
* possibly due to a brief outage between monitoring intervals - attempt to
* reset it.
*/
log_notice(_("upstream is available but upstream connection has gone away, resetting"));
PQfinish(*conn);
*conn = establish_db_connection_quiet(conninfo);
if (PQstatus(*conn) == CONNECTION_OK)
{
if (paired_conn != NULL)
{
log_debug("resetting paired connection");
*paired_conn = *conn;
}
return true;
}
return false;
}
for (;;)
{
if (PQstatus(*conn) != CONNECTION_OK)
{
log_debug("check_upstream_connection(): upstream connection has gone away, resetting");
if (twice)
return false;
/* reconnect */
PQfinish(*conn);
*conn = establish_db_connection_quiet(conninfo);
if (paired_conn != NULL)
{
log_debug("resetting paired connection");
*paired_conn = *conn;
}
twice = true;
}
else
{
if (!cancel_query(*conn, config_file_options.async_query_timeout))
goto failed;
if (wait_connection_availability(*conn, config_file_options.async_query_timeout) != 1)
goto failed;
/* execute a simple query to verify connection availability */
if (PQsendQuery(*conn, config_file_options.connection_check_query) == 0)
{
log_warning(_("unable to send query to upstream"));
log_detail("%s", PQerrorMessage(*conn));
goto failed;
}
if (wait_connection_availability(*conn, config_file_options.async_query_timeout) != 1)
goto failed;
break;
failed:
/* retry once */
if (twice)
return false;
/* reconnect */
log_debug("check_upstream_connection(): upstream connection not available, resetting");
PQfinish(*conn);
*conn = establish_db_connection_quiet(conninfo);
if (paired_conn != NULL)
{
log_debug("resetting paired connection");
*paired_conn = *conn;
}
twice = true;
}
}
return true;
}
void
try_reconnect(PGconn **conn, t_node_info *node_info)
{
PGconn *our_conn;
t_conninfo_param_list conninfo_params = T_CONNINFO_PARAM_LIST_INITIALIZER;
int i;
int max_attempts = config_file_options.reconnect_attempts;
initialize_conninfo_params(&conninfo_params, false);
/* we assume by now the conninfo string is parseable */
(void) parse_conninfo_string(node_info->conninfo, &conninfo_params, NULL, false);
/* set some default values if not explicitly provided */
param_set_ine(&conninfo_params, "connect_timeout", "2");
param_set_ine(&conninfo_params, "fallback_application_name", "repmgr");
for (i = 0; i < max_attempts; i++)
{
log_info(_("checking state of node %i, %i of %i attempts"),
node_info->node_id, i + 1, max_attempts);
if (is_server_available_params(&conninfo_params) == true)
{
log_notice(_("node %i has recovered, reconnecting"), node_info->node_id);
/*
* Note: we could also handle the case where node is pingable but
* connection denied due to connection exhaustion, by falling back to
* degraded monitoring (make configurable)
*/
our_conn = establish_db_connection_by_params(&conninfo_params, false);
if (PQstatus(our_conn) == CONNECTION_OK)
{
free_conninfo_params(&conninfo_params);
log_info(_("connection to node %i succeeded"), node_info->node_id);
if (PQstatus(*conn) == CONNECTION_BAD)
{
log_verbose(LOG_INFO, _("original connection handle returned CONNECTION_BAD, using new connection"));
close_connection(conn);
*conn = our_conn;
}
else
{
ExecStatusType ping_result;
ping_result = connection_ping(*conn);
if (ping_result != PGRES_TUPLES_OK)
{
log_info(_("original connection no longer available, using new connection"));
close_connection(conn);
*conn = our_conn;
}
else
{
log_info(_("original connection is still available"));
PQfinish(our_conn);
}
}
node_info->node_status = NODE_STATUS_UP;
return;
}
close_connection(&our_conn);
log_notice(_("unable to reconnect to node \"%s\" (ID: %i)"),
node_info->node_name,
node_info->node_id);
}
if (i + 1 < max_attempts)
{
log_info(_("sleeping %i seconds until next reconnection attempt"),
config_file_options.reconnect_interval);
sleep(config_file_options.reconnect_interval);
}
}
log_warning(_("unable to reconnect to node %i after %i attempts"),
node_info->node_id,
max_attempts);
node_info->node_status = NODE_STATUS_DOWN;
free_conninfo_params(&conninfo_params);
return;
}
int
calculate_elapsed(instr_time start_time)
{
instr_time current_time;
INSTR_TIME_SET_CURRENT(current_time);
INSTR_TIME_SUBTRACT(current_time, start_time);
return (int) INSTR_TIME_GET_DOUBLE(current_time);
}
const char *
print_monitoring_state(MonitoringState monitoring_state)
{
switch (monitoring_state)
{
case MS_NORMAL:
return "normal";
case MS_DEGRADED:
return "degraded";
}
/* should never reach here */
return "UNKNOWN";
}
void
terminate(int retval)
{
if (PQstatus(local_conn) == CONNECTION_OK)
repmgrd_set_pid(local_conn, UNKNOWN_PID, NULL);
logger_shutdown();
if (pid_file[0] != '\0')
{
unlink(pid_file);
}
log_info(_("%s terminating..."), progname());
exit(retval);
}
Loading...
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/mirrors/repmgr.git
git@gitee.com:mirrors/repmgr.git
mirrors
repmgr
repmgr
master

搜索帮助