public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: 'Shinya Kato' <[email protected]>
Cc: [email protected] <[email protected]>
Subject: RE: [Proposal] Add foreign-server health checks infrastructure
Date: Wed, 24 Nov 2021 04:57:23 +0000
Message-ID: <TYAPR01MB58661992162BA7708191EC79F5619@TYAPR01MB5866.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <[email protected]>
References: <TYAPR01MB58662809E678253B90E82CE5F5889@TYAPR01MB5866.jpnprd01.prod.outlook.com>
<[email protected]>
<TYAPR01MB58662CD4FD98AA475B3D10F9F59B9@TYAPR01MB5866.jpnprd01.prod.outlook.com>
<[email protected]>
Dear Kato-san,
Thank you for reviewing!
> Thank you for sending the patches!
> I confirmed that they can be compiled and tested successfully on CentOS
> 8.
Thanks!
> + {
> + {"remote_servers_connection_check_interval", PGC_USERSET,
> CONN_AUTH_SETTINGS,
> + gettext_noop("Sets the time interval between checks
> for
> disconnection of remote servers."),
> + NULL,
> + GUC_UNIT_MS
> + },
> + &remote_servers_connection_check_interval,
> + 0, 0, INT_MAX,
> + },
>
> If you don't use check_hook, assign_hook and show_hook, you should
> explicitly write "NULL, NULL, NULL", as show below.
Yeah I forgot the line. Fixed.
> + ereport(ERROR,
> +
> errcode(ERRCODE_CONNECTION_FAILURE),
> + errmsg("Postgres foreign server %s
> might be down.",
> +
> server->servername));
>
> According to [1], error messages should start with a lowercase letter
> and not use a period.
> Also, along with the rest of the code, it is a good idea to enclose the
> server name in double quotes.
I confirmed the postgres error-reporting policy and fixed to follow that.
How do you think?
Attached are the latest version.
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
Attachments:
[application/x-zip-compressed] 20211124.zip (6.2K, ../TYAPR01MB58661992162BA7708191EC79F5619@TYAPR01MB5866.jpnprd01.prod.outlook.com/2-20211124.zip)
download
[application/octet-stream] v02_add_checking_infrastracture.patch (11.9K, ../TYAPR01MB58661992162BA7708191EC79F5619@TYAPR01MB5866.jpnprd01.prod.outlook.com/3-v02_add_checking_infrastracture.patch)
download | inline diff:
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index aac10165ec..0c4d61a34c 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -269,6 +269,7 @@ CopyGetData(CopyFromState cstate, void *databuf, int minread, int maxread)
readmessage:
HOLD_CANCEL_INTERRUPTS();
+ HOLD_CHECKING_REMOTE_SERVERS_INTERRUPTS();
pq_startmsgread();
mtype = pq_getbyte();
if (mtype == EOF)
@@ -300,6 +301,7 @@ CopyGetData(CopyFromState cstate, void *databuf, int minread, int maxread)
ereport(ERROR,
(errcode(ERRCODE_CONNECTION_FAILURE),
errmsg("unexpected EOF on client connection with an open transaction")));
+ RESUME_CHECKING_REMOTE_SERVERS_INTERRUPTS();
RESUME_CANCEL_INTERRUPTS();
/* ... and process it */
switch (mtype)
diff --git a/src/backend/foreign/foreign.c b/src/backend/foreign/foreign.c
index e07cc57431..eb9c4f3f05 100644
--- a/src/backend/foreign/foreign.c
+++ b/src/backend/foreign/foreign.c
@@ -26,7 +26,11 @@
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/syscache.h"
+#include "utils/timeout.h"
+/* for checking remote servers */
+int remote_servers_connection_check_interval = 0;
+static CheckingRemoteServersCallbackItem *fdw_callbacks = NULL;
/*
* GetForeignDataWrapper - look up the foreign-data wrapper by OID.
@@ -836,3 +840,51 @@ GetExistingLocalJoinPath(RelOptInfo *joinrel)
}
return NULL;
}
+
+/*
+ * Register callbacks for checking remote servers.
+ *
+ * This function is intended for use by FDW extensions.
+ * The checking timeout will be fired after registering the first callback.
+ */
+void
+RegisterCheckingRemoteServersCallback(CheckingRemoteServersCallback callback, void *arg)
+{
+ CheckingRemoteServersCallbackItem *item;
+ /* should we start checking timeout? */
+ bool first_exec = HaveCheckingRemoteServersCallbacks();
+
+ item = (CheckingRemoteServersCallbackItem *)
+ MemoryContextAlloc(TopMemoryContext, sizeof(CheckingRemoteServersCallbackItem));
+ item->callback = callback;
+ item->arg = arg;
+ item->next = fdw_callbacks;
+ fdw_callbacks = item;
+
+ if (first_exec && remote_servers_connection_check_interval > 0)
+ enable_timeout_after(CHECKING_REMOTE_SERVERS_TIMEOUT, remote_servers_connection_check_interval);
+}
+
+
+/*
+ * Call callbacks for checking remote servers.
+ *
+ * Note that this function will not return anything.
+ * Callback functions must throw ereport(ERROR) if disconnection has been detected.
+ */
+void
+CallCheckingRemoteServersCallbacks(void)
+{
+ CheckingRemoteServersCallbackItem *item;
+ for (item = fdw_callbacks; item; item = item->next)
+ item->callback(item->arg);
+}
+
+/*
+ * Check whether any callbacks has been registered.
+ */
+bool
+HaveCheckingRemoteServersCallbacks(void)
+{
+ return fdw_callbacks != NULL;
+}
diff --git a/src/backend/libpq/pqcomm.c b/src/backend/libpq/pqcomm.c
index 9ebba025cc..5130a8340b 100644
--- a/src/backend/libpq/pqcomm.c
+++ b/src/backend/libpq/pqcomm.c
@@ -1171,7 +1171,6 @@ pq_startmsgread(void)
ereport(FATAL,
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg("terminating connection because protocol synchronization was lost")));
-
PqCommReadingMsg = true;
}
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 0775abe35d..d7890b5307 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -43,6 +43,7 @@
#include "commands/async.h"
#include "commands/prepare.h"
#include "executor/spi.h"
+#include "foreign/foreign.h"
#include "jit/jit.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
@@ -349,6 +350,7 @@ SocketBackend(StringInfo inBuf)
* Get message type code from the frontend.
*/
HOLD_CANCEL_INTERRUPTS();
+ HOLD_CHECKING_REMOTE_SERVERS_INTERRUPTS();
pq_startmsgread();
qtype = pq_getbyte();
@@ -455,6 +457,7 @@ SocketBackend(StringInfo inBuf)
*/
if (pq_getmessage(inBuf, maxmsglen))
return EOF; /* suitable message already logged */
+ RESUME_CHECKING_REMOTE_SERVERS_INTERRUPTS();
RESUME_CANCEL_INTERRUPTS();
return qtype;
@@ -2708,6 +2711,13 @@ start_xact_command(void)
!get_timeout_active(CLIENT_CONNECTION_CHECK_TIMEOUT))
enable_timeout_after(CLIENT_CONNECTION_CHECK_TIMEOUT,
client_connection_check_interval);
+ if (remote_servers_connection_check_interval > 0 &&
+ IsUnderPostmaster &&
+ MyProcPort &&
+ HaveCheckingRemoteServersCallbacks() &&
+ !get_timeout_active(CHECKING_REMOTE_SERVERS_TIMEOUT))
+ enable_timeout_after(CHECKING_REMOTE_SERVERS_TIMEOUT,
+ remote_servers_connection_check_interval);
}
static void
@@ -3213,6 +3223,24 @@ ProcessInterrupts(void)
}
}
+ if (CheckingRemoteServersTimeoutPending && CheckingRemoteServersHoldoffCount != 0)
+ {
+ /*
+ * Skip checking foreign servers while reading messages.
+ */
+ InterruptPending = true;
+ }
+ else if (CheckingRemoteServersTimeoutPending)
+ {
+ CheckingRemoteServersTimeoutPending = false;
+
+ CallCheckingRemoteServersCallbacks();
+
+ if (remote_servers_connection_check_interval > 0)
+ enable_timeout_after(CHECKING_REMOTE_SERVERS_TIMEOUT,
+ remote_servers_connection_check_interval);
+ }
+
if (ClientConnectionLost)
{
QueryCancelPending = false; /* lost connection trumps QueryCancel */
diff --git a/src/backend/utils/init/globals.c b/src/backend/utils/init/globals.c
index 381d9e548d..15b0c2727b 100644
--- a/src/backend/utils/init/globals.c
+++ b/src/backend/utils/init/globals.c
@@ -36,9 +36,11 @@ volatile sig_atomic_t IdleInTransactionSessionTimeoutPending = false;
volatile sig_atomic_t IdleSessionTimeoutPending = false;
volatile sig_atomic_t ProcSignalBarrierPending = false;
volatile sig_atomic_t LogMemoryContextPending = false;
+volatile sig_atomic_t CheckingRemoteServersTimeoutPending = false;
volatile uint32 InterruptHoldoffCount = 0;
volatile uint32 QueryCancelHoldoffCount = 0;
volatile uint32 CritSectionCount = 0;
+volatile uint32 CheckingRemoteServersHoldoffCount = 0;
int MyProcPid;
pg_time_t MyStartTime;
diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c
index 646126edee..5962b1e395 100644
--- a/src/backend/utils/init/postinit.c
+++ b/src/backend/utils/init/postinit.c
@@ -73,6 +73,7 @@ static void LockTimeoutHandler(void);
static void IdleInTransactionSessionTimeoutHandler(void);
static void IdleSessionTimeoutHandler(void);
static void ClientCheckTimeoutHandler(void);
+static void CheckingRemoteServersTimeoutHandler(void);
static bool ThereIsAtLeastOneRole(void);
static void process_startup_options(Port *port, bool am_superuser);
static void process_settings(Oid databaseid, Oid roleid);
@@ -621,6 +622,7 @@ InitPostgres(const char *in_dbname, Oid dboid, const char *username,
IdleInTransactionSessionTimeoutHandler);
RegisterTimeout(IDLE_SESSION_TIMEOUT, IdleSessionTimeoutHandler);
RegisterTimeout(CLIENT_CONNECTION_CHECK_TIMEOUT, ClientCheckTimeoutHandler);
+ RegisterTimeout(CHECKING_REMOTE_SERVERS_TIMEOUT, CheckingRemoteServersTimeoutHandler);
}
/*
@@ -1250,6 +1252,14 @@ ClientCheckTimeoutHandler(void)
SetLatch(MyLatch);
}
+static void
+CheckingRemoteServersTimeoutHandler(void)
+{
+ CheckingRemoteServersTimeoutPending = true;
+ InterruptPending = true;
+ SetLatch(MyLatch);
+}
+
/*
* Returns true if at least one role is defined in this database cluster.
*/
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index e91d5a3cfd..1faeb59b87 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -51,6 +51,7 @@
#include "commands/vacuum.h"
#include "commands/variable.h"
#include "common/string.h"
+#include "foreign/foreign.h"
#include "funcapi.h"
#include "jit/jit.h"
#include "libpq/auth.h"
@@ -105,6 +106,7 @@
#include "utils/queryjumble.h"
#include "utils/rls.h"
#include "utils/snapmgr.h"
+#include "utils/timeout.h"
#include "utils/tzparser.h"
#include "utils/inval.h"
#include "utils/varlena.h"
@@ -3583,6 +3585,18 @@ static struct config_int ConfigureNamesInt[] =
NULL, NULL, NULL
},
+
+ {
+ {"remote_servers_connection_check_interval", PGC_USERSET, CONN_AUTH_SETTINGS,
+ gettext_noop("Sets the time interval between checks for disconnection of remote servers."),
+ NULL,
+ GUC_UNIT_MS
+ },
+ &remote_servers_connection_check_interval,
+ 0, 0, INT_MAX,
+ NULL, NULL, NULL
+ },
+
/* End-of-list marker */
{
{NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 1cbc9feeb6..9b70761d9c 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -90,6 +90,10 @@
# disconnection while running queries;
# 0 for never
+#remote_servers_connection_check_interval = 0 # time between time between checks for
+ # foreign server disconnection;
+ # 0 for never
+
# - Authentication -
#authentication_timeout = 1min # 1s-600s
diff --git a/src/include/foreign/foreign.h b/src/include/foreign/foreign.h
index 8169eb76b1..98f59c1153 100644
--- a/src/include/foreign/foreign.h
+++ b/src/include/foreign/foreign.h
@@ -81,4 +81,20 @@ extern List *GetForeignColumnOptions(Oid relid, AttrNumber attnum);
extern Oid get_foreign_data_wrapper_oid(const char *fdwname, bool missing_ok);
extern Oid get_foreign_server_oid(const char *servername, bool missing_ok);
+
+/* functions and variables for fdw checking. */
+typedef void (*CheckingRemoteServersCallback) (void *arg);
+typedef struct CheckingRemoteServersCallbackItem
+{
+ struct CheckingRemoteServersCallbackItem *next;
+ CheckingRemoteServersCallback callback;
+ void *arg;
+} CheckingRemoteServersCallbackItem;
+
+extern void RegisterCheckingRemoteServersCallback(CheckingRemoteServersCallback callback, void *arg);
+extern void CallCheckingRemoteServersCallbacks(void);
+extern bool HaveCheckingRemoteServersCallbacks(void);
+
+extern int remote_servers_connection_check_interval;
+
#endif /* FOREIGN_H */
diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h
index 90a3016065..377fc68aaa 100644
--- a/src/include/miscadmin.h
+++ b/src/include/miscadmin.h
@@ -98,10 +98,13 @@ extern PGDLLIMPORT volatile sig_atomic_t LogMemoryContextPending;
extern PGDLLIMPORT volatile sig_atomic_t CheckClientConnectionPending;
extern PGDLLIMPORT volatile sig_atomic_t ClientConnectionLost;
+extern PGDLLIMPORT volatile sig_atomic_t CheckingRemoteServersTimeoutPending;
+
/* these are marked volatile because they are examined by signal handlers: */
extern PGDLLIMPORT volatile uint32 InterruptHoldoffCount;
extern PGDLLIMPORT volatile uint32 QueryCancelHoldoffCount;
extern PGDLLIMPORT volatile uint32 CritSectionCount;
+extern PGDLLIMPORT volatile uint32 CheckingRemoteServersHoldoffCount;
/* in tcop/postgres.c */
extern void ProcessInterrupts(void);
@@ -126,7 +129,7 @@ do { \
/* Is ProcessInterrupts() guaranteed to clear InterruptPending? */
#define INTERRUPTS_CAN_BE_PROCESSED() \
(InterruptHoldoffCount == 0 && CritSectionCount == 0 && \
- QueryCancelHoldoffCount == 0)
+ QueryCancelHoldoffCount == 0 && CheckingRemoteServersHoldoffCount == 0)
#define HOLD_INTERRUPTS() (InterruptHoldoffCount++)
@@ -152,6 +155,13 @@ do { \
CritSectionCount--; \
} while(0)
+#define HOLD_CHECKING_REMOTE_SERVERS_INTERRUPTS() (CheckingRemoteServersHoldoffCount++)
+
+#define RESUME_CHECKING_REMOTE_SERVERS_INTERRUPTS() \
+do { \
+ Assert(CheckingRemoteServersHoldoffCount > 0); \
+ CheckingRemoteServersHoldoffCount--; \
+} while(0)
/*****************************************************************************
* globals.h -- *
diff --git a/src/include/utils/timeout.h b/src/include/utils/timeout.h
index 2cbc5de4d9..ceb6b1c12c 100644
--- a/src/include/utils/timeout.h
+++ b/src/include/utils/timeout.h
@@ -34,6 +34,7 @@ typedef enum TimeoutId
IDLE_SESSION_TIMEOUT,
CLIENT_CONNECTION_CHECK_TIMEOUT,
STARTUP_PROGRESS_TIMEOUT,
+ CHECKING_REMOTE_SERVERS_TIMEOUT,
/* First user-definable timeout reason */
USER_TIMEOUT,
/* Maximum number of timeout reasons */
[application/octet-stream] v02_add_helth_check_for_postgres_fdw.patch (3.1K, ../TYAPR01MB58661992162BA7708191EC79F5619@TYAPR01MB5866.jpnprd01.prod.outlook.com/4-v02_add_helth_check_for_postgres_fdw.patch)
download | inline diff:
diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c
index 4aff315b7c..30becdeddb 100644
--- a/contrib/postgres_fdw/connection.c
+++ b/contrib/postgres_fdw/connection.c
@@ -109,6 +109,7 @@ static void pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql,
bool toplevel);
static bool UserMappingPasswordRequired(UserMapping *user);
static bool disconnect_cached_connections(Oid serverid);
+static bool pgfdw_connection_check_internal(PGconn *conn);
/*
* Get a PGconn which can be used to execute queries on the remote PostgreSQL
@@ -153,6 +154,7 @@ GetConnection(UserMapping *user, bool will_prep_stmt, PgFdwConnState **state)
pgfdw_inval_callback, (Datum) 0);
CacheRegisterSyscacheCallback(USERMAPPINGOID,
pgfdw_inval_callback, (Datum) 0);
+ RegisterCheckingRemoteServersCallback(pgfdw_connection_check, NULL);
}
/* Set flag that we did GetConnection during the current transaction */
@@ -1638,3 +1640,66 @@ disconnect_cached_connections(Oid serverid)
return result;
}
+
+
+/*
+ * Callback function for checking remote servers.
+ *
+ * This function searches the hash table from the beginning
+ * and performs a health-check on each entry.
+ *
+ * Note that this might be expensive because create and wait
+ * eventset many times.
+ */
+void
+pgfdw_connection_check(void *args)
+{
+ HASH_SEQ_STATUS scan;
+ ConnCacheEntry *entry;
+
+ /* Exit immediately if hash is not initialized. */
+ if (!ConnectionHash)
+ return;
+
+ hash_seq_init(&scan, ConnectionHash);
+ while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
+ {
+ if (entry->conn == NULL || entry->xact_depth == 0)
+ continue;
+ if (!pgfdw_connection_check_internal(entry->conn))
+ {
+ ForeignServer *server = GetForeignServer(entry->serverid);
+ disconnect_pg_server(entry);
+ ereport(ERROR,
+ errcode(ERRCODE_CONNECTION_FAILURE),
+ errmsg("foreign server \"%s\"disconnected due to the health-check failure",
+ server->servername),
+ errdetail("Foreign server might be down."),
+ errhint("Please check the server and network health."));
+ }
+ }
+}
+
+/*
+ * helper fucntion for pgfdw_connection_check
+ */
+static bool
+pgfdw_connection_check_internal(PGconn *conn)
+{
+ WaitEventSet *eventset;
+ WaitEvent events;
+
+ eventset = CreateWaitEventSet(CurrentMemoryContext, 1);
+ AddWaitEventToSet(eventset, WL_SOCKET_CLOSED, PQsocket(conn), NULL, NULL);
+
+ WaitEventSetWait(eventset, 0, &events, 1, 0);
+
+ if (events.events & WL_SOCKET_CLOSED)
+ {
+ FreeWaitEventSet(eventset);
+ return false;
+ }
+ FreeWaitEventSet(eventset);
+
+ return true;
+}
diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h
index 90b72e9ec5..4630ca05d7 100644
--- a/contrib/postgres_fdw/postgres_fdw.h
+++ b/contrib/postgres_fdw/postgres_fdw.h
@@ -151,6 +151,7 @@ extern PGresult *pgfdw_exec_query(PGconn *conn, const char *query,
PgFdwConnState *state);
extern void pgfdw_report_error(int elevel, PGresult *res, PGconn *conn,
bool clear, const char *sql);
+extern void pgfdw_connection_check(void *args);
/* in option.c */
extern int ExtractConnectionOptions(List *defelems,
view thread (39+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected]
Subject: RE: [Proposal] Add foreign-server health checks infrastructure
In-Reply-To: <TYAPR01MB58661992162BA7708191EC79F5619@TYAPR01MB5866.jpnprd01.prod.outlook.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox