stub for s3 integration

Author: funny-falcon

Makefile

@@ -76,6 +76,19 @@ include $(top_builddir)/src/Makefile.global
 include $(top_srcdir)/contrib/contrib-global.mk
 endif
 
+ifndef S3_DIR
+  ifneq ("$(wildcard $(abspath $(top_pbk_srcdir))/../s3)", "")
+    S3_DIR = $(abspath $(CURDIR))/../s3
+  endif
+endif
+
+ifdef S3_DIR
+  LDFLAGS += -lcurl
+  CFLAGS += $(shell pkg-config --cflags libxml-2.0) -DPBCKP_S3=1
+  LDFLAGS += $(shell pkg-config --libs libxml-2.0)
+  OBJS += $(S3_DIR)/s3.o
+endif
+
 #
 PG_CPPFLAGS = -I$(libpq_srcdir) ${PTHREAD_CFLAGS} -I$(top_pbk_srcdir)src -I$(BORROW_DIR)
 PG_CPPFLAGS += -I$(top_pbk_srcdir)src/fu_util -Wno-declaration-after-statement
@@ -87,6 +100,9 @@ PG_LIBS_INTERNAL = $(libpq_pgport) ${PTHREAD_CFLAGS}
 
 # additional dependencies on borrowed files
 src/backup.o src/catchup.o src/pg_probackup.o: $(BORROW_DIR)/streamutil.h
+ifdef S3_DIR
+  src/backup.o src/catchup.o src/pg_probackup.o: $(S3_DIR)/s3.o
+endif
 src/stream.o $(BORROW_DIR)/receivelog.o $(BORROW_DIR)/streamutil.o $(BORROW_DIR)/walmethods.o: $(BORROW_DIR)/receivelog.h
 $(BORROW_DIR)/receivelog.h: $(BORROW_DIR)/walmethods.h
 

src/backup.c

@@ -46,7 +46,6 @@ static void do_backup_pg(InstanceState *instanceState, PGconn *backup_conn,
 
 static void pg_switch_wal(PGconn *conn);
 
-static void pg_stop_backup(InstanceState *instanceState, pgBackup *backup, PGconn *pg_startbackup_conn, PGNodeInfo *nodeInfo);
 
 static void check_external_for_tablespaces(parray *external_list,
 										   PGconn *backup_conn);
@@ -1827,7 +1826,7 @@ pg_stop_backup_write_file_helper(const char *path, const char *filename, const c
 /*
  * Notify end of backup to PostgreSQL server.
  */
-static void
+void
 pg_stop_backup(InstanceState *instanceState, pgBackup *backup, PGconn *pg_startbackup_conn,
 				PGNodeInfo *nodeInfo)
 {

src/pg_probackup.c

@@ -164,6 +164,12 @@ int64 ttl = -1;
 static char *expire_time_string = NULL;
 static pgSetBackupParams *set_backup_params = NULL;
 
+#ifdef PBCKP_S3
+/* S3 options */
+S3_protocol s3_protocol;
+char* s3_target_bucket = NULL;
+#endif
+
 /* ================ backupState =========== */
 static char *backup_id_string = NULL;
 pgBackup	current;
@@ -174,6 +180,9 @@ static bool help_opt = false;
 static void opt_incr_restore_mode(ConfigOption *opt, const char *arg);
 static void opt_backup_mode(ConfigOption *opt, const char *arg);
 static void opt_show_format(ConfigOption *opt, const char *arg);
+#ifdef PBCKP_S3
+static void opt_s3_protocol(ConfigOption *opt, const char *arg);
+#endif
 
 static void compress_init(ProbackupSubcmd const subcmd);
 
@@ -270,6 +279,12 @@ static ConfigOption cmd_options[] =
 	{ 'I', 170, "ttl", &ttl, SOURCE_CMD_STRICT, SOURCE_DEFAULT, 0, OPTION_UNIT_S, option_get_value},
 	{ 's', 171, "expire-time",		&expire_time_string,	SOURCE_CMD_STRICT },
 
+#ifdef PBCKP_S3
+	/* S3 options */
+	{ 'f', 245, "s3",				opt_s3_protocol,	SOURCE_CMD_STRICT },
+	{ 's', 246, "target-bucket", 	&s3_target_bucket,	SOURCE_CMD_STRICT },
+#endif
+
 	/* options for backward compatibility
 	 * TODO: remove in 3.0.0
 	 */
@@ -952,6 +967,19 @@ main(int argc, char *argv[])
 
 	compress_init(backup_subcmd);
 
+#ifdef PBCKP_S3
+	if (s3_protocol != S3_INVALID_PROTOCOL)
+	{
+		char* s3_config_file="";
+		read_s3_config(s3_config_file);
+	}
+	else
+	{
+		if (s3_target_bucket != NULL)
+			elog(WARNING, "You cannot specify s3-target without using --s3 option with name of protocol");
+	}
+#endif
+
 	/* do actual operation */
 	switch (backup_subcmd)
 	{
@@ -964,11 +992,27 @@ main(int argc, char *argv[])
 						   wal_file_path, wal_file_name, batch_size, !no_validate_wal);
 			break;
 		case ADD_INSTANCE_CMD:
-			return do_add_instance(instanceState, &instance_config);
+		{
+			int err = 0;
+			err = do_add_instance(instanceState, &instance_config);
+#ifdef PBCKP_S3
+			if (err == 0 && s3_protocol != S3_INVALID_PROTOCOL)
+				err = do_S3_write_config(&instance_config);
+#endif
+			return err;
+		}
 		case DELETE_INSTANCE_CMD:
 			return do_delete_instance(instanceState);
 		case INIT_CMD:
-			return do_init(catalogState);
+		{
+			int err = 0;
+			err = do_init(catalogState);
+#ifdef PBCKP_S3
+			if (err == 0 && s3_protocol != S3_INVALID_PROTOCOL)
+				err = S3_pre_start_check(config);
+#endif
+			return err;
+		}
 		case BACKUP_CMD:
 			{
 				current.stream = stream_wal;
@@ -983,13 +1027,21 @@ main(int argc, char *argv[])
 					elog(ERROR, "required parameter not specified: BACKUP_MODE "
 						 "(-b, --backup-mode)");
 
+#ifdef PBCKP_S3
+				if (s3_protocol != S3_INVALID_PROTOCOL)
+					return do_S3_backup(instanceState, set_backup_params, start_time);
+#endif
 				return do_backup(instanceState, set_backup_params,
 								 no_validate, no_sync, backup_logs, start_time);
 			}
 		case CATCHUP_CMD:
 			return do_catchup(catchup_source_pgdata, catchup_destination_pgdata, num_threads, !no_sync,
 				exclude_absolute_paths_list, exclude_relative_paths_list);
 		case RESTORE_CMD:
+#ifdef PBCKP_S3
+			if (s3_protocol != S3_INVALID_PROTOCOL)
+				return do_S3_restore(instanceState, current.backup_id);
+#endif
 			return do_restore_or_validate(instanceState, current.backup_id,
 							recovery_target_options,
 							restore_params, no_sync);
@@ -1009,6 +1061,10 @@ main(int argc, char *argv[])
 						  restore_params,
 						  no_sync);
 		case SHOW_CMD:
+#ifdef PBCKP_S3
+			if (s3_protocol != S3_INVALID_PROTOCOL)
+				return do_S3_show(instanceState);
+#endif
 			return do_show(catalogState, instanceState, current.backup_id, show_archive);
 		case DELETE_CMD:
 
@@ -1197,3 +1253,34 @@ opt_exclude_path(ConfigOption *opt, const char *arg)
 	else
 		opt_parser_add_to_parray_helper(&exclude_relative_paths_list, arg);
 }
+
+#ifdef PBCKP_S3
+static S3_protocol
+parse_s3_protocol(const char *value)
+{
+	const char *v = value;
+	size_t		len;
+
+	/* Skip all spaces detected */
+	while (IsSpace(*v))
+		v++;
+	len = strlen(v);
+
+	if (len > 0 && pg_strncasecmp("MINIO", v, len) == 0)
+		return S3_MINIO_PROTOCOL;
+	if (len > 0 && pg_strncasecmp("AWS", v, len) == 0)
+		return S3_AWS_PROTOCOL;
+	else if (len > 0 && pg_strncasecmp("GOOGLE", v, len) == 0)
+		return S3_GOOGLE_PROTOCOL;
+	else if (len > 0 && pg_strncasecmp("VK", v, len) == 0)
+		return S3_VK_PROTOCOL;
+	else
+		return S3_INVALID_PROTOCOL;
+}
+
+static void
+opt_s3_protocol(ConfigOption *opt, const char *arg)
+{
+	s3_protocol = parse_s3_protocol(arg);
+}
+#endif

src/pg_probackup.h

@@ -43,6 +43,10 @@
 
 #include "pg_probackup_state.h"
 
+#ifdef PGPRO_S3
+#include "../s3/s3.h"
+#endif
+
 #if defined(WIN32) && !(defined(_UCRT) && defined(__MINGW64__))
 #error Windows port requires compilation in MinGW64 UCRT environment
 #endif
@@ -800,6 +804,11 @@ extern pgBackup current;
 /* argv of the process */
 extern char** commands_args;
 
+#ifdef PBCKP_S3
+/* S3 options */
+extern S3_protocol s3_protocol;
+#endif
+
 /* in backup.c */
 extern int do_backup(InstanceState *instanceState, pgSetBackupParams *set_backup_params,
 					 bool no_validate, bool no_sync, bool backup_logs, time_t start_time);
@@ -810,6 +819,15 @@ extern const char *deparse_backup_mode(BackupMode mode);
 extern void process_block_change(ForkNumber forknum, RelFileNode rnode,
 								 BlockNumber blkno);
 
+#ifdef PBCKP_S3
+/* in s3.c */
+extern int do_S3_backup(InstanceState *instanceState,
+								 pgSetBackupParams *set_backup_params,time_t start_time);
+extern int do_S3_show(InstanceState *instanceState);
+extern int do_S3_restore(InstanceState *instanceState, time_t target_backup_id);
+extern int do_S3_write_config(InstanceConfig *instance);
+#endif
+
 /* in catchup.c */
 extern int do_catchup(const char *source_pgdata, const char *dest_pgdata, int num_threads, bool sync_dest_files,
 	parray *exclude_absolute_paths_list, parray *exclude_relative_paths_list);
@@ -1248,6 +1266,7 @@ extern parray *backup_files_list;
 
 extern void pg_start_backup(const char *label, bool smooth, pgBackup *backup,
 							PGNodeInfo *nodeInfo, PGconn *conn);
+extern void pg_stop_backup(InstanceState *instanceState, pgBackup *backup, PGconn *pg_startbackup_conn, PGNodeInfo *nodeInfo);
 extern void pg_silent_client_messages(PGconn *conn);
 extern void pg_create_restore_point(PGconn *conn, time_t backup_start_time);
 extern void pg_stop_backup_send(PGconn *conn, int server_version, bool is_started_on_replica, char **query_text);

src/utils/file.c

@@ -8,6 +8,10 @@
 #include "file.h"
 #include "storage/checksum.h"
 
+#ifdef PBCKP_S3
+#include "../s3/s3.h"
+#endif
+
 #define PRINTF_BUF_SIZE  1024
 
 static __thread unsigned long fio_fdset = 0;
@@ -3417,6 +3421,13 @@ static pioDrive_i remoteDrive;
 pioDrive_i
 pioDriveForLocation(fio_location loc)
 {
+    if (loc == FIO_CLOUD_HOST)
+#ifdef PBCKP_S3
+		return cloudDrive;
+#else
+		elog(ERROR, "NO CLOUD DRIVE YET");
+#endif
+
     if (fio_is_remote(loc))
         return remoteDrive;
     else
@@ -4915,4 +4926,7 @@ init_pio_objects(void)
 
     localDrive = bindref_pioDrive($alloc(pioLocalDrive));
     remoteDrive = bindref_pioDrive($alloc(pioRemoteDrive));
+#ifdef PBCKP_S3
+    create_pioCloudeDrive();
+#endif
 }

src/utils/file.h

@@ -88,7 +88,8 @@ typedef enum
 	FIO_LOCAL_HOST,  /* data is locate at local host */
 	FIO_DB_HOST,     /* data is located at Postgres server host */
 	FIO_BACKUP_HOST, /* data is located at backup host */
-	FIO_REMOTE_HOST  /* date is located at remote host */
+	FIO_REMOTE_HOST,  /* date is located at remote host */
+	FIO_CLOUD_HOST  /* date is located at cloud (S3) */
 } fio_location;
 
 extern fio_location MyLocation;