Implent snapshots support (#313)

This PR implements the necessary support to generate the RDB snapshots
in the background.

A new fiber has been introduced which takes care of generating a
snapshot, depending on the settings. A new set of parameters have been
introduced, example below with explanation in line.

```
  snapshots:
    # The path where the snapshot file will be stored, if the rotation is enabled, the path will be used as prefix and
    # the timestamp of the start of the snapshot will be appended to the file name.
    path: /var/lib/cachegrand/dump.rdb
    # The interval between the snapshots, the allowed units are s, m, h, if not specified the default is seconds.
    interval: 5m
    # The number of keys that must be changed before a snapshot is taken, 0 means no limit
    min_keys_changed: 1000
    # The amount of data that must be changed before a snapshot is taken, the allowed units are b, k, m, g, if not
    min_data_changed: 100mb
    # Rotation settings, optional, if missing the snapshots rotation will be disabled
    rotation:
      # The max number of snapshots files to keep, minimum 2
      max_files: 10
```

The example is from cachegrand.yaml.skel

The new mechanism takes care of reporting every 3 seconds a status
update.

Closes #293 

Notes:
- Currently the redis commands SAVE and BGSAVE are not implemented, so
it's not possible to trigger a backup on demand (#314)
- The SHUTDOWN command needs to be updated to support SAVE and NOSAVE
(#315)
- The shutdown logic needs to be updated to trigger a dump at the
shutdown unless the SHUTDOWN NOSAVE command has been issued (#316)
- The current implementation also doesn't compress strings with the LZF
algorithm as liblzf is causing segfault and the issue has to be
investigated (#312)
- Tests for the high level snapshotting process (implemented in
storage_db_snapshot.c mostly) are not included in this PR (#317)

etc/cachegrand.yaml.skel

@@ -125,6 +125,23 @@ database:
 #    # When database.limit.hard.max_memory_usage is set to 0, this value will be ignored
 #    max_keys: 999999
 
+  # The snapshot settings, optional, if missing the snapshots will be disabled. Snapshots are compatible with the Redis
+  # RDB format and can be used to restore the database state or import it from/to another instance.
+  snapshots:
+    # The path where the snapshot file will be stored, if the rotation is enabled, the path will be used as prefix and
+    # the timestamp of the start of the snapshot will be appended to the file name.
+    path: /var/lib/cachegrand/dump.rdb
+    # The interval between the snapshots, the allowed units are s, m, h, if not specified the default is seconds.
+    interval: 5m
+    # The number of keys that must be changed before a snapshot is taken, 0 means no limit
+    min_keys_changed: 1000
+    # The amount of data that must be changed before a snapshot is taken, the allowed units are b, k, m, g, if not
+    min_data_changed: 100mb
+    # Rotation settings, optional, if missing the snapshots rotation will be disabled
+    rotation:
+      # The max number of snapshots files to keep, minimum 2
+      max_files: 10
+
   backend: memory
   memory:
     # Limits:

src/clock.c

@@ -25,4 +25,39 @@ int64_t clock_realtime_coarse_get_resolution_ms() {
 
 int64_t clock_monotonic_coarse_get_resolution_ms() {
     return 1;
-}
+}
+
+char *clock_timespan_human_readable(
+        uint64_t timespan_ms,
+        char *buffer,
+        size_t buffer_length) {
+    assert(buffer_length >= CLOCK_TIMESPAN_MIN_LENGTH);
+
+    size_t offset = 0;
+    uint64_t days = timespan_ms / (1000 * 60 * 60 * 24);
+    uint64_t hours = (timespan_ms / (1000 * 60 * 60)) % 24;
+    uint64_t minutes = (timespan_ms / (1000 * 60)) % 60;
+    uint64_t seconds = (timespan_ms / 1000) % 60;
+
+    if (days > 0) {
+        offset += snprintf(buffer + offset, buffer_length - offset, "%lu days", days);
+    }
+
+    if (hours > 0) {
+        offset += snprintf(buffer + offset, buffer_length - offset, "%s%lu hours", offset > 0 ? " " : "", hours);
+    }
+
+    if (minutes > 0) {
+        offset += snprintf(buffer + offset, buffer_length - offset, "%s%lu minutes", offset > 0 ? " " : "", minutes);
+    }
+
+    if (seconds > 0) {
+        offset += snprintf(buffer + offset, buffer_length - offset, "%s%lu seconds", offset > 0 ? " " : "", seconds);
+    }
+
+    if (offset == 0) {
+        snprintf(buffer + offset, buffer_length - offset, "0 seconds");
+    }
+
+    return buffer;
+}

src/clock.h

@@ -22,12 +22,20 @@ extern "C" {
 #include "misc.h"
 #include "fatal.h"
 
+#define CLOCK_TIMESPAN_MIN_LENGTH (38)
+#define CLOCK_TIMESPAN_MAX_LENGTH (64)
+
 typedef struct timespec timespec_t;
 
 int64_t clock_monotonic_coarse_get_resolution_ms();
 
 int64_t clock_realtime_coarse_get_resolution_ms();
 
+char *clock_timespan_human_readable(
+        uint64_t timespan_ms,
+        char *buffer,
+        size_t buffer_length);
+
 static inline __attribute__((always_inline)) int64_t clock_timespec_to_int64_ms(
         timespec_t *timespec) {
     time_t s = timespec->tv_sec * 1000;

src/config.c

@@ -17,6 +17,7 @@
 #include <sys/statvfs.h>
 #include <sys/sysinfo.h>
 #include <cyaml/cyaml.h>
+#include <limits.h>
 
 #include "exttypes.h"
 #include "misc.h"
@@ -105,6 +106,98 @@ cyaml_err_t config_internal_cyaml_load(
             NULL);
 }
 
+bool config_parse_string_time(
+        char *string,
+        size_t string_len,
+        bool allow_negative,
+        bool allow_zero,
+        bool allow_time_suffix,
+        int64_t *return_value) {
+    bool result = false;
+    char *string_end;
+    size_t string_end_len;
+    int64_t string_value;
+
+    // Skip any leading space
+    while (isspace(string[0]) && string_len > 0) {
+        string++;
+        string_len--;
+    }
+
+    // Skip any trailing space
+    while (isspace(string[string_len - 1]) && string_len > 0) {
+        string_len--;
+    }
+
+    // If there are only spaces, skip them
+    if (string_len == 0) {
+        return false;
+    }
+
+    // As strtoll doesn't support non-null terminated strings, duplicate the string and null terminate it using strndup
+    string = strndup(string, string_len);
+    if (string == NULL) {
+        return false;
+    }
+
+    // Try to parse the number
+    string_value = strtoll(string, &string_end, 10);
+    string_end_len = strlen(string_end);
+
+    // Skip any leading space after parsing string_end
+    while (isspace(string_end[0]) && string_end_len > 0) {
+        string_end++;
+        string_end_len--;
+    }
+
+    // Check if the end of the string matches the beginning of the string, if's true then the string is not a number
+    if (string_end == string) {
+        goto end;
+    }
+
+    // Check if string_value is negative
+    if (!allow_negative && string_value < 0) {
+        goto end;
+    }
+
+    // Check if string_value is zero
+    if (!allow_zero && string_value == 0) {
+        goto end;
+    }
+
+    if (!allow_time_suffix && string_end_len == 0) {
+        *return_value = string_value;
+        result = true;
+        goto end;
+    }
+
+    // Check if the string is a size suffix
+    if (allow_time_suffix && string_end_len == 1) {
+        switch (string_end[0]) {
+            case 's':
+                string_value *= 1;
+                break;
+            case 'm':
+                string_value *= 60;
+                break;
+            case 'h':
+                string_value *= 60 * 60;
+                break;
+            case 'd':
+                string_value *= 60 * 60 * 24;
+                break;
+        }
+
+        *return_value = string_value;
+        result = true;
+        goto end;
+    }
+
+end:
+    free(string);
+    return result;
+}
+
 bool config_parse_string_absolute_or_percent(
         char *string,
         size_t string_len,
@@ -341,6 +434,38 @@ bool config_validate_after_load_database_backend_memory(
     return return_result;
 }
 
+bool config_validate_after_load_database_snapshots(
+        config_t* config) {
+    bool return_result = true;
+
+    if (!config->database->snapshots) {
+        return return_result;
+    }
+
+    // Ensure that the path is not longer than PATH_MAX
+    if (strlen(config->database->snapshots->path) > PATH_MAX) {
+        LOG_E(TAG, "The path for the snapshots is too long");
+        return_result = false;
+    }
+
+    // Check that the maximum allowed interval is 7 days
+    if (config->database->snapshots->interval_ms > 7 * 24 * 60 * 60 * 1000) {
+        LOG_E(TAG, "The maximum allowed interval for the snapshots is <7> days");
+        return_result = false;
+    }
+
+    // Ensure that if rotation is enabled, the maximum number of max_files is equal or greater than 0 and smaller than
+    // uint16_t
+    if (config->database->snapshots->rotation && (
+            config->database->snapshots->rotation->max_files < 2 ||
+            config->database->snapshots->rotation->max_files > 65535)) {
+        LOG_E(TAG, "The maximum number of files for the snapshots rotation must be between <2> and <65535>");
+        return_result = false;
+    }
+
+    return return_result;
+}
+
 bool config_validate_after_load_database_limits(
         config_t* config) {
     bool return_result = true;
@@ -529,6 +654,7 @@ bool config_validate_after_load(
     if (config_validate_after_load_cpus(config) == false
         || config_validate_after_load_database_backend_file(config) == false
         || config_validate_after_load_database_backend_memory(config) == false
+        || config_validate_after_load_database_snapshots(config) == false
         || config_validate_after_load_database_limits(config) == false
         || config_validate_after_load_database_keys_eviction(config) == false
         || config_validate_after_load_modules(config) == false
@@ -779,6 +905,47 @@ bool config_process_string_values(
         config_t *config) {
     config_parse_string_absolute_or_percent_return_value_t return_value_type;
 
+    // Check if snapshots are enabled
+    if (config->database->snapshots) {
+        config->database->snapshots->min_data_changed = 0;
+
+        bool result = config_parse_string_time(
+                config->database->snapshots->interval_str,
+                strlen(config->database->snapshots->interval_str),
+                false,
+                false,
+                true,
+                &config->database->snapshots->interval_ms);
+
+        // The returned time is in seconds, so multiply by 1000 to convert to ms
+        config->database->snapshots->interval_ms *= 1000;
+
+        if (!result) {
+            LOG_E(TAG, "Failed to parse the snapshot interval");
+            config_free(config);
+            return NULL;
+        }
+
+        if (config->database->snapshots->min_data_changed_str) {
+            result = config_parse_string_absolute_or_percent(
+                    config->database->snapshots->min_data_changed_str,
+                    strlen(config->database->snapshots->min_data_changed_str),
+                    false,
+                    false,
+                    false,
+                    true,
+                    true,
+                    &config->database->snapshots->min_data_changed,
+                    &return_value_type);
+
+            if (!result) {
+                LOG_E(TAG, "Failed to parse the snapshot minimum data changed");
+                config_free(config);
+                return NULL;
+            }
+        }
+    }
+
     // Check if the memory backend for the database is defined in the config, if yes process the hard and soft memory
     // limits, the latter is optional
     if (config->database->memory) {

src/config.h

@@ -239,10 +239,27 @@ struct config_database_limits {
 };
 typedef struct config_database_limits config_database_limits_t;
 
+struct config_database_snapshots_rotation {
+    int64_t max_files;
+};
+typedef struct config_database_snapshots_rotation config_database_snapshots_rotation_t;
+
+struct config_database_snapshots {
+    char *path;
+    char *interval_str;
+    char *min_data_changed_str;
+    int64_t interval_ms;
+    int64_t min_keys_changed;
+    int64_t min_data_changed;
+    config_database_snapshots_rotation_t *rotation;
+};
+typedef struct config_database_snapshots config_database_snapshots_t;
+
 struct config_database {
     config_database_limits_t *limits;
     config_database_keys_eviction_t *keys_eviction;
     config_database_backend_t backend;
+    config_database_snapshots_t *snapshots;
     config_database_file_t *file;
     config_database_memory_t *memory;
 };
@@ -277,6 +294,14 @@ enum config_cpus_validate_error {
 };
 typedef enum config_cpus_validate_error config_cpus_validate_error_t;
 
+bool config_parse_string_time(
+        char *string,
+        size_t string_len,
+        bool allow_negative,
+        bool allow_zero,
+        bool allow_time_suffix,
+        int64_t *return_value);
+
 bool config_parse_string_absolute_or_percent(
         char *string,
         size_t string_len,
@@ -297,6 +322,9 @@ void config_free(
 bool config_validate_after_load_cpus(
         config_t* config);
 
+bool config_validate_after_load_database_snapshots(
+        config_t* config);
+
 bool config_validate_after_load_database_backend_file(
         config_t* config);
 

src/config_cyaml_schema.c

@@ -302,11 +302,42 @@ const cyaml_schema_field_t config_database_limits_schema[] = {
         CYAML_FIELD_END
 };
 
+// Schema for config -> database -> snapshots -> rotation
+const cyaml_schema_field_t config_database_snapshots_rotation_schema[] = {
+        CYAML_FIELD_UINT(
+                "max_files", CYAML_FLAG_DEFAULT | CYAML_FLAG_OPTIONAL,
+                config_database_snapshots_rotation_t, max_files),
+        CYAML_FIELD_END
+};
+
+// Schema for config -> database -> snapshots
+const cyaml_schema_field_t config_database_snapshots_schema[] = {
+        CYAML_FIELD_STRING_PTR(
+                "path", CYAML_FLAG_DEFAULT,
+                config_database_snapshots_t, path, 0, CYAML_UNLIMITED),
+        CYAML_FIELD_STRING_PTR(
+                "interval", CYAML_FLAG_DEFAULT,
+                config_database_snapshots_t, interval_str, 0, 20),
+        CYAML_FIELD_UINT(
+                "min_keys_changed", CYAML_FLAG_DEFAULT | CYAML_FLAG_OPTIONAL,
+                config_database_snapshots_t, min_keys_changed),
+        CYAML_FIELD_STRING_PTR(
+                "min_data_changed", CYAML_FLAG_DEFAULT | CYAML_FLAG_OPTIONAL,
+                config_database_snapshots_t, min_data_changed_str, 0, 20),
+        CYAML_FIELD_MAPPING_PTR(
+                "rotation", CYAML_FLAG_POINTER | CYAML_FLAG_OPTIONAL,
+                config_database_snapshots_t, rotation, config_database_snapshots_rotation_schema),
+        CYAML_FIELD_END
+};
+
 // Schema for config -> database
 const cyaml_schema_field_t config_database_schema[] = {
         CYAML_FIELD_MAPPING_PTR(
                 "limits", CYAML_FLAG_POINTER,
                 config_database_t, limits, config_database_limits_schema),
+        CYAML_FIELD_MAPPING_PTR(
+                "snapshots", CYAML_FLAG_POINTER | CYAML_FLAG_OPTIONAL,
+                config_database_t, snapshots, config_database_snapshots_schema),
         CYAML_FIELD_MAPPING_PTR(
                 "keys_eviction", CYAML_FLAG_POINTER | CYAML_FLAG_OPTIONAL,
                 config_database_t, keys_eviction, config_database_keys_eviction_schema),