public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: 'Fujii Masao' <[email protected]>
To: 'Kyotaro Horiguchi' <[email protected]>
Cc: [email protected] <[email protected]>
Cc: [email protected] <[email protected]>
Cc: [email protected] <[email protected]>
Subject: RE: [Proposal] Add foreign-server health checks infrastructure
Date: Tue, 22 Feb 2022 05:25:59 +0000
Message-ID: <TYAPR01MB5866FF28C802916242F1B492F53B9@TYAPR01MB5866.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <[email protected]>
References: <[email protected]>
<[email protected]>
<TYAPR01MB58661B088B6066282824ADD9F5369@TYAPR01MB5866.jpnprd01.prod.outlook.com>
<[email protected]>
<TYAPR01MB58664E79C4728A6FFEA46F94F53B9@TYAPR01MB5866.jpnprd01.prod.outlook.com>
<[email protected]>
Dear Fujii-san,
> cfbot is reporting that the 0002 patch fails to be applied cleanly. Could you update
> the patch?
> http://cfbot.cputube.org/patch_37_3388.log
Thanks for reporting and sorry for inconvenience.
I repo was not latest version. Attached can be applied to 52e4f0c
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
Attachments:
[application/octet-stream] v11_0001_add_checking_infrastracture.patch (1.4K, ../TYAPR01MB5866FF28C802916242F1B492F53B9@TYAPR01MB5866.jpnprd01.prod.outlook.com/2-v11_0001_add_checking_infrastracture.patch)
download | inline diff:
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 3c7d08209f..206f7e1d59 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -106,6 +106,8 @@ int PostAuthDelay = 0;
/* Time between checks that the client is still connected. */
int client_connection_check_interval = 0;
+char *QeuryCancelMessage = NULL;
+
/* ----------------
* private typedefs etc
* ----------------
@@ -3327,6 +3329,8 @@ ProcessInterrupts(void)
LockErrorCleanup();
ereport(ERROR,
(errcode(ERRCODE_QUERY_CANCELED),
+ QeuryCancelMessage ?
+ errmsg("%s", QeuryCancelMessage) :
errmsg("canceling statement due to user request")));
}
}
@@ -4248,6 +4252,9 @@ PostgresMain(const char *dbname, const char *username)
/* Report the error to the client and/or server log */
EmitErrorReport();
+ /* Make sure QeuryCancelMessage is reset. */
+ QeuryCancelMessage = NULL;
+
/*
* Make sure debug_query_string gets reset before we possibly clobber
* the storage it points at.
diff --git a/src/include/tcop/tcopprot.h b/src/include/tcop/tcopprot.h
index 15a11bc3ff..54ea1de9de 100644
--- a/src/include/tcop/tcopprot.h
+++ b/src/include/tcop/tcopprot.h
@@ -30,6 +30,7 @@ extern PGDLLIMPORT const char *debug_query_string;
extern int max_stack_depth;
extern int PostAuthDelay;
extern int client_connection_check_interval;
+extern char* QeuryCancelMessage;
/* GUC-configurable parameters */
[application/octet-stream] v11_0002_add_health_check.patch (6.7K, ../TYAPR01MB5866FF28C802916242F1B492F53B9@TYAPR01MB5866.jpnprd01.prod.outlook.com/3-v11_0002_add_health_check.patch)
download | inline diff:
diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c
index f753c6e232..d55243fb7e 100644
--- a/contrib/postgres_fdw/connection.c
+++ b/contrib/postgres_fdw/connection.c
@@ -23,12 +23,14 @@
#include "postgres_fdw.h"
#include "storage/fd.h"
#include "storage/latch.h"
+#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/datetime.h"
#include "utils/hsearch.h"
#include "utils/inval.h"
#include "utils/memutils.h"
#include "utils/syscache.h"
+#include "utils/timeout.h"
/*
* Connection cache hash table entry
@@ -109,6 +111,10 @@ 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 void pgfdw_connection_check(void);
+static bool pgfdw_connection_check_internal(PGconn *conn);
+static TimeoutId pgfdw_health_check_timeout = MAX_TIMEOUTS;
+int pgfdw_health_check_interval;
/*
* Get a PGconn which can be used to execute queries on the remote PostgreSQL
@@ -153,6 +159,9 @@ GetConnection(UserMapping *user, bool will_prep_stmt, PgFdwConnState **state)
pgfdw_inval_callback, (Datum) 0);
CacheRegisterSyscacheCallback(USERMAPPINGOID,
pgfdw_inval_callback, (Datum) 0);
+
+ /* Register a timeout for checking remote servers */
+ pgfdw_health_check_timeout = RegisterTimeout(USER_TIMEOUT, pgfdw_connection_check);
}
/* Set flag that we did GetConnection during the current transaction */
@@ -276,6 +285,12 @@ GetConnection(UserMapping *user, bool will_prep_stmt, PgFdwConnState **state)
if (state)
*state = &entry->state;
+ /* Fire timeout if needed */
+ if (pgfdw_health_check_interval > 0 &&
+ !get_timeout_active(pgfdw_health_check_timeout))
+ enable_timeout_after(pgfdw_health_check_timeout,
+ pgfdw_health_check_interval);
+
return entry->conn;
}
@@ -1702,3 +1717,133 @@ disconnect_cached_connections(Oid serverid)
return result;
}
+
+/*
+ * Signal handler for checking remote servers.
+ *
+ * This function searches the hash table from the beginning
+ * and performs a health-check on each entry.
+ *
+ * Raise SIGINT if someone might be down, otherwise do nothing.
+ */
+void
+pgfdw_connection_check(void)
+{
+ HASH_SEQ_STATUS scan;
+ ConnCacheEntry *entry;
+ bool raised = false;
+
+ Assert(ConnectionHash);
+
+ /*
+ * checking will be done by waiting WL_SOCKET_CLOSED event,
+ * so exit immediately if it cannot be used in this system.
+ */
+ if (!WaitEventSetCanReportClosed())
+ return;
+
+ /* Is there any cancel messages? */
+ if (QeuryCancelMessage != NULL)
+ return;
+
+ hash_seq_init(&scan, ConnectionHash);
+ while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)) && !raised)
+ {
+ if (entry->conn == NULL || entry->xact_depth == 0)
+ continue;
+ if (!pgfdw_connection_check_internal(entry->conn))
+ {
+ /*
+ * Foreign server might be down, so raise SIGINT.
+ * Note that error message is passed to QeuryCancelMessage
+ * for reporting error in ProcessInterrupts().
+ */
+ char msg[31 + MAXDATELEN];
+ MemoryContext old;
+ ForeignServer *server;
+
+ /*
+ * Switch to CurTransactionContext in order to
+ * make sure that the lifetime of palloc'd is transaction.
+ */
+ old = MemoryContextSwitchTo(CurTransactionContext);
+ server = GetForeignServer(entry->serverid);
+ snprintf(msg, sizeof(msg), "Foreign Server %s might be down.", server->servername);
+ QeuryCancelMessage = pstrdup(msg);
+ MemoryContextSwitchTo(old);
+
+ disconnect_pg_server(entry);
+ raise(SIGINT);
+ raised = true;
+ break;
+ }
+ }
+
+ /* re-schedule timer if needed. */
+ if (!raised && pgfdw_health_check_interval > 0)
+ enable_timeout_after(pgfdw_health_check_timeout,
+ pgfdw_health_check_interval);
+
+ return;
+}
+
+/*
+ * helper function for pgfdw_connection_check
+ */
+static bool
+pgfdw_connection_check_internal(PGconn *conn)
+{
+ WaitEventSet *eventset;
+ WaitEvent events;
+
+ Assert(WaitEventSetCanReportClosed());
+
+ 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;
+}
+
+bool
+check_pgfdw_health_check_interval(int *newval, void **extra, GucSource source)
+{
+ if (!WaitEventSetCanReportClosed() && *newval != 0)
+ {
+ GUC_check_errdetail("pgfdw_health_check_interval must be set to 0 on this platform");
+ return false;
+ }
+ return true;
+}
+
+void
+assign_pgfdw_health_check_interval(int newval, void *extra)
+{
+ /* Quick return if timeout is not registered yet. */
+ if (pgfdw_health_check_timeout == MAX_TIMEOUTS)
+ return;
+
+ if (get_timeout_active(pgfdw_health_check_timeout))
+ {
+ if (newval == 0)
+ disable_timeout(pgfdw_health_check_timeout, false);
+
+ /*
+ * we don't have to do anything because
+ * new value will be used in pgfdw_connection_check().
+ */
+ return;
+ }
+
+ /* Start timeout if wants to */
+ if (newval > 0)
+ enable_timeout_after(pgfdw_health_check_timeout, newval);
+}
diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c
index 2c6b2894b9..75a910b0ff 100644
--- a/contrib/postgres_fdw/option.c
+++ b/contrib/postgres_fdw/option.c
@@ -538,5 +538,18 @@ _PG_init(void)
NULL,
NULL);
+ DefineCustomIntVariable("postgres_fdw.health_check_interval",
+ "Sets the time interval between checks of remote servers.",
+ NULL,
+ &pgfdw_health_check_interval,
+ 0,
+ 0,
+ INT_MAX,
+ PGC_USERSET,
+ GUC_UNIT_MS,
+ check_pgfdw_health_check_interval,
+ assign_pgfdw_health_check_interval,
+ NULL);
+
MarkGUCPrefixReserved("postgres_fdw");
}
diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h
index 8ae79e97e4..c129af5082 100644
--- a/contrib/postgres_fdw/postgres_fdw.h
+++ b/contrib/postgres_fdw/postgres_fdw.h
@@ -18,6 +18,7 @@
#include "libpq-fe.h"
#include "nodes/execnodes.h"
#include "nodes/pathnodes.h"
+#include "utils/guc.h"
#include "utils/relcache.h"
/*
@@ -151,6 +152,10 @@ 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 bool check_pgfdw_health_check_interval(int *newval, void **extra,
+ GucSource source);
+extern void assign_pgfdw_health_check_interval(int newval, void *extra);
+extern int pgfdw_health_check_interval;
/* in option.c */
extern int ExtractConnectionOptions(List *defelems,
[application/octet-stream] v11_0003_add_doc.patch (1.4K, ../TYAPR01MB5866FF28C802916242F1B492F53B9@TYAPR01MB5866.jpnprd01.prod.outlook.com/4-v11_0003_add_doc.patch)
download | inline diff:
diff --git a/doc/src/sgml/postgres-fdw.sgml b/doc/src/sgml/postgres-fdw.sgml
index dc57fe4b0d..21532e19d9 100644
--- a/doc/src/sgml/postgres-fdw.sgml
+++ b/doc/src/sgml/postgres-fdw.sgml
@@ -1028,6 +1028,32 @@ postgres=# SELECT postgres_fdw_disconnect_all();
</listitem>
</varlistentry>
+ <varlistentry id="guc-pgfdw-health-check-interval" xreflabel="postgres_fdw.health_check_interval">
+ <term>
+ <varname>postgres_fdw.health_check_interval</varname> (<type>integer</type>)
+ <indexterm>
+ <primary><varname>postgres_fdw.health_check_interval</varname> configuration parameter</primary>
+ </indexterm>
+ </term>
+ <listitem>
+ <para>
+ Sets the time interval between optional checks that remote servers
+ are still alive. When losing a remote connection is detected,
+ the running transaction is aborted. This feature is performed
+ by polling the socket.
+ </para>
+ <para>
+ this option relies on kernel events exposed by Linux, macOS,
+ illumos and the BSD family of operating systems, and is not currently available
+ on other systems.
+ </para>
+ <para>
+ If the value is specified without units, it is taken as milliseconds.
+ The default value is <literal>0</literal>, which disables connection
+ checks.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist>
</sect2>
[application/x-zip-compressed] v11_0004_add_test.zip (855B, ../TYAPR01MB5866FF28C802916242F1B492F53B9@TYAPR01MB5866.jpnprd01.prod.outlook.com/5-v11_0004_add_test.zip)
download
view thread (30+ 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], [email protected], [email protected], [email protected]
Subject: RE: [Proposal] Add foreign-server health checks infrastructure
In-Reply-To: <TYAPR01MB5866FF28C802916242F1B492F53B9@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