[PBCKP-236] first-stage compatibility protocol impl with stubs

Author: Ivan Lazarev

src/pg_probackup.h

@@ -341,8 +341,8 @@ typedef enum ShowFormat
 #define PROGRAM_VERSION	"2.5.8"
 
 /* update when remote agent API or behaviour changes */
-#define AGENT_PROTOCOL_VERSION 20501
-#define AGENT_PROTOCOL_VERSION_STR "2.5.1"
+#define AGENT_PROTOCOL_VERSION 20509
+#define AGENT_PROTOCOL_VERSION_STR "2.5.9"
 
 /* update only when changing storage format */
 #define STORAGE_FORMAT_VERSION "2.4.4"
@@ -881,6 +881,10 @@ extern bool tliIsPartOfHistory(const parray *timelines, TimeLineID tli);
 extern DestDirIncrCompatibility check_incremental_compatibility(const char *pgdata, uint64 system_identifier,
 																IncrRestoreMode incremental_mode);
 
+/* in remote.c */
+extern void check_remote_agent_compatibility(int agent_version, char *compatibility_str);
+extern size_t prepare_remote_agent_compatibility_str(char* compatibility_buf, size_t buf_size);
+
 /* in merge.c */
 extern void do_merge(InstanceState *instanceState, time_t backup_id, bool no_validate, bool no_sync);
 extern void merge_backups(pgBackup *backup, pgBackup *next_backup);

src/utils/file.c

@@ -268,18 +268,24 @@ fio_write_all(int fd, void const* buf, size_t size)
 	return offs;
 }
 
+//TODO REVIEW XXX move to remote.c????
 /* Get version of remote agent */
-int
-fio_get_agent_version(void)
+void
+fio_get_agent_version(int* protocol, char* payload_buf, size_t payload_buf_size)
 {
 	fio_header hdr;
 	hdr.cop = FIO_AGENT_VERSION;
 	hdr.size = 0;
 
 	IO_CHECK(fio_write_all(fio_stdout, &hdr, sizeof(hdr)), sizeof(hdr));
 	IO_CHECK(fio_read_all(fio_stdin, &hdr, sizeof(hdr)), sizeof(hdr));
+	if (hdr.size > payload_buf_size)
+	{
+		elog(ERROR, "Bad protocol, insufficient payload_buf_size=%u", payload_buf_size);
+	}
 
-	return hdr.arg;
+	*protocol = hdr.arg;
+	IO_CHECK(fio_read_all(fio_stdin, payload_buf, hdr.size), hdr.size);
 }
 
 /* Open input stream. Remote file is fetched to the in-memory buffer and then accessed through Linux fmemopen */
@@ -3210,6 +3216,7 @@ fio_delete_impl(mode_t mode, char *buf)
 }
 
 /* Execute commands at remote host */
+//TODO REVIEW XXX move to remote.c?
 void
 fio_communicate(int in, int out)
 {
@@ -3316,6 +3323,13 @@ fio_communicate(int in, int out)
 		  case FIO_AGENT_VERSION:
 			hdr.arg = AGENT_PROTOCOL_VERSION;
 			IO_CHECK(fio_write_all(out, &hdr, sizeof(hdr)), sizeof(hdr));
+			//TODO REVIEW XXX is it allowed by ANSI C to declare new scope inside???
+			{
+				size_t payload_size = prepare_remote_agent_compatibility_str(buf, buf_size);
+				IO_CHECK(fio_write_all(out, buf, payload_size), payload_size);
+				//TODO REVIEW XXX make INFO to LOG or VERBOSE
+				elog(INFO, "TODO REVIEW XXX sent agent compatibility\n %s", buf);
+			}
 			break;
 		  case FIO_STAT: /* Get information about file with specified path */
 			hdr.size = sizeof(st);

src/utils/file.h

@@ -91,7 +91,7 @@ extern fio_location MyLocation;
 extern void    fio_redirect(int in, int out, int err);
 extern void    fio_communicate(int in, int out);
 
-extern int     fio_get_agent_version(void);
+extern void    fio_get_agent_version(int* protocol, char* payload_buf, size_t payload_buf_size);
 extern FILE*   fio_fopen(char const* name, char const* mode, fio_location location);
 extern size_t  fio_fwrite(FILE* f, void const* buf, size_t size);
 extern ssize_t fio_fwrite_async_compressed(FILE* f, void const* buf, size_t size, int compress_alg);

src/utils/remote.c

@@ -117,6 +117,9 @@ bool launch_agent(void)
 	int infd[2];
 	int errfd[2];
 	int agent_version;
+	//TODO REVIEW XXX review buf_size
+	size_t payload_buf_size = 1024 * 8;
+	char payload_buf[payload_buf_size];
 
 	ssh_argc = 0;
 #ifdef WIN32
@@ -238,10 +241,31 @@ bool launch_agent(void)
 		fio_redirect(infd[0], outfd[1], errfd[0]); /* write to stdout */
 	}
 
-	/* Make sure that remote agent has the same version
-	 * TODO: we must also check PG version and fork edition
+	/* Make sure that remote agent has the same version, fork and other features to be binary compatible
 	 */
-	agent_version = fio_get_agent_version();
+	fio_get_agent_version(&agent_version, payload_buf, payload_buf_size);
+	check_remote_agent_compatibility(0, payload_buf);
+
+	return true;
+}
+
+//TODO REVIEW XXX review macro
+#define STR(macro) #macro
+size_t prepare_remote_agent_compatibility_str(char* compatibility_buf, size_t buf_size)
+{
+	size_t payload_size = snprintf(compatibility_buf, buf_size,
+								   "%s\n%s\n%s\n%s\n",
+								   STR(PG_MAJORVERSION), PG_MAJORVERSION,
+								   STR(PGPRO_EDN), PGPRO_EDN);
+	if (payload_size >= buf_size)
+	{
+		elog(ERROR, "TODO REVIEW XXX too bad message buffer exhaust");
+	}
+	return payload_size + 1;
+}
+
+void check_remote_agent_compatibility(int agent_version, char *compatibility_str)
+{
 	if (agent_version != AGENT_PROTOCOL_VERSION)
 	{
 		char agent_version_str[1024];
@@ -254,6 +278,6 @@ bool launch_agent(void)
 					"consider to upgrade pg_probackup binary",
 			agent_version_str, AGENT_PROTOCOL_VERSION_STR);
 	}
-
-	return true;
+	assert(false);
+	elog(ERROR, " check_remote_agent_compatibility() not implemented");
 }