pgpool2.git / summary / log / commit / refs
commit 87e6e11c1931555e90a1cc65cc3ee79759ebaaab
Author: Muhammad Usama <m.usama@highgo.ca>
Commit: Muhammad Usama <m.usama@gmail.com>
Date: Mon May 04 11:38:39 2020 +0000
New feature: cluster scope for pcp_stop_pgpool utility command
Make pcp_stop_pgpool utility to terminate (optional) all the pgpool nodes in the
watchdog cluster
For that purpose added "--scope=cluster/local" option in the pcp_stop_pgpool
utility to set the scope of the command. The scope could be either
'local' or 'cluster'
Along with that, the commit adds the infrastructure in watchdog for broadcasting
the custom commands over the watchdog network. This will make it easier to
implement the --scope=cluster for other PCP utilities.
doc/src/sgml/ref/pcp_stop_pgpool.sgml | 17 ++++
src/include/pcp/libpcp_ext.h | 2 +-
src/include/watchdog/watchdog.h | 12 +++
src/include/watchdog/wd_internal_commands.h | 4 +-
src/include/watchdog/wd_ipc_defines.h | 15 +++-
src/include/watchdog/wd_json_data.h | 10 +++
src/libs/pcp/pcp.c | 7 +-
src/pcp_con/pcp_worker.c | 23 ++++-
src/tools/pcp/pcp_frontend_client.c | 29 ++++++-
src/watchdog/watchdog.c | 130 ++++++++++++++++++++++++++++
src/watchdog/wd_internal_commands.c | 50 +++++++++++
src/watchdog/wd_json_data.c | 118 +++++++++++++++++++++++++
12 files changed, 404 insertions(+), 13 deletions(-)
diff --git a/doc/src/sgml/ref/pcp_stop_pgpool.sgml b/doc/src/sgml/ref/pcp_stop_pgpool.sgml
index d8fb83382..cdec14d89 100644
--- a/doc/src/sgml/ref/pcp_stop_pgpool.sgml
+++ b/doc/src/sgml/ref/pcp_stop_pgpool.sgml
@@ -59,6 +59,23 @@ Pgpool-II documentation
</listitem>
</varlistentry>
+ <varlistentry>
+ <term><option>-s <replaceable class="parameter">scope</replaceable></option></term>
+ <term><option>--scope=<replaceable class="parameter">scope</replaceable></option></term>
+ <listitem>
+ <para>
+ Shutdown mode for terminating the Pgpool-II process.
+ </para>
+ <para>
+ The supported command scopes are as follows:
+ <itemizedlist>
+ <listitem><para>c, cluster : terminates all Pgpool-II nodes part of the cluster </para></listitem>
+ <listitem><para>l, local : terminates local Pgpool-II node only </para></listitem>
+ </itemizedlist>
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry>
<term><option>Other options </option></term>
<listitem>
diff --git a/src/include/pcp/libpcp_ext.h b/src/include/pcp/libpcp_ext.h
index caf05bbdc..179f15ab4 100644
--- a/src/include/pcp/libpcp_ext.h
+++ b/src/include/pcp/libpcp_ext.h
@@ -313,7 +313,7 @@ struct WdInfo;
extern PCPConnInfo * pcp_connect(char *hostname, int port, char *username, char *password, FILE *Pfdebug);
extern void pcp_disconnect(PCPConnInfo * pcpConn);
-extern PCPResultInfo * pcp_terminate_pgpool(PCPConnInfo * pcpCon, char mode);
+extern PCPResultInfo * pcp_terminate_pgpool(PCPConnInfo * pcpConn, char mode, char command_scope);
extern PCPResultInfo * pcp_node_count(PCPConnInfo * pcpCon);
extern PCPResultInfo * pcp_node_info(PCPConnInfo * pcpCon, int nid);
extern PCPResultInfo * pcp_health_check_stats(PCPConnInfo * pcpCon, int nid);
diff --git a/src/include/watchdog/watchdog.h b/src/include/watchdog/watchdog.h
index 83f6651b2..cee44e050 100644
--- a/src/include/watchdog/watchdog.h
+++ b/src/include/watchdog/watchdog.h
@@ -187,6 +187,18 @@ typedef struct WatchdogNode
* initiated by local */
} WatchdogNode;
+/*
+ * Argument for WD Exec cluster command
+ */
+#define WD_MAX_ARG_NAME_LEN 64
+#define WD_MAX_ARG_VALUE_LEN 64
+
+typedef struct WDExecCommandArg
+{
+ char arg_name[WD_MAX_ARG_NAME_LEN];
+ char arg_value[WD_MAX_ARG_VALUE_LEN];
+} WDExecCommandArg;
+
extern pid_t initialize_watchdog(void);
#endif /* WATCHDOG_H */
diff --git a/src/include/watchdog/wd_internal_commands.h b/src/include/watchdog/wd_internal_commands.h
index e22401830..d8d701475 100644
--- a/src/include/watchdog/wd_internal_commands.h
+++ b/src/include/watchdog/wd_internal_commands.h
@@ -37,6 +37,9 @@ extern WDFailoverCMDResults wd_send_failback_request(int node_id, unsigned char
extern WDFailoverCMDResults wd_degenerate_backend_set(int *node_id_set, int count, unsigned char flags);
extern WDFailoverCMDResults wd_promote_backend(int node_id, unsigned char flags);
+extern WdCommandResult wd_execute_cluster_command(char* clusterCommand,
+ int nArgs, WDExecCommandArg *wdExecCommandArg);
+
extern WDPGBackendStatus * get_pg_backend_status_from_master_wd_node(void);
extern WD_STATES wd_internal_get_watchdog_local_node_state(void);
@@ -57,5 +60,4 @@ extern void set_watchdog_node_escalated(void);
extern void reset_watchdog_node_escalated(void);
extern bool get_watchdog_node_escalation_state(void);
-
#endif /* WD_INTERNAL_COMMANDS_H */
diff --git a/src/include/watchdog/wd_ipc_defines.h b/src/include/watchdog/wd_ipc_defines.h
index 0bda9c1bb..74f9b85c9 100644
--- a/src/include/watchdog/wd_ipc_defines.h
+++ b/src/include/watchdog/wd_ipc_defines.h
@@ -6,7 +6,7 @@
* pgpool: a language independent connection pool server for PostgreSQL
* written by Tatsuo Ishii
*
- * Copyright (c) 2003-2019 PgPool Global Development Group
+ * Copyright (c) 2003-2020 PgPool Global Development Group
*
* Permission to use, copy, modify, and distribute this software and
* its documentation for any purpose and without fee is hereby
@@ -62,13 +62,20 @@ typedef enum WDValueDataType
#define WD_IPC_CMD_RESULT_OK '7'
#define WD_IPC_CMD_TIMEOUT '8'
+#define WD_EXECUTE_CLUSTER_COMMAND 'c'
#define WD_IPC_FAILOVER_COMMAND 'f'
-#define WD_IPC_ONLINE_RECOVERY_COMMAND 'r'
-#define WD_FAILOVER_LOCKING_REQUEST 's'
+#define WD_IPC_ONLINE_RECOVERY_COMMAND 'r'
+#define WD_FAILOVER_LOCKING_REQUEST 's'
#define WD_GET_MASTER_DATA_REQUEST 'd'
#define WD_GET_RUNTIME_VARIABLE_VALUE 'v'
#define WD_FAILOVER_INDICATION 'i'
+
+#define WD_COMMAND_RESTART_CLUSTER "RESTART_CLUSTER"
+#define WD_COMMAND_REELECT_MASTER "REELECT_MASTER"
+#define WD_COMMAND_SHUTDOWN_CLUSTER "SHUTDOWN_CLUSTER"
+
+
#define WD_FUNCTION_START_RECOVERY "START_RECOVERY"
#define WD_FUNCTION_END_RECOVERY "END_RECOVERY"
#define WD_FUNCTION_FAILBACK_REQUEST "FAILBACK_REQUEST"
@@ -112,7 +119,7 @@ typedef enum WDValueDataType
#define WD_RUNTIME_VAR_ESCALATION_STATE "Escalated"
/* Use to inform node new node status by lifecheck */
-#define WD_LIFECHECK_NODE_STATUS_DEAD 1
+#define WD_LIFECHECK_NODE_STATUS_DEAD 1
#define WD_LIFECHECK_NODE_STATUS_ALIVE 2
diff --git a/src/include/watchdog/wd_json_data.h b/src/include/watchdog/wd_json_data.h
index 0080a217e..6bd460632 100644
--- a/src/include/watchdog/wd_json_data.h
+++ b/src/include/watchdog/wd_json_data.h
@@ -69,4 +69,14 @@ extern char *get_simple_request_json(char *key, char *value, unsigned int shared
extern bool parse_data_request_json(char *json_data, int data_len, char **request_type);
extern char *get_data_request_json(char *request_type, unsigned int sharedKey, char *authKey);
+extern bool
+parse_wd_exec_cluster_command_json(char *json_data, int data_len,
+ char **clusterCommand,
+ int *nArgs, WDExecCommandArg **wdExecCommandArg);
+
+extern char *
+get_wd_exec_cluster_command_json(char *clusterCommand,int nArgs,
+ WDExecCommandArg *wdExecCommandArg,
+ unsigned int sharedKey, char *authKey);
+
#endif
diff --git a/src/libs/pcp/pcp.c b/src/libs/pcp/pcp.c
index 788d7224a..0e25ff844 100644
--- a/src/libs/pcp/pcp.c
+++ b/src/libs/pcp/pcp.c
@@ -608,7 +608,7 @@ pcp_disconnect(PCPConnInfo * pcpConn)
* --------------------------------
*/
PCPResultInfo *
-pcp_terminate_pgpool(PCPConnInfo * pcpConn, char mode)
+pcp_terminate_pgpool(PCPConnInfo * pcpConn, char mode, char command_scope)
{
int wsize;
@@ -617,7 +617,10 @@ pcp_terminate_pgpool(PCPConnInfo * pcpConn, char mode)
pcp_internal_error(pcpConn, "invalid PCP connection");
return NULL;
}
- pcp_write(pcpConn->pcpConn, "T", 1);
+ if (command_scope == 'l') /*local only*/
+ pcp_write(pcpConn->pcpConn, "T", 1);
+ else
+ pcp_write(pcpConn->pcpConn, "t", 1);
wsize = htonl(sizeof(int) + sizeof(char));
pcp_write(pcpConn->pcpConn, &wsize, sizeof(int));
pcp_write(pcpConn->pcpConn, &mode, sizeof(char));
diff --git a/src/pcp_con/pcp_worker.c b/src/pcp_con/pcp_worker.c
index 7dc6b85b2..c547d2a75 100644
--- a/src/pcp_con/pcp_worker.c
+++ b/src/pcp_con/pcp_worker.c
@@ -87,7 +87,7 @@ static void process_attach_node(PCP_CONNECTION * frontend, char *buf);
static void process_recovery_request(PCP_CONNECTION * frontend, char *buf);
static void process_status_request(PCP_CONNECTION * frontend);
static void process_promote_node(PCP_CONNECTION * frontend, char *buf, char tos);
-static void process_shutown_request(PCP_CONNECTION * frontend, char mode);
+static void process_shutown_request(PCP_CONNECTION * frontend, char mode, char tos);
static void process_set_configration_parameter(PCP_CONNECTION * frontend, char *buf, int len);
static void pcp_worker_will_go_down(int code, Datum arg);
@@ -304,8 +304,9 @@ pcp_process_command(char tos, char *buf, int buf_len)
break;
case 'T':
+ case 't':
set_ps_display("PCP: processing shutdown request", false);
- process_shutown_request(pcp_frontend, buf[0]);
+ process_shutown_request(pcp_frontend, buf[0], tos);
break;
case 'O': /* recovery request */
@@ -1267,7 +1268,7 @@ send_md5salt(PCP_CONNECTION * frontend, char *salt)
}
static void
-process_shutown_request(PCP_CONNECTION * frontend, char mode)
+process_shutown_request(PCP_CONNECTION * frontend, char mode, char tos)
{
char code[] = "CommandComplete";
int len;
@@ -1286,6 +1287,22 @@ process_shutown_request(PCP_CONNECTION * frontend, char mode)
errdetail("invalid shutdown mode \"%c\"", mode)));
}
+ if (tos == 't' && pool_config->use_watchdog)
+ {
+ WDExecCommandArg wdExecCommandArg;
+
+ strncpy(wdExecCommandArg.arg_name, "mode", sizeof(wdExecCommandArg.arg_name) - 1);
+ snprintf(wdExecCommandArg.arg_value, sizeof(wdExecCommandArg.arg_name) - 1, "%c",mode);
+
+ ereport(LOG,
+ (errmsg("PCP: sending command to watchdog to shutdown cluster")));
+
+ if (wd_execute_cluster_command(WD_COMMAND_SHUTDOWN_CLUSTER,1, &wdExecCommandArg) != COMMAND_OK)
+ ereport(ERROR,
+ (errmsg("PCP: error while processing shutdown cluster request"),
+ errdetail("failed to propogate shutdown command through watchdog")));
+ }
+
pcp_write(frontend, "t", 1);
len = htonl(sizeof(code) + sizeof(int));
pcp_write(frontend, &len, sizeof(int));
diff --git a/src/tools/pcp/pcp_frontend_client.c b/src/tools/pcp/pcp_frontend_client.c
index b89c5030f..6637d2499 100644
--- a/src/tools/pcp/pcp_frontend_client.c
+++ b/src/tools/pcp/pcp_frontend_client.c
@@ -90,7 +90,7 @@ struct AppTypes AllAppTypes[] =
{"pcp_proc_info", PCP_PROC_INFO, "h:p:P:U:awWvd", "display a pgpool-II child process' information"},
{"pcp_promote_node", PCP_PROMOTE_NODE, "n:h:p:U:gwWvd", "promote a node as new master from pgpool-II"},
{"pcp_recovery_node", PCP_RECOVERY_NODE, "n:h:p:U:wWvd", "recover a node"},
- {"pcp_stop_pgpool", PCP_STOP_PGPOOL, "m:h:p:U:wWvd", "terminate pgpool-II"},
+ {"pcp_stop_pgpool", PCP_STOP_PGPOOL, "m:h:p:U:s:wWvda", "terminate pgpool-II"},
{"pcp_watchdog_info", PCP_WATCHDOG_INFO, "n:h:p:U:wWvd", "display a pgpool-II watchdog's information"},
{NULL, UNKNOWN, NULL, NULL},
};
@@ -107,6 +107,7 @@ main(int argc, char **argv)
int processID = 0;
int ch;
char shutdown_mode = 's';
+ char command_scope = 'l';
int optindex;
int i;
bool all = false;
@@ -129,6 +130,7 @@ main(int argc, char **argv)
{"no-password", no_argument, NULL, 'w'},
{"password", no_argument, NULL, 'W'},
{"mode", required_argument, NULL, 'm'},
+ {"scope", required_argument, NULL, 's'},
{"gracefully", no_argument, NULL, 'g'},
{"verbose", no_argument, NULL, 'v'},
{"all", no_argument, NULL, 'a'},
@@ -193,6 +195,26 @@ main(int argc, char **argv)
gracefully = true;
break;
+ case 's':
+ if (current_app_type->app_type == PCP_STOP_PGPOOL)
+ {
+ if (strcmp(optarg, "c") == 0 || strcmp(optarg, "cluster") == 0)
+ command_scope = 'c';
+ else if (strcmp(optarg, "l") == 0 || strcmp(optarg, "local") == 0)
+ command_scope = 'l';
+ else
+ {
+ fprintf(stderr, "%s: Invalid command socpe \"%s\", must be either \"cluster\" or \"local\" \n", progname, optarg);
+ exit(1);
+ }
+ }
+ else
+ {
+ fprintf(stderr, "Invalid argument \"%s\", Try \"%s --help\" for more information.\n", optarg, progname);
+ exit(1);
+ }
+ break;
+
case 'm':
if (current_app_type->app_type == PCP_STOP_PGPOOL)
{
@@ -404,7 +426,7 @@ main(int argc, char **argv)
else if (current_app_type->app_type == PCP_STOP_PGPOOL)
{
- pcpResInfo = pcp_terminate_pgpool(pcpConn, shutdown_mode);
+ pcpResInfo = pcp_terminate_pgpool(pcpConn, shutdown_mode, command_scope);
}
else if (current_app_type->app_type == PCP_WATCHDOG_INFO)
@@ -819,6 +841,9 @@ usage(void)
if (current_app_type->app_type == PCP_STOP_PGPOOL)
{
fprintf(stderr, " -m, --mode=MODE MODE can be \"smart\", \"fast\", or \"immediate\"\n");
+ fprintf(stderr, " -s, --scope=SCOPE SCOPE can be \"cluster\", or \"local\"\n");
+ fprintf(stderr, " cluster scope terminates all Pgpool-II nodes part\n");
+ fprintf(stderr, " of the watchdog cluster\n");
}
if (current_app_type->app_type == PCP_PROMOTE_NODE ||
current_app_type->app_type == PCP_DETACH_NODE)
diff --git a/src/watchdog/watchdog.c b/src/watchdog/watchdog.c
index ea5173182..f13e7632d 100644
--- a/src/watchdog/watchdog.c
+++ b/src/watchdog/watchdog.c
@@ -134,6 +134,8 @@ typedef enum IPC_CMD_PREOCESS_RES
#define WD_CMD_REPLY_IN_DATA '-'
#define WD_CLUSTER_SERVICE_MESSAGE '#'
+#define WD_EXECUTE_COMMAND_REQUEST '!'
+
#define WD_FAILOVER_START 'F'
#define WD_FAILOVER_END 'H'
#define WD_FAILOVER_WAITING_FOR_CONSENSUS 'K'
@@ -151,6 +153,7 @@ typedef enum IPC_CMD_PREOCESS_RES
#define CLUSTER_NODE_APPEARING_LOST 'Y'
#define CLUSTER_NODE_APPEARING_FOUND 'Z'
+
#define WD_MASTER_NODE getMasterWatchdogNode()
typedef struct packet_types
@@ -175,6 +178,7 @@ packet_types all_packet_types[] = {
{WD_STAND_FOR_COORDINATOR_MESSAGE, "STAND FOR COORDINATOR"},
{WD_REMOTE_FAILOVER_REQUEST, "REPLICATE FAILOVER REQUEST"},
{WD_IPC_ONLINE_RECOVERY_COMMAND, "ONLINE RECOVERY REQUEST"},
+ {WD_EXECUTE_CLUSTER_COMMAND, "EXECUTE CLUSTER COMMAND"},
{WD_IPC_FAILOVER_COMMAND, "FAILOVER FUNCTION COMMAND"},
{WD_INFORM_I_AM_GOING_DOWN, "INFORM I AM GOING DOWN"},
{WD_ASK_FOR_POOL_CONFIG, "ASK FOR POOL CONFIG"},
@@ -192,6 +196,7 @@ packet_types all_packet_types[] = {
{WD_IPC_CMD_RESULT_BAD, "IPC RESPONSE BAD"},
{WD_IPC_CMD_RESULT_OK, "IPC RESPONSE GOOD"},
{WD_IPC_CMD_TIMEOUT, "IPC TIMEOUT"},
+ {WD_EXECUTE_COMMAND_REQUEST, "WD EXECUTE COMMAND"},
{WD_NO_MESSAGE, ""}
};
@@ -540,6 +545,7 @@ static IPC_CMD_PREOCESS_RES process_IPC_failover_indication(WDCommandData * ipcC
static IPC_CMD_PREOCESS_RES process_IPC_data_request_from_master(WDCommandData * ipcCommand);
static IPC_CMD_PREOCESS_RES process_IPC_failover_command(WDCommandData * ipcCommand);
static IPC_CMD_PREOCESS_RES process_failover_command_on_coordinator(WDCommandData * ipcCommand);
+static IPC_CMD_PREOCESS_RES process_IPC_execute_cluster_command(WDCommandData * ipcCommand);
static bool write_ipc_command_with_result_data(WDCommandData * ipcCommand, char type, char *data, int len);
@@ -578,6 +584,7 @@ static void clear_standby_nodes_list(void);
static int standby_node_left_cluster(WatchdogNode * wdNode);
static int standby_node_join_cluster(WatchdogNode * wdNode);
static void update_missed_beacon_count(WDCommandData* ipcCommand, bool clear);
+static void wd_execute_cluster_command_processor(WatchdogNode * wdNode, WDPacketData * pkt);
/* global variables */
wd_cluster g_cluster;
@@ -2000,12 +2007,20 @@ static IPC_CMD_PREOCESS_RES process_IPC_command(WDCommandData * ipcCommand)
case WD_FAILOVER_INDICATION:
return process_IPC_failover_indication(ipcCommand);
+ break;
case WD_GET_MASTER_DATA_REQUEST:
return process_IPC_data_request_from_master(ipcCommand);
+ break;
case WD_GET_RUNTIME_VARIABLE_VALUE:
return process_IPC_get_runtime_variable_value_request(ipcCommand);
+ break;
+
+ case WD_EXECUTE_CLUSTER_COMMAND:
+ return process_IPC_execute_cluster_command(ipcCommand);
+ break;
+
default:
ipcCommand->errorMessage = MemoryContextStrdup(ipcCommand->memoryContext, "unknown IPC command type");
break;
@@ -2013,6 +2028,56 @@ static IPC_CMD_PREOCESS_RES process_IPC_command(WDCommandData * ipcCommand)
return IPC_CMD_ERROR;
}
+static IPC_CMD_PREOCESS_RES
+process_IPC_execute_cluster_command(WDCommandData * ipcCommand)
+{
+ /* get the json for node list */
+ char *clusterCommand = NULL;
+ int nArgs;
+ WDExecCommandArg *wdExecCommandArg = NULL;
+
+ if (ipcCommand->sourcePacket.len <= 0 || ipcCommand->sourcePacket.data == NULL)
+ return IPC_CMD_ERROR;
+
+ if (!parse_wd_exec_cluster_command_json(ipcCommand->sourcePacket.data, ipcCommand->sourcePacket.len,
+ &clusterCommand, &nArgs, &wdExecCommandArg))
+ {
+ goto ERROR_EXIT;
+ }
+ if (strcasecmp(WD_COMMAND_SHUTDOWN_CLUSTER, clusterCommand) == 0)
+ {
+ ereport(LOG,
+ (errmsg("Watchdog has received shutdown cluster command from IPC channel")));
+ }
+ else
+ {
+ ipcCommand->errorMessage = MemoryContextStrdup(ipcCommand->memoryContext,
+ "unknown cluster command requested");
+ goto ERROR_EXIT;
+ }
+
+ /*
+ * Just broadcast the execute command request to destination node
+ * Processing the command on the local node is the responsibility of caller
+ * process
+ */
+ reply_with_message(NULL, WD_EXECUTE_COMMAND_REQUEST,
+ ipcCommand->sourcePacket.data, ipcCommand->sourcePacket.len,
+ NULL);
+
+ if (wdExecCommandArg)
+ pfree(wdExecCommandArg);
+ if (clusterCommand)
+ pfree(clusterCommand);
+ return IPC_CMD_OK;
+
+ERROR_EXIT:
+ if (wdExecCommandArg)
+ pfree(wdExecCommandArg);
+ if (clusterCommand)
+ pfree(clusterCommand);
+ return IPC_CMD_ERROR;
+}
static IPC_CMD_PREOCESS_RES process_IPC_get_runtime_variable_value_request(WDCommandData * ipcCommand)
{
@@ -3921,6 +3986,66 @@ cluster_service_message_processor(WatchdogNode * wdNode, WDPacketData * pkt)
}
}
+static void
+wd_execute_cluster_command_processor(WatchdogNode * wdNode, WDPacketData * pkt)
+{
+ /* get the json for node list */
+ char *clusterCommand = NULL;
+ int nArgs;
+ WDExecCommandArg *wdExecCommandArg = NULL;
+
+ if (pkt->type != WD_EXECUTE_COMMAND_REQUEST)
+ return;
+
+ if (pkt->len <= 0 || pkt->data == NULL)
+ {
+ ereport(LOG,
+ (errmsg("node \"%s\" sent an empty execute cluster command message", wdNode->nodeName)));
+ return;
+ }
+
+ if (!parse_wd_exec_cluster_command_json(pkt->data, pkt->len,
+ &clusterCommand, &nArgs, &wdExecCommandArg))
+ {
+ ereport(LOG,
+ (errmsg("node \"%s\" sent an invalid JSON data in cluster command message", wdNode->nodeName)));
+ return;
+ }
+
+ ereport(DEBUG1,
+ (errmsg("received \"%s\" command from node \"%s\"",clusterCommand, wdNode->nodeName)));
+
+ if (strcasecmp(WD_COMMAND_SHUTDOWN_CLUSTER, clusterCommand) == 0)
+ {
+ int i;
+ char mode = 's';
+ for ( i =0; i < nArgs; i++)
+ {
+ if (strcmp(wdExecCommandArg[i].arg_name, "mode") == 0)
+ {
+ mode = wdExecCommandArg[i].arg_value[0];
+ }
+ else
+ ereport(LOG,
+ (errmsg("unsupported argument \"%s\" in shutdown command from remote node \"%s\"", wdExecCommandArg[i].arg_name, wdNode->nodeName)));
+ }
+ ereport(LOG,
+ (errmsg("processing shutdown command from remote node \"%s\"", wdNode->nodeName)));
+ terminate_pgpool(mode, false);
+ }
+ else
+ {
+ ereport(WARNING,
+ (errmsg("received \"%s\" command from node \"%s\" is not supported",clusterCommand, wdNode->nodeName)));
+ }
+
+ if (wdExecCommandArg)
+ pfree(wdExecCommandArg);
+ if (clusterCommand)
+ pfree(clusterCommand);
+ return;
+}
+
static int
standard_packet_processor(WatchdogNode * wdNode, WDPacketData * pkt)
{
@@ -3934,6 +4059,10 @@ standard_packet_processor(WatchdogNode * wdNode, WDPacketData * pkt)
register_inform_quarantine_nodes_req();
break;
+ case WD_EXECUTE_COMMAND_REQUEST:
+ wd_execute_cluster_command_processor(wdNode, pkt);
+ break;
+
case WD_CLUSTER_SERVICE_MESSAGE:
cluster_service_message_processor(wdNode, pkt);
break;
@@ -7419,6 +7548,7 @@ check_and_report_IPC_authentication(WDCommandData * ipcCommand)
case WD_IPC_FAILOVER_COMMAND:
case WD_IPC_ONLINE_RECOVERY_COMMAND:
+ case WD_EXECUTE_CLUSTER_COMMAND:
case WD_GET_MASTER_DATA_REQUEST:
/* only allowed internaly. */
internal_client_only = true;
diff --git a/src/watchdog/wd_internal_commands.c b/src/watchdog/wd_internal_commands.c
index 9c121290e..e85b49cda 100644
--- a/src/watchdog/wd_internal_commands.c
+++ b/src/watchdog/wd_internal_commands.c
@@ -266,6 +266,56 @@ wd_end_recovery(void)
return COMMAND_FAILED;
}
+WdCommandResult
+wd_execute_cluster_command(char* clusterCommand,
+ int nArgs, WDExecCommandArg *wdExecCommandArg)
+{
+ char type;
+ unsigned int *shared_key = get_ipc_shared_key();
+
+ char *func = get_wd_exec_cluster_command_json(clusterCommand, nArgs, wdExecCommandArg,
+ shared_key ? *shared_key : 0, pool_config->wd_authkey);
+
+ WDIPCCmdResult *result = issue_command_to_watchdog(WD_EXECUTE_CLUSTER_COMMAND,
+ WD_DEFAULT_IPC_COMMAND_TIMEOUT,
+ func, strlen(func), true);
+
+ pfree(func);
+
+ if (result == NULL)
+ {
+ ereport(WARNING,
+ (errmsg("execute cluster command failed"),
+ errdetail("issue command to watchdog returned NULL")));
+ return COMMAND_FAILED;
+ }
+
+ type = result->type;
+ FreeCmdResult(result);
+
+ if (type == WD_IPC_CMD_CLUSTER_IN_TRAN)
+ {
+ ereport(WARNING,
+ (errmsg("execute cluster command failed"),
+ errdetail("watchdog cluster is not in stable state"),
+ errhint("try again when the cluster is fully initialized")));
+ return CLUSTER_IN_TRANSATIONING;
+ }
+ else if (type == WD_IPC_CMD_TIMEOUT)
+ {
+ ereport(WARNING,
+ (errmsg("execute cluster command failed"),
+ errdetail("ipc command timeout")));
+ return COMMAND_TIMEOUT;
+ }
+ else if (type == WD_IPC_CMD_RESULT_OK)
+ {
+ return COMMAND_OK;
+ }
+ return COMMAND_FAILED;
+}
+
+
static char *
get_wd_failover_state_json(bool start)
{
diff --git a/src/watchdog/wd_json_data.c b/src/watchdog/wd_json_data.c
index 61db6d71a..3d60bd9e4 100644
--- a/src/watchdog/wd_json_data.c
+++ b/src/watchdog/wd_json_data.c
@@ -805,3 +805,121 @@ get_wd_simple_message_json(char *message)
jw_destroy(jNode);
return json_str;
}
+
+char *
+get_wd_exec_cluster_command_json(char *clusterCommand, int nArgs,
+ WDExecCommandArg *wdExecCommandArg,
+ unsigned int sharedKey, char *authKey)
+{
+ int i;
+ char *json_str;
+ JsonNode *jNode = jw_create_with_object(true);
+
+ jw_put_int(jNode, WD_IPC_SHARED_KEY, sharedKey); /* put the shared key */
+
+ if (authKey != NULL && strlen(authKey) > 0)
+ jw_put_string(jNode, WD_IPC_AUTH_KEY, authKey); /* put the auth key */
+
+ jw_put_string(jNode, "Command", clusterCommand);
+
+ jw_put_int(jNode, "nArgs", nArgs);
+
+ /* Array of arguments */
+ jw_start_array(jNode, "argument_list");
+ for (i = 0; i < nArgs; i++)
+ {
+ jw_start_object(jNode, "Arg");
+ jw_put_string(jNode, "arg_name", wdExecCommandArg[i].arg_name);
+ jw_put_string(jNode, "arg_value", wdExecCommandArg[i].arg_value);
+ jw_end_element(jNode);
+ }
+ jw_end_element(jNode); /* argument_list array End */
+
+ jw_finish_document(jNode);
+ json_str = pstrdup(jw_get_json_string(jNode));
+ jw_destroy(jNode);
+ return json_str;
+}
+
+bool
+parse_wd_exec_cluster_command_json(char *json_data, int data_len,
+ char **clusterCommand,
+ int *nArgs, WDExecCommandArg **wdExecCommandArg)
+{
+ json_value *root;
+ char *ptr = NULL;
+ int i;
+
+ root = json_parse(json_data, data_len);
+
+ /* The root node must be object */
+ if (root == NULL || root->type != json_object)
+ {
+ json_value_free(root);
+ ereport(LOG,
+ (errmsg("watchdog is unable to parse exec cluster command json"),
+ errdetail("invalid json data \"%.*s\"", data_len, json_data)));
+ return false;
+ }
+ ptr = json_get_string_value_for_key(root, "Command");
+ if (ptr == NULL)
+ {
+ json_value_free(root);
+ ereport(LOG,
+ (errmsg("watchdog is unable to parse exec cluster command json"),
+ errdetail("command node not found in json data \"%s\"", json_data)));
+ return false;
+ }
+ *clusterCommand = pstrdup(ptr);
+
+ if (json_get_int_value_for_key(root, "nArgs", nArgs))
+ {
+ /* nArgs not found, Just ignore it */
+ *nArgs = 0;
+ /* it may be from the old version */
+ }
+ if (*nArgs > 0)
+ {
+ json_value *value;
+
+ *wdExecCommandArg = palloc0(sizeof(WDExecCommandArg) * *nArgs);
+
+ /* backend_desc array */
+ value = json_get_value_for_key(root, "argument_list");
+ if (value == NULL || value->type != json_array)
+ goto ERROR_EXIT;
+
+ if (*nArgs!= value->u.array.length)
+ {
+ ereport(LOG,
+ (errmsg("watchdog is unable to parse exec cluster command json"),
+ errdetail("nArgs is different than argument array length \"%s\"", json_data)));
+ goto ERROR_EXIT;
+ }
+ for (i = 0; i < *nArgs; i++)
+ {
+ json_value *arr_value = value->u.array.values[i];
+ char *ptr;
+
+ ptr = json_get_string_value_for_key(arr_value, "arg_name");
+ if (ptr == NULL)
+ goto ERROR_EXIT;
+ strncpy(wdExecCommandArg[i]->arg_name, ptr, sizeof(wdExecCommandArg[i]->arg_name) - 1);
+
+ ptr = json_get_string_value_for_key(arr_value, "arg_value");
+ if (ptr == NULL)
+ goto ERROR_EXIT;
+ strncpy(wdExecCommandArg[i]->arg_value, ptr, sizeof(wdExecCommandArg[i]->arg_value) - 1);
+ }
+ }
+
+ json_value_free(root);
+ return true;
+
+ ERROR_EXIT:
+ if (root)
+ json_value_free(root);
+ if (*wdExecCommandArg )
+ pfree(*wdExecCommandArg);
+ return false;
+}
[parent: 3c68ae5a3681]