From cb2b1b9d07cce3cb13d9ed2276b683466d6a73fa Mon Sep 17 00:00:00 2001 From: Greg Nancarrow Date: Mon, 30 Sep 2019 13:23:57 +1000 Subject: [PATCH v14 2/3] Enhance libpq target_session_attrs: primary/prefer-standby/standby Add new "primary" target_session_attrs option value, to support connecting to a server which is not in recovery mode, if available from the list of hosts (otherwise the connection attempt fails). Add new "prefer-standby" target_session_attrs option value, to support connecting to a server which is in recovery mode, if available from the list of hosts (otherwise connect to a server which is not in recovery mode). Add new "standby" target_session_attrs option value, to support connecting to a server which is in recovery mode, if available from the list of hosts (otherwise the connection attempt fails). To determine if running in recovery mode, the server is sent the query 'SELECT pg_is_in_recovery()'. Discussion: https://www.postgresql.org/message-id/flat/CAF3+xM+8-ztOkaV9gHiJ3wfgENTq97QcjXQt+rbFQ6F7oNzt9A@mail.gmail.com --- doc/src/sgml/libpq.sgml | 26 ++- src/interfaces/libpq/fe-connect.c | 313 ++++++++++++++++++++++++++-------- src/interfaces/libpq/libpq-fe.h | 10 +- src/interfaces/libpq/libpq-int.h | 4 +- src/test/recovery/t/001_stream_rep.pl | 18 +- 5 files changed, 291 insertions(+), 80 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index 0d3edfc..5f31fd0 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -1675,7 +1675,8 @@ postgresql://%2Fvar%2Flib%2Fpostgresql/dbname The supported options for this parameter are any, - read-write, prefer-read and read-only. + read-write, prefer-read, read-only, + primary, prefer-standby and standby. The default value of this parameter, any, regards all connections as acceptable. If multiple hosts are specified in the connection string, each host is tried in the order given until a connection @@ -1712,6 +1713,29 @@ postgresql://%2Fvar%2Flib%2Fpostgresql/dbname in which read-only transactions are accepted by default is considered acceptable. + + + If this parameter is set to primary, only a connection in which + the server is not in recovery mode is considered acceptable. + + + + If this parameter is set to prefer-standby, a connection in which + the server is in recovery mode is preferred. If no such connections can be found, + then a connection in which the server is not in recovery mode will be considered. + + + + If this parameter is set to standby, only a connection in which + the server is in recovery mode is considered acceptable. + + + + To determine whether the server is in recovery mode, the query + SELECT pg_is_in_recovery() will be sent upon any successful connection; + if it returns t, it means the server is in recovery mode. + + diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c index 7058f5d..0c096d1 100644 --- a/src/interfaces/libpq/fe-connect.c +++ b/src/interfaces/libpq/fe-connect.c @@ -350,7 +350,7 @@ static const internalPQconninfoOption PQconninfoOptions[] = { {"target_session_attrs", "PGTARGETSESSIONATTRS", DefaultTargetSessionAttrs, NULL, - "Target-Session-Attrs", "", 12, /* sizeof("prefer-read") = 12 */ + "Target-Session-Attrs", "", 15, /* sizeof("prefer-standby") = 15 */ offsetof(struct pg_conn, target_session_attrs)}, /* Terminating entry --- MUST BE LAST */ @@ -1335,6 +1335,12 @@ connectOptions2(PGconn *conn) conn->requested_session_type = SESSION_TYPE_PREFER_READ; else if (strcmp(conn->target_session_attrs, "read-only") == 0) conn->requested_session_type = SESSION_TYPE_READ_ONLY; + else if (strcmp(conn->target_session_attrs, "primary") == 0) + conn->requested_session_type = SESSION_TYPE_PRIMARY; + else if (strcmp(conn->target_session_attrs, "prefer-standby") == 0) + conn->requested_session_type = SESSION_TYPE_PREFER_STANDBY; + else if (strcmp(conn->target_session_attrs, "standby") == 0) + conn->requested_session_type = SESSION_TYPE_STANDBY; else { conn->status = CONNECTION_BAD; @@ -2147,6 +2153,60 @@ restoreErrorMessage(PGconn *conn, PQExpBuffer savedMessage) termPQExpBuffer(savedMessage); } +/* + * Internal helper function used for rejecting (and closing) a connection that + * doesn't satisfy the requested session type. The connection state is set to + * try the next host (if any). + * In the case of SESSION_TYPE_PREFER_READ, if the read-write host-index hasn't + * been set, then it is set to the index of this connection's host, so that a + * connection to this host can be made again in the event that no connection to + * a read-only host could be made after the first host scan. + */ +static void +reject_checked_read_or_write_connection(PGconn *conn) +{ + /* Not a requested type; fail this connection. */ + const char *displayed_host; + const char *displayed_port; + + /* Append error report to conn->errorMessage. */ + if (conn->connhost[conn->whichhost].type == CHT_HOST_ADDRESS) + displayed_host = conn->connhost[conn->whichhost].hostaddr; + else + displayed_host = conn->connhost[conn->whichhost].host; + displayed_port = conn->connhost[conn->whichhost].port; + if (displayed_port == NULL || displayed_port[0] == '\0') + displayed_port = DEF_PGPORT_STR; + + if (conn->requested_session_type == SESSION_TYPE_READ_WRITE) + appendPQExpBuffer(&conn->errorMessage, + libpq_gettext("could not make a writable " + "connection to server " + "\"%s:%s\"\n"), + displayed_host, displayed_port); + else + appendPQExpBuffer(&conn->errorMessage, + libpq_gettext("could not make a readonly " + "connection to server " + "\"%s:%s\"\n"), + displayed_host, displayed_port); + + /* Close connection politely. */ + conn->status = CONNECTION_OK; + sendTerminateConn(conn); + + /* Record read-write host index */ + if (conn->requested_session_type == SESSION_TYPE_PREFER_READ && + conn->read_write_or_primary_host_index == -1) + conn->read_write_or_primary_host_index = conn->whichhost; + + /* + * Try next host if any, but we don't want to consider additional + * addresses for this host. + */ + conn->try_next_host = true; +} + /* ---------------- * PQconnectPoll * @@ -2229,6 +2289,7 @@ PQconnectPoll(PGconn *conn) case CONNECTION_CHECK_WRITABLE: case CONNECTION_CONSUME: case CONNECTION_GSS_STARTUP: + case CONNECTION_CHECK_RECOVERY: break; default: @@ -2268,19 +2329,19 @@ keep_going: /* We will come back to here until there is if (conn->whichhost + 1 >= conn->nconnhost) { - if (conn->read_write_host_index >= 0) + if (conn->read_write_or_primary_host_index >= 0) { /* * Getting here means we failed to connect to read-only servers * and should now try to connect to a read-write server again. */ - conn->whichhost = conn->read_write_host_index; + conn->whichhost = conn->read_write_or_primary_host_index; /* * Reset the host index value to avoid recursion during the * second connection attempt. */ - conn->read_write_host_index = -2; + conn->read_write_or_primary_host_index = -2; } else { @@ -3507,7 +3568,9 @@ keep_going: /* We will come back to here until there is * may just skip the test in that case. */ if (conn->sversion >= 70400 && - conn->requested_session_type != SESSION_TYPE_ANY) + (conn->requested_session_type == SESSION_TYPE_READ_WRITE || + conn->requested_session_type == SESSION_TYPE_PREFER_READ || + conn->requested_session_type == SESSION_TYPE_READ_ONLY)) { if (conn->sversion < 130000) { @@ -3540,55 +3603,16 @@ keep_going: /* We will come back to here until there is (conn->requested_session_type == SESSION_TYPE_PREFER_READ || conn->requested_session_type == SESSION_TYPE_READ_ONLY))) { - /* Not a requested type; fail this connection. */ - const char *displayed_host; - const char *displayed_port; - /* * The following scenario is possible only for the * prefer-read mode for the next pass of the list of * connections as it couldn't find any servers that * are default read-only. */ - if (conn->read_write_host_index == -2) + if (conn->read_write_or_primary_host_index == -2) goto consume_checked_target_connection; - /* Append error report to conn->errorMessage. */ - if (conn->connhost[conn->whichhost].type == CHT_HOST_ADDRESS) - displayed_host = conn->connhost[conn->whichhost].hostaddr; - else - displayed_host = conn->connhost[conn->whichhost].host; - displayed_port = conn->connhost[conn->whichhost].port; - if (displayed_port == NULL || displayed_port[0] == '\0') - displayed_port = DEF_PGPORT_STR; - - if (conn->requested_session_type == SESSION_TYPE_READ_WRITE) - appendPQExpBuffer(&conn->errorMessage, - libpq_gettext("could not make a writable " - "connection to server " - "\"%s:%s\"\n"), - displayed_host, displayed_port); - else - appendPQExpBuffer(&conn->errorMessage, - libpq_gettext("could not make a readonly " - "connection to server " - "\"%s:%s\"\n"), - displayed_host, displayed_port); - - /* Close connection politely. */ - conn->status = CONNECTION_OK; - sendTerminateConn(conn); - - /* Record read-write host index */ - if (conn->requested_session_type == SESSION_TYPE_PREFER_READ && - conn->read_write_host_index == -1) - conn->read_write_host_index = conn->whichhost; - - /* - * Try next host if any, but we don't want to consider - * additional addresses for this host. - */ - conn->try_next_host = true; + reject_checked_read_or_write_connection(conn); goto keep_going; } @@ -3597,30 +3621,70 @@ keep_going: /* We will come back to here until there is } /* - * Requested type is prefer-read, then record this host index - * and try the other before considering it later. If requested - * type of connection is read-only, ignore this connection. + * Servers before 9.0 don't support recovery, skip the check + * when the requested type of connection is primary, + * prefer-standby or standby. + */ + else if ((conn->sversion >= 90000 && + (conn->requested_session_type == SESSION_TYPE_PRIMARY || + conn->requested_session_type == SESSION_TYPE_PREFER_STANDBY || + conn->requested_session_type == SESSION_TYPE_STANDBY))) + { + /* + * Save existing error messages across the PQsendQuery + * attempt. This is necessary because PQsendQuery is + * going to reset conn->errorMessage, so we would lose + * error messages related to previous hosts we have tried + * and failed to connect to. + */ + if (!saveErrorMessage(conn, &savedMessage)) + goto error_return; + + conn->status = CONNECTION_OK; + if (!PQsendQuery(conn, "SELECT pg_is_in_recovery()")) + { + restoreErrorMessage(conn, &savedMessage); + goto error_return; + } + + conn->status = CONNECTION_CHECK_RECOVERY; + + restoreErrorMessage(conn, &savedMessage); + return PGRES_POLLING_READING; + } + + /* + * Requested type is prefer-read or prefer-standby, then + * record this host index and try the other before considering + * it later. If requested type of connection is read-only or + * standby, ignore this connection. */ + if (conn->requested_session_type == SESSION_TYPE_PREFER_READ || - conn->requested_session_type == SESSION_TYPE_READ_ONLY) + conn->requested_session_type == SESSION_TYPE_READ_ONLY || + conn->requested_session_type == SESSION_TYPE_PREFER_STANDBY || + conn->requested_session_type == SESSION_TYPE_STANDBY) { /* * The following scenario is possible only for the - * prefer-read mode for the next pass of the list of - * connections as it couldn't find any servers that are - * default read-only. + * prefer-read or prefer-standby mode for the next pass of + * the list of connections as it couldn't find any servers + * that are default read-only or in recovery mode. */ - if (conn->read_write_host_index == -2) - goto target_accept_connection; + if (conn->read_write_or_primary_host_index == -2) + goto consume_checked_target_connection; /* Close connection politely. */ conn->status = CONNECTION_OK; sendTerminateConn(conn); /* Record read-write host index */ - if (conn->requested_session_type == SESSION_TYPE_PREFER_READ && - conn->read_write_host_index == -1) - conn->read_write_host_index = conn->whichhost; + if (conn->requested_session_type == SESSION_TYPE_PREFER_READ || + conn->requested_session_type == SESSION_TYPE_PREFER_STANDBY) + { + if (conn->read_write_or_primary_host_index == -1) + conn->read_write_or_primary_host_index = conn->whichhost; + } /* * Try next host if any, but we don't want to consider @@ -3755,13 +3819,119 @@ keep_going: /* We will come back to here until there is * connections as it couldn't find any servers that * are default read-only. */ - if (conn->read_write_host_index == -2) + if (conn->read_write_or_primary_host_index == -2) goto consume_checked_write_connection; /* Not a requested type; fail this connection. */ PQclear(res); restoreErrorMessage(conn, &savedMessage); + reject_checked_read_or_write_connection(conn); + goto keep_going; + } + + consume_checked_write_connection: + /* Session is requested type, so we're good. */ + PQclear(res); + termPQExpBuffer(&savedMessage); + + /* + * Finish reading any remaining messages before being + * considered as ready. + */ + conn->status = CONNECTION_CONSUME; + goto keep_going; + } + + /* + * Something went wrong with "SHOW transaction_read_only". We + * should try next addresses. + */ + if (res) + PQclear(res); + restoreErrorMessage(conn, &savedMessage); + + /* Append error report to conn->errorMessage. */ + if (conn->connhost[conn->whichhost].type == CHT_HOST_ADDRESS) + displayed_host = conn->connhost[conn->whichhost].hostaddr; + else + displayed_host = conn->connhost[conn->whichhost].host; + displayed_port = conn->connhost[conn->whichhost].port; + if (displayed_port == NULL || displayed_port[0] == '\0') + displayed_port = DEF_PGPORT_STR; + appendPQExpBuffer(&conn->errorMessage, + libpq_gettext("test \"SHOW transaction_read_only\" failed " + "on server \"%s:%s\"\n"), + displayed_host, displayed_port); + + /* Close connection politely. */ + conn->status = CONNECTION_OK; + sendTerminateConn(conn); + + /* Try next address */ + conn->try_next_addr = true; + goto keep_going; + } + + case CONNECTION_CHECK_RECOVERY: + { + const char *displayed_host; + const char *displayed_port; + + if (!saveErrorMessage(conn, &savedMessage)) + goto error_return; + + conn->status = CONNECTION_OK; + if (!PQconsumeInput(conn)) + { + restoreErrorMessage(conn, &savedMessage); + goto error_return; + } + + if (PQisBusy(conn)) + { + conn->status = CONNECTION_CHECK_RECOVERY; + restoreErrorMessage(conn, &savedMessage); + return PGRES_POLLING_READING; + } + + res = PQgetResult(conn); + if (res && (PQresultStatus(res) == PGRES_TUPLES_OK) && + PQntuples(res) == 1) + { + char *val; + bool standby_server; + + val = PQgetvalue(res, 0, 0); + standby_server = (strncmp(val, "t", 1) == 0); + + /* + * Server is in recovery mode and requested mode is + * primary, ignore it. Server is not in recovery mode and + * requested mode is prefer-standby, record it for the + * first time and try to consume in the next scan (it + * means no standby server is found in the first scan). + */ + if ((standby_server && + conn->requested_session_type == SESSION_TYPE_PRIMARY) || + (!standby_server && + (conn->requested_session_type == SESSION_TYPE_PREFER_STANDBY || + conn->requested_session_type == SESSION_TYPE_STANDBY))) + { + + /* + * The following scenario is possible only for the + * prefer-standby mode for the next pass of the list + * of connections as it couldn't find any servers that + * are in recovery. + */ + if (conn->read_write_or_primary_host_index == -2) + goto consume_checked_recovery_connection; + + /* Not a requested type; fail this connection. */ + PQclear(res); + restoreErrorMessage(conn, &savedMessage); + /* Append error report to conn->errorMessage. */ if (conn->connhost[conn->whichhost].type == CHT_HOST_ADDRESS) displayed_host = conn->connhost[conn->whichhost].hostaddr; @@ -3771,16 +3941,14 @@ keep_going: /* We will come back to here until there is if (displayed_port == NULL || displayed_port[0] == '\0') displayed_port = DEF_PGPORT_STR; - if (conn->requested_session_type == SESSION_TYPE_READ_WRITE) + if (conn->requested_session_type == SESSION_TYPE_PRIMARY) appendPQExpBuffer(&conn->errorMessage, - libpq_gettext("could not make a writable " - "connection to server " + libpq_gettext("server is in recovery mode " "\"%s:%s\"\n"), displayed_host, displayed_port); else appendPQExpBuffer(&conn->errorMessage, - libpq_gettext("could not make a readonly " - "connection to server " + libpq_gettext("server is not in recovery mode " "\"%s:%s\"\n"), displayed_host, displayed_port); @@ -3788,10 +3956,10 @@ keep_going: /* We will come back to here until there is conn->status = CONNECTION_OK; sendTerminateConn(conn); - /* Record read-write host index */ - if (conn->requested_session_type == SESSION_TYPE_PREFER_READ && - conn->read_write_host_index == -1) - conn->read_write_host_index = conn->whichhost; + /* Record primary host index */ + if (conn->requested_session_type == SESSION_TYPE_PREFER_STANDBY && + conn->read_write_or_primary_host_index == -1) + conn->read_write_or_primary_host_index = conn->whichhost; /* * Try next host if any, but we don't want to consider @@ -3801,7 +3969,7 @@ keep_going: /* We will come back to here until there is goto keep_going; } - consume_checked_write_connection: + consume_checked_recovery_connection: /* Session is requested type, so we're good. */ PQclear(res); termPQExpBuffer(&savedMessage); @@ -3815,7 +3983,7 @@ keep_going: /* We will come back to here until there is } /* - * Something went wrong with "SHOW transaction_read_only". We + * Something went wrong with "SELECT pg_is_in_recovery()". We * should try next addresses. */ if (res) @@ -3831,7 +3999,7 @@ keep_going: /* We will come back to here until there is if (displayed_port == NULL || displayed_port[0] == '\0') displayed_port = DEF_PGPORT_STR; appendPQExpBuffer(&conn->errorMessage, - libpq_gettext("test \"SHOW transaction_read_only\" failed " + libpq_gettext("test \"SELECT pg_is_in_recovery()\" failed " "on server \"%s:%s\"\n"), displayed_host, displayed_port); @@ -3843,7 +4011,6 @@ keep_going: /* We will come back to here until there is conn->try_next_addr = true; goto keep_going; } - default: appendPQExpBuffer(&conn->errorMessage, libpq_gettext("invalid connection state %d, " @@ -3989,7 +4156,7 @@ makeEmptyPGconn(void) #endif conn->requested_session_type = SESSION_TYPE_ANY; - conn->read_write_host_index = -1; + conn->read_write_or_primary_host_index = -1; /* * We try to send at least 8K at a time, which is the usual size of pipe diff --git a/src/interfaces/libpq/libpq-fe.h b/src/interfaces/libpq/libpq-fe.h index aa6f22f..30b181a 100644 --- a/src/interfaces/libpq/libpq-fe.h +++ b/src/interfaces/libpq/libpq-fe.h @@ -68,7 +68,8 @@ typedef enum CONNECTION_CONSUME, /* Wait for any pending message and consume * them. */ CONNECTION_GSS_STARTUP, /* Negotiating GSSAPI. */ - CONNECTION_CHECK_TARGET /* Check if we have a proper target connection */ + CONNECTION_CHECK_TARGET, /* Check if we have a proper target connection */ + CONNECTION_CHECK_RECOVERY /* Check whether server is in recovery */ } ConnStatusType; typedef enum @@ -76,8 +77,11 @@ typedef enum SESSION_TYPE_ANY = 0, /* Any session (default) */ SESSION_TYPE_READ_WRITE, /* Read-write session */ SESSION_TYPE_PREFER_READ, /* Prefer read only session */ - SESSION_TYPE_READ_ONLY /* Read only session */ -} TargetSessionAttrsType; + SESSION_TYPE_READ_ONLY, /* Read only session */ + SESSION_TYPE_PRIMARY, /* Primary server */ + SESSION_TYPE_PREFER_STANDBY, /* Prefer Standby server */ + SESSION_TYPE_STANDBY /* Standby server */ +} TargetSessionAttrsType; typedef enum { diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h index eb687d8..0efe16f 100644 --- a/src/interfaces/libpq/libpq-int.h +++ b/src/interfaces/libpq/libpq-int.h @@ -369,7 +369,7 @@ struct pg_conn /* * Type of connection to make. Possible values: any, read-write, - * prefer-read and read-only. + * prefer-read, read-only, primary, prefer-standby and standby. */ char *target_session_attrs; TargetSessionAttrsType requested_session_type; @@ -413,7 +413,7 @@ struct pg_conn * Initial value is -1, then the index of the first read-write host, -2 * during the second attempt of connection to avoid recursion. */ - int read_write_host_index; + int read_write_or_primary_host_index; /* Connection data */ pgsocket sock; /* FD for socket, PGINVALID_SOCKET if diff --git a/src/test/recovery/t/001_stream_rep.pl b/src/test/recovery/t/001_stream_rep.pl index ac1e11e..8fa28da 100644 --- a/src/test/recovery/t/001_stream_rep.pl +++ b/src/test/recovery/t/001_stream_rep.pl @@ -3,7 +3,7 @@ use strict; use warnings; use PostgresNode; use TestLib; -use Test::More tests => 37; +use Test::More tests => 41; # Initialize master node my $node_master = get_new_node('master'); @@ -141,6 +141,22 @@ test_target_session_attrs($node_master, $node_standby_1, $node_standby_1, test_target_session_attrs($node_standby_1, $node_master, $node_standby_1, "read-only", 0); +# Connect to master in "primary" mode with standby1,master list. +test_target_session_attrs($node_standby_1, $node_master, $node_master, + "primary", 0); + +# Connect to master in "prefer-standby" mode with master,master list. +test_target_session_attrs($node_master, $node_master, $node_master, + "prefer-standby", 0); + +# Connect to standby1 in "prefer-standby" mode with master,standby1 list. +test_target_session_attrs($node_master, $node_standby_1, $node_standby_1, + "prefer-standby", 0); + +# Connect to standby1 in "standby" mode with master,standby1 list. +test_target_session_attrs($node_master, $node_standby_1, $node_standby_1, + "standby", 0); + # Test for SHOW commands using a WAL sender connection with a replication # role. note "testing SHOW commands for replication connection"; -- 1.8.3.1