From 337a444e888920e6942a427132501e95485a8186 Mon Sep 17 00:00:00 2001
From: Nathan Bossart <bossartn@amazon.com>
Date: Mon, 25 Oct 2021 19:17:59 +0000
Subject: [PATCH v6 1/2] Introduce archive module framework.

---
 src/backend/access/transam/xlog.c      |   2 +-
 src/backend/postmaster/Makefile        |   1 +
 src/backend/postmaster/pgarch.c        | 182 +++++++++++----------------------
 src/backend/postmaster/postmaster.c    |   2 +
 src/backend/postmaster/shell_archive.c | 156 ++++++++++++++++++++++++++++
 src/backend/utils/init/miscinit.c      |  27 +++++
 src/backend/utils/misc/guc.c           |  15 ++-
 src/include/access/xlog.h              |   1 -
 src/include/miscadmin.h                |   2 +
 src/include/postmaster/pgarch.h        |  45 ++++++++
 10 files changed, 308 insertions(+), 125 deletions(-)
 create mode 100644 src/backend/postmaster/shell_archive.c

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index cd553d6e12..70e87af284 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -8795,7 +8795,7 @@ ShutdownXLOG(int code, Datum arg)
 		 * process one more time at the end of shutdown). The checkpoint
 		 * record will go to the next XLOG file and won't be archived (yet).
 		 */
-		if (XLogArchivingActive() && XLogArchiveCommandSet())
+		if (XLogArchivingActive())
 			RequestXLogSwitch(false);
 
 		CreateCheckPoint(CHECKPOINT_IS_SHUTDOWN | CHECKPOINT_IMMEDIATE);
diff --git a/src/backend/postmaster/Makefile b/src/backend/postmaster/Makefile
index 787c6a2c3b..dbbeac5a82 100644
--- a/src/backend/postmaster/Makefile
+++ b/src/backend/postmaster/Makefile
@@ -23,6 +23,7 @@ OBJS = \
 	pgarch.o \
 	pgstat.o \
 	postmaster.o \
+	shell_archive.o \
 	startup.o \
 	syslogger.o \
 	walwriter.o
diff --git a/src/backend/postmaster/pgarch.c b/src/backend/postmaster/pgarch.c
index 74a7d7c4d0..f0e437f820 100644
--- a/src/backend/postmaster/pgarch.c
+++ b/src/backend/postmaster/pgarch.c
@@ -25,18 +25,12 @@
  */
 #include "postgres.h"
 
-#include <fcntl.h>
-#include <signal.h>
-#include <time.h>
 #include <sys/stat.h>
-#include <sys/time.h>
-#include <sys/wait.h>
 #include <unistd.h>
 
 #include "access/xlog.h"
 #include "access/xlog_internal.h"
 #include "libpq/pqsignal.h"
-#include "miscadmin.h"
 #include "pgstat.h"
 #include "postmaster/interrupt.h"
 #include "postmaster/pgarch.h"
@@ -78,6 +72,8 @@ typedef struct PgArchData
 	int			pgprocno;		/* pgprocno of archiver process */
 } PgArchData;
 
+char *XLogArchiveLibrary = "";
+
 
 /* ----------
  * Local data
@@ -85,6 +81,8 @@ typedef struct PgArchData
  */
 static time_t last_sigterm_time = 0;
 static PgArchData *PgArch = NULL;
+static ArchiveModuleCallbacks *ArchiveContext = NULL;
+
 
 /*
  * Flags set by interrupt handlers for later service in the main loop.
@@ -103,6 +101,7 @@ static bool pgarch_readyXlog(char *xlog);
 static void pgarch_archiveDone(char *xlog);
 static void pgarch_die(int code, Datum arg);
 static void HandlePgArchInterrupts(void);
+static void LoadArchiveLibrary(void);
 
 /* Report shared memory space needed by PgArchShmemInit */
 Size
@@ -198,6 +197,11 @@ PgArchiverMain(void)
 	 */
 	PgArch->pgprocno = MyProc->pgprocno;
 
+	/*
+	 * Load the archive_library.
+	 */
+	LoadArchiveLibrary();
+
 	pgarch_MainLoop();
 
 	proc_exit(0);
@@ -358,11 +362,11 @@ pgarch_ArchiverCopyLoop(void)
 			 */
 			HandlePgArchInterrupts();
 
-			/* can't do anything if no command ... */
-			if (!XLogArchiveCommandSet())
+			/* can't do anything if not configured ... */
+			if (!ArchiveContext->check_configured_cb())
 			{
 				ereport(WARNING,
-						(errmsg("archive_mode enabled, yet archive_command is not set")));
+						(errmsg("archive_mode enabled, yet archiving is not configured")));
 				return;
 			}
 
@@ -443,136 +447,31 @@ pgarch_ArchiverCopyLoop(void)
 /*
  * pgarch_archiveXlog
  *
- * Invokes system(3) to copy one archive file to wherever it should go
+ * Invokes archive_file_cb to copy one archive file to wherever it should go
  *
  * Returns true if successful
  */
 static bool
 pgarch_archiveXlog(char *xlog)
 {
-	char		xlogarchcmd[MAXPGPATH];
 	char		pathname[MAXPGPATH];
 	char		activitymsg[MAXFNAMELEN + 16];
-	char	   *dp;
-	char	   *endp;
-	const char *sp;
-	int			rc;
+	bool		ret;
 
 	snprintf(pathname, MAXPGPATH, XLOGDIR "/%s", xlog);
 
-	/*
-	 * construct the command to be executed
-	 */
-	dp = xlogarchcmd;
-	endp = xlogarchcmd + MAXPGPATH - 1;
-	*endp = '\0';
-
-	for (sp = XLogArchiveCommand; *sp; sp++)
-	{
-		if (*sp == '%')
-		{
-			switch (sp[1])
-			{
-				case 'p':
-					/* %p: relative path of source file */
-					sp++;
-					strlcpy(dp, pathname, endp - dp);
-					make_native_path(dp);
-					dp += strlen(dp);
-					break;
-				case 'f':
-					/* %f: filename of source file */
-					sp++;
-					strlcpy(dp, xlog, endp - dp);
-					dp += strlen(dp);
-					break;
-				case '%':
-					/* convert %% to a single % */
-					sp++;
-					if (dp < endp)
-						*dp++ = *sp;
-					break;
-				default:
-					/* otherwise treat the % as not special */
-					if (dp < endp)
-						*dp++ = *sp;
-					break;
-			}
-		}
-		else
-		{
-			if (dp < endp)
-				*dp++ = *sp;
-		}
-	}
-	*dp = '\0';
-
-	ereport(DEBUG3,
-			(errmsg_internal("executing archive command \"%s\"",
-							 xlogarchcmd)));
-
 	/* Report archive activity in PS display */
 	snprintf(activitymsg, sizeof(activitymsg), "archiving %s", xlog);
 	set_ps_display(activitymsg);
 
-	rc = system(xlogarchcmd);
-	if (rc != 0)
-	{
-		/*
-		 * If either the shell itself, or a called command, died on a signal,
-		 * abort the archiver.  We do this because system() ignores SIGINT and
-		 * SIGQUIT while waiting; so a signal is very likely something that
-		 * should have interrupted us too.  Also die if the shell got a hard
-		 * "command not found" type of error.  If we overreact it's no big
-		 * deal, the postmaster will just start the archiver again.
-		 */
-		int			lev = wait_result_is_any_signal(rc, true) ? FATAL : LOG;
-
-		if (WIFEXITED(rc))
-		{
-			ereport(lev,
-					(errmsg("archive command failed with exit code %d",
-							WEXITSTATUS(rc)),
-					 errdetail("The failed archive command was: %s",
-							   xlogarchcmd)));
-		}
-		else if (WIFSIGNALED(rc))
-		{
-#if defined(WIN32)
-			ereport(lev,
-					(errmsg("archive command was terminated by exception 0x%X",
-							WTERMSIG(rc)),
-					 errhint("See C include file \"ntstatus.h\" for a description of the hexadecimal value."),
-					 errdetail("The failed archive command was: %s",
-							   xlogarchcmd)));
-#else
-			ereport(lev,
-					(errmsg("archive command was terminated by signal %d: %s",
-							WTERMSIG(rc), pg_strsignal(WTERMSIG(rc))),
-					 errdetail("The failed archive command was: %s",
-							   xlogarchcmd)));
-#endif
-		}
-		else
-		{
-			ereport(lev,
-					(errmsg("archive command exited with unrecognized status %d",
-							rc),
-					 errdetail("The failed archive command was: %s",
-							   xlogarchcmd)));
-		}
-
+	ret = ArchiveContext->archive_file_cb(xlog, pathname);
+	if (ret)
+		snprintf(activitymsg, sizeof(activitymsg), "last was %s", xlog);
+	else
 		snprintf(activitymsg, sizeof(activitymsg), "failed on %s", xlog);
-		set_ps_display(activitymsg);
-
-		return false;
-	}
-	elog(DEBUG1, "archived write-ahead log file \"%s\"", xlog);
-
-	snprintf(activitymsg, sizeof(activitymsg), "last was %s", xlog);
 	set_ps_display(activitymsg);
 
-	return true;
+	return ret;
 }
 
 /*
@@ -716,3 +615,44 @@ HandlePgArchInterrupts(void)
 		ProcessConfigFile(PGC_SIGHUP);
 	}
 }
+
+/*
+ * LoadArchiveLibrary
+ *
+ * Loads the archiving callbacks into our local ArchiveContext.
+ */
+static void
+LoadArchiveLibrary(void)
+{
+	ArchiveContext = palloc0(sizeof(ArchiveModuleCallbacks));
+
+	/*
+	 * If shell archiving is enabled, use our special initialization
+	 * function.  Otherwise, load the library and call its
+	 * _PG_archive_module_init().
+	 */
+	if (ShellArchivingEnabled())
+		shell_archive_init(ArchiveContext);
+	else
+	{
+		ArchiveModuleInit archive_init;
+
+		archive_init = (ArchiveModuleInit)
+			load_external_function(XLogArchiveLibrary,
+								   "_PG_archive_module_init", false, NULL);
+
+		if (archive_init == NULL)
+			ereport(ERROR,
+					(errmsg("archive modules have to declare the "
+							"_PG_archive_module_init symbol")));
+
+		archive_init(ArchiveContext);
+	}
+
+	if (ArchiveContext->check_configured_cb == NULL)
+		ereport(ERROR,
+				(errmsg("archive modules must register a check callback")));
+	if (ArchiveContext->archive_file_cb == NULL)
+		ereport(ERROR,
+				(errmsg("archive modules must register an archive callback")));
+}
diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c
index e2a76ba055..f43c6b4cdc 100644
--- a/src/backend/postmaster/postmaster.c
+++ b/src/backend/postmaster/postmaster.c
@@ -1024,6 +1024,7 @@ PostmasterMain(int argc, char *argv[])
 	 * process any libraries that should be preloaded at postmaster start
 	 */
 	process_shared_preload_libraries();
+	process_archive_library();
 
 	/*
 	 * Initialize SSL library, if specified.
@@ -5011,6 +5012,7 @@ SubPostmasterMain(int argc, char *argv[])
 	 * non-EXEC_BACKEND behavior.
 	 */
 	process_shared_preload_libraries();
+	process_archive_library();
 
 	/* Run backend or appropriate child */
 	if (strcmp(argv[1], "--forkbackend") == 0)
diff --git a/src/backend/postmaster/shell_archive.c b/src/backend/postmaster/shell_archive.c
new file mode 100644
index 0000000000..7298dda6ee
--- /dev/null
+++ b/src/backend/postmaster/shell_archive.c
@@ -0,0 +1,156 @@
+/*-------------------------------------------------------------------------
+ *
+ * shell_archive.c
+ *
+ * This archiving function uses a user-specified shell command (the
+ * archive_command GUC) to copy write-ahead log files.  It is used as the
+ * default, but other modules may define their own custom archiving logic.
+ *
+ * Copyright (c) 2021, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/postmaster/shell_archive.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include <sys/wait.h>
+
+#include "access/xlog.h"
+#include "postmaster/pgarch.h"
+
+static bool shell_archive_configured(void);
+static bool shell_archive_file(const char *file, const char *path);
+
+void
+shell_archive_init(ArchiveModuleCallbacks *cb)
+{
+	AssertVariableIsOfType(&shell_archive_init, ArchiveModuleInit);
+
+	cb->check_configured_cb = shell_archive_configured;
+	cb->archive_file_cb = shell_archive_file;
+}
+
+static bool
+shell_archive_configured(void)
+{
+	return XLogArchiveCommand[0] != '\0';
+}
+
+static bool
+shell_archive_file(const char *file, const char *path)
+{
+	char		xlogarchcmd[MAXPGPATH];
+	char	   *dp;
+	char	   *endp;
+	const char *sp;
+	int			rc;
+
+	Assert(file != NULL);
+	Assert(path != NULL);
+
+	/*
+	 * construct the command to be executed
+	 */
+	dp = xlogarchcmd;
+	endp = xlogarchcmd + MAXPGPATH - 1;
+	*endp = '\0';
+
+	for (sp = XLogArchiveCommand; *sp; sp++)
+	{
+		if (*sp == '%')
+		{
+			switch (sp[1])
+			{
+				case 'p':
+					/* %p: relative path of source file */
+					sp++;
+					strlcpy(dp, path, endp - dp);
+					make_native_path(dp);
+					dp += strlen(dp);
+					break;
+				case 'f':
+					/* %f: filename of source file */
+					sp++;
+					strlcpy(dp, file, endp - dp);
+					dp += strlen(dp);
+					break;
+				case '%':
+					/* convert %% to a single % */
+					sp++;
+					if (dp < endp)
+						*dp++ = *sp;
+					break;
+				default:
+					/* otherwise treat the % as not special */
+					if (dp < endp)
+						*dp++ = *sp;
+					break;
+			}
+		}
+		else
+		{
+			if (dp < endp)
+				*dp++ = *sp;
+		}
+	}
+	*dp = '\0';
+
+	ereport(DEBUG3,
+			(errmsg_internal("executing archive command \"%s\"",
+							 xlogarchcmd)));
+
+	rc = system(xlogarchcmd);
+	if (rc != 0)
+	{
+		/*
+		 * If either the shell itself, or a called command, died on a signal,
+		 * abort the archiver.  We do this because system() ignores SIGINT and
+		 * SIGQUIT while waiting; so a signal is very likely something that
+		 * should have interrupted us too.  Also die if the shell got a hard
+		 * "command not found" type of error.  If we overreact it's no big
+		 * deal, the postmaster will just start the archiver again.
+		 */
+		int			lev = wait_result_is_any_signal(rc, true) ? FATAL : LOG;
+
+		if (WIFEXITED(rc))
+		{
+			ereport(lev,
+					(errmsg("archive command failed with exit code %d",
+							WEXITSTATUS(rc)),
+					 errdetail("The failed archive command was: %s",
+							   xlogarchcmd)));
+		}
+		else if (WIFSIGNALED(rc))
+		{
+#if defined(WIN32)
+			ereport(lev,
+					(errmsg("archive command was terminated by exception 0x%X",
+							WTERMSIG(rc)),
+					 errhint("See C include file \"ntstatus.h\" for a description of the hexadecimal value."),
+					 errdetail("The failed archive command was: %s",
+							   xlogarchcmd)));
+#else
+			ereport(lev,
+					(errmsg("archive command was terminated by signal %d: %s",
+							WTERMSIG(rc), pg_strsignal(WTERMSIG(rc))),
+					 errdetail("The failed archive command was: %s",
+							   xlogarchcmd)));
+#endif
+		}
+		else
+		{
+			ereport(lev,
+					(errmsg("archive command exited with unrecognized status %d",
+							rc),
+					 errdetail("The failed archive command was: %s",
+							   xlogarchcmd)));
+		}
+
+		return false;
+	}
+
+	elog(DEBUG1, "archived write-ahead log file \"%s\"", file);
+	return true;
+}
diff --git a/src/backend/utils/init/miscinit.c b/src/backend/utils/init/miscinit.c
index 88801374b5..9f2766ed04 100644
--- a/src/backend/utils/init/miscinit.c
+++ b/src/backend/utils/init/miscinit.c
@@ -38,6 +38,7 @@
 #include "pgstat.h"
 #include "postmaster/autovacuum.h"
 #include "postmaster/interrupt.h"
+#include "postmaster/pgarch.h"
 #include "postmaster/postmaster.h"
 #include "storage/fd.h"
 #include "storage/ipc.h"
@@ -1614,6 +1615,9 @@ char	   *local_preload_libraries_string = NULL;
 /* Flag telling that we are loading shared_preload_libraries */
 bool		process_shared_preload_libraries_in_progress = false;
 
+/* Flag telling that we are loading archive_library */
+bool		process_archive_library_in_progress = false;
+
 /*
  * load the shared libraries listed in 'libraries'
  *
@@ -1696,6 +1700,29 @@ process_session_preload_libraries(void)
 				   true);
 }
 
+/*
+ * process the archive library
+ */
+void
+process_archive_library(void)
+{
+	process_archive_library_in_progress = true;
+
+	/*
+	 * The shell archiving code is in the core server, so there's nothing
+	 * to load for that.
+	 */
+	if (!ShellArchivingEnabled())
+	{
+		load_file(XLogArchiveLibrary, false);
+		ereport(DEBUG1,
+				(errmsg_internal("loaded archive library \"%s\"",
+								 XLogArchiveLibrary)));
+	}
+
+	process_archive_library_in_progress = false;
+}
+
 void
 pg_bindtextdomain(const char *domain)
 {
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index e91d5a3cfd..9204f608fc 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -3864,13 +3864,23 @@ static struct config_string ConfigureNamesString[] =
 	{
 		{"archive_command", PGC_SIGHUP, WAL_ARCHIVING,
 			gettext_noop("Sets the shell command that will be called to archive a WAL file."),
-			NULL
+			gettext_noop("This is unused if \"archive_library\" does not indicate archiving via shell is enabled.")
 		},
 		&XLogArchiveCommand,
 		"",
 		NULL, NULL, show_archive_command
 	},
 
+	{
+		{"archive_library", PGC_POSTMASTER, WAL_ARCHIVING,
+			gettext_noop("Sets the library that will be called to archive a WAL file."),
+			gettext_noop("A value of \"shell\" or an empty string indicates that \"archive_command\" should be used.")
+		},
+		&XLogArchiveLibrary,
+		"shell",
+		NULL, NULL, NULL
+	},
+
 	{
 		{"restore_command", PGC_SIGHUP, WAL_ARCHIVE_RECOVERY,
 			gettext_noop("Sets the shell command that will be called to retrieve an archived WAL file."),
@@ -8961,7 +8971,8 @@ init_custom_variable(const char *name,
 	 * module might already have hooked into.
 	 */
 	if (context == PGC_POSTMASTER &&
-		!process_shared_preload_libraries_in_progress)
+		!process_shared_preload_libraries_in_progress &&
+		!process_archive_library_in_progress)
 		elog(FATAL, "cannot create PGC_POSTMASTER variables after startup");
 
 	/*
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 5e2c94a05f..7093e3390f 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -157,7 +157,6 @@ extern PGDLLIMPORT int wal_level;
 /* Is WAL archiving enabled always (even during recovery)? */
 #define XLogArchivingAlways() \
 	(AssertMacro(XLogArchiveMode == ARCHIVE_MODE_OFF || wal_level >= WAL_LEVEL_REPLICA), XLogArchiveMode == ARCHIVE_MODE_ALWAYS)
-#define XLogArchiveCommandSet() (XLogArchiveCommand[0] != '\0')
 
 /*
  * Is WAL-logging necessary for archival or log-shipping, or can we skip
diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h
index 90a3016065..8717fed0dc 100644
--- a/src/include/miscadmin.h
+++ b/src/include/miscadmin.h
@@ -464,6 +464,7 @@ extern void BaseInit(void);
 /* in utils/init/miscinit.c */
 extern bool IgnoreSystemIndexes;
 extern PGDLLIMPORT bool process_shared_preload_libraries_in_progress;
+extern PGDLLIMPORT bool process_archive_library_in_progress;
 extern char *session_preload_libraries_string;
 extern char *shared_preload_libraries_string;
 extern char *local_preload_libraries_string;
@@ -477,6 +478,7 @@ extern bool RecheckDataDirLockFile(void);
 extern void ValidatePgVersion(const char *path);
 extern void process_shared_preload_libraries(void);
 extern void process_session_preload_libraries(void);
+extern void process_archive_library(void);
 extern void pg_bindtextdomain(const char *domain);
 extern bool has_rolreplication(Oid roleid);
 
diff --git a/src/include/postmaster/pgarch.h b/src/include/postmaster/pgarch.h
index 1e47a143e1..7d09d2665e 100644
--- a/src/include/postmaster/pgarch.h
+++ b/src/include/postmaster/pgarch.h
@@ -32,4 +32,49 @@ extern bool PgArchCanRestart(void);
 extern void PgArchiverMain(void) pg_attribute_noreturn();
 extern void PgArchWakeup(void);
 
+/*
+ * The value of the archive_library GUC.
+ */
+extern char *XLogArchiveLibrary;
+
+/*
+ * Callback that gets called to determine if the archive module is
+ * configured.
+ */
+typedef bool (*ArchiveCheckConfiguredCB) (void);
+
+/*
+ * Callback called to archive a single WAL file.
+ */
+typedef bool (*ArchiveFileCB) (const char *file, const char *path);
+
+/*
+ * Archive module callbacks
+ */
+typedef struct ArchiveModuleCallbacks
+{
+	ArchiveCheckConfiguredCB check_configured_cb;
+	ArchiveFileCB archive_file_cb;
+} ArchiveModuleCallbacks;
+
+/*
+ * Type of the shared library symbol _PG_archive_module_init that is looked
+ * up when loading an archive library.
+ */
+typedef void (*ArchiveModuleInit) (ArchiveModuleCallbacks *cb);
+
+/*
+ * Since the logic for archiving via a shell command is in the core server
+ * and does not need to be loaded via a shared library, it has a special
+ * initialization function.
+ */
+extern void shell_archive_init(ArchiveModuleCallbacks *cb);
+
+/*
+ * We consider archiving via shell to be enabled if archive_library is
+ * empty or if archive_library is set to "shell".
+ */
+#define ShellArchivingEnabled() \
+	(XLogArchiveLibrary[0] == '\0' || strcmp(XLogArchiveLibrary, "shell") == 0)
+
 #endif							/* _PGARCH_H */
-- 
2.16.6

