github/repmgr
1
0
Fork 0
mirror of https://github.com/EnterpriseDB/repmgr.git synced 2026-03-26 08:36:30 +00:00
Code Issues Packages Projects Releases 1 Wiki Activity

Reformat all source code using astyle

Browse Source 
astyle --style=ansi --indent=tab
This commit is contained in:
Greg Smith
2010-12-22 19:07:01 -05:00
parent 1ddb78ddd8
commit 9843205a4f
9 changed files with 878 additions and 871 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
check_dir.c
View File

@@ -6,15 +6,15 @@
* it under the terms of the GNU General Public License as published by * it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or * the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version. * (at your option) any later version.
* *
* This program is distributed in the hope that it will be useful, * This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of * but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details. * GNU General Public License for more details.
* *
* You should have received a copy of the GNU General Public License * You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
* *
*/ */
#include <sys/stat.h> #include <sys/stat.h>
@@ -43,63 +43,63 @@ static int mkdir_p(char *path, mode_t omode);
int int
check_dir(char *dir) check_dir(char *dir)
{ {
DIR *chkdir; DIR *chkdir;
struct dirent *file; struct dirent *file;
int result = 1; int result = 1;
errno = 0; errno = 0;
chkdir = opendir(dir); chkdir = opendir(dir);
if (!chkdir) if (!chkdir)
return (errno == ENOENT) ? 0 : -1; return (errno == ENOENT) ? 0 : -1;
while ((file = readdir(chkdir)) != NULL) while ((file = readdir(chkdir)) != NULL)
{ {
if (strcmp(".", file->d_name) == 0 || if (strcmp(".", file->d_name) == 0 ||
strcmp("..", file->d_name) == 0) strcmp("..", file->d_name) == 0)
{ {
/* skip this and parent directory */ /* skip this and parent directory */
continue; continue;
} }
else else
{ {
result = 2; /* not empty */ result = 2; /* not empty */
break; break;
} }
} }
#ifdef WIN32 #ifdef WIN32
/* /*
* This fix is in mingw cvs (runtime/mingwex/dirent.c rev 1.4), but not in * This fix is in mingw cvs (runtime/mingwex/dirent.c rev 1.4), but not in
* released version * released version
*/ */
if (GetLastError() == ERROR_NO_MORE_FILES) if (GetLastError() == ERROR_NO_MORE_FILES)
errno = 0; errno = 0;
#endif #endif
closedir(chkdir); closedir(chkdir);
if (errno != 0) if (errno != 0)
return -1; /* some kind of I/O error? */ return -1; /* some kind of I/O error? */
return result; return result;
} }
/* /*
* Create directory * Create directory
*/ */
bool bool
create_directory(char *dir) create_directory(char *dir)
{ {
if (mkdir_p(dir, 0700) == 0) if (mkdir_p(dir, 0700) == 0)
return true; return true;
fprintf(stderr, _("Could not create directory \"%s\": %s\n"), fprintf(stderr, _("Could not create directory \"%s\": %s\n"),
dir, strerror(errno)); dir, strerror(errno));
return false; return false;
} }
bool bool
@@ -126,10 +126,10 @@ mkdir_p(char *path, mode_t omode)
{ {
struct stat sb; struct stat sb;
mode_t numask, mode_t numask,
oumask; oumask;
int first, int first,
last, last,
retval; retval;
char *p; char *p;
p = path; p = path;
@@ -148,8 +148,8 @@ mkdir_p(char *path, mode_t omode)
return 1; return 1;
} }
else if (p[1] == ':' && else if (p[1] == ':' &&
((p[0] >= 'a' && p[0] <= 'z') || ((p[0] >= 'a' && p[0] <= 'z') ||
(p[0] >= 'A' && p[0] <= 'Z'))) (p[0] >= 'A' && p[0] <= 'Z')))
{ {
/* local drive */ /* local drive */
p += 2; p += 2;
@@ -224,5 +224,5 @@ is_pg_dir(char *dir)
sprintf(path, "%s/PG_VERSION", dir); sprintf(path, "%s/PG_VERSION", dir);
return (stat(path, &sb) == 0) ? true : false; return (stat(path, &sb) == 0) ? true : false;
} }

2
check_dir.h
View File

@@ -1,7 +1,7 @@
/* /*
* check_dir.h * check_dir.h
* Copyright (c) 2ndQuadrant, 2010 * Copyright (c) 2ndQuadrant, 2010
* *
* This program is free software: you can redistribute it and/or modify * 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 * it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or * the Free Software Foundation, either version 3 of the License, or

36
config.c
View File

@@ -6,15 +6,15 @@
* it under the terms of the GNU General Public License as published by * it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or * the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version. * (at your option) any later version.
* *
* This program is distributed in the hope that it will be useful, * This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of * but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details. * GNU General Public License for more details.
* *
* You should have received a copy of the GNU General Public License * You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
* *
*/ */
#include "repmgr.h" #include "repmgr.h"
@@ -55,28 +55,28 @@ parse_config(const char *config_file, char *cluster_name, int *node, char *conni
} }
/* Close file */ /* Close file */
fclose (fp); fclose (fp);
} }
char * char *
trim (char *s) trim (char *s)
{ {
/* Initialize start, end pointers */ /* Initialize start, end pointers */
char *s1 = s, *s2 = &s[strlen (s) - 1]; char *s1 = s, *s2 = &s[strlen (s) - 1];
/* Trim and delimit right side */ /* Trim and delimit right side */
while ( (isspace (*s2)) && (s2 >= s1) ) while ( (isspace (*s2)) && (s2 >= s1) )
--s2; --s2;
*(s2+1) = '\0'; *(s2+1) = '\0';
/* Trim left side */ /* Trim left side */
while ( (isspace (*s1)) && (s1 < s2) ) while ( (isspace (*s1)) && (s1 < s2) )
++s1; ++s1;
/* Copy finished string */ /* Copy finished string */
strcpy (s, s1); strcpy (s, s1);
return s; return s;
} }
void void
@@ -99,7 +99,7 @@ parse_line(char *buff, char *name, char *value)
/* /*
* Now the value * Now the value
*/ */
j = 0; j = 0;
for ( ++i ; i < MAXLEN; ++i) for ( ++i ; i < MAXLEN; ++i)
if (buff[i] == '\'') if (buff[i] == '\'')

2
config.h
View File

@@ -1,7 +1,7 @@
/* /*
* config.h * config.h
* Copyright (c) 2ndQuadrant, 2010 * Copyright (c) 2ndQuadrant, 2010
* *
* This program is free software: you can redistribute it and/or modify * 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 * it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or * the Free Software Foundation, either version 3 of the License, or

120
dbutils.c
View File

@@ -6,15 +6,15 @@
* it under the terms of the GNU General Public License as published by * it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or * the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version. * (at your option) any later version.
* *
* This program is distributed in the hope that it will be useful, * This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of * but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details. * GNU General Public License for more details.
* *
* You should have received a copy of the GNU General Public License * You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
* *
*/ */
#include "repmgr.h" #include "repmgr.h"
@@ -31,8 +31,8 @@ establishDBConnection(const char *conninfo, const bool exit_on_error)
/* Check to see that the backend connection was successfully made */ /* Check to see that the backend connection was successfully made */
if ((PQstatus(conn) != CONNECTION_OK)) if ((PQstatus(conn) != CONNECTION_OK))
{ {
fprintf(stderr, "Connection to database failed: %s", fprintf(stderr, "Connection to database failed: %s",
PQerrorMessage(conn)); PQerrorMessage(conn));
if (exit_on_error) if (exit_on_error)
{ {
PQfinish(conn); PQfinish(conn);
@@ -63,14 +63,14 @@ is_standby(PGconn *conn)
if (strcmp(PQgetvalue(res, 0, 0), "f") == 0) if (strcmp(PQgetvalue(res, 0, 0), "f") == 0)
result = false; result = false;
else else
result = true; result = true;
PQclear(res); PQclear(res);
return result; return result;
} }
/* /*
* If postgreSQL version is 9 or superior returns the major version * If postgreSQL version is 9 or superior returns the major version
* if 8 or inferior returns an empty string * if 8 or inferior returns an empty string
*/ */
@@ -82,18 +82,18 @@ pg_version(PGconn *conn, char* major_version)
int major_version1; int major_version1;
char *major_version2; char *major_version2;
res = PQexec(conn, "WITH pg_version(ver) AS (SELECT split_part(version(), ' ', 2)) " res = PQexec(conn, "WITH pg_version(ver) AS (SELECT split_part(version(), ' ', 2)) "
"SELECT split_part(ver, '.', 1), split_part(ver, '.', 2) FROM pg_version"); "SELECT split_part(ver, '.', 1), split_part(ver, '.', 2) FROM pg_version");
if (PQresultStatus(res) != PGRES_TUPLES_OK) if (PQresultStatus(res) != PGRES_TUPLES_OK)
{ {
fprintf(stderr, "PQexec failed: %s", PQerrorMessage(conn)); fprintf(stderr, "PQexec failed: %s", PQerrorMessage(conn));
PQclear(res); PQclear(res);
PQfinish(conn); PQfinish(conn);
exit(1); exit(1);
} }
major_version1 = atoi(PQgetvalue(res, 0, 0)); major_version1 = atoi(PQgetvalue(res, 0, 0));
major_version2 = PQgetvalue(res, 0, 1); major_version2 = PQgetvalue(res, 0, 1);
PQclear(res); PQclear(res);
if (major_version1 >= 9) if (major_version1 >= 9)
{ {
@@ -114,21 +114,21 @@ guc_setted(PGconn *conn, const char *parameter, const char *op, const char *valu
char sqlquery[MAXQUERY]; char sqlquery[MAXQUERY];
sprintf(sqlquery, "SELECT true FROM pg_settings " sprintf(sqlquery, "SELECT true FROM pg_settings "
" WHERE name = '%s' AND setting %s '%s'", " WHERE name = '%s' AND setting %s '%s'",
parameter, op, value); parameter, op, value);
res = PQexec(conn, sqlquery); res = PQexec(conn, sqlquery);
if (PQresultStatus(res) != PGRES_TUPLES_OK) if (PQresultStatus(res) != PGRES_TUPLES_OK)
{ {
fprintf(stderr, "PQexec failed: %s", PQerrorMessage(conn)); fprintf(stderr, "PQexec failed: %s", PQerrorMessage(conn));
PQclear(res); PQclear(res);
PQfinish(conn); PQfinish(conn);
exit(1); exit(1);
} }
if (PQntuples(res) == 0) if (PQntuples(res) == 0)
{ {
PQclear(res); PQclear(res);
return false; return false;
} }
PQclear(res); PQclear(res);
@@ -144,51 +144,51 @@ get_cluster_size(PGconn *conn)
char sqlquery[MAXQUERY]; char sqlquery[MAXQUERY];
sprintf(sqlquery, "SELECT pg_size_pretty(SUM(pg_database_size(oid))::bigint) " sprintf(sqlquery, "SELECT pg_size_pretty(SUM(pg_database_size(oid))::bigint) "
" FROM pg_database "); " FROM pg_database ");
res = PQexec(conn, sqlquery); res = PQexec(conn, sqlquery);
if (PQresultStatus(res) != PGRES_TUPLES_OK) if (PQresultStatus(res) != PGRES_TUPLES_OK)
{ {
fprintf(stderr, "PQexec failed: %s", PQerrorMessage(conn)); fprintf(stderr, "PQexec failed: %s", PQerrorMessage(conn));
PQclear(res); PQclear(res);
PQfinish(conn); PQfinish(conn);
exit(1); exit(1);
} }
size = PQgetvalue(res, 0, 0); size = PQgetvalue(res, 0, 0);
PQclear(res); PQclear(res);
return size; return size;
} }
/* /*
* get a connection to master by reading repl_nodes, creating a connection * get a connection to master by reading repl_nodes, creating a connection
* to each node (one at a time) and finding if it is a master or a standby * to each node (one at a time) and finding if it is a master or a standby
*/ */
PGconn * PGconn *
getMasterConnection(PGconn *standby_conn, int id, char *cluster, int *master_id) getMasterConnection(PGconn *standby_conn, int id, char *cluster, int *master_id)
{ {
PGconn *master_conn = NULL; PGconn *master_conn = NULL;
PGresult *res1; PGresult *res1;
PGresult *res2; PGresult *res2;
char sqlquery[MAXQUERY]; char sqlquery[MAXQUERY];
char master_conninfo[MAXCONNINFO]; char master_conninfo[MAXCONNINFO];
int i; int i;
/* find all nodes belonging to this cluster */ /* find all nodes belonging to this cluster */
sprintf(sqlquery, "SELECT * FROM repmgr_%s.repl_nodes " sprintf(sqlquery, "SELECT * FROM repmgr_%s.repl_nodes "
" WHERE cluster = '%s' and id <> %d", " WHERE cluster = '%s' and id <> %d",
cluster, cluster, id); cluster, cluster, id);
res1 = PQexec(standby_conn, sqlquery); res1 = PQexec(standby_conn, sqlquery);
if (PQresultStatus(res1) != PGRES_TUPLES_OK) if (PQresultStatus(res1) != PGRES_TUPLES_OK)
{ {
fprintf(stderr, "Can't get nodes info: %s\n", PQerrorMessage(standby_conn)); fprintf(stderr, "Can't get nodes info: %s\n", PQerrorMessage(standby_conn));
PQclear(res1); PQclear(res1);
PQfinish(standby_conn); PQfinish(standby_conn);
exit(1); exit(1);
} }
for (i = 0; i < PQntuples(res1); i++) for (i = 0; i < PQntuples(res1); i++)
{ {
/* initialize with the values of the current node being processed */ /* initialize with the values of the current node being processed */
*master_id = atoi(PQgetvalue(res1, i, 0)); *master_id = atoi(PQgetvalue(res1, i, 0));
strncpy(master_conninfo, PQgetvalue(res1, i, 2), MAXCONNINFO); strncpy(master_conninfo, PQgetvalue(res1, i, 2), MAXCONNINFO);
@@ -196,19 +196,19 @@ getMasterConnection(PGconn *standby_conn, int id, char *cluster, int *master_id)
if (PQstatus(master_conn) != CONNECTION_OK) if (PQstatus(master_conn) != CONNECTION_OK)
continue; continue;
/* /*
* I can't use the is_standby() function here because on error that * I can't use the is_standby() function here because on error that
* function closes the connection i pass and exit, but i still need to close * function closes the connection i pass and exit, but i still need to close
* standby_conn * standby_conn
*/ */
res2 = PQexec(master_conn, "SELECT pg_is_in_recovery()"); res2 = PQexec(master_conn, "SELECT pg_is_in_recovery()");
if (PQresultStatus(res2) != PGRES_TUPLES_OK) if (PQresultStatus(res2) != PGRES_TUPLES_OK)
{ {
fprintf(stderr, "Can't get recovery state from this node: %s\n", PQerrorMessage(master_conn)); fprintf(stderr, "Can't get recovery state from this node: %s\n", PQerrorMessage(master_conn));
PQclear(res2); PQclear(res2);
PQfinish(master_conn); PQfinish(master_conn);
continue; continue;
} }
/* if false, this is the master */ /* if false, this is the master */
if (strcmp(PQgetvalue(res2, 0, 0), "f") == 0) if (strcmp(PQgetvalue(res2, 0, 0), "f") == 0)
@@ -224,16 +224,16 @@ getMasterConnection(PGconn *standby_conn, int id, char *cluster, int *master_id)
PQfinish(master_conn); PQfinish(master_conn);
*master_id = -1; *master_id = -1;
} }
} }
/* If we finish this loop without finding a master then /* If we finish this loop without finding a master then
* we doesn't have the info or the master has failed (or we * we doesn't have the info or the master has failed (or we
* reached max_connections or superuser_reserved_connections, * reached max_connections or superuser_reserved_connections,
* anything else i'm missing?), * anything else i'm missing?),
* Probably we will need to check the error to know if we need * Probably we will need to check the error to know if we need
* to start failover procedure or just fix some situation on the * to start failover procedure or just fix some situation on the
* standby. * standby.
*/ */
PQclear(res1); PQclear(res1);
return NULL; return NULL;
} }

2
dbutils.h
View File

@@ -1,7 +1,7 @@
/* /*
* dbutils.h * dbutils.h
* Copyright (c) 2ndQuadrant, 2010 * Copyright (c) 2ndQuadrant, 2010
* *
* This program is free software: you can redistribute it and/or modify * 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 * it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or * the Free Software Foundation, either version 3 of the License, or

1240
repmgr.c
View File

File diff suppressed because it is too large Load Diff

2
repmgr.h
View File

@@ -1,7 +1,7 @@
/* /*
* repmgr.h * repmgr.h
* Copyright (c) 2ndQuadrant, 2010 * Copyright (c) 2ndQuadrant, 2010
* *
* This program is free software: you can redistribute it and/or modify * 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 * it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or * the Free Software Foundation, either version 3 of the License, or

255
repmgrd.c
View File

@@ -3,8 +3,8 @@
* Copyright (C) 2ndQuadrant, 2010 * Copyright (C) 2ndQuadrant, 2010
* *
* This module connects to the nodes of a replication cluster and monitors * This module connects to the nodes of a replication cluster and monitors
* how far are they from master * how far are they from master
* *
* This program is free software: you can redistribute it and/or modify * 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 * it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or * the Free Software Foundation, either version 3 of the License, or
@@ -78,13 +78,14 @@ static void setup_cancel_handler(void);
{ \ { \
MonitorExecute(); \ MonitorExecute(); \
sleep(3); \ sleep(3); \
} }
int int
main(int argc, char **argv) main(int argc, char **argv)
{ {
static struct option long_options[] = { static struct option long_options[] =
{
{"config", required_argument, NULL, 'f'}, {"config", required_argument, NULL, 'f'},
{"verbose", no_argument, NULL, 'v'}, {"verbose", no_argument, NULL, 'v'},
{NULL, 0, NULL, 0} {NULL, 0, NULL, 0}
@@ -93,7 +94,7 @@ main(int argc, char **argv)
int optindex; int optindex;
int c; int c;
char conninfo[MAXLEN]; char conninfo[MAXLEN];
char standby_version[MAXVERSIONSTR]; char standby_version[MAXVERSIONSTR];
progname = get_progname(argv[0]); progname = get_progname(argv[0]);
@@ -117,20 +118,20 @@ main(int argc, char **argv)
{ {
switch (c) switch (c)
{ {
case 'f': case 'f':
config_file = optarg; config_file = optarg;
break; break;
case 'v': case 'v':
verbose = true; verbose = true;
break; break;
default: default:
fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname); fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname);
exit(1); exit(1);
} }
} }
setup_cancel_handler(); setup_cancel_handler();
if (config_file == NULL) if (config_file == NULL)
{ {
config_file = malloc(5 + sizeof(CONFIG_FILE)); config_file = malloc(5 + sizeof(CONFIG_FILE));
@@ -139,30 +140,30 @@ main(int argc, char **argv)
/* /*
* Read the configuration file: repmgr.conf * Read the configuration file: repmgr.conf
*/ */
parse_config(config_file, myClusterName, &myLocalId, conninfo); parse_config(config_file, myClusterName, &myLocalId, conninfo);
if (myLocalId == -1) if (myLocalId == -1)
{ {
fprintf(stderr, "Node information is missing. " fprintf(stderr, "Node information is missing. "
"Check the configuration file.\n"); "Check the configuration file.\n");
exit(1); exit(1);
} }
myLocalConn = establishDBConnection(conninfo, true); myLocalConn = establishDBConnection(conninfo, true);
/* should be v9 or better */ /* should be v9 or better */
pg_version(myLocalConn, standby_version); pg_version(myLocalConn, standby_version);
if (strcmp(standby_version, "") == 0) if (strcmp(standby_version, "") == 0)
{ {
PQfinish(myLocalConn); PQfinish(myLocalConn);
fprintf(stderr, _("%s needs standby to be PostgreSQL 9.0 or better\n"), progname); fprintf(stderr, _("%s needs standby to be PostgreSQL 9.0 or better\n"), progname);
exit(1); exit(1);
} }
/* /*
* Set my server mode, establish a connection to primary * Set my server mode, establish a connection to primary
* and start monitor * and start monitor
*/ */
myLocalMode = is_standby(myLocalConn) ? STANDBY_MODE : PRIMARY_MODE; myLocalMode = is_standby(myLocalConn) ? STANDBY_MODE : PRIMARY_MODE;
if (myLocalMode == PRIMARY_MODE) if (myLocalMode == PRIMARY_MODE)
{ {
@@ -182,13 +183,13 @@ main(int argc, char **argv)
checkNodeConfiguration(conninfo); checkNodeConfiguration(conninfo);
if (myLocalMode == STANDBY_MODE) if (myLocalMode == STANDBY_MODE)
{ {
MonitorCheck(); MonitorCheck();
} }
/* close the connection to the database and cleanup */ /* close the connection to the database and cleanup */
CloseConnections(); CloseConnections();
return 0; return 0;
} }
@@ -200,7 +201,7 @@ main(int argc, char **argv)
static void static void
MonitorExecute(void) MonitorExecute(void)
{ {
PGresult *res; PGresult *res;
char monitor_standby_timestamp[MAXLEN]; char monitor_standby_timestamp[MAXLEN];
char last_wal_primary_location[MAXLEN]; char last_wal_primary_location[MAXLEN];
char last_wal_standby_received[MAXLEN]; char last_wal_standby_received[MAXLEN];
@@ -212,9 +213,9 @@ MonitorExecute(void)
int connection_retries; int connection_retries;
/* /*
* Check if the master is still available, if after 5 minutes of retries * Check if the master is still available, if after 5 minutes of retries
* we cannot reconnect, try to get a new master. * we cannot reconnect, try to get a new master.
*/ */
for (connection_retries = 0; connection_retries < 15; connection_retries++) for (connection_retries = 0; connection_retries < 15; connection_retries++)
{ {
@@ -225,7 +226,7 @@ MonitorExecute(void)
sleep(20); sleep(20);
PQreset(primaryConn); PQreset(primaryConn);
} }
else else
{ {
fprintf(stderr, "\n%s: Connection to master has been restored, continue monitoring.\n", progname); fprintf(stderr, "\n%s: Connection to master has been restored, continue monitoring.\n", progname);
@@ -261,53 +262,53 @@ MonitorExecute(void)
/* Check if we still are a standby, we could have been promoted */ /* Check if we still are a standby, we could have been promoted */
if (!is_standby(myLocalConn)) if (!is_standby(myLocalConn))
{ {
fprintf(stderr, "\n%s: seems like we have been promoted, so exit from monitoring...\n", fprintf(stderr, "\n%s: seems like we have been promoted, so exit from monitoring...\n",
progname); progname);
CloseConnections(); CloseConnections();
exit(1); exit(1);
} }
/* /*
* first check if there is a command being executed, * first check if there is a command being executed,
* and if that is the case, cancel the query so i can * and if that is the case, cancel the query so i can
* insert the current record * insert the current record
*/ */
if (PQisBusy(primaryConn) == 1) if (PQisBusy(primaryConn) == 1)
CancelQuery(); CancelQuery();
/* Get local xlog info */ /* Get local xlog info */
sprintf(sqlquery, sprintf(sqlquery,
"SELECT CURRENT_TIMESTAMP, pg_last_xlog_receive_location(), " "SELECT CURRENT_TIMESTAMP, pg_last_xlog_receive_location(), "
"pg_last_xlog_replay_location()"); "pg_last_xlog_replay_location()");
res = PQexec(myLocalConn, sqlquery); res = PQexec(myLocalConn, sqlquery);
if (PQresultStatus(res) != PGRES_TUPLES_OK) if (PQresultStatus(res) != PGRES_TUPLES_OK)
{ {
fprintf(stderr, "PQexec failed: %s\n", PQerrorMessage(myLocalConn)); fprintf(stderr, "PQexec failed: %s\n", PQerrorMessage(myLocalConn));
PQclear(res); PQclear(res);
/* if there is any error just let it be and retry in next loop */ /* if there is any error just let it be and retry in next loop */
return; return;
} }
strcpy(monitor_standby_timestamp, PQgetvalue(res, 0, 0)); strcpy(monitor_standby_timestamp, PQgetvalue(res, 0, 0));
strcpy(last_wal_standby_received , PQgetvalue(res, 0, 1)); strcpy(last_wal_standby_received , PQgetvalue(res, 0, 1));
strcpy(last_wal_standby_applied , PQgetvalue(res, 0, 2)); strcpy(last_wal_standby_applied , PQgetvalue(res, 0, 2));
PQclear(res); PQclear(res);
/* Get primary xlog info */ /* Get primary xlog info */
sprintf(sqlquery, "SELECT pg_current_xlog_location() "); sprintf(sqlquery, "SELECT pg_current_xlog_location() ");
res = PQexec(primaryConn, sqlquery); res = PQexec(primaryConn, sqlquery);
if (PQresultStatus(res) != PGRES_TUPLES_OK) if (PQresultStatus(res) != PGRES_TUPLES_OK)
{ {
fprintf(stderr, "PQexec failed: %s\n", PQerrorMessage(primaryConn)); fprintf(stderr, "PQexec failed: %s\n", PQerrorMessage(primaryConn));
PQclear(res); PQclear(res);
return; return;
} }
strcpy(last_wal_primary_location, PQgetvalue(res, 0, 0)); strcpy(last_wal_primary_location, PQgetvalue(res, 0, 0));
PQclear(res); PQclear(res);
/* Calculate the lag */ /* Calculate the lag */
lsn_primary = walLocationToBytes(last_wal_primary_location); lsn_primary = walLocationToBytes(last_wal_primary_location);
@@ -318,15 +319,15 @@ MonitorExecute(void)
* Build the SQL to execute on primary * Build the SQL to execute on primary
*/ */
sprintf(sqlquery, sprintf(sqlquery,
"INSERT INTO repmgr_%s.repl_monitor " "INSERT INTO repmgr_%s.repl_monitor "
"VALUES(%d, %d, '%s'::timestamp with time zone, " "VALUES(%d, %d, '%s'::timestamp with time zone, "
" '%s', '%s', " " '%s', '%s', "
" %lld, %lld)", myClusterName, " %lld, %lld)", myClusterName,
primaryId, myLocalId, monitor_standby_timestamp, primaryId, myLocalId, monitor_standby_timestamp,
last_wal_primary_location, last_wal_primary_location,
last_wal_standby_received, last_wal_standby_received,
(lsn_primary - lsn_standby_received), (lsn_primary - lsn_standby_received),
(lsn_standby_received - lsn_standby_applied)); (lsn_standby_received - lsn_standby_applied));
/* /*
* Execute the query asynchronously, but don't check for a result. We * Execute the query asynchronously, but don't check for a result. We
@@ -334,27 +335,27 @@ MonitorExecute(void)
*/ */
if (PQsendQuery(primaryConn, sqlquery) == 0) if (PQsendQuery(primaryConn, sqlquery) == 0)
fprintf(stderr, "Query could not be sent to primary. %s\n", fprintf(stderr, "Query could not be sent to primary. %s\n",
PQerrorMessage(primaryConn)); PQerrorMessage(primaryConn));
} }
static void static void
checkClusterConfiguration(void) checkClusterConfiguration(void)
{ {
PGresult *res; PGresult *res;
sprintf(sqlquery, "SELECT oid FROM pg_class " sprintf(sqlquery, "SELECT oid FROM pg_class "
" WHERE oid = 'repmgr_%s.repl_nodes'::regclass", " WHERE oid = 'repmgr_%s.repl_nodes'::regclass",
myClusterName); myClusterName);
res = PQexec(myLocalConn, sqlquery); res = PQexec(myLocalConn, sqlquery);
if (PQresultStatus(res) != PGRES_TUPLES_OK) if (PQresultStatus(res) != PGRES_TUPLES_OK)
{ {
fprintf(stderr, "PQexec failed: %s\n", PQerrorMessage(myLocalConn)); fprintf(stderr, "PQexec failed: %s\n", PQerrorMessage(myLocalConn));
PQclear(res); PQclear(res);
PQfinish(myLocalConn); PQfinish(myLocalConn);
PQfinish(primaryConn); PQfinish(primaryConn);
exit(1); exit(1);
} }
/* /*
* If there isn't any results then we have not configured a primary node yet * If there isn't any results then we have not configured a primary node yet
@@ -363,10 +364,10 @@ checkClusterConfiguration(void)
*/ */
if (PQntuples(res) == 0) if (PQntuples(res) == 0)
{ {
fprintf(stderr, "The replication cluster is not configured\n"); fprintf(stderr, "The replication cluster is not configured\n");
PQclear(res); PQclear(res);
PQfinish(myLocalConn); PQfinish(myLocalConn);
PQfinish(primaryConn); PQfinish(primaryConn);
exit(1); exit(1);
} }
PQclear(res); PQclear(res);
@@ -376,41 +377,41 @@ checkClusterConfiguration(void)
static void static void
checkNodeConfiguration(char *conninfo) checkNodeConfiguration(char *conninfo)
{ {
PGresult *res; PGresult *res;
/* /*
* Check if we have my node information in repl_nodes * Check if we have my node information in repl_nodes
*/ */
sprintf(sqlquery, "SELECT * FROM repmgr_%s.repl_nodes " sprintf(sqlquery, "SELECT * FROM repmgr_%s.repl_nodes "
" WHERE id = %d AND cluster = '%s' ", " WHERE id = %d AND cluster = '%s' ",
myClusterName, myLocalId, myClusterName); myClusterName, myLocalId, myClusterName);
res = PQexec(myLocalConn, sqlquery); res = PQexec(myLocalConn, sqlquery);
if (PQresultStatus(res) != PGRES_TUPLES_OK) if (PQresultStatus(res) != PGRES_TUPLES_OK)
{ {
fprintf(stderr, "PQexec failed: %s\n", PQerrorMessage(myLocalConn)); fprintf(stderr, "PQexec failed: %s\n", PQerrorMessage(myLocalConn));
PQclear(res); PQclear(res);
PQfinish(myLocalConn); PQfinish(myLocalConn);
PQfinish(primaryConn); PQfinish(primaryConn);
exit(1); exit(1);
} }
/* /*
* If there isn't any results then we have not configured this node yet * If there isn't any results then we have not configured this node yet
* in repmgr, if that is the case we will insert the node to the cluster * in repmgr, if that is the case we will insert the node to the cluster
*/ */
if (PQntuples(res) == 0) if (PQntuples(res) == 0)
{ {
PQclear(res); PQclear(res);
/* Adding the node */ /* Adding the node */
sprintf(sqlquery, "INSERT INTO repmgr_%s.repl_nodes " sprintf(sqlquery, "INSERT INTO repmgr_%s.repl_nodes "
"VALUES (%d, '%s', '%s')", "VALUES (%d, '%s', '%s')",
myClusterName, myLocalId, myClusterName, conninfo); myClusterName, myLocalId, myClusterName, conninfo);
if (!PQexec(primaryConn, sqlquery)) if (!PQexec(primaryConn, sqlquery))
{ {
fprintf(stderr, "Cannot insert node details, %s\n", fprintf(stderr, "Cannot insert node details, %s\n",
PQerrorMessage(primaryConn)); PQerrorMessage(primaryConn));
PQfinish(myLocalConn); PQfinish(myLocalConn);
PQfinish(primaryConn); PQfinish(primaryConn);
exit(1); exit(1);
@@ -420,33 +421,33 @@ checkNodeConfiguration(char *conninfo)
} }
static unsigned long long int static unsigned long long int
walLocationToBytes(char *wal_location) walLocationToBytes(char *wal_location)
{ {
unsigned int xlogid; unsigned int xlogid;
unsigned int xrecoff; unsigned int xrecoff;
if (sscanf(wal_location, "%X/%X", &xlogid, &xrecoff) != 2) if (sscanf(wal_location, "%X/%X", &xlogid, &xrecoff) != 2)
{ {
fprintf(stderr, "wrong log location format: %s\n", wal_location); fprintf(stderr, "wrong log location format: %s\n", wal_location);
return 0; return 0;
} }
return ((xlogid * 16 * 1024 * 1024 * 255) + xrecoff); return ((xlogid * 16 * 1024 * 1024 * 255) + xrecoff);
} }
static void static void
help(const char *progname) help(const char *progname)
{ {
printf(_("\n%s: Replicator manager daemon \n"), progname); printf(_("\n%s: Replicator manager daemon \n"), progname);
printf(_("Usage:\n")); printf(_("Usage:\n"));
printf(_(" %s [OPTIONS]\n"), progname); printf(_(" %s [OPTIONS]\n"), progname);
printf(_("\nOptions:\n")); printf(_("\nOptions:\n"));
printf(_(" --help show this help, then exit\n")); printf(_(" --help show this help, then exit\n"));
printf(_(" --version output version information, then exit\n")); printf(_(" --version output version information, then exit\n"));
printf(_(" --verbose output verbose activity information\n")); printf(_(" --verbose output verbose activity information\n"));
printf(_(" -f, --config_file=PATH database to connect to\n")); printf(_(" -f, --config_file=PATH database to connect to\n"));
printf(_("\n%s monitors a cluster of servers.\n"), progname); printf(_("\n%s monitors a cluster of servers.\n"), progname);
} }
@@ -455,13 +456,13 @@ help(const char *progname)
static void static void
handle_sigint(SIGNAL_ARGS) handle_sigint(SIGNAL_ARGS)
{ {
CloseConnections(); CloseConnections();
} }
static void static void
setup_cancel_handler(void) setup_cancel_handler(void)
{ {
pqsignal(SIGINT, handle_sigint); pqsignal(SIGINT, handle_sigint);
} }
#endif #endif
@@ -469,13 +470,13 @@ setup_cancel_handler(void)
static void static void
CancelQuery(void) CancelQuery(void)
{ {
char errbuf[256]; char errbuf[256];
PGcancel *pgcancel; PGcancel *pgcancel;
pgcancel = PQgetCancel(primaryConn); pgcancel = PQgetCancel(primaryConn);
if (!pgcancel || PQcancel(pgcancel, errbuf, 256) == 0) if (!pgcancel || PQcancel(pgcancel, errbuf, 256) == 0)
fprintf(stderr, "Can't stop current query: %s", errbuf); fprintf(stderr, "Can't stop current query: %s", errbuf);
PQfreeCancel(pgcancel); PQfreeCancel(pgcancel);
} }