From 5755feee5de09651200721b17473bba7d15c00f5 Mon Sep 17 00:00:00 2001 From: xue_meng_en <1836611252@qq.com> Date: Thu, 7 Apr 2022 14:50:24 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=91=E5=B8=83=E8=AE=A2=E9=98=85=E6=94=AF?= =?UTF-8?q?=E6=8C=81=E4=B8=BB=E5=A4=87=E5=88=87=E6=8D=A2=E4=B8=8D=E6=96=AD?= =?UTF-8?q?=E5=BC=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/bin/pg_dump/pg_dump.cpp | 2 +- .../backend/catalog/pg_subscription.cpp | 27 +- .../optimizer/commands/subscriptioncmds.cpp | 287 +++++++++++++++--- .../storage/replication/libpqwalreceiver.cpp | 8 +- .../storage/replication/logical/proto.cpp | 28 +- .../storage/replication/logical/worker.cpp | 78 ++++- .../storage/replication/pgoutput/pgoutput.cpp | 60 +++- src/include/catalog/pg_subscription.h | 3 +- src/include/commands/subscriptioncmds.h | 11 + src/include/replication/libpqwalreceiver.h | 1 + src/include/replication/logicalproto.h | 2 + src/test/regress/input/subscription.source | 6 +- src/test/regress/output/subscription.source | 19 +- 13 files changed, 436 insertions(+), 96 deletions(-) diff --git a/src/bin/pg_dump/pg_dump.cpp b/src/bin/pg_dump/pg_dump.cpp index a6d00d505..e734947b5 100644 --- a/src/bin/pg_dump/pg_dump.cpp +++ b/src/bin/pg_dump/pg_dump.cpp @@ -233,7 +233,7 @@ char* all_data_nodename_list = NULL; const uint32 USTORE_UPGRADE_VERSION = 92368; const uint32 PACKAGE_ENHANCEMENT = 92444; const uint32 SUBSCRIPTION_VERSION = 92580; -const uint32 SUBSCRIPTION_BINARY_VERSION_NUM = 92607; +const uint32 SUBSCRIPTION_BINARY_VERSION_NUM = 92606; #ifdef DUMPSYSLOG char* syslogpath = NULL; diff --git a/src/common/backend/catalog/pg_subscription.cpp b/src/common/backend/catalog/pg_subscription.cpp index f2d31b218..ce7eaf981 100644 --- a/src/common/backend/catalog/pg_subscription.cpp +++ b/src/common/backend/catalog/pg_subscription.cpp @@ -28,7 +28,6 @@ #include "utils/builtins.h" #include "utils/fmgroids.h" #include "utils/syscache.h" -#include "replication/worker_internal.h" static List *textarray_to_stringlist(ArrayType *textarray); @@ -92,7 +91,10 @@ Subscription *GetSubscription(Oid subid, bool missing_ok) sub->publications = textarray_to_stringlist(DatumGetArrayTypeP(datum)); datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup, Anum_pg_subscription_subbinary, &isnull); - Assert(!isnull); + if (unlikely(isnull)) { + ereport(ERROR, (errcode(ERRCODE_UNEXPECTED_NULL_VALUE), + errmsg("null binary for subscription %u", subid))); + } sub->binary = DatumGetBool(datum); ReleaseSysCache(tup); @@ -187,7 +189,7 @@ char *get_subscription_name(Oid subid, bool missing_ok) } /* Clear the list content, only deal with DefElem and string content */ -static void ClearListContent(List *list) +void ClearListContent(List *list) { ListCell *cell = NULL; foreach(cell, list) { @@ -207,25 +209,6 @@ static void ClearListContent(List *list) } } -/* - * Decrypt conninfo for subscription. - * IMPORTANT: caller should clear and free the memory after using it immediately - */ -char *DecryptConninfo(char *encryptConninfo) -{ - const char* sensitiveOptionsArray[] = {"password"}; - const int sensitiveArrayLength = lengthof(sensitiveOptionsArray); - List *defList = ConninfoToDefList(encryptConninfo); - DecryptOptions(defList, sensitiveOptionsArray, sensitiveArrayLength, SUBSCRIPTION_MODE); - char *decryptConninfo = DefListToString(defList); - - /* defList has plain content, clear it before free */ - ClearListContent(defList); - list_free_ext(defList); - /* IMPORTANT: caller should clear and free the memory after using it immediately */ - return decryptConninfo; -} - /* * Convert text array to list of strings. * diff --git a/src/gausskernel/optimizer/commands/subscriptioncmds.cpp b/src/gausskernel/optimizer/commands/subscriptioncmds.cpp index 15ced8475..656ffbe8a 100644 --- a/src/gausskernel/optimizer/commands/subscriptioncmds.cpp +++ b/src/gausskernel/optimizer/commands/subscriptioncmds.cpp @@ -44,7 +44,7 @@ #include "utils/array.h" #include "utils/acl.h" -static void ConnectPublisher(char *conninfo, char* slotname); +static bool ConnectPublisher(char* conninfo, char* slotname); static void CreateSlotInPublisher(char *slotname); static void ValidateReplicationSlot(char *slotname, List *publications); @@ -210,26 +210,82 @@ static Datum publicationListToArray(List *publist) } /* - * connect publisher and create slot. - * the input conninfo should be encrypt, we will decrypt password inside + * Parse the original connection string which is encrypted, poll all hosts and ports, + * and try to connect to the publisher. + * When checkRemoteMode is true, the remotemode must be normal or primary. + * Return true to indicate successful connection. */ -static void ConnectPublisher(char *conninfo, char *slotname) +bool AttemptConnectPublisher(const char *conninfoOriginal, char* slotname, bool checkRemoteMode) +{ + size_t conninfoLen = strlen(conninfoOriginal) + 1; + + char* conninfo = NULL; + StringInfoData conninfoWithoutHostport; + initStringInfo(&conninfoWithoutHostport); + HostPort* hostPortList[MAX_REPLNODE_NUM] = {NULL}; + ParseConninfo(conninfoOriginal, &conninfoWithoutHostport, hostPortList); + if (hostPortList[0] == NULL) { + ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg( + "invalid connection string syntax, missing host and port"))); + } + bool connectSuccess = false; + conninfo = (char*)palloc(conninfoLen * sizeof(char)); + for (int i = 0; i < MAX_REPLNODE_NUM; ++i) { + if (hostPortList[i] == NULL) { + break; + } + int ret = snprintf_s(conninfo, conninfoLen, conninfoLen - 1, + "%s host=%s port=%s", conninfoWithoutHostport.data, + hostPortList[i]->host, hostPortList[i]->port); + securec_check_ss(ret, "\0", "\0"); + + connectSuccess = ConnectPublisher(conninfo, slotname); + if (!connectSuccess) { + /* try next host */ + continue; + } + if (!checkRemoteMode) { + break; + } + ServerMode publisherServerMde = IdentifyRemoteMode(); + if (publisherServerMde == NORMAL_MODE || publisherServerMde == PRIMARY_MODE) { + break; + } + /* it's a standby, try next host */ + (WalReceiverFuncTable[GET_FUNC_IDX]).walrcv_disconnect(); + connectSuccess = false; + } + pfree_ext(conninfo); + + /* clean up */ + FreeStringInfo(&conninfoWithoutHostport); + for (int i = 0; i < MAX_REPLNODE_NUM; ++i) { + if (hostPortList[i] == NULL) { + break; + } + pfree_ext(hostPortList[i]->host); + pfree_ext(hostPortList[i]->port); + pfree_ext(hostPortList[i]); + } + return connectSuccess; +} + +/* + * connect to publisher with conninfo + */ +static bool ConnectPublisher(char* conninfo, char* slotname) { /* Try to connect to the publisher. */ volatile WalRcvData *walrcv = t_thrd.walreceiverfuncs_cxt.WalRcv; SpinLockAcquire(&walrcv->mutex); walrcv->conn_target = REPCONNTARGET_PUBLICATION; SpinLockRelease(&walrcv->mutex); - - char *decryptConninfo = DecryptConninfo(conninfo); + char* decryptConninfo = EncryptOrDecryptConninfo(conninfo, 'D'); bool connectSuccess = (WalReceiverFuncTable[GET_FUNC_IDX]).walrcv_connect(decryptConninfo, NULL, slotname, -1); int rc = memset_s(decryptConninfo, strlen(decryptConninfo), 0, strlen(decryptConninfo)); securec_check(rc, "", ""); pfree_ext(decryptConninfo); - - if (!connectSuccess) { - ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE), errmsg("could not connect to the publisher"))); - } + return connectSuccess; } /* @@ -306,7 +362,6 @@ ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) bool enabled_given = false; bool enabled = true; char *synchronous_commit; - char *conninfo; char *slotname; bool slotname_given; bool binary; @@ -348,11 +403,10 @@ ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) synchronous_commit = "off"; } - conninfo = stmt->conninfo; publications = stmt->publication; /* Check the connection info string. */ - libpqrcv_check_conninfo(conninfo); + libpqrcv_check_conninfo(stmt->conninfo); /* Everything ok, form a new tuple. */ rc = memset_s(values, sizeof(values), 0, sizeof(values)); @@ -367,16 +421,9 @@ ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) values[Anum_pg_subscription_subbinary - 1] = BoolGetDatum(binary); /* encrypt conninfo */ - List *conninfoList = ConninfoToDefList(stmt->conninfo); - /* Sensitive options for subscription, will be encrypted when saved to catalog. */ - const char* sensitiveOptionsArray[] = {"password"}; - const int sensitiveArrayLength = lengthof(sensitiveOptionsArray); - EncryptGenericOptions(conninfoList, sensitiveOptionsArray, sensitiveArrayLength, SUBSCRIPTION_MODE); - char *encryptConninfo = DefListToString(conninfoList); - + char *encryptConninfo = EncryptOrDecryptConninfo(stmt->conninfo, 'E'); values[Anum_pg_subscription_subconninfo - 1] = CStringGetTextDatum(encryptConninfo); - pfree_ext(conninfoList); if (enabled) { if (!slotname_given) { slotname = stmt->subname; @@ -412,11 +459,14 @@ ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) */ if (enabled) { Assert(slotname); - ConnectPublisher(encryptConninfo, slotname); + + if (!AttemptConnectPublisher(encryptConninfo, slotname, true)) { + ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE), errmsg("Failed to connect to publisher."))); + } + CreateSlotInPublisher(slotname); (WalReceiverFuncTable[GET_FUNC_IDX]).walrcv_disconnect(); } - pfree_ext(encryptConninfo); heap_close(rel, RowExclusiveLock); rc = memset_s(stmt->conninfo, strlen(stmt->conninfo), 0, strlen(stmt->conninfo)); @@ -508,23 +558,15 @@ ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt) if (conninfo) { /* Check the connection info string. */ libpqrcv_check_conninfo(conninfo); - - /* encrypt conninfo */ - List *conninfoList = ConninfoToDefList(conninfo); - /* Sensitive options for subscription, will be encrypted when saved to catalog. */ - const char* sensitiveOptionsArray[] = {"password"}; - const int sensitiveArrayLength = lengthof(sensitiveOptionsArray); - EncryptGenericOptions(conninfoList, sensitiveOptionsArray, sensitiveArrayLength, SUBSCRIPTION_MODE); - encryptConninfo = DefListToString(conninfoList); - needFreeConninfo = true; - + encryptConninfo = EncryptOrDecryptConninfo(conninfo, 'E'); + rc = memset_s(conninfo, strlen(conninfo), 0, strlen(conninfo)); + securec_check(rc, "\0", "\0"); values[Anum_pg_subscription_subconninfo - 1] = CStringGetTextDatum(encryptConninfo); replaces[Anum_pg_subscription_subconninfo - 1] = true; + needFreeConninfo = true; - pfree_ext(conninfoList); - + /* need to check whether new conninfo can be used to connect to new publisher */ if (sub->enabled || (enabled_given && enabled)) { - /* we need to check whether new conninfo can be used to connect to new publisher */ checkConn = true; } } @@ -592,16 +634,18 @@ ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt) if (sub->enabled && !enabled) { ereport(ERROR, (errmsg("If you want to deactivate this subscription, use DROP SUBSCRIPTION."))); } - /* enable subscription */ - if (!sub->enabled && enabled) { - /* if slot hasn't been created, then create it */ - if (!sub->slotname || !*(sub->slotname)) { + /* enabling subscription, but slot hasn't been created, + * then mark createSlot to true. + */ + if (!sub->enabled && enabled && (!sub->slotname || !*(sub->slotname))) { createSlot = true; - } } if (checkConn || createSlot || validateSlot) { - ConnectPublisher(encryptConninfo, finalSlotName); + if (!AttemptConnectPublisher(encryptConninfo, finalSlotName, true)) { + ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE), errmsg( + checkConn ? "The new conninfo cannot connect to new publisher." : "Failed to connect to publisher."))); + } if (createSlot) { CreateSlotInPublisher(finalSlotName); @@ -619,12 +663,6 @@ ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt) if (needFreeConninfo) { pfree_ext(encryptConninfo); } - - if (conninfo) { - rc = memset_s(conninfo, strlen(conninfo), 0, strlen(conninfo)); - securec_check(rc, "", ""); - } - return myself; } @@ -775,7 +813,11 @@ void DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) initStringInfo(&cmd); appendStringInfo(&cmd, "DROP_REPLICATION_SLOT %s", quote_identifier(slotname)); - ConnectPublisher(conninfo, slotname); + if (!AttemptConnectPublisher(conninfo, slotname, true)) { + ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE), errmsg( + "could not connect to publisher."))); + } + PG_TRY(); { int sqlstate = 0; @@ -801,6 +843,7 @@ void DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) (WalReceiverFuncTable[GET_FUNC_IDX]).walrcv_disconnect(); + pfree_ext(conninfo); pfree(cmd.data); heap_close(rel, NoLock); } @@ -930,3 +973,149 @@ void RenameSubscription(List *oldname, const char *newname) return; } + +/* + * Parse the host or port string into a string array, + * where host and port are separated by ",". + * input: conn --- host or port string separated by "," + * output: connArray --- host or port string array + * return: the length of connArray + * for example: + * (1): + * conn = 1.1.1.1,2.2.2.2,...,9.9.9.9 + * connArray = { + * 1,.1.1.1, + * 2.2.2.2, + * ..., + * 9.9.9.9 + * } + * return 9 + * (2): + * conn = 1,2,...,9 + * connArray = {1,2,...,9} + * return 9 + */ +static int HostsPortsToArray(const char* conn, char** connArray) +{ + if (conn == NULL) { + return 0; + } + char* cp = NULL; + char* cur = NULL; + char *buf = pstrdup(conn); + + cp = buf; + int i = 0; + while (*cp) { + cur = cp; + while (*cp && *cp != ',') { + ++cp; + } + if (*cp == ',') { + *cp = '\0'; + ++cp; + } + if (i >= MAX_REPLNODE_NUM) { + ereport(ERROR, (errmsg("Currently, a maximum of %d servers are " + "supported.", MAX_REPLNODE_NUM))); + } + connArray[i++] = pstrdup(cur); + + if (*cp == 0) { + break; + } + } + pfree(buf); + return i; +} + +/* + * parse host and port + */ +static void ParseHostPort(char* hoststr, char* portstr, HostPort** hostPortList) +{ + char* hosts[MAX_REPLNODE_NUM] = {NULL}; + char* ports[MAX_REPLNODE_NUM] = {NULL}; + int hostNum = HostsPortsToArray(hoststr, hosts); + int portNum = HostsPortsToArray(portstr, ports); + if (hostNum != portNum) { + ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("The number of host and port are inconsistent."))); + } + + for (int i = 0; i < hostNum; ++i) { + hostPortList[i] = (HostPort*)palloc(sizeof(HostPort)); + hostPortList[i]->host = hosts[i]; + hostPortList[i]->port = ports[i]; + } +} + +/* + * Parse conninfo + * conninfo format: + * 'dbname=abc user=username password=xxxx host=ip1,ip2,...,ip9 port=p1,p2,...,p9' + * after parsing: + * conninfoWithoutHostPort: + * 'dbname=abc user=username password=xxxx' + * hostPortList: + * { + * {host=ip1, port=p1}, + * {host=ip2, port=p2}, + * ... + * {host=ip9, port=p9} + * } + */ +void ParseConninfo(const char* conninfo, StringInfoData* conninfoWithoutHostPort, HostPort** hostPortList) +{ + List* conninfoList = ConninfoToDefList(conninfo); + ListCell* l = NULL; + + char* hostStr = NULL; + char* portStr = NULL; + foreach (l, conninfoList) { + DefElem* defel = (DefElem*)lfirst(l); + if (pg_strcasecmp(defel->defname, "host") == 0) { + hostStr = defGetString(defel); + } else if (pg_strcasecmp(defel->defname, "port") == 0) { + portStr = defGetString(defel); + } else { + appendStringInfo(conninfoWithoutHostPort, "%s=%s ", defel->defname, defGetString(defel)); + } + } + if (hostPortList != NULL) { + ParseHostPort(hostStr, portStr, hostPortList); + } +} + +/* + * encrypt conninfo when action = 'E' + * decrypt conninfo when action = 'D' + * conninfoNew: encrypted or decrypted conninfo + */ +char* EncryptOrDecryptConninfo(const char* conninfo, const char action) +{ + /* parse conninfo to list */ + List *conninfoList = ConninfoToDefList(conninfo); + /* Sensitive options for subscription */ + const char* sensitiveOptionsArray[] = {"password"}; + const int sensitiveArrayLength = lengthof(sensitiveOptionsArray); + switch (action) { + /* Encrypt */ + case 'E': + EncryptGenericOptions(conninfoList, sensitiveOptionsArray, sensitiveArrayLength, SUBSCRIPTION_MODE); + break; + + /* Decrypt */ + case 'D': + DecryptOptions(conninfoList, sensitiveOptionsArray, sensitiveArrayLength, SUBSCRIPTION_MODE); + break; + + default: + break; + } + + char* conninfoNew = DefListToString(conninfoList); + ClearListContent(conninfoList); + list_free_ext(conninfoList); + + return conninfoNew; +} diff --git a/src/gausskernel/storage/replication/libpqwalreceiver.cpp b/src/gausskernel/storage/replication/libpqwalreceiver.cpp index 22cf77ee5..30a338a1a 100755 --- a/src/gausskernel/storage/replication/libpqwalreceiver.cpp +++ b/src/gausskernel/storage/replication/libpqwalreceiver.cpp @@ -259,7 +259,7 @@ void StartRemoteStreaming(const LibpqrcvConnectParam *options) if (options->binary && PQserverVersion(t_thrd.libwalreceiver_cxt.streamConn) >= 90204) { appendStringInfoString(&cmd, ", binary 'true'"); - ereport(DEBUG5, ( errmsg("append binary true"))); + ereport(DEBUG5, (errmsg("append binary true"))); } appendStringInfoChar(&cmd, ')'); @@ -426,7 +426,7 @@ void IdentifyRemoteSystem(bool checkRemote) } /* identify remote mode, should do this after connect success. */ -static ServerMode IdentifyRemoteMode() +ServerMode IdentifyRemoteMode() { Assert(t_thrd.libwalreceiver_cxt.streamConn != NULL); volatile WalRcvData *walrcv = t_thrd.walreceiverfuncs_cxt.WalRcv; @@ -449,7 +449,9 @@ static ServerMode IdentifyRemoteMode() num_fields))); } remoteMode = (ServerMode)pg_strtoint32(PQgetvalue(res, 0, 0)); - if (!t_thrd.walreceiver_cxt.AmWalReceiverForFailover && (!IS_PRIMARY_NORMAL(remoteMode)) && + if (walrcv->conn_target != REPCONNTARGET_PUBLICATION && + !t_thrd.walreceiver_cxt.AmWalReceiverForFailover && + (!IS_PRIMARY_NORMAL(remoteMode)) && /* remoteMode of cascade standby is a standby */ !t_thrd.xlog_cxt.is_cascade_standby && !IS_SHARED_STORAGE_MODE) { PQclear(res); diff --git a/src/gausskernel/storage/replication/logical/proto.cpp b/src/gausskernel/storage/replication/logical/proto.cpp index c8d993ca3..2c537dd4c 100644 --- a/src/gausskernel/storage/replication/logical/proto.cpp +++ b/src/gausskernel/storage/replication/logical/proto.cpp @@ -406,7 +406,7 @@ static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple pfree(outputbytes); } } else { - char *outputstr; + char* outputstr = NULL; pq_sendbyte(out, LOGICALREP_COLUMN_TEXT); if (!typclass->typbyval && typclass->typlen == -1) { /* definitely detoasted Datum */ @@ -464,10 +464,8 @@ static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple) len = pq_getmsgint(in, sizeof(uint32)); /* read length */ /* and data */ - value->data = (char *) palloc((len + 1) * sizeof(char)); + value->data = (char *)palloc0((len + 1) * sizeof(char)); pq_copymsgbytes(in, value->data, len); - /* not strictly necessary but per StringInfo practice */ - value->data[len] = '\0'; /* make StringInfo fully valid */ value->len = len; value->cursor = 0; @@ -603,3 +601,25 @@ static const char *logicalrep_read_namespace(StringInfo in) return nspname; } + +/* + * Write conninfo to the output stream. + */ +void logicalrep_write_conninfo(StringInfo out, char* conninfo) +{ + pq_sendbyte(out, 'S'); /* action */ + + pq_writestring(out, conninfo); /* conninfo follows */ +} + +/* + * Read conninfo from stream. + */ +void logicalrep_read_conninfo(StringInfo in, char** conninfo) +{ + const char* conninfoTemp = pq_getmsgstring(in); + size_t conninfoLen = strlen(conninfoTemp) + 1; + *conninfo = (char*)palloc(conninfoLen); + int rc = strcpy_s(*conninfo, conninfoLen, conninfoTemp); + securec_check(rc, "", ""); +} diff --git a/src/gausskernel/storage/replication/logical/worker.cpp b/src/gausskernel/storage/replication/logical/worker.cpp index f6d41dece..f91d90ef3 100644 --- a/src/gausskernel/storage/replication/logical/worker.cpp +++ b/src/gausskernel/storage/replication/logical/worker.cpp @@ -41,6 +41,7 @@ #include "catalog/pg_partition_fn.h" #include "commands/trigger.h" +#include "commands/subscriptioncmds.h" #include "executor/executor.h" #include "executor/node/nodeModifyTable.h" @@ -111,6 +112,8 @@ static void store_flush_position(XLogRecPtr remote_lsn); static void reread_subscription(void); static void ApplyWorkerProcessMsg(char type, StringInfo s, XLogRecPtr *lastRcv); static void apply_dispatch(StringInfo s); +static void apply_handle_conninfo(StringInfo s); +static void UpdateConninfo(char* standbysInfo); /* SIGHUP: set flag to re-read config file at next convenient time */ static void LogicalrepWorkerSighub(SIGNAL_ARGS) @@ -882,6 +885,9 @@ static void apply_dispatch(StringInfo s) case 'O': apply_handle_origin(s); break; + case 'S': + apply_handle_conninfo(s); + break; default: ereport(ERROR, (errcode(ERRCODE_PROTOCOL_VIOLATION), errmsg("invalid logical replication message type \"%c\"", action))); @@ -1491,14 +1497,9 @@ void ApplyWorkerMain() CommitTransactionCommand(); - char *decryptConnInfo = DecryptConninfo(t_thrd.applyworker_cxt.mySubscription->conninfo); - bool connectSuccess = (WalReceiverFuncTable[GET_FUNC_IDX]).walrcv_connect(decryptConnInfo, NULL, - t_thrd.applyworker_cxt.mySubscription->name, -1); - rc = memset_s(decryptConnInfo, strlen(decryptConnInfo), 0, strlen(decryptConnInfo)); - securec_check(rc, "", ""); - pfree_ext(decryptConnInfo); - if (!connectSuccess) { - ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE), errmsg("could not connect to the publisher"))); + if (!AttemptConnectPublisher(t_thrd.applyworker_cxt.mySubscription->conninfo, + t_thrd.applyworker_cxt.mySubscription->name, true)) { + ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE), errmsg("Failed to connect to publisher."))); } /* @@ -1665,3 +1666,64 @@ char* DefListToString(const List *defList) return buf.data; } +/* + * Handle conninfo update message. + */ +static void apply_handle_conninfo(StringInfo s) +{ + char* standbysInfo = NULL; + logicalrep_read_conninfo(s, &standbysInfo); + UpdateConninfo(standbysInfo); + pfree_ext(standbysInfo); +} + +static void UpdateConninfo(char* standbysInfo) +{ + Relation rel; + bool nulls[Natts_pg_subscription]; + bool replaces[Natts_pg_subscription]; + Datum values[Natts_pg_subscription]; + HeapTuple tup; + Subscription* sub = t_thrd.applyworker_cxt.mySubscription; + Oid subid = sub->oid; + + StartTransactionCommand(); + rel = heap_open(SubscriptionRelationId, RowExclusiveLock); + /* Fetch the existing tuple. */ + tup = SearchSysCacheCopy2(SUBSCRIPTIONNAME, u_sess->proc_cxt.MyDatabaseId, + CStringGetDatum(t_thrd.applyworker_cxt.mySubscription->name)); + if (!HeapTupleIsValid(tup)) { + ereport(ERROR, (errcode(ERRCODE_UNDEFINED_OBJECT), errmsg("subscription \"%s\" does not exist", + t_thrd.applyworker_cxt.mySubscription->name))); + } + subid = HeapTupleGetOid(tup); + + /* Form a new tuple. */ + int rc = memset_s(nulls, sizeof(nulls), false, sizeof(nulls)); + securec_check(rc, "", ""); + rc = memset_s(values, sizeof(values), 0, sizeof(values)); + securec_check(rc, "", ""); + rc = memset_s(replaces, sizeof(replaces), false, sizeof(replaces)); + securec_check(rc, "", ""); + + /* get conninfoWithoutHostport */ + StringInfoData conninfoWithoutHostport; + initStringInfo(&conninfoWithoutHostport); + ParseConninfo(sub->conninfo, &conninfoWithoutHostport, (HostPort**)NULL); + + /* join conninfoWithoutHostport together with standbysinfo */ + appendStringInfo(&conninfoWithoutHostport, " %s", standbysInfo); + /* Replace connection information */ + values[Anum_pg_subscription_subconninfo - 1] = CStringGetTextDatum(conninfoWithoutHostport.data); + replaces[Anum_pg_subscription_subconninfo - 1] = true; + tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls, replaces); + + /* Update the catalog. */ + simple_heap_update(rel, &tup->t_self, tup); + CatalogUpdateIndexes(rel, tup); + + heap_close(rel, RowExclusiveLock); + CommitTransactionCommand(); + + ereport(LOG, (errmsg("Update conninfo successfully, new conninfo %s.", standbysInfo))); +} diff --git a/src/gausskernel/storage/replication/pgoutput/pgoutput.cpp b/src/gausskernel/storage/replication/pgoutput/pgoutput.cpp index 9f4fa4407..f9f070473 100644 --- a/src/gausskernel/storage/replication/pgoutput/pgoutput.cpp +++ b/src/gausskernel/storage/replication/pgoutput/pgoutput.cpp @@ -46,6 +46,8 @@ static bool pgoutput_origin_filter(LogicalDecodingContext *ctx, RepOriginId orig static List *LoadPublications(List *pubnames); static void publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue); +static bool ReplconninfoChanged(); +static void GetConninfo(StringInfoData* standbysInfo); /* Entry in the map used to remember which relation schemas we sent. */ typedef struct RelationSyncEntry { @@ -76,7 +78,7 @@ void _PG_output_plugin_init(OutputPluginCallbacks *cb) cb->shutdown_cb = pgoutput_shutdown; } -static void parse_output_parameters(List* options, PGOutputData* data) +static void parse_output_parameters(List *options, PGOutputData *data) { ListCell *lc; bool protocol_version_given = false; @@ -217,6 +219,22 @@ static void pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *t OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit(ctx->out, txn, commit_lsn); OutputPluginWrite(ctx, true); + + /* + * Send the newest connecttion information to the subscriber, + * when the connection information about the standby changes. + */ + if (ReplconninfoChanged()) { + StringInfoData standbysInfo; + initStringInfo(&standbysInfo); + + GetConninfo(&standbysInfo); + OutputPluginPrepareWrite(ctx, true); + logicalrep_write_conninfo(ctx->out, standbysInfo.data); + OutputPluginWrite(ctx, true); + + FreeStringInfo(&standbysInfo); + } } /* @@ -616,3 +634,43 @@ static void rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashval entry->pubactions.pubdelete = false; } } + +static void GetConninfo(StringInfoData* standbysInfo) +{ + bool primaryJoined = false; + StringInfoData hosts; + StringInfoData ports; + initStringInfo(&hosts); + initStringInfo(&ports); + for (int i = 1; i < MAX_REPLNODE_NUM + 1; ++i) { + t_thrd.postmaster_cxt.ReplConnChangeType[i] = 0; + if (t_thrd.postmaster_cxt.ReplConnArray[i] == NULL) { + continue; + } + if (!primaryJoined) { + appendStringInfo(&hosts, "%s,%s", + t_thrd.postmaster_cxt.ReplConnArray[i]->localhost, + t_thrd.postmaster_cxt.ReplConnArray[i]->remotehost); + appendStringInfo(&ports, "%d,%d", + t_thrd.postmaster_cxt.ReplConnArray[i]->localport, + t_thrd.postmaster_cxt.ReplConnArray[i]->remoteport); + primaryJoined = true; + } else { + appendStringInfo(&hosts, ",%s", + t_thrd.postmaster_cxt.ReplConnArray[i]->remotehost); + appendStringInfo(&ports, ",%d", + t_thrd.postmaster_cxt.ReplConnArray[i]->remoteport); + } + } + appendStringInfo(standbysInfo, "host=%s port=%s", hosts.data, ports.data); +} + +static inline bool ReplconninfoChanged() +{ + for (int i = 1; i < MAX_REPLNODE_NUM; ++i) { + if (t_thrd.postmaster_cxt.ReplConnChangeType[i]) { + return true; + } + } + return false; +} diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h index 2b7ecd05e..6765cd38c 100644 --- a/src/include/catalog/pg_subscription.h +++ b/src/include/catalog/pg_subscription.h @@ -90,6 +90,7 @@ extern Oid get_subscription_oid(const char *subname, bool missing_ok); extern char *get_subscription_name(Oid subid, bool missing_ok); extern int CountDBSubscriptions(Oid dbid); -extern char *DecryptConninfo(char *encryptConninfo); +extern void ClearListContent(List *list); + #endif /* PG_SUBSCRIPTION_H */ diff --git a/src/include/commands/subscriptioncmds.h b/src/include/commands/subscriptioncmds.h index 9ee8cf177..4f1498225 100644 --- a/src/include/commands/subscriptioncmds.h +++ b/src/include/commands/subscriptioncmds.h @@ -17,6 +17,11 @@ #include "nodes/parsenodes.h" +typedef struct HostPort { + char* host; + char* port; +} HostPort; + extern ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel); extern ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt); extern void DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel); @@ -24,6 +29,12 @@ extern void DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel); extern ObjectAddress AlterSubscriptionOwner(const char *name, Oid newOwnerId); extern void AlterSubscriptionOwner_oid(Oid subid, Oid newOwnerId); extern void RenameSubscription(List* oldname, const char* newname); +extern void AddStandbysInfo(char* standbysInfo); +extern void DropStandbysInfo(char* standbysInfo); + +extern void ParseConninfo(const char* conninfo, StringInfoData* conninfoWithoutHostPort, HostPort** hostPortList); +extern char* EncryptOrDecryptConninfo(const char* conninfo, const char action); +extern bool AttemptConnectPublisher(const char *conninfoOriginal, char* slotname, bool checkRemoteMode); #endif /* SUBSCRIPTIONCMDS_H */ diff --git a/src/include/replication/libpqwalreceiver.h b/src/include/replication/libpqwalreceiver.h index 96e15fc98..ecd6a1a92 100755 --- a/src/include/replication/libpqwalreceiver.h +++ b/src/include/replication/libpqwalreceiver.h @@ -54,5 +54,6 @@ extern bool libpqrcv_command(const char *cmd, char **err, int *sqlstate); extern void IdentifyRemoteSystem(bool checkRemote); extern void CreateRemoteReplicationSlot(XLogRecPtr startpoint, const char* slotname, bool isLogical); extern void StartRemoteStreaming(const LibpqrcvConnectParam *options); +extern ServerMode IdentifyRemoteMode(); #endif diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h index 5ca74b1bf..4e2876b86 100644 --- a/src/include/replication/logicalproto.h +++ b/src/include/replication/logicalproto.h @@ -101,5 +101,7 @@ extern void logicalrep_write_rel(StringInfo out, Relation rel); extern LogicalRepRelation *logicalrep_read_rel(StringInfo in); extern void logicalrep_write_typ(StringInfo out, Oid typoid); extern void logicalrep_read_typ(StringInfo out, LogicalRepTyp *ltyp); +extern void logicalrep_write_conninfo(StringInfo out, char* conninfo); +extern void logicalrep_read_conninfo(StringInfo in, char** conninfo); #endif /* LOGICALREP_PROTO_H */ diff --git a/src/test/regress/input/subscription.source b/src/test/regress/input/subscription.source index f47d0a501..2d5a4c4e9 100644 --- a/src/test/regress/input/subscription.source +++ b/src/test/regress/input/subscription.source @@ -24,6 +24,7 @@ CREATE SUBSCRIPTION testsub CONNECTION 'foo'; CREATE SUBSCRIPTION testsub PUBLICATION foo; -- fail - could not connect to the publisher create subscription testsub2 connection 'host=abc' publication pub; +create subscription testsub2 connection 'host=abc port=12345' publication pub; set client_min_messages to error; -- fail - syntax error, invalid connection string syntax: missing "=" CREATE SUBSCRIPTION testsub CONNECTION 'testconn' PUBLICATION testpub; @@ -32,7 +33,10 @@ CREATE SUBSCRIPTION testsub CONNECTION 'dbname=doesnotexist' PUBLICATION testpub CREATE SUBSCRIPTION testsub CONNECTION 'dbname=doesnotexist' PUBLICATION testpub WITH (ENABLED=false, slot_name='testsub', synchronous_commit=off); -- create SUBSCRIPTION with conninfo in two single quote, used to check mask string bug CREATE SUBSCRIPTION testsub_maskconninfo CONNECTION 'host=''1.2.3.4'' port=''12345'' user=''username'' dbname=''postgres'' password=''password_1234''' PUBLICATION testpub WITH (ENABLED=false, slot_name='testsub', synchronous_commit=off); - +-- fail - The number of host and port are inconsistent +create subscription sub1 connection 'dbname=postgres user=pubusr password=Huawei@123 host=192.168.0.38,192.168.0.38,192.168.0.38 port=14001,14501' publication pub1; +-- fail - a maximum of 9 servers are supported +create subscription sub1 connection 'dbname=postgres user=pubusr password=Huawei@123 host=192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38 port=14001,14501' publication pub1; -- alter connection ALTER SUBSCRIPTION testsub CONNECTION 'host=''1.2.3.4'' port=''12345'' user=''username'' dbname=''postgres'' password=''password_1234'''; ALTER SUBSCRIPTION testsub CONNECTION 'dbname=does_not_exist'; diff --git a/src/test/regress/output/subscription.source b/src/test/regress/output/subscription.source index 11145e895..7bf363408 100644 --- a/src/test/regress/output/subscription.source +++ b/src/test/regress/output/subscription.source @@ -60,8 +60,10 @@ LINE 1: CREATE SUBSCRIPTION testsub PUBLICATION foo; ^ -- fail - could not connect to the publisher create subscription testsub2 connection 'host=abc' publication pub; +ERROR: The number of host and port are inconsistent. +create subscription testsub2 connection 'host=abc port=12345' publication pub; WARNING: apply worker could not connect to the remote server -ERROR: could not connect to the publisher +ERROR: Failed to connect to publisher. set client_min_messages to error; -- fail - syntax error, invalid connection string syntax: missing "=" CREATE SUBSCRIPTION testsub CONNECTION 'testconn' PUBLICATION testpub; @@ -72,14 +74,20 @@ ERROR: unrecognized subscription parameter: create_slot CREATE SUBSCRIPTION testsub CONNECTION 'dbname=doesnotexist' PUBLICATION testpub WITH (ENABLED=false, slot_name='testsub', synchronous_commit=off); -- create SUBSCRIPTION with conninfo in two single quote, used to check mask string bug CREATE SUBSCRIPTION testsub_maskconninfo CONNECTION 'host=''1.2.3.4'' port=''12345'' user=''username'' dbname=''postgres'' password=''password_1234''' PUBLICATION testpub WITH (ENABLED=false, slot_name='testsub', synchronous_commit=off); +-- fail - The number of host and port are inconsistent +create subscription sub1 connection 'dbname=postgres user=pubusr password=Huawei@123 host=192.168.0.38,192.168.0.38,192.168.0.38 port=14001,14501' publication pub1; +ERROR: The number of host and port are inconsistent. +-- fail - a maximum of 9 servers are supported +create subscription sub1 connection 'dbname=postgres user=pubusr password=Huawei@123 host=192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38 port=14001,14501' publication pub1; +ERROR: Currently, a maximum of 9 servers are supported. -- alter connection ALTER SUBSCRIPTION testsub CONNECTION 'host=''1.2.3.4'' port=''12345'' user=''username'' dbname=''postgres'' password=''password_1234'''; ALTER SUBSCRIPTION testsub CONNECTION 'dbname=does_not_exist'; reset client_min_messages; select subname, pg_get_userbyid(subowner) as Owner, subenabled, subconninfo, subpublications, subbinary from pg_subscription where subname='testsub'; - subname | owner | subenabled | subconninfo | subpublications | subbinary ----------+---------------------------+------------+----------------------+-----------------+----------- - testsub | regress_subscription_user | f | dbname=doesnotexist | {testpub} | f + subname | owner | subenabled | subconninfo | subpublications | subbinary +---------+---------------------------+------------+------------------------+-----------------+----------- + testsub | regress_subscription_user | f | dbname=does_not_exist | {testpub} | f (1 row) --- alter subscription @@ -142,8 +150,7 @@ COMMIT; -- -- active SUBSCRIPTION BEGIN; ALTER SUBSCRIPTION testsub_rename ENABLE; -WARNING: apply worker could not connect to the remote server -ERROR: could not connect to the publisher +ERROR: invalid connection string syntax, missing host and port select subname, subenabled from pg_subscription where subname='testsub_rename'; ERROR: current transaction is aborted, commands ignored until end of transaction block, firstChar[Q] ALTER SUBSCRIPTION testsub_rename SET (ENABLED=false);