[pgmoneta#123] Native backup: Foundation

Based upon a patch by jesper.pedersen@redhat.com

src/include/memory.h

@@ -69,6 +69,33 @@ pgmoneta_memory_free(void);
 void
 pgmoneta_memory_destroy(void);
 
+/**
+ * Create a dynamic memory segment
+ * @param size The new size
+ * @return The segment
+ */
+void*
+pgmoneta_memory_dynamic_create(size_t* size);
+
+/**
+ * Destroy a dynamic memory segment
+ * @param data The segment
+ */
+void
+pgmoneta_memory_dynamic_destroy(void* data);
+
+/**
+ * Append a dynamic memory segment
+ * @param orig The original memory segment
+ * @param orig_size The original size
+ * @param append The append memory segment
+ * @param append_size The append size
+ * @param new_size The new size
+ * @return The new segment
+ */
+void*
+pgmoneta_memory_dynamic_append(void* orig, size_t orig_size, void* append, size_t append_size, size_t* new_size);
+
 #ifdef __cplusplus
 }
 #endif

src/include/message.h

@@ -55,6 +55,26 @@ struct message
    void* data;        /**< The message data */
 } __attribute__ ((aligned (64)));
 
+/** @struct
+ * Defines a tuple
+ */
+struct tuple
+{
+   char** data;                   /**< The data */
+   struct tuple* next;            /**< The next tuple */
+} __attribute__ ((aligned (64)));
+
+/** @struct
+ * Defines the response to a query
+ */
+struct query_response
+{
+   char names[MAX_NUMBER_OF_COLUMNS][MISC_LENGTH]; /**< The column names */
+   int number_of_columns;                          /**< The number of columns */
+
+   struct tuple* tuples;
+} __attribute__ ((aligned (64)));
+
 /**
  * Read a message in blocking mode
  * @param ssl The SSL struct
@@ -256,11 +276,105 @@ pgmoneta_create_ssl_message(struct message** msg);
  * Create a startup message
  * @param username The user name
  * @param database The database
+ * @param replication Should replication be enabled
+ * @param msg The resulting message
+ * @return 0 upon success, otherwise 1
+ */
+int
+pgmoneta_create_startup_message(char* username, char* database, bool replication, struct message** msg);
+
+/**
+ * Create an IDENTIFY SYSTEM message
+ * @param msg The resulting message
+ * @return 0 upon success, otherwise 1
+ */
+int
+pgmoneta_create_identify_system_message(struct message** msg);
+
+/**
+ * Create a TIMELINE_HISTORY message
+ * @param timeline The timeline
+ * @param msg The resulting message
+ * @return 0 upon success, otherwise 1
+ */
+int
+pgmoneta_create_timeline_history_message(int timeline, struct message** msg);
+
+/**
+ * Create a READ_REPLICATION_SLOT message
+ * @param slot The slot
+ * @param msg The resulting message
+ * @return 0 upon success, otherwise 1
+ */
+int
+pgmoneta_create_read_replication_slot_message(char* slot, struct message** msg);
+
+/**
+ * Create a START_REPLICATION message
+ * @param xlogpos The WAL position (can be NULL)
+ * @param timeline The timeline (can be -1)
+ * @param slot The slot
+ * @param msg The resulting message
+ * @return 0 upon success, otherwise 1
+ */
+int
+pgmoneta_create_start_replication_message(char* xlogpos, int timeline, char* slot, struct message** msg);
+
+/**
+ * Create a standby status update message
+ * @param received The received position
+ * @param flushed The flushed position
+ * @param applied The applied position
  * @param msg The resulting message
  * @return 0 upon success, otherwise 1
  */
 int
-pgmoneta_create_startup_message(char* username, char* database, struct message** msg);
+pgmoneta_create_standby_status_update_message(int64_t received, int64_t flushed, int64_t applied, struct message** msg);
+
+/**
+ * Has a message
+ * @param type The message type to be extracted
+ * @param data The data
+ * @param data_size The data size
+ * @return true if found, otherwise false
+ */
+bool
+pgmoneta_has_message(char type, void* data, size_t data_size);
+
+/**
+ * Query execute
+ * @param socket The socket
+ * @param msg The query message
+ * @param query The query response
+ * @return 0 upon success, otherwise 1
+ */
+int
+pgmoneta_query_execute(int socket, struct message* msg, struct query_response** response);
+
+/**
+ * Get data from a query response
+ * @param response The response
+ * @param column The column
+ * @return The data
+ */
+char*
+pgmoneta_query_response_get_data(struct query_response* response, int column);
+
+/**
+ * Free query
+ * @param query The query
+ * @return 0 upon success, otherwise 1
+ */
+int
+pgmoneta_free_query_response(struct query_response* response);
+
+/**
+ * Debug query
+ * @param query The resulting query
+ * @return 0 upon success, otherwise 1
+ */
+void
+pgmoneta_query_response_debug(struct query_response* response);
 
 #ifdef __cplusplus
 }

src/include/pgmoneta.h

@@ -65,6 +65,8 @@ extern "C" {
 #define NUMBER_OF_USERS    64
 #define NUMBER_OF_ADMINS    8
 
+#define MAX_NUMBER_OF_COLUMNS 8
+
 #define STATE_FREE        0
 #define STATE_IN_USE      1
 
@@ -236,6 +238,8 @@ struct prometheus
  */
 struct configuration
 {
+   bool running; /**< Is pgmoneta running */
+
    char configuration_path[MAX_PATH]; /**< The configuration path */
    char users_path[MAX_PATH];         /**< The users path */
    char admins_path[MAX_PATH];        /**< The admins path */

src/include/security.h

@@ -45,11 +45,12 @@ extern "C" {
  * @param database The database
  * @param username The username
  * @param password The password
+ * @param replication Is replication enabled
  * @param fd The resulting socket
  * @return AUTH_SUCCESS, AUTH_BAD_PASSWORD or AUTH_ERROR
  */
 int
-pgmoneta_server_authenticate(int server, char* database, char* username, char* password, int* fd);
+pgmoneta_server_authenticate(int server, char* database, char* username, char* password, bool replication, int* fd);
 
 /**
  * Authenticate a remote management user

src/include/server.h

@@ -42,6 +42,15 @@ extern "C" {
 void
 pgmoneta_server_info(int srv);
 
+/**
+ * Get the WAL size for a server
+ * @param socket The socket
+ * @param ws The WAL size
+ * @return 0 upon success, otherwise 1
+ */
+int
+pgmoneta_server_get_wal_size(int socket, int* ws);
+
 #ifdef __cplusplus
 }
 #endif

src/include/utils.h

@@ -80,6 +80,27 @@ pgmoneta_extract_username_database(struct message* msg, char** username, char**
 int
 pgmoneta_extract_message(char type, struct message* msg, struct message** extracted);
 
+/**
+ * Extract a message based on an offset
+ * @param offset The offset
+ * @param data The data segment
+ * @param extracted The resulting message
+ * @return The next offset
+ */
+size_t
+pgmoneta_extract_message_offset(size_t offset, void* data, struct message** extracted);
+
+/**
+ * Extract a message based on a type
+ * @param type The type
+ * @param data The data segment
+ * @param data_size The data size
+ * @param extracted The resulting message
+ * @return 0 upon success, otherwise 1
+ */
+int
+pgmoneta_extract_message_from_data(char type, void* data, size_t data_size, struct message** extracted);
+
 /**
  * Read a byte
  * @param data Pointer to the data
@@ -88,6 +109,14 @@ pgmoneta_extract_message(char type, struct message* msg, struct message** extrac
 signed char
 pgmoneta_read_byte(void* data);
 
+/**
+ * Read an int16
+ * @param data Pointer to the data
+ * @return The int16
+ */
+int16_t
+pgmoneta_read_int16(void* data);
+
 /**
  * Read an int32
  * @param data Pointer to the data
@@ -642,6 +671,20 @@ pgmoneta_get_timestamp_ISO8601_format(char* short_date, char* long_date);
 int
 pgmoneta_get_timestamp_UTC_format(char* utc_date);
 
+/**
+ * Get the current local time
+ * @return The microseconds
+ */
+int64_t
+pgmoneta_get_current_timestamp(void);
+
+/**
+ * Get the local time since 2000-01-01 at midnight
+ * @return The microseconds
+ */
+int64_t
+pgmoneta_get_y2000_timestamp(void);
+
 /**
  * Convert base32 to hexadecimal.
  * @param base32 The base32.

src/libpgmoneta/aes.c

@@ -359,7 +359,7 @@ aes_encrypt(char* plaintext, unsigned char* key, unsigned char* iv, char** ciphe
    size_t size;
    unsigned char* ct = NULL;
    int ct_length;
-   const EVP_CIPHER* (*cipher_fp)(void) = get_cipher(mode);
+   const EVP_CIPHER* (* cipher_fp)(void) = get_cipher(mode);
    if (!(ctx = EVP_CIPHER_CTX_new()))
    {
       goto error;
@@ -417,7 +417,7 @@ aes_decrypt(char* ciphertext, int ciphertext_length, unsigned char* key, unsigne
    int length;
    size_t size;
    char* pt = NULL;
-   const EVP_CIPHER* (*cipher_fp)(void) = get_cipher(mode);
+   const EVP_CIPHER* (* cipher_fp)(void) = get_cipher(mode);
 
    if (!(ctx = EVP_CIPHER_CTX_new()))
    {
@@ -505,7 +505,7 @@ encrypt_file(char* from, char* to, int enc)
    char* master_key = NULL;
    EVP_CIPHER_CTX* ctx = NULL;
    struct configuration* config;
-   const EVP_CIPHER* (*cipher_fp)(void) = NULL;
+   const EVP_CIPHER* (* cipher_fp)(void) = NULL;
    int cipher_block_size = 0;
    int inbuf_size = 0;
    int outbuf_size = 0;

src/libpgmoneta/configuration.c

@@ -91,6 +91,8 @@ pgmoneta_init_configuration(void* shm)
 
    config = (struct configuration*)shm;
 
+   config->running = true;
+
    config->compression_type = COMPRESSION_ZSTD;
    config->compression_level = 3;
 

src/libpgmoneta/hashmap.c

@@ -46,7 +46,7 @@
 #define HASHMAP_PTR_CAST(type, x) ((type)x)
 #define HASHMAP_MAX_CHAIN_LENGTH 8
 
-static int hashmap_iterate(struct hashmap* hashmap, int (*f)(void*, struct hashmap_element*), void* context);
+static int hashmap_iterate(struct hashmap* hashmap, int (* f)(void*, struct hashmap_element*), void* context);
 static unsigned int hashmap_crc32_helper(char* s, unsigned int len);
 static unsigned int hashmap_hash_helper_int_helper(struct hashmap* m, char* keystring, unsigned int len);
 static int hashmap_match_helper(struct hashmap_element* element, char* key, unsigned int len);
@@ -282,7 +282,7 @@ pgmoneta_hashmap_destroy(struct hashmap* hashmap)
 
 static int
 hashmap_iterate(struct hashmap* hashmap,
-                int (*f)(void*, struct hashmap_element*),
+                int (* f)(void*, struct hashmap_element*),
                 void* context)
 {
    struct hashmap_element* p;

src/libpgmoneta/memory.c

@@ -122,3 +122,40 @@ pgmoneta_memory_destroy(void)
    data = NULL;
    message = NULL;
 }
+
+void*
+pgmoneta_memory_dynamic_create(size_t* size)
+{
+   *size = 0;
+
+   return NULL;
+}
+
+void
+pgmoneta_memory_dynamic_destroy(void* data)
+{
+   free(data);
+}
+
+void*
+pgmoneta_memory_dynamic_append(void* orig, size_t orig_size, void* append, size_t append_size, size_t* new_size)
+{
+   void* d = NULL;
+   size_t s;
+
+   if (append != NULL)
+   {
+      s = orig_size + append_size;
+      d = realloc(orig, s);
+      memcpy(d + orig_size, append, append_size);
+   }
+   else
+   {
+      s = orig_size;
+      d = orig;
+   }
+
+   *new_size = s;
+
+   return d;
+}