From 7ef4caf5bba57e9696417952e174a41a7dfe6c76 Mon Sep 17 00:00:00 2001 From: luozihao <1165977584@qq.com> Date: Wed, 24 Nov 2021 11:28:39 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8F=90=E4=BE=9B=E8=A7=86=E5=9B=BE=E6=9F=A5?= =?UTF-8?q?=E7=9C=8B=E5=BD=93=E5=89=8D=E6=98=AF=E5=93=AA=E5=8F=B0=E5=A4=87?= =?UTF-8?q?=E6=9C=BA=E5=9C=A8=E5=81=9A=E5=BD=92=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/common/backend/catalog/builtin_funcs.ini | 4 + src/common/backend/catalog/system_views.sql | 1 + src/gausskernel/process/postmaster/pgarch.cpp | 97 +++++++++++++++++-- .../process/threadpool/knl_instance.cpp | 3 + .../rollback_catalog_maindb_92_300.sql | 1 + .../rollback_catalog_otherdb_92_300.sql | 1 + .../upgrade-post_catalog_maindb_92_300.sql | 1 + .../upgrade-post_catalog_otherdb_92_300.sql | 1 + src/include/knl/knl_instance.h | 5 + src/include/postmaster/pgarch.h | 1 + .../expected/single_node_opr_sanity.out | 1 + 11 files changed, 109 insertions(+), 7 deletions(-) create mode 100644 src/include/catalog/upgrade_sql/rollback_catalog_maindb/rollback_catalog_maindb_92_300.sql create mode 100644 src/include/catalog/upgrade_sql/rollback_catalog_otherdb/rollback_catalog_otherdb_92_300.sql create mode 100644 src/include/catalog/upgrade_sql/upgrade_catalog_maindb/upgrade-post_catalog_maindb_92_300.sql create mode 100644 src/include/catalog/upgrade_sql/upgrade_catalog_otherdb/upgrade-post_catalog_otherdb_92_300.sql diff --git a/src/common/backend/catalog/builtin_funcs.ini b/src/common/backend/catalog/builtin_funcs.ini index 0fa4d5d35..3bce48d3f 100755 --- a/src/common/backend/catalog/builtin_funcs.ini +++ b/src/common/backend/catalog/builtin_funcs.ini @@ -3191,6 +3191,10 @@ "gs_fault_inject", 1, AddBuiltinFunc(_0(4000), _1("gs_fault_inject"), _2(6), _3(true), _4(false), _5(gs_fault_inject), _6(20), _7(PG_CATALOG_NAMESPACE), _8(BOOTSTRAP_SUPERUSERID), _9(INTERNALlanguageId), _10(1), _11(0), _12(0), _13(0), _14(false), _15(false), _16(false), _17(false), _18('v'), _19(0), _20(6, 20, 25, 25, 25, 25, 25), _21(NULL), _22(NULL), _23(NULL), _24(NULL), _25("gs_fault_inject"), _26(NULL), _27(NULL), _28(NULL), _29(0), _30(false), _31(NULL), _32(false), _33(NULL), _34('f')) ), + AddFuncGroup( + "gs_get_archive_status", 1, + AddBuiltinFunc(_0(3796), _1("gs_get_archive_status"), _2(0), _3(true), _4(true), _5(gs_get_archive_status), _6(2249), _7(PG_CATALOG_NAMESPACE), _8(BOOTSTRAP_SUPERUSERID), _9(INTERNALlanguageId), _10(1), _11(1), _12(0), _13(0), _14(false), _15(false), _16(false), _17(false), _18('s'), _19(0), _20(0), _21(4, 25, 25, 1184, 25), _22(4, 'o', 'o', 'o', 'o'), _23(4, "archive_standby", "archive_lsn", "last_arch_time", "archive_path"), _24(NULL), _25("gs_get_archive_status"), _26(NULL), _27(NULL), _28(NULL), _29(0), _30(false), _31(NULL), _32(false), _33(NULL), _34('f')) + ), AddFuncGroup( "gs_get_global_barrier_status", 1, AddBuiltinFunc(_0(9032), _1("gs_get_global_barrier_status"), _2(0), _3(true), _4(false), _5(gs_get_global_barrier_status), _6(2249), _7(PG_CATALOG_NAMESPACE), _8(BOOTSTRAP_SUPERUSERID), _9(INTERNALlanguageId), _10(1), _11(0), _12(0), _13(0), _14(false), _15(false), _16(false), _17(false), _18('v'), _19(0), _20(0), _21(2, 25, 25), _22(2, 'o', 'o'), _23(2, "global_barrier_id", "global_achive_barrier_id"), _24(NULL), _25("gs_get_global_barrier_status"), _26(NULL), _27(NULL), _28(NULL), _29(0), _30(false), _31(NULL), _32(false), _33(NULL), _34('f')) diff --git a/src/common/backend/catalog/system_views.sql b/src/common/backend/catalog/system_views.sql index b60f2bd17..791ac3003 100644 --- a/src/common/backend/catalog/system_views.sql +++ b/src/common/backend/catalog/system_views.sql @@ -3453,3 +3453,4 @@ CREATE unlogged table statement_history( ); REVOKE ALL on table pg_catalog.statement_history FROM public; create index statement_history_time_idx on pg_catalog.statement_history USING btree (start_time, is_slow_sql); +CREATE OR REPLACE VIEW pg_catalog.gs_archive_status AS SELECT * FROM gs_get_archive_status(); \ No newline at end of file diff --git a/src/gausskernel/process/postmaster/pgarch.cpp b/src/gausskernel/process/postmaster/pgarch.cpp index 877db984c..35426e162 100644 --- a/src/gausskernel/process/postmaster/pgarch.cpp +++ b/src/gausskernel/process/postmaster/pgarch.cpp @@ -24,6 +24,7 @@ * * ------------------------------------------------------------------------- */ +#include "funcapi.h" #include "postgres.h" #include "knl/knl_variable.h" @@ -39,6 +40,7 @@ #include "access/xact.h" #include "libpq/pqsignal.h" #include "miscadmin.h" +#include "nodes/execnodes.h" #include "postmaster/fork_process.h" #include "postmaster/pgarch.h" #include "postmaster/postmaster.h" @@ -50,6 +52,7 @@ #include "storage/pmsignal.h" #include "utils/guc.h" #include "utils/ps_status.h" +#include "utils/builtins.h" #include "gssignal/gs_signal.h" #include "alarm/alarm.h" @@ -668,6 +671,16 @@ static XLogRecPtr getRestartLsnFromSlot() return result; } +static inline void pgarch_UpdateArchiveLastLsn(XLogRecPtr targetLsn) +{ + struct timeval tv; + gettimeofday(&tv, NULL); + t_thrd.arch.last_arch_time = TIME_GET_MILLISEC(tv); + t_thrd.arch.pitr_task_last_lsn = targetLsn; + g_instance.archive_obs_cxt.pitr_task_last_lsn = t_thrd.arch.pitr_task_last_lsn; + g_instance.archive_obs_cxt.last_arch_time = t_thrd.arch.last_arch_time; +} + /* * pgarch_ArchiverObsCopyLoop * @@ -675,7 +688,6 @@ static XLogRecPtr getRestartLsnFromSlot() */ static void pgarch_ArchiverObsCopyLoop(XLogRecPtr flushPtr, doArchive fun) { - struct timeval tv; bool time_to_stop = false; /* @@ -743,9 +755,7 @@ static void pgarch_ArchiverObsCopyLoop(XLogRecPtr flushPtr, doArchive fun) ereport(WARNING, (errmsg("transaction log file \"%X/%X\" does not existed, it may be archived by the old primary.", (uint32)(targetLsn >> 32), (uint32)(targetLsn)))); - gettimeofday(&tv, NULL); - t_thrd.arch.last_arch_time = TIME_GET_MILLISEC(tv); - t_thrd.arch.pitr_task_last_lsn = targetLsn; + pgarch_UpdateArchiveLastLsn(targetLsn); continue; } @@ -755,9 +765,7 @@ static void pgarch_ArchiverObsCopyLoop(XLogRecPtr flushPtr, doArchive fun) (uint32)(targetLsn >> 32), (uint32)(targetLsn)))); pg_usleep(1000000L); /* wait a bit before retrying */ } else { - gettimeofday(&tv, NULL); - t_thrd.arch.last_arch_time = TIME_GET_MILLISEC(tv); - t_thrd.arch.pitr_task_last_lsn = targetLsn; + pgarch_UpdateArchiveLastLsn(targetLsn); ResetLatch(&t_thrd.arch.mainloop_latch); ereport(LOG, (errmsg("pgarch_ArchiverObsCopyLoop time change to %ld", t_thrd.arch.last_arch_time))); @@ -1127,6 +1135,7 @@ static WalSnd* pgarch_chooseWalsnd(XLogRecPtr targetLsn) SpinLockRelease(&walsnd->mutex); walsnd->arch_task_lsn = targetLsn; t_thrd.arch.sync_walsender_idx = i; + g_instance.archive_obs_cxt.sync_walsender_idx = i; g_instance.archive_obs_cxt.sync_walsender_term++; ereport(LOG, (errmsg("pgarch_chooseWalsnd has change from %d to %d , sub_term:%d", @@ -1189,3 +1198,77 @@ static void initArchiveCxt() { g_instance.archive_obs_cxt.archive_task.targetLsn = InvalidXLogRecPtr; g_instance.archive_obs_cxt.pitr_task_status = PITR_TASK_NONE; } + +Datum gs_get_archive_status(PG_FUNCTION_ARGS) +{ +#define ARCHIVE_STATUS_COLS 4 +#define MAX_LSN_LENGTH 18 + ReplicationSlot* slot = getObsReplicationSlot(); + ReturnSetInfo* rsinfo = (ReturnSetInfo*) fcinfo->resultinfo; + MemoryContext oldcontext = MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory); + + int col_number = 0; + bool nulls[ARCHIVE_STATUS_COLS]; + nulls[col_number++] = false; + nulls[col_number++] = false; + nulls[col_number++] = false; + nulls[col_number++] = false; + + col_number = 0; + TupleDesc tupdesc = CreateTemplateTupleDesc(ARCHIVE_STATUS_COLS, false); + TupleDescInitEntry(tupdesc, (AttrNumber)++col_number, "archive_standby", TEXTOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++col_number, "archive_lsn", TEXTOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++col_number, "last_arch_time", TIMESTAMPTZOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++col_number, "archive_path", TEXTOID, -1, 0); + rsinfo->returnMode = SFRM_Materialize; + rsinfo->setResult = tuplestore_begin_heap(true, false, u_sess->attr.attr_memory.work_mem); + rsinfo->setDesc = BlessTupleDesc(tupdesc); + + MemoryContextSwitchTo(oldcontext); + volatile WalSnd* walsnd = NULL; + int sync_walsender_idx = g_instance.archive_obs_cxt.sync_walsender_idx; + if (slot != NULL && sync_walsender_idx >= 0) { + walsnd = &t_thrd.walsender_cxt.WalSndCtl->walsnds[sync_walsender_idx]; + SpinLockAcquire(&walsnd->mutex); + if (walsnd->pid != 0 && ((walsnd->sendRole & SNDROLE_PRIMARY_STANDBY) == walsnd->sendRole) + && !XLogRecPtrIsInvalid(walsnd->flush)) { + int ret = 0; + Datum values[ARCHIVE_STATUS_COLS]; + XLogRecPtr last_arch_lsn = g_instance.archive_obs_cxt.pitr_task_last_lsn; + long last_arch_time = g_instance.archive_obs_cxt.last_arch_time; + /* get the ip address of standby */ + char remoteip[IP_LEN] = {0}; + errno_t rc = strncpy_s(remoteip, IP_LEN, (char *)walsnd->wal_sender_channel.remotehost, IP_LEN - 1); + securec_check(rc, "\0", "\0"); + remoteip[IP_LEN - 1] = '\0'; + + /* Construct the standby node identifier. The format is "remoteip_remoteport". */ + char archive_standby[MAXFNAMELEN] = {0}; + char archive_path[MAXFNAMELEN] = {0}; + int localport = walsnd->wal_sender_channel.localport; + ret = snprintf_s(archive_standby, sizeof(archive_standby), sizeof(archive_standby) - 1, "%s:%d", + remoteip, localport); + securec_check_ss(ret, "\0", "\0"); + + /* get the path of archiving xlog files */ + ret = snprintf_s(archive_path, sizeof(archive_path), sizeof(archive_path) - 1, "%s", + slot->archive_obs->obs_prefix); + securec_check_ss(ret, "\0", "\0"); + + char archive_lsn[MAX_LSN_LENGTH]; + ret = snprintf_s(archive_lsn, sizeof(archive_lsn), sizeof(archive_lsn) - 1, "%X/%X", + (uint32)(last_arch_lsn >> 32), (uint32)(last_arch_lsn)); + securec_check_ss(ret, "\0", "\0"); + + col_number = 0; + values[col_number++] = CStringGetTextDatum(archive_standby); + values[col_number++] = CStringGetTextDatum(archive_lsn); + values[col_number++] = TimestampTzGetDatum(last_arch_time); + values[col_number++] = CStringGetTextDatum(archive_path); + tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls); + } + SpinLockRelease(&walsnd->mutex); + } + tuplestore_donestoring(rsinfo->setResult); + return (Datum)0; +} diff --git a/src/gausskernel/process/threadpool/knl_instance.cpp b/src/gausskernel/process/threadpool/knl_instance.cpp index ccd7537c4..c09bc2556 100644 --- a/src/gausskernel/process/threadpool/knl_instance.cpp +++ b/src/gausskernel/process/threadpool/knl_instance.cpp @@ -488,6 +488,9 @@ static void knl_g_archive_obs_init(knl_g_archive_obs_context *archive_obs_cxt) archive_obs_cxt->sync_walsender_term = 0; archive_obs_cxt->archive_slot = (ReplicationSlot*)MemoryContextAllocZero( INSTANCE_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_STORAGE), sizeof(ReplicationSlot)); + archive_obs_cxt->sync_walsender_idx = -1; + archive_obs_cxt->last_arch_time = 0; + archive_obs_cxt->pitr_task_last_lsn = InvalidXLogRecPtr; } #ifdef ENABLE_MOT diff --git a/src/include/catalog/upgrade_sql/rollback_catalog_maindb/rollback_catalog_maindb_92_300.sql b/src/include/catalog/upgrade_sql/rollback_catalog_maindb/rollback_catalog_maindb_92_300.sql new file mode 100644 index 000000000..be358abde --- /dev/null +++ b/src/include/catalog/upgrade_sql/rollback_catalog_maindb/rollback_catalog_maindb_92_300.sql @@ -0,0 +1 @@ +DROP VIEW IF EXISTS pg_catalog.gs_archive_status; \ No newline at end of file diff --git a/src/include/catalog/upgrade_sql/rollback_catalog_otherdb/rollback_catalog_otherdb_92_300.sql b/src/include/catalog/upgrade_sql/rollback_catalog_otherdb/rollback_catalog_otherdb_92_300.sql new file mode 100644 index 000000000..be358abde --- /dev/null +++ b/src/include/catalog/upgrade_sql/rollback_catalog_otherdb/rollback_catalog_otherdb_92_300.sql @@ -0,0 +1 @@ +DROP VIEW IF EXISTS pg_catalog.gs_archive_status; \ No newline at end of file diff --git a/src/include/catalog/upgrade_sql/upgrade_catalog_maindb/upgrade-post_catalog_maindb_92_300.sql b/src/include/catalog/upgrade_sql/upgrade_catalog_maindb/upgrade-post_catalog_maindb_92_300.sql new file mode 100644 index 000000000..a21d7c85d --- /dev/null +++ b/src/include/catalog/upgrade_sql/upgrade_catalog_maindb/upgrade-post_catalog_maindb_92_300.sql @@ -0,0 +1 @@ +CREATE OR REPLACE VIEW pg_catalog.gs_archive_status AS SELECT * FROM gs_get_archive_status(); \ No newline at end of file diff --git a/src/include/catalog/upgrade_sql/upgrade_catalog_otherdb/upgrade-post_catalog_otherdb_92_300.sql b/src/include/catalog/upgrade_sql/upgrade_catalog_otherdb/upgrade-post_catalog_otherdb_92_300.sql new file mode 100644 index 000000000..a21d7c85d --- /dev/null +++ b/src/include/catalog/upgrade_sql/upgrade_catalog_otherdb/upgrade-post_catalog_otherdb_92_300.sql @@ -0,0 +1 @@ +CREATE OR REPLACE VIEW pg_catalog.gs_archive_status AS SELECT * FROM gs_get_archive_status(); \ No newline at end of file diff --git a/src/include/knl/knl_instance.h b/src/include/knl/knl_instance.h index 4c50e6953..d903c968f 100644 --- a/src/include/knl/knl_instance.h +++ b/src/include/knl/knl_instance.h @@ -735,6 +735,11 @@ typedef struct knl_g_archive_obs_context { struct ReplicationSlot* archive_slot; volatile int obs_slot_num; int sync_walsender_term; + + /* used to show the archive status */ + volatile int sync_walsender_idx; + volatile long last_arch_time; + volatile XLogRecPtr pitr_task_last_lsn; } knl_g_archive_obs_context; #ifdef ENABLE_MOT diff --git a/src/include/postmaster/pgarch.h b/src/include/postmaster/pgarch.h index e854facad..ce2c7f301 100644 --- a/src/include/postmaster/pgarch.h +++ b/src/include/postmaster/pgarch.h @@ -18,6 +18,7 @@ * ---------- */ extern ThreadId pgarch_start(void); +extern Datum gs_get_archive_status(PG_FUNCTION_ARGS); #ifdef EXEC_BACKEND extern void PgArchiverMain(); #endif diff --git a/src/test/regress/expected/single_node_opr_sanity.out b/src/test/regress/expected/single_node_opr_sanity.out index a35bd49d5..77493e298 100644 --- a/src/test/regress/expected/single_node_opr_sanity.out +++ b/src/test/regress/expected/single_node_opr_sanity.out @@ -2192,6 +2192,7 @@ WHERE d.classoid IS NULL AND p1.oid <= 9999 order by 1; 3793 | zhprs_getlexeme 3794 | zhprs_end 3795 | zhprs_lextype + 3796 | gs_get_archive_status 3798 | pound_start 3799 | pound_nexttoken 3800 | pound_end