Module Configurations (#10285)

This feature adds the ability to add four different types (Bool, Numeric,
String, Enum) of configurations to a module to be accessed via the redis
config file, and the CONFIG command.

**Configuration Names**:

We impose a restriction that a module configuration always starts with the
module name and contains a '.' followed by the config name. If a module passes
"config1" as the name to a register function, it will be registered as MODULENAME.config1.

**Configuration Persistence**:

Module Configurations exist only as long as a module is loaded. If a module is
unloaded, the configurations are removed.
There is now also a minimal core API for removal of standardConfig objects
from configs by name.

**Get and Set Callbacks**:

Storage of config values is owned by the module that registers them, and provides
callbacks for Redis to access and manipulate the values.
This is exposed through a GET and SET callback.

The get callback returns a typed value of the config to redis. The callback takes
the name of the configuration, and also a privdata pointer. Note that these only
take the CONFIGNAME portion of the config, not the entire MODULENAME.CONFIGNAME.

```
 typedef RedisModuleString * (*RedisModuleConfigGetStringFunc)(const char *name, void *privdata);
 typedef long long (*RedisModuleConfigGetNumericFunc)(const char *name, void *privdata);
 typedef int (*RedisModuleConfigGetBoolFunc)(const char *name, void *privdata);
 typedef int (*RedisModuleConfigGetEnumFunc)(const char *name, void *privdata);
```

Configs must also must specify a set callback, i.e. what to do on a CONFIG SET XYZ 123
or when loading configurations from cli/.conf file matching these typedefs. *name* is
again just the CONFIGNAME portion, *val* is the parsed value from the core,
*privdata* is the registration time privdata pointer, and *err* is for providing errors to a client.

```
typedef int (*RedisModuleConfigSetStringFunc)(const char *name, RedisModuleString *val, void *privdata, RedisModuleString **err);
typedef int (*RedisModuleConfigSetNumericFunc)(const char *name, long long val, void *privdata, RedisModuleString **err);
typedef int (*RedisModuleConfigSetBoolFunc)(const char *name, int val, void *privdata, RedisModuleString **err);
typedef int (*RedisModuleConfigSetEnumFunc)(const char *name, int val, void *privdata, RedisModuleString **err);
```

Modules can also specify an optional apply callback that will be called after
value(s) have been set via CONFIG SET:

```
typedef int (*RedisModuleConfigApplyFunc)(RedisModuleCtx *ctx, void *privdata, RedisModuleString **err);
```

**Flags:**
We expose 7 new flags to the module, which are used as part of the config registration.

```
#define REDISMODULE_CONFIG_MODIFIABLE 0 /* This is the default for a module config. */
#define REDISMODULE_CONFIG_IMMUTABLE (1ULL<<0) /* Can this value only be set at startup? */
#define REDISMODULE_CONFIG_SENSITIVE (1ULL<<1) /* Does this value contain sensitive information */
#define REDISMODULE_CONFIG_HIDDEN (1ULL<<4) /* This config is hidden in `config get <pattern>` (used for tests/debugging) */
#define REDISMODULE_CONFIG_PROTECTED (1ULL<<5) /* Becomes immutable if enable-protected-configs is enabled. */
#define REDISMODULE_CONFIG_DENY_LOADING (1ULL<<6) /* This config is forbidden during loading. */
/* Numeric Specific Configs */
#define REDISMODULE_CONFIG_MEMORY (1ULL<<7) /* Indicates if this value can be set as a memory value */
```

**Module Registration APIs**:

```
int (*RedisModule_RegisterBoolConfig)(RedisModuleCtx *ctx, char *name, int default_val, unsigned int flags, RedisModuleConfigGetBoolFunc getfn, RedisModuleConfigSetBoolFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata);
int (*RedisModule_RegisterNumericConfig)(RedisModuleCtx *ctx, const char *name, long long default_val, unsigned int flags, long long min, long long max, RedisModuleConfigGetNumericFunc getfn, RedisModuleConfigSetNumericFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata);
int (*RedisModule_RegisterStringConfig)(RedisModuleCtx *ctx, const char *name, const char *default_val, unsigned int flags, RedisModuleConfigGetStringFunc getfn, RedisModuleConfigSetStringFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata);
int (*RedisModule_RegisterEnumConfig)(RedisModuleCtx *ctx, const char *name, int default_val, unsigned int flags, const char **enum_values, const int *int_values, int num_enum_vals, RedisModuleConfigGetEnumFunc getfn, RedisModuleConfigSetEnumFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata);
int (*RedisModule_LoadConfigs)(RedisModuleCtx *ctx);
```

The module name will be auto appended along with a "." to the front of the name of the config.

**What RM_Register[...]Config does**:

A RedisModule struct now keeps a list of ModuleConfig objects which look like:
```
typedef struct ModuleConfig {
    sds name; /* Name of config without the module name appended to the front */
    void *privdata; /* Optional data passed into the module config callbacks */
    union get_fn { /* The get callback specificed by the module */
        RedisModuleConfigGetStringFunc get_string;
        RedisModuleConfigGetNumericFunc get_numeric;
        RedisModuleConfigGetBoolFunc get_bool;
        RedisModuleConfigGetEnumFunc get_enum;
    } get_fn;
    union set_fn { /* The set callback specified by the module */
        RedisModuleConfigSetStringFunc set_string;
        RedisModuleConfigSetNumericFunc set_numeric;
        RedisModuleConfigSetBoolFunc set_bool;
        RedisModuleConfigSetEnumFunc set_enum;
    } set_fn;
    RedisModuleConfigApplyFunc apply_fn;
    RedisModule *module;
} ModuleConfig;
```
It also registers a standardConfig in the configs array, with a pointer to the
ModuleConfig object associated with it.

**What happens on a CONFIG GET/SET MODULENAME.MODULECONFIG:**

For CONFIG SET, we do the same parsing as is done in config.c and pass that
as the argument to the module set callback. For CONFIG GET, we call the
module get callback and return that value to config.c to return to a client.

**CONFIG REWRITE**:

Starting up a server with module configurations in a .conf file but no module load
directive will fail. The flip side is also true, specifying a module load and a bunch
of module configurations will load those configurations in using the module defined
set callbacks on a RM_LoadConfigs call. Configs being rewritten works the same
way as it does for standard configs, as the module has the ability to specify a
default value. If a module is unloaded with configurations specified in the .conf file
those configurations will be commented out from the .conf file on the next config rewrite.

**RM_LoadConfigs:**

`RedisModule_LoadConfigs(RedisModuleCtx *ctx);`

This last API is used to make configs available within the onLoad() after they have
been registered. The expected usage is that a module will register all of its configs,
then call LoadConfigs to trigger all of the set callbacks, and then can error out if any
of them were malformed. LoadConfigs will attempt to set all configs registered to
either a .conf file argument/loadex argument or their default value if an argument is
not specified. **LoadConfigs is a required function if configs are registered.
** Also note that LoadConfigs **does not** call the apply callbacks, but a module
can do that directly after the LoadConfigs call.

**New Command: MODULE LOADEX [CONFIG NAME VALUE] [ARGS ...]:**

This command provides the ability to provide startup context information to a module.
LOADEX stands for "load extended" similar to GETEX. Note that provided config
names need the full MODULENAME.MODULECONFIG name. Any additional
arguments a module might want are intended to be specified after ARGS.
Everything after ARGS is passed to onLoad as RedisModuleString **argv.

Co-authored-by: Madelyn Olson <madelyneolson@gmail.com>
Co-authored-by: Madelyn Olson <matolson@amazon.com>
Co-authored-by: sundb <sundbcn@gmail.com>
Co-authored-by: Madelyn Olson <34459052+madolson@users.noreply.github.com>
Co-authored-by: Oran Agra <oran@redislabs.com>
Co-authored-by: Yossi Gottlieb <yossigo@gmail.com>

8 files changed, 1117 insertions, 232 deletions

diff --git a/src/commands.c b/src/commands.c
index d7a62a71c..0730a71c1 100644
--- a/src/commands.c
+++ b/src/commands.c
@@ -4776,6 +4776,35 @@ struct redisCommandArg MODULE_LOAD_Args[] = {
 {0}
 };
 
+/********** MODULE LOADEX ********************/
+
+/* MODULE LOADEX history */
+#define MODULE_LOADEX_History NULL
+
+/* MODULE LOADEX tips */
+#define MODULE_LOADEX_tips NULL
+
+/* MODULE LOADEX configs argument table */
+struct redisCommandArg MODULE_LOADEX_configs_Subargs[] = {
+{"name",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_NONE},
+{"value",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_NONE},
+{0}
+};
+
+/* MODULE LOADEX args argument table */
+struct redisCommandArg MODULE_LOADEX_args_Subargs[] = {
+{"arg",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_NONE},
+{0}
+};
+
+/* MODULE LOADEX argument table */
+struct redisCommandArg MODULE_LOADEX_Args[] = {
+{"path",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_NONE},
+{"configs",ARG_TYPE_BLOCK,-1,"CONFIG",NULL,NULL,CMD_ARG_OPTIONAL|CMD_ARG_MULTIPLE,.subargs=MODULE_LOADEX_configs_Subargs},
+{"args",ARG_TYPE_BLOCK,-1,"ARGS",NULL,NULL,CMD_ARG_OPTIONAL|CMD_ARG_MULTIPLE,.subargs=MODULE_LOADEX_args_Subargs},
+{0}
+};
+
 /********** MODULE UNLOAD ********************/
 
 /* MODULE UNLOAD history */
@@ -4795,6 +4824,7 @@ struct redisCommand MODULE_Subcommands[] = {
 {"help","Show helpful text about the different subcommands","O(1)","5.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SERVER,MODULE_HELP_History,MODULE_HELP_tips,moduleCommand,2,CMD_LOADING|CMD_STALE,0},
 {"list","List all modules loaded by the server","O(N) where N is the number of loaded modules.","4.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SERVER,MODULE_LIST_History,MODULE_LIST_tips,moduleCommand,2,CMD_ADMIN|CMD_NOSCRIPT,0},
 {"load","Load a module","O(1)","4.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SERVER,MODULE_LOAD_History,MODULE_LOAD_tips,moduleCommand,-3,CMD_NO_ASYNC_LOADING|CMD_ADMIN|CMD_NOSCRIPT|CMD_PROTECTED,0,.args=MODULE_LOAD_Args},
+{"loadex","Load a module with extended parameters","O(1)","7.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SERVER,MODULE_LOADEX_History,MODULE_LOADEX_tips,moduleCommand,-3,CMD_NO_ASYNC_LOADING|CMD_ADMIN|CMD_NOSCRIPT|CMD_PROTECTED,0,.args=MODULE_LOADEX_Args},
 {"unload","Unload a module","O(1)","4.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SERVER,MODULE_UNLOAD_History,MODULE_UNLOAD_tips,moduleCommand,3,CMD_NO_ASYNC_LOADING|CMD_ADMIN|CMD_NOSCRIPT|CMD_PROTECTED,0,.args=MODULE_UNLOAD_Args},
 {0}
 };
diff --git a/src/commands/module-loadex.json b/src/commands/module-loadex.json
new file mode 100644
index 000000000..e772cbfe4
--- /dev/null
+++ b/src/commands/module-loadex.json
@@ -0,0 +1,53 @@
+{
+    "LOADEX": {
+        "summary": "Load a module with extended parameters",
+        "complexity": "O(1)",
+        "group": "server",
+        "since": "7.0.0",
+        "arity": -3,
+        "container": "MODULE",
+        "function": "moduleCommand",
+        "command_flags": [
+            "NO_ASYNC_LOADING",
+            "ADMIN",
+            "NOSCRIPT",
+            "PROTECTED"
+        ],
+        "arguments": [
+            {
+                "name": "path",
+                "type": "string"
+            },
+            {
+                "name": "configs",
+                "token": "CONFIG",
+                "type": "block",
+                "multiple": true,
+                "optional": true,
+                "arguments": [
+                    {
+                        "name": "name",
+                        "type": "string"
+                    },
+                    {
+                        "name": "value",
+                        "type": "string"
+                    }
+                ]
+            },
+            {
+                "name": "args",
+                "token": "ARGS",
+                "type": "block",
+                "multiple": true,
+                "optional": true,
+                "arguments": [
+                    {
+                        "name": "arg",
+                        "type": "string"
+                    }
+                ]
+            }
+        ]
+    }
+}
\ No newline at end of file
diff --git a/src/config.c b/src/config.c
index e36315743..3b5d4d349 100644
--- a/src/config.c
+++ b/src/config.c
@@ -40,11 +40,6 @@
  * Config file name-value maps.
  *----------------------------------------------------------------------------*/
 
-typedef struct configEnum {
-    const char *name;
-    const int val;
-} configEnum;
-
 typedef struct deprecatedConfig {
     const char *name;
     const int argc_min;
@@ -168,7 +163,7 @@ int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT] = { 0, 200, 800 };
  * rewrite. */
 typedef struct boolConfigData {
     int *config; /* The pointer to the server config this value is stored in */
-    const int default_value; /* The default value of the config on rewrite */
+    int default_value; /* The default value of the config on rewrite */
     int (*is_valid_fn)(int val, const char **err); /* Optional function to check validity of new value (generic doc above) */
 } boolConfigData;
 
@@ -182,7 +177,7 @@ typedef struct stringConfigData {
 
 typedef struct sdsConfigData {
     sds *config; /* Pointer to the server config this value is stored in. */
-    const char *default_value; /* Default value of the config on rewrite. */
+    char *default_value; /* Default value of the config on rewrite. */
     int (*is_valid_fn)(sds val, const char **err); /* Optional function to check validity of new value (generic doc above) */
     int convert_empty_to_null; /* Boolean indicating if empty SDS strings should
                                   be stored as a NULL value. */
@@ -191,7 +186,7 @@ typedef struct sdsConfigData {
 typedef struct enumConfigData {
     int *config; /* The pointer to the server config this value is stored in */
     configEnum *enum_value; /* The underlying enum type this data represents */
-    const int default_value; /* The default value of the config on rewrite */
+    int default_value; /* The default value of the config on rewrite */
     int (*is_valid_fn)(int val, const char **err); /* Optional function to check validity of new value (generic doc above) */
 } enumConfigData;
 
@@ -208,11 +203,6 @@ typedef enum numericType {
     NUMERIC_TYPE_TIME_T,
 } numericType;
 
-#define INTEGER_CONFIG 0 /* No flags means a simple integer configuration */
-#define MEMORY_CONFIG (1<<0) /* Indicates if this value can be loaded as a memory value */
-#define PERCENT_CONFIG (1<<1) /* Indicates if this value can be loaded as a percent (and stored as a negative int) */
-#define OCTAL_CONFIG (1<<2) /* This value uses octal representation */
-
 typedef struct numericConfigData {
     union {
         int *i;
@@ -230,7 +220,7 @@ typedef struct numericConfigData {
     numericType numeric_type; /* An enum indicating the type of this value */
     long long lower_bound; /* The lower bound of this numeric value */
     long long upper_bound; /* The upper bound of this numeric value */
-    const long long default_value; /* The default value of the config on rewrite */
+    long long default_value; /* The default value of the config on rewrite */
     int (*is_valid_fn)(long long val, const char **err); /* Optional function to check validity of new value (generic doc above) */
 } numericConfigData;
 
@@ -242,42 +232,35 @@ typedef union typeData {
     numericConfigData numeric;
 } typeData;
 
+typedef struct standardConfig standardConfig;
+
 typedef int (*apply_fn)(const char **err);
 typedef struct typeInterface {
     /* Called on server start, to init the server with default value */
-    void (*init)(typeData data);
+    void (*init)(standardConfig *config);
     /* Called on server startup and CONFIG SET, returns 1 on success,
      * 2 meaning no actual change done, 0 on error and can set a verbose err
      * string */
-    int (*set)(typeData data, sds *argv, int argc, const char **err);
+    int (*set)(standardConfig *config, sds *argv, int argc, const char **err);
     /* Optional: called after `set()` to apply the config change. Used only in
      * the context of CONFIG SET. Returns 1 on success, 0 on failure.
      * Optionally set err to a static error string. */
     apply_fn apply;
     /* Called on CONFIG GET, returns sds to be used in reply */
-    sds (*get)(typeData data);
+    sds (*get)(standardConfig *config);
     /* Called on CONFIG REWRITE, required to rewrite the config state */
-    void (*rewrite)(typeData data, const char *name, struct rewriteConfigState *state);
+    void (*rewrite)(standardConfig *config, const char *name, struct rewriteConfigState *state);
 } typeInterface;
 
-typedef struct standardConfig {
+struct standardConfig {
     const char *name; /* The user visible name of this config */
     const char *alias; /* An alias that can also be used for this config */
     unsigned int flags; /* Flags for this specific config */
     typeInterface interface; /* The function pointers that define the type interface */
     typeData data; /* The type specific data exposed used by the interface */
-} standardConfig;
-
-#define MODIFIABLE_CONFIG 0 /* This is the implied default for a standard 
-                             * config, which is mutable. */
-#define IMMUTABLE_CONFIG (1ULL<<0) /* Can this value only be set at startup? */
-#define SENSITIVE_CONFIG (1ULL<<1) /* Does this value contain sensitive information */
-#define DEBUG_CONFIG (1ULL<<2) /* Values that are useful for debugging. */
-#define MULTI_ARG_CONFIG (1ULL<<3) /* This config receives multiple arguments. */
-#define HIDDEN_CONFIG (1ULL<<4) /* This config is hidden in `config get <pattern>` (used for tests/debugging) */
-#define PROTECTED_CONFIG (1ULL<<5) /* Becomes immutable if enable-protected-configs is enabled. */
-#define DENY_LOADING_CONFIG (1ULL<<6) /* This config is forbidden during loading. */
-#define ALIAS_CONFIG (1ULL<<7) /* For configs with multiple names, this flag is set on the alias. */
+    configType type; /* The type of config this is. */
+    void *privdata; /* privdata for this config, for module configs this is a ModuleConfig struct */
+};
 
 dict *configs = NULL; /* Runtime config values */
 
@@ -469,7 +452,7 @@ void loadServerConfigFromString(char *config) {
                 goto loaderr;
             }
             /* Set config using all arguments that follows */
-            if (!config->interface.set(config->data, &argv[1], argc-1, &err)) {
+            if (!config->interface.set(config, &argv[1], argc-1, &err)) {
                 goto loaderr;
             }
 
@@ -530,6 +513,13 @@ void loadServerConfigFromString(char *config) {
             }
         } else if (!strcasecmp(argv[0],"loadmodule") && argc >= 2) {
             queueLoadModule(argv[1],&argv[2],argc-2);
+        } else if (strchr(argv[0], '.')) {
+            if (argc != 2) {
+                err = "Module config specified without value";
+                goto loaderr;
+            }
+            sds name = sdsdup(argv[0]);
+            if (!dictReplace(server.module_configs_queue, name, sdsdup(argv[1]))) sdsfree(name);
         } else if (!strcasecmp(argv[0],"sentinel")) {
             /* argc == 1 is handled by main() as we need to enter the sentinel
              * mode ASAP. */
@@ -678,12 +668,45 @@ static int performInterfaceSet(standardConfig *config, sds value, const char **e
     }
 
     /* Set the config */
-    res = config->interface.set(config->data, argv, argc, errstr);
+    res = config->interface.set(config, argv, argc, errstr);
     if (config->flags & MULTI_ARG_CONFIG) sdsfreesplitres(argv, argc);
     return res;
 }
 
-static void restoreBackupConfig(standardConfig **set_configs, sds *old_values, int count, apply_fn *apply_fns) {
+/* Find the config by name and attempt to set it to value. */
+int performModuleConfigSetFromName(sds name, sds value, const char **err) {
+    standardConfig *config = lookupConfig(name);
+    if (!config || !(config->flags & MODULE_CONFIG)) {
+        *err = "Config name not found";
+        return 0;
+    }
+    return performInterfaceSet(config, value, err);
+}
+
+/* Find config by name and attempt to set it to its default value. */
+int performModuleConfigSetDefaultFromName(sds name, const char **err) {
+    standardConfig *config = lookupConfig(name);
+    serverAssert(config);
+    if (!(config->flags & MODULE_CONFIG)) {
+        *err = "Config name not found";
+        return 0;
+    }
+    switch (config->type) {
+        case BOOL_CONFIG:
+            return setModuleBoolConfig(config->privdata, config->data.yesno.default_value, err);
+        case SDS_CONFIG:
+            return setModuleStringConfig(config->privdata, config->data.sds.default_value, err);
+        case NUMERIC_CONFIG:
+            return setModuleNumericConfig(config->privdata, config->data.numeric.default_value, err);
+        case ENUM_CONFIG:
+            return setModuleEnumConfig(config->privdata, config->data.enumd.default_value, err);
+        default:
+            serverPanic("Config type of module config is not allowed.");
+    }
+    return 0;
+}
+
+static void restoreBackupConfig(standardConfig **set_configs, sds *old_values, int count, apply_fn *apply_fns, list *module_configs) {
     int i;
     const char *errstr = "unknown error";
     /* Set all backup values */
@@ -699,6 +722,10 @@ static void restoreBackupConfig(standardConfig **set_configs, sds *old_values, i
                 serverLog(LL_WARNING, "Failed applying restored failed CONFIG SET command: %s", errstr);
         }
     }
+    if (module_configs) {
+        if (!moduleConfigApplyConfig(module_configs, &errstr, NULL))
+            serverLog(LL_WARNING, "Failed applying restored failed CONFIG SET command: %s", errstr);
+    }
 }
 
 /*-----------------------------------------------------------------------------
@@ -710,6 +737,7 @@ void configSetCommand(client *c) {
     const char *invalid_arg_name = NULL;
     const char *err_arg_name = NULL;
     standardConfig **set_configs; /* TODO: make this a dict for better performance */
+    list *module_configs_apply;
     const char **config_names;
     sds *new_values;
     sds *old_values = NULL;
@@ -725,6 +753,7 @@ void configSetCommand(client *c) {
     }
     config_count = (c->argc - 2) / 2;
 
+    module_configs_apply = listCreate();
     set_configs = zcalloc(sizeof(standardConfig*)*config_count);
     config_names = zcalloc(sizeof(char*)*config_count);
     new_values = zmalloc(sizeof(sds*)*config_count);
@@ -790,18 +819,20 @@ void configSetCommand(client *c) {
 
     /* Backup old values before setting new ones */
     for (i = 0; i < config_count; i++)
-        old_values[i] = set_configs[i]->interface.get(set_configs[i]->data);
+        old_values[i] = set_configs[i]->interface.get(set_configs[i]);
 
     /* Set all new values (don't apply yet) */
     for (i = 0; i < config_count; i++) {
         int res = performInterfaceSet(set_configs[i], new_values[i], &errstr);
         if (!res) {
-            restoreBackupConfig(set_configs, old_values, i+1, NULL);
+            restoreBackupConfig(set_configs, old_values, i+1, NULL, NULL);
             err_arg_name = set_configs[i]->name;
             goto err;
         } else if (res == 1) {
             /* A new value was set, if this config has an apply function then store it for execution later */
-            if (set_configs[i]->interface.apply) {
+            if (set_configs[i]->flags & MODULE_CONFIG) {
+                addModuleConfigApply(module_configs_apply, set_configs[i]->privdata);
+            } else if (set_configs[i]->interface.apply) {
                 /* Check if this apply function is already stored */
                 int exists = 0;
                 for (j = 0; apply_fns[j] != NULL && j <= i; j++) {
@@ -823,11 +854,18 @@ void configSetCommand(client *c) {
     for (i = 0; i < config_count && apply_fns[i] != NULL; i++) {
         if (!apply_fns[i](&errstr)) {
             serverLog(LL_WARNING, "Failed applying new configuration. Possibly related to new %s setting. Restoring previous settings.", set_configs[config_map_fns[i]]->name);
-            restoreBackupConfig(set_configs, old_values, config_count, apply_fns);
+            restoreBackupConfig(set_configs, old_values, config_count, apply_fns, NULL);
             err_arg_name = set_configs[config_map_fns[i]]->name;
             goto err;
         }
     }
+    /* Apply all module configs that were set. */
+    if (!moduleConfigApplyConfig(module_configs_apply, &errstr, &err_arg_name)) {
+        serverLogRaw(LL_WARNING, "Failed applying new module configuration. Restoring previous settings.");
+        restoreBackupConfig(set_configs, old_values, config_count, apply_fns, module_configs_apply);
+        goto err;
+    }
+
     RedisModuleConfigChangeV1 cc = {.num_changes = config_count, .config_names = config_names};
     moduleFireServerEvent(REDISMODULE_EVENT_CONFIG, REDISMODULE_SUBEVENT_CONFIG_CHANGE, &cc);
     addReply(c,shared.ok);
@@ -853,6 +891,7 @@ end:
     zfree(old_values);
     zfree(apply_fns);
     zfree(config_map_fns);
+    listRelease(module_configs_apply);
 }
 
 /*-----------------------------------------------------------------------------
@@ -901,7 +940,7 @@ void configGetCommand(client *c) {
     while ((de = dictNext(di)) != NULL) {
         standardConfig *config = (standardConfig *) dictGetVal(de);
         addReplyBulkCString(c, de->key);
-        addReplyBulkSds(c, config->interface.get(config->data));
+        addReplyBulkSds(c, config->interface.get(config));
     }
     dictReleaseIterator(di);
     dictRelease(matches);
@@ -1050,12 +1089,13 @@ struct rewriteConfigState *rewriteConfigReadOldFile(char *path) {
 
         /* Not a comment, split into arguments. */
         argv = sdssplitargs(line,&argc);
-        if (argv == NULL) {
+        if (argv == NULL || (!server.sentinel_mode && !lookupConfig(argv[0]))) {
             /* Apparently the line is unparsable for some reason, for
-             * instance it may have unbalanced quotes. Load it as a
-             * comment. */
+             * instance it may have unbalanced quotes, or may contain a
+             * config that doesn't exist anymore. Load it as a comment. */
             sds aux = sdsnew("# ??? ");
             aux = sdscatsds(aux,line);
+            if (argv) sdsfreesplitres(argv, argc);
             sdsfree(line);
             rewriteConfigAppendLine(state,aux);
             continue;
@@ -1267,8 +1307,8 @@ void rewriteConfigEnumOption(struct rewriteConfigState *state, const char *optio
 }
 
 /* Rewrite the save option. */
-void rewriteConfigSaveOption(typeData data, const char *name, struct rewriteConfigState *state) {
-    UNUSED(data);
+void rewriteConfigSaveOption(standardConfig *config, const char *name, struct rewriteConfigState *state) {
+    UNUSED(config);
     int j;
     sds line;
 
@@ -1328,8 +1368,8 @@ void rewriteConfigUserOption(struct rewriteConfigState *state) {
 }
 
 /* Rewrite the dir option, always using absolute paths.*/
-void rewriteConfigDirOption(typeData data, const char *name, struct rewriteConfigState *state) {
-    UNUSED(data);
+void rewriteConfigDirOption(standardConfig *config, const char *name, struct rewriteConfigState *state) {
+    UNUSED(config);
     char cwd[1024];
 
     if (getcwd(cwd,sizeof(cwd)) == NULL) {
@@ -1340,8 +1380,8 @@ void rewriteConfigDirOption(typeData data, const char *name, struct rewriteConfi
 }
 
 /* Rewrite the slaveof option. */
-void rewriteConfigReplicaOfOption(typeData data, const char *name, struct rewriteConfigState *state) {
-    UNUSED(data);
+void rewriteConfigReplicaOfOption(standardConfig *config, const char *name, struct rewriteConfigState *state) {
+    UNUSED(config);
     sds line;
 
     /* If this is a master, we want all the slaveof config options
@@ -1357,8 +1397,8 @@ void rewriteConfigReplicaOfOption(typeData data, const char *name, struct rewrit
 }
 
 /* Rewrite the notify-keyspace-events option. */
-void rewriteConfigNotifyKeyspaceEventsOption(typeData data, const char *name, struct rewriteConfigState *state) {
-    UNUSED(data);
+void rewriteConfigNotifyKeyspaceEventsOption(standardConfig *config, const char *name, struct rewriteConfigState *state) {
+    UNUSED(config);
     int force = server.notify_keyspace_events != 0;
     sds line, flags;
 
@@ -1371,8 +1411,8 @@ void rewriteConfigNotifyKeyspaceEventsOption(typeData data, const char *name, st
 }
 
 /* Rewrite the client-output-buffer-limit option. */
-void rewriteConfigClientOutputBufferLimitOption(typeData data, const char *name, struct rewriteConfigState *state) {
-    UNUSED(data);
+void rewriteConfigClientOutputBufferLimitOption(standardConfig *config, const char *name, struct rewriteConfigState *state) {
+    UNUSED(config);
     int j;
     for (j = 0; j < CLIENT_TYPE_OBUF_COUNT; j++) {
         int force = (server.client_obuf_limits[j].hard_limit_bytes !=
@@ -1399,8 +1439,8 @@ void rewriteConfigClientOutputBufferLimitOption(typeData data, const char *name,
 }
 
 /* Rewrite the oom-score-adj-values option. */
-void rewriteConfigOOMScoreAdjValuesOption(typeData data, const char *name, struct rewriteConfigState *state) {
-    UNUSED(data);
+void rewriteConfigOOMScoreAdjValuesOption(standardConfig *config, const char *name, struct rewriteConfigState *state) {
+    UNUSED(config);
     int force = 0;
     int j;
     sds line;
@@ -1419,8 +1459,8 @@ void rewriteConfigOOMScoreAdjValuesOption(typeData data, const char *name, struc
 }
 
 /* Rewrite the bind option. */
-void rewriteConfigBindOption(typeData data, const char *name, struct rewriteConfigState *state) {
-    UNUSED(data);
+void rewriteConfigBindOption(standardConfig *config, const char *name, struct rewriteConfigState *state) {
+    UNUSED(config);
     int force = 1;
     sds line, addresses;
     int is_default = 0;
@@ -1545,7 +1585,7 @@ sds getConfigDebugInfo() {
     while ((de = dictNext(di)) != NULL) {
         standardConfig *config = dictGetVal(de);
         if (!(config->flags & DEBUG_CONFIG)) continue;
-        config->interface.rewrite(config->data, config->name, state);
+        config->interface.rewrite(config, config->name, state);
     }
     dictReleaseIterator(di);
     sds info = rewriteConfigGetContentFromState(state);
@@ -1641,7 +1681,7 @@ int rewriteConfig(char *path, int force_write) {
         standardConfig *config = dictGetVal(de);
         /* Only rewrite the primary names */
         if (config->flags & ALIAS_CONFIG) continue;
-        if (config->interface.rewrite) config->interface.rewrite(config->data, de->key, state);
+        if (config->interface.rewrite) config->interface.rewrite(config, de->key, state);
     }
     dictReleaseIterator(di);
 
@@ -1698,38 +1738,46 @@ static char loadbuf[LOADBUF_SIZE];
  */
 
 /* Bool Configs */
-static void boolConfigInit(typeData data) {
-    *data.yesno.config = data.yesno.default_value;
+static void boolConfigInit(standardConfig *config) {
+    *config->data.yesno.config = config->data.yesno.default_value;
 }
 
-static int boolConfigSet(typeData data, sds *argv, int argc, const char **err) {
+static int boolConfigSet(standardConfig *config, sds *argv, int argc, const char **err) {
     UNUSED(argc);
     int yn = yesnotoi(argv[0]);
     if (yn == -1) {
         *err = "argument must be 'yes' or 'no'";
         return 0;
     }
-    if (data.yesno.is_valid_fn && !data.yesno.is_valid_fn(yn, err))
+    if (config->data.yesno.is_valid_fn && !config->data.yesno.is_valid_fn(yn, err))
         return 0;
-    int prev = *(data.yesno.config);
+    int prev = config->flags & MODULE_CONFIG ? getModuleBoolConfig(config->privdata) : *(config->data.yesno.config);
     if (prev != yn) {
-        *(data.yesno.config) = yn;
+        if (config->flags & MODULE_CONFIG) {
+            return setModuleBoolConfig(config->privdata, yn, err);
+        }
+        *(config->data.yesno.config) = yn;
         return 1;
     }
     return 2;
 }
 
-static sds boolConfigGet(typeData data) {
-    return sdsnew(*data.yesno.config ? "yes" : "no");
+static sds boolConfigGet(standardConfig *config) {
+    if (config->flags & MODULE_CONFIG) {
+        return sdsnew(getModuleBoolConfig(config->privdata) ? "yes" : "no");
+    }
+    return sdsnew(*config->data.yesno.config ? "yes" : "no");
 }
 
-static void boolConfigRewrite(typeData data, const char *name, struct rewriteConfigState *state) {
-    rewriteConfigYesNoOption(state, name,*(data.yesno.config), data.yesno.default_value);
+static void boolConfigRewrite(standardConfig *config, const char *name, struct rewriteConfigState *state) {
+    int val = config->flags & MODULE_CONFIG ? getModuleBoolConfig(config->privdata) : *(config->data.yesno.config);
+    rewriteConfigYesNoOption(state, name, val, config->data.yesno.default_value);
 }
 
 #define createBoolConfig(name, alias, flags, config_addr, default, is_valid, apply) { \
     embedCommonConfig(name, alias, flags) \
     embedConfigInterface(boolConfigInit, boolConfigSet, boolConfigGet, boolConfigRewrite, apply) \
+    .type = BOOL_CONFIG, \
     .data.yesno = { \
         .config = &(config_addr), \
         .default_value = (default), \
@@ -1738,61 +1786,69 @@ static void boolConfigRewrite(typeData data, const char *name, struct rewriteCon
 }
 
 /* String Configs */
-static void stringConfigInit(typeData data) {
-    *data.string.config = (data.string.convert_empty_to_null && !data.string.default_value) ? NULL : zstrdup(data.string.default_value);
+static void stringConfigInit(standardConfig *config) {
+    *config->data.string.config = (config->data.string.convert_empty_to_null && !config->data.string.default_value) ? NULL : zstrdup(config->data.string.default_value);
 }
 
-static int stringConfigSet(typeData data, sds *argv, int argc, const char **err) {
+static int stringConfigSet(standardConfig *config, sds *argv, int argc, const char **err) {
     UNUSED(argc);
-    if (data.string.is_valid_fn && !data.string.is_valid_fn(argv[0], err))
+    if (config->data.string.is_valid_fn && !config->data.string.is_valid_fn(argv[0], err))
         return 0;
-    char *prev = *data.string.config;
-    char *new = (data.string.convert_empty_to_null && !argv[0][0]) ? NULL : argv[0];
+    char *prev = *config->data.string.config;
+    char *new = (config->data.string.convert_empty_to_null && !argv[0][0]) ? NULL : argv[0];
     if (new != prev && (new == NULL || prev == NULL || strcmp(prev, new))) {
-        *data.string.config = new != NULL ? zstrdup(new) : NULL;
+        *config->data.string.config = new != NULL ? zstrdup(new) : NULL;
         zfree(prev);
         return 1;
     }
     return 2;
 }
 
-static sds stringConfigGet(typeData data) {
-    return sdsnew(*data.string.config ? *data.string.config : "");
+static sds stringConfigGet(standardConfig *config) {
+    return sdsnew(*config->data.string.config ? *config->data.string.config : "");
 }
 
-static void stringConfigRewrite(typeData data, const char *name, struct rewriteConfigState *state) {
-    rewriteConfigStringOption(state, name,*(data.string.config), data.string.default_value);
+static void stringConfigRewrite(standardConfig *config, const char *name, struct rewriteConfigState *state) {
+    rewriteConfigStringOption(state, name,*(config->data.string.config), config->data.string.default_value);
 }
 
 /* SDS Configs */
-static void sdsConfigInit(typeData data) {
-    *data.sds.config = (data.sds.convert_empty_to_null && !data.sds.default_value) ? NULL: sdsnew(data.sds.default_value);
+static void sdsConfigInit(standardConfig *config) {
+    *config->data.sds.config = (config->data.sds.convert_empty_to_null && !config->data.sds.default_value) ? NULL : sdsnew(config->data.sds.default_value);
 }
 
-static int sdsConfigSet(typeData data, sds *argv, int argc, const char **err) {
+static int sdsConfigSet(standardConfig *config, sds *argv, int argc, const char **err) {
     UNUSED(argc);
-    if (data.sds.is_valid_fn && !data.sds.is_valid_fn(argv[0], err))
+    if (config->data.sds.is_valid_fn && !config->data.sds.is_valid_fn(argv[0], err))
         return 0;
-    sds prev = *data.sds.config;
-    sds new = (data.string.convert_empty_to_null && (sdslen(argv[0]) == 0)) ? NULL : argv[0];
+    sds prev = config->flags & MODULE_CONFIG ? getModuleStringConfig(config->privdata) : *config->data.sds.config;
+    sds new = (config->data.string.convert_empty_to_null && (sdslen(argv[0]) == 0)) ? NULL : argv[0];
     if (new != prev && (new == NULL || prev == NULL || sdscmp(prev, new))) {
-        *data.sds.config = new != NULL ? sdsdup(new) : NULL;
         sdsfree(prev);
+        if (config->flags & MODULE_CONFIG) {
+            return setModuleStringConfig(config->privdata, new, err);
+        }
+        *config->data.sds.config = new != NULL ? sdsdup(new) : NULL;
         return 1;
     }
+    if (config->flags & MODULE_CONFIG && prev) sdsfree(prev);
     return 2;
 }
 
-static sds sdsConfigGet(typeData data) {
-    if (*data.sds.config) {
-        return sdsdup(*data.sds.config);
+static sds sdsConfigGet(standardConfig *config) {
+    sds val = config->flags & MODULE_CONFIG ? getModuleStringConfig(config->privdata) : *config->data.sds.config;
+    if (val) {
+        if (config->flags & MODULE_CONFIG) return val;
+        return sdsdup(val);
     } else {
         return sdsnew("");
     }
 }
 
-static void sdsConfigRewrite(typeData data, const char *name, struct rewriteConfigState *state) {
-    rewriteConfigSdsOption(state, name, *(data.sds.config), data.sds.default_value);
+static void sdsConfigRewrite(standardConfig *config, const char *name, struct rewriteConfigState *state) {
+    sds val = config->flags & MODULE_CONFIG ? getModuleStringConfig(config->privdata) : *config->data.sds.config;
+    rewriteConfigSdsOption(state, name, val, config->data.sds.default_value);
+    if (val) sdsfree(val);
 }
 
 
@@ -1802,6 +1858,7 @@ static void sdsConfigRewrite(typeData data, const char *name, struct rewriteConf
 #define createStringConfig(name, alias, flags, empty_to_null, config_addr, default, is_valid, apply) { \
     embedCommonConfig(name, alias, flags) \
     embedConfigInterface(stringConfigInit, stringConfigSet, stringConfigGet, stringConfigRewrite, apply) \
+    .type = STRING_CONFIG, \
     .data.string = { \
         .config = &(config_addr), \
         .default_value = (default), \
@@ -1813,6 +1870,7 @@ static void sdsConfigRewrite(typeData data, const char *name, struct rewriteConf
 #define createSDSConfig(name, alias, flags, empty_to_null, config_addr, default, is_valid, apply) { \
     embedCommonConfig(name, alias, flags) \
     embedConfigInterface(sdsConfigInit, sdsConfigSet, sdsConfigGet, sdsConfigRewrite, apply) \
+    .type = SDS_CONFIG, \
     .data.sds = { \
         .config = &(config_addr), \
         .default_value = (default), \
@@ -1822,16 +1880,16 @@ static void sdsConfigRewrite(typeData data, const char *name, struct rewriteConf
 }
 
 /* Enum configs */
-static void enumConfigInit(typeData data) {
-    *data.enumd.config = data.enumd.default_value;
+static void enumConfigInit(standardConfig *config) {
+    *config->data.enumd.config = config->data.enumd.default_value;
 }
 
-static int enumConfigSet(typeData data, sds *argv, int argc, const char **err) {
+static int enumConfigSet(standardConfig *config, sds *argv, int argc, const char **err) {
     UNUSED(argc);
-    int enumval = configEnumGetValue(data.enumd.enum_value, argv[0]);
+    int enumval = configEnumGetValue(config->data.enumd.enum_value, argv[0]);
     if (enumval == INT_MIN) {
         sds enumerr = sdsnew("argument must be one of the following: ");
-        configEnum *enumNode = data.enumd.enum_value;
+        configEnum *enumNode = config->data.enumd.enum_value;
         while(enumNode->name != NULL) {
             enumerr = sdscatlen(enumerr, enumNode->name,
                                 strlen(enumNode->name));
@@ -1847,27 +1905,32 @@ static int enumConfigSet(typeData data, sds *argv, int argc, const char **err) {
         *err = loadbuf;
         return 0;
     }
-    if (data.enumd.is_valid_fn && !data.enumd.is_valid_fn(enumval, err))
+    if (config->data.enumd.is_valid_fn && !config->data.enumd.is_valid_fn(enumval, err))
         return 0;
-    int prev = *(data.enumd.config);
+    int prev = config->flags & MODULE_CONFIG ? getModuleEnumConfig(config->privdata) : *(config->data.enumd.config);
     if (prev != enumval) {
-        *(data.enumd.config) = enumval;
+        if (config->flags & MODULE_CONFIG)
+            return setModuleEnumConfig(config->privdata, enumval, err);
+        *(config->data.enumd.config) = enumval;
         return 1;
     }
     return 2;
 }
 
-static sds enumConfigGet(typeData data) {
-    return sdsnew(configEnumGetNameOrUnknown(data.enumd.enum_value,*data.enumd.config));
+static sds enumConfigGet(standardConfig *config) {
+    int val = config->flags & MODULE_CONFIG ? getModuleEnumConfig(config->privdata) : *(config->data.enumd.config);
+    return sdsnew(configEnumGetNameOrUnknown(config->data.enumd.enum_value,val));
 }
 
-static void enumConfigRewrite(typeData data, const char *name, struct rewriteConfigState *state) {
-    rewriteConfigEnumOption(state, name,*(data.enumd.config), data.enumd.enum_value, data.enumd.default_value);
+static void enumConfigRewrite(standardConfig *config, const char *name, struct rewriteConfigState *state) {
+    int val = config->flags & MODULE_CONFIG ? getModuleEnumConfig(config->privdata) : *(config->data.enumd.config);
+    rewriteConfigEnumOption(state, name, val, config->data.enumd.enum_value, config->data.enumd.default_value);
 }
 
 #define createEnumConfig(name, alias, flags, enum, config_addr, default, is_valid, apply) { \
     embedCommonConfig(name, alias, flags) \
     embedConfigInterface(enumConfigInit, enumConfigSet, enumConfigGet, enumConfigRewrite, apply) \
+    .type = ENUM_CONFIG, \
     .data.enumd = { \
         .config = &(config_addr), \
         .default_value = (default), \
@@ -1878,69 +1941,74 @@ static void enumConfigRewrite(typeData data, const char *name, struct rewriteCon
 
 /* Gets a 'long long val' and sets it into the union, using a macro to get
  * compile time type check. */
-#define SET_NUMERIC_TYPE(val) \
-    if (data.numeric.numeric_type == NUMERIC_TYPE_INT) { \
-        *(data.numeric.config.i) = (int) val; \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_UINT) { \
-        *(data.numeric.config.ui) = (unsigned int) val; \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_LONG) { \
-        *(data.numeric.config.l) = (long) val; \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_ULONG) { \
-        *(data.numeric.config.ul) = (unsigned long) val; \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_LONG_LONG) { \
-        *(data.numeric.config.ll) = (long long) val; \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_ULONG_LONG) { \
-        *(data.numeric.config.ull) = (unsigned long long) val; \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_SIZE_T) { \
-        *(data.numeric.config.st) = (size_t) val; \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_SSIZE_T) { \
-        *(data.numeric.config.sst) = (ssize_t) val; \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_OFF_T) { \
-        *(data.numeric.config.ot) = (off_t) val; \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_TIME_T) { \
-        *(data.numeric.config.tt) = (time_t) val; \
+int setNumericType(standardConfig *config, long long val, const char **err) {
+    if (config->data.numeric.numeric_type == NUMERIC_TYPE_INT) {
+        *(config->data.numeric.config.i) = (int) val;
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_UINT) {
+        *(config->data.numeric.config.ui) = (unsigned int) val;
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_LONG) {
+        *(config->data.numeric.config.l) = (long) val;
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_ULONG) {
+        *(config->data.numeric.config.ul) = (unsigned long) val;
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_LONG_LONG) {
+        if (config->flags & MODULE_CONFIG)
+            return setModuleNumericConfig(config->privdata, val, err);
+        else *(config->data.numeric.config.ll) = (long long) val;
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_ULONG_LONG) {
+        *(config->data.numeric.config.ull) = (unsigned long long) val;
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_SIZE_T) {
+        *(config->data.numeric.config.st) = (size_t) val;
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_SSIZE_T) {
+        *(config->data.numeric.config.sst) = (ssize_t) val;
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_OFF_T) {
+        *(config->data.numeric.config.ot) = (off_t) val;
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_TIME_T) {
+        *(config->data.numeric.config.tt) = (time_t) val;
     }
+    return 1;
+}
 
 /* Gets a 'long long val' and sets it with the value from the union, using a
  * macro to get compile time type check. */
 #define GET_NUMERIC_TYPE(val) \
-    if (data.numeric.numeric_type == NUMERIC_TYPE_INT) { \
-        val = *(data.numeric.config.i); \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_UINT) { \
-        val = *(data.numeric.config.ui); \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_LONG) { \
-        val = *(data.numeric.config.l); \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_ULONG) { \
-        val = *(data.numeric.config.ul); \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_LONG_LONG) { \
-        val = *(data.numeric.config.ll); \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_ULONG_LONG) { \
-        val = *(data.numeric.config.ull); \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_SIZE_T) { \
-        val = *(data.numeric.config.st); \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_SSIZE_T) { \
-        val = *(data.numeric.config.sst); \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_OFF_T) { \
-        val = *(data.numeric.config.ot); \
-    } else if (data.numeric.numeric_type == NUMERIC_TYPE_TIME_T) { \
-        val = *(data.numeric.config.tt); \
+    if (config->data.numeric.numeric_type == NUMERIC_TYPE_INT) { \
+        val = *(config->data.numeric.config.i); \
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_UINT) { \
+        val = *(config->data.numeric.config.ui); \
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_LONG) { \
+        val = *(config->data.numeric.config.l); \
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_ULONG) { \
+        val = *(config->data.numeric.config.ul); \
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_LONG_LONG) { \
+        if (config->flags & MODULE_CONFIG) val = getModuleNumericConfig(config->privdata); \
+        else val = *(config->data.numeric.config.ll); \
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_ULONG_LONG) { \
+        val = *(config->data.numeric.config.ull); \
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_SIZE_T) { \
+        val = *(config->data.numeric.config.st); \
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_SSIZE_T) { \
+        val = *(config->data.numeric.config.sst); \
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_OFF_T) { \
+        val = *(config->data.numeric.config.ot); \
+    } else if (config->data.numeric.numeric_type == NUMERIC_TYPE_TIME_T) { \
+        val = *(config->data.numeric.config.tt); \
     }
 
 /* Numeric configs */
-static void numericConfigInit(typeData data) {
-    SET_NUMERIC_TYPE(data.numeric.default_value)
+static void numericConfigInit(standardConfig *config) {
+    setNumericType(config, config->data.numeric.default_value, NULL);
 }
 
-static int numericBoundaryCheck(typeData data, long long ll, const char **err) {
-    if (data.numeric.numeric_type == NUMERIC_TYPE_ULONG_LONG ||
-        data.numeric.numeric_type == NUMERIC_TYPE_UINT ||
-        data.numeric.numeric_type == NUMERIC_TYPE_SIZE_T) {
+static int numericBoundaryCheck(standardConfig *config, long long ll, const char **err) {
+    if (config->data.numeric.numeric_type == NUMERIC_TYPE_ULONG_LONG ||
+        config->data.numeric.numeric_type == NUMERIC_TYPE_UINT ||
+        config->data.numeric.numeric_type == NUMERIC_TYPE_SIZE_T) {
         /* Boundary check for unsigned types */
         unsigned long long ull = ll;
-        unsigned long long upper_bound = data.numeric.upper_bound;
-        unsigned long long lower_bound = data.numeric.lower_bound;
+        unsigned long long upper_bound = config->data.numeric.upper_bound;
+        unsigned long long lower_bound = config->data.numeric.lower_bound;
         if (ull > upper_bound || ull < lower_bound) {
-            if (data.numeric.flags & OCTAL_CONFIG) {
+            if (config->data.numeric.flags & OCTAL_CONFIG) {
                 snprintf(loadbuf, LOADBUF_SIZE,
                     "argument must be between %llo and %llo inclusive",
                     lower_bound,
@@ -1956,21 +2024,21 @@ static int numericBoundaryCheck(typeData data, long long ll, const char **err) {
         }
     } else {
         /* Boundary check for percentages */
-        if (data.numeric.flags & PERCENT_CONFIG && ll < 0) {
-            if (ll < data.numeric.lower_bound) {
+        if (config->data.numeric.flags & PERCENT_CONFIG && ll < 0) {
+            if (ll < config->data.numeric.lower_bound) {
                 snprintf(loadbuf, LOADBUF_SIZE,
                          "percentage argument must be less or equal to %lld",
-                         -data.numeric.lower_bound);
+                         -config->data.numeric.lower_bound);
                 *err = loadbuf;
                 return 0;
             }
         }
         /* Boundary check for signed types */
-        else if (ll > data.numeric.upper_bound || ll < data.numeric.lower_bound) {
+        else if (ll > config->data.numeric.upper_bound || ll < config->data.numeric.lower_bound) {
             snprintf(loadbuf, LOADBUF_SIZE,
                 "argument must be between %lld and %lld inclusive",
-                data.numeric.lower_bound,
-                data.numeric.upper_bound);
+                config->data.numeric.lower_bound,
+                config->data.numeric.upper_bound);
             *err = loadbuf;
             return 0;
         }
@@ -1978,9 +2046,9 @@ static int numericBoundaryCheck(typeData data, long long ll, const char **err) {
     return 1;
 }
 
-static int numericParseString(typeData data, sds value, const char **err, long long *res) {
+static int numericParseString(standardConfig *config, sds value, const char **err, long long *res) {
     /* First try to parse as memory */
-    if (data.numeric.flags & MEMORY_CONFIG) {
+    if (config->data.numeric.flags & MEMORY_CONFIG) {
         int memerr;
         *res = memtoull(value, &memerr);
         if (!memerr)
@@ -1988,7 +2056,7 @@ static int numericParseString(typeData data, sds value, const char **err, long l
     }
 
     /* Attempt to parse as percent */
-    if (data.numeric.flags & PERCENT_CONFIG &&
+    if (config->data.numeric.flags & PERCENT_CONFIG &&
         sdslen(value) > 1 && value[sdslen(value)-1] == '%' &&
         string2ll(value, sdslen(value)-1, res) &&
         *res >= 0) {
@@ -1998,7 +2066,7 @@ static int numericParseString(typeData data, sds value, const char **err, long l
     }
 
     /* Attempt to parse as an octal number */
-    if (data.numeric.flags & OCTAL_CONFIG) {
+    if (config->data.numeric.flags & OCTAL_CONFIG) {
         char *endptr;
         errno = 0;
         *res = strtoll(value, &endptr, 8);
@@ -2007,58 +2075,57 @@ static int numericParseString(typeData data, sds value, const char **err, long l
     }
 
     /* Attempt a simple number (no special flags set) */
-    if (!data.numeric.flags && string2ll(value, sdslen(value), res))
+    if (!config->data.numeric.flags && string2ll(value, sdslen(value), res))
         return 1;
 
     /* Select appropriate error string */
-    if (data.numeric.flags & MEMORY_CONFIG &&
-        data.numeric.flags & PERCENT_CONFIG)
+    if (config->data.numeric.flags & MEMORY_CONFIG &&
+        config->data.numeric.flags & PERCENT_CONFIG)
         *err = "argument must be a memory or percent value" ;
-    else if (data.numeric.flags & MEMORY_CONFIG)
+    else if (config->data.numeric.flags & MEMORY_CONFIG)
         *err = "argument must be a memory value";
-    else if (data.numeric.flags & OCTAL_CONFIG)
+    else if (config->data.numeric.flags & OCTAL_CONFIG)
         *err = "argument couldn't be parsed as an octal number";
     else
         *err = "argument couldn't be parsed into an integer";
     return 0;
 }
 
-static int numericConfigSet(typeData data, sds *argv, int argc, const char **err) {
+static int numericConfigSet(standardConfig *config, sds *argv, int argc, const char **err) {
     UNUSED(argc);
     long long ll, prev = 0;
 
-    if (!numericParseString(data, argv[0], err, &ll))
+    if (!numericParseString(config, argv[0], err, &ll))
         return 0;
 
-    if (!numericBoundaryCheck(data, ll, err))
+    if (!numericBoundaryCheck(config, ll, err))
         return 0;
 
-    if (data.numeric.is_valid_fn && !data.numeric.is_valid_fn(ll, err))
+    if (config->data.numeric.is_valid_fn && !config->data.numeric.is_valid_fn(ll, err))
         return 0;
 
     GET_NUMERIC_TYPE(prev)
     if (prev != ll) {
-        SET_NUMERIC_TYPE(ll)
-        return 1;
+        return setNumericType(config, ll, err);
     }
 
     return 2;
 }
 
-static sds numericConfigGet(typeData data) {
+static sds numericConfigGet(standardConfig *config) {
     char buf[128];
 
     long long value = 0;
     GET_NUMERIC_TYPE(value)
 
-    if (data.numeric.flags & PERCENT_CONFIG && value < 0) {
+    if (config->data.numeric.flags & PERCENT_CONFIG && value < 0) {
         int len = ll2string(buf, sizeof(buf), -value);
         buf[len] = '%';
         buf[len+1] = '\0';
     }
-    else if (data.numeric.flags & MEMORY_CONFIG) {
+    else if (config->data.numeric.flags & MEMORY_CONFIG) {
         ull2string(buf, sizeof(buf), value);
-    } else if (data.numeric.flags & OCTAL_CONFIG) {
+    } else if (config->data.numeric.flags & OCTAL_CONFIG) {
         snprintf(buf, sizeof(buf), "%llo", value);
     } else {
         ll2string(buf, sizeof(buf), value);
@@ -2066,25 +2133,26 @@ static sds numericConfigGet(typeData data) {
     return sdsnew(buf);
 }
 
-static void numericConfigRewrite(typeData data, const char *name, struct rewriteConfigState *state) {
+static void numericConfigRewrite(standardConfig *config, const char *name, struct rewriteConfigState *state) {
     long long value = 0;
 
     GET_NUMERIC_TYPE(value)
 
-    if (data.numeric.flags & PERCENT_CONFIG && value < 0) {
-        rewriteConfigPercentOption(state, name, -value, data.numeric.default_value);
-    } else if (data.numeric.flags & MEMORY_CONFIG) {
-        rewriteConfigBytesOption(state, name, value, data.numeric.default_value);
-    } else if (data.numeric.flags & OCTAL_CONFIG) {
-        rewriteConfigOctalOption(state, name, value, data.numeric.default_value);
+    if (config->data.numeric.flags & PERCENT_CONFIG && value < 0) {
+        rewriteConfigPercentOption(state, name, -value, config->data.numeric.default_value);
+    } else if (config->data.numeric.flags & MEMORY_CONFIG) {
+        rewriteConfigBytesOption(state, name, value, config->data.numeric.default_value);
+    } else if (config->data.numeric.flags & OCTAL_CONFIG) {
+        rewriteConfigOctalOption(state, name, value, config->data.numeric.default_value);
     } else {
-        rewriteConfigNumericalOption(state, name, value, data.numeric.default_value);
+        rewriteConfigNumericalOption(state, name, value, config->data.numeric.default_value);
     }
 }
 
 #define embedCommonNumericalConfig(name, alias, _flags, lower, upper, config_addr, default, num_conf_flags, is_valid, apply) { \
     embedCommonConfig(name, alias, _flags) \
     embedConfigInterface(numericConfigInit, numericConfigSet, numericConfigGet, numericConfigRewrite, apply) \
+    .type = NUMERIC_CONFIG, \
     .data.numeric = { \
         .lower_bound = (lower), \
         .upper_bound = (upper), \
@@ -2163,6 +2231,7 @@ static void numericConfigRewrite(typeData data, const char *name, struct rewrite
 }
 
 #define createSpecialConfig(name, alias, modifiable, setfn, getfn, rewritefn, applyfn) { \
+    .type = SPECIAL_CONFIG, \
     embedCommonConfig(name, alias, modifiable) \
     embedConfigInterface(NULL, setfn, getfn, rewritefn, applyfn) \
 }
@@ -2439,8 +2508,8 @@ static int applyTLSPort(const char **err) {
 
 #endif  /* USE_OPENSSL */
 
-static int setConfigDirOption(typeData data, sds *argv, int argc, const char **err) {
-    UNUSED(data);
+static int setConfigDirOption(standardConfig *config, sds *argv, int argc, const char **err) {
+    UNUSED(config);
     if (argc != 1) {
         *err = "wrong number of arguments";
         return 0;
@@ -2452,8 +2521,8 @@ static int setConfigDirOption(typeData data, sds *argv, int argc, const char **e
     return 1;
 }
 
-static sds getConfigDirOption(typeData data) {
-    UNUSED(data);
+static sds getConfigDirOption(standardConfig *config) {
+    UNUSED(config);
     char buf[1024];
 
     if (getcwd(buf,sizeof(buf)) == NULL)
@@ -2462,8 +2531,8 @@ static sds getConfigDirOption(typeData data) {
     return sdsnew(buf);
 }
 
-static int setConfigSaveOption(typeData data, sds *argv, int argc, const char **err) {
-    UNUSED(data);
+static int setConfigSaveOption(standardConfig *config, sds *argv, int argc, const char **err) {
+    UNUSED(config);
     int j;
 
     /* Special case: treat single arg "" as zero args indicating empty save configuration */
@@ -2515,8 +2584,8 @@ static int setConfigSaveOption(typeData data, sds *argv, int argc, const char **
     return 1;
 }
 
-static sds getConfigSaveOption(typeData data) {
-    UNUSED(data);
+static sds getConfigSaveOption(standardConfig *config) {
+    UNUSED(config);
     sds buf = sdsempty();
     int j;
 
@@ -2531,13 +2600,13 @@ static sds getConfigSaveOption(typeData data) {
     return buf;
 }
 
-static int setConfigClientOutputBufferLimitOption(typeData data, sds *argv, int argc, const char **err) {
-    UNUSED(data);
+static int setConfigClientOutputBufferLimitOption(standardConfig *config, sds *argv, int argc, const char **err) {
+    UNUSED(config);
     return updateClientOutputBufferLimit(argv, argc, err);
 }
 
-static sds getConfigClientOutputBufferLimitOption(typeData data) {
-    UNUSED(data);
+static sds getConfigClientOutputBufferLimitOption(standardConfig *config) {
+    UNUSED(config);
     sds buf = sdsempty();
     int j;
     for (j = 0; j < CLIENT_TYPE_OBUF_COUNT; j++) {
@@ -2555,11 +2624,11 @@ static sds getConfigClientOutputBufferLimitOption(typeData data) {
 /* Parse an array of CONFIG_OOM_COUNT sds strings, validate and populate
  * server.oom_score_adj_values if valid.
  */
-static int setConfigOOMScoreAdjValuesOption(typeData data, sds *argv, int argc, const char **err) {
+static int setConfigOOMScoreAdjValuesOption(standardConfig *config, sds *argv, int argc, const char **err) {
     int i;
     int values[CONFIG_OOM_COUNT];
     int change = 0;
-    UNUSED(data);
+    UNUSED(config);
 
     if (argc != CONFIG_OOM_COUNT) {
         *err = "wrong number of arguments";
@@ -2600,8 +2669,8 @@ static int setConfigOOMScoreAdjValuesOption(typeData data, sds *argv, int argc,
     return change ? 1 : 2;
 }
 
-static sds getConfigOOMScoreAdjValuesOption(typeData data) {
-    UNUSED(data);
+static sds getConfigOOMScoreAdjValuesOption(standardConfig *config) {
+    UNUSED(config);
     sds buf = sdsempty();
     int j;
 
@@ -2614,8 +2683,8 @@ static sds getConfigOOMScoreAdjValuesOption(typeData data) {
     return buf;
 }
 
-static int setConfigNotifyKeyspaceEventsOption(typeData data, sds *argv, int argc, const char **err) {
-    UNUSED(data);
+static int setConfigNotifyKeyspaceEventsOption(standardConfig *config, sds *argv, int argc, const char **err) {
+    UNUSED(config);
     if (argc != 1) {
         *err = "wrong number of arguments";
         return 0;
@@ -2629,13 +2698,13 @@ static int setConfigNotifyKeyspaceEventsOption(typeData data, sds *argv, int arg
     return 1;
 }
 
-static sds getConfigNotifyKeyspaceEventsOption(typeData data) {
-    UNUSED(data);
+static sds getConfigNotifyKeyspaceEventsOption(standardConfig *config) {
+    UNUSED(config);
     return keyspaceEventsFlagsToString(server.notify_keyspace_events);
 }
 
-static int setConfigBindOption(typeData data, sds* argv, int argc, const char **err) {
-    UNUSED(data);
+static int setConfigBindOption(standardConfig *config, sds* argv, int argc, const char **err) {
+    UNUSED(config);
     int j;
 
     if (argc > CONFIG_BINDADDR_MAX) {
@@ -2657,8 +2726,8 @@ static int setConfigBindOption(typeData data, sds* argv, int argc, const char **
     return 1;
 }
 
-static int setConfigReplicaOfOption(typeData data, sds* argv, int argc, const char **err) {
-    UNUSED(data);
+static int setConfigReplicaOfOption(standardConfig *config, sds* argv, int argc, const char **err) {
+    UNUSED(config);
 
     if (argc != 2) {
         *err = "wrong number of arguments";
@@ -2681,13 +2750,13 @@ static int setConfigReplicaOfOption(typeData data, sds* argv, int argc, const ch
     return 1;
 }
 
-static sds getConfigBindOption(typeData data) {
-    UNUSED(data);
+static sds getConfigBindOption(standardConfig *config) {
+    UNUSED(config);
     return sdsjoin(server.bindaddr,server.bindaddr_count," ");
 }
 
-static sds getConfigReplicaOfOption(typeData data) {
-    UNUSED(data);
+static sds getConfigReplicaOfOption(standardConfig *config) {
+    UNUSED(config);
     char buf[256];
     if (server.masterhost)
         snprintf(buf,sizeof(buf),"%s %d",
@@ -2703,8 +2772,8 @@ int allowProtectedAction(int config, client *c) {
 }
 
 
-static int setConfigLatencyTrackingInfoPercentilesOutputOption(typeData data, sds *argv, int argc, const char **err) {
-    UNUSED(data);
+static int setConfigLatencyTrackingInfoPercentilesOutputOption(standardConfig *config, sds *argv, int argc, const char **err) {
+    UNUSED(config);
     zfree(server.latency_tracking_info_percentiles);
     server.latency_tracking_info_percentiles = NULL;
     server.latency_tracking_info_percentiles_len = argc;
@@ -2736,8 +2805,8 @@ configerr:
     return 0;
 }
 
-static sds getConfigLatencyTrackingInfoPercentilesOutputOption(typeData data) {
-    UNUSED(data);
+static sds getConfigLatencyTrackingInfoPercentilesOutputOption(standardConfig *config) {
+    UNUSED(config);
     sds buf = sdsempty();
     for (int j = 0; j < server.latency_tracking_info_percentiles_len; j++) {
         char fbuf[128];
@@ -2751,8 +2820,8 @@ static sds getConfigLatencyTrackingInfoPercentilesOutputOption(typeData data) {
 }
 
 /* Rewrite the latency-tracking-info-percentiles option. */
-void rewriteConfigLatencyTrackingInfoPercentilesOutputOption(typeData data, const char *name, struct rewriteConfigState *state) {
-    UNUSED(data);
+void rewriteConfigLatencyTrackingInfoPercentilesOutputOption(standardConfig *config, const char *name, struct rewriteConfigState *state) {
+    UNUSED(config);
     sds line = sdsnew(name);
     /* Rewrite latency-tracking-info-percentiles parameters,
      * or an empty 'latency-tracking-info-percentiles ""' line to avoid the
@@ -2972,7 +3041,7 @@ standardConfig static_configs[] = {
     createSpecialConfig("replicaof", "slaveof", IMMUTABLE_CONFIG | MULTI_ARG_CONFIG, setConfigReplicaOfOption, getConfigReplicaOfOption, rewriteConfigReplicaOfOption, NULL),
     createSpecialConfig("latency-tracking-info-percentiles", NULL, MODIFIABLE_CONFIG | MULTI_ARG_CONFIG, setConfigLatencyTrackingInfoPercentilesOutputOption, getConfigLatencyTrackingInfoPercentilesOutputOption, rewriteConfigLatencyTrackingInfoPercentilesOutputOption, NULL),
 
-    /* NULL Terminator */
+    /* NULL Terminator, this is dropped when we convert to the runtime array. */
     {NULL}
 };
 
@@ -2996,7 +3065,7 @@ void initConfigValues() {
     configs = dictCreate(&sdsHashDictType);
     dictExpand(configs, sizeof(static_configs) / sizeof(standardConfig));
     for (standardConfig *config = static_configs; config->name != NULL; config++) {
-        if (config->interface.init) config->interface.init(config->data);
+        if (config->interface.init) config->interface.init(config);
         /* Add the primary config to the dictionary. */
         int ret = registerConfigValue(config->name, config, 0);
         serverAssert(ret);
@@ -3010,6 +3079,67 @@ void initConfigValues() {
     }
 }
 
+/* Remove a config by name from the configs dict. */
+void removeConfig(sds name) {
+    standardConfig *config = lookupConfig(name);
+    if (!config) return;
+    if (config->flags & MODULE_CONFIG) {
+        sdsfree((sds) config->name);
+        if (config->type == ENUM_CONFIG) {
+            configEnum *enumNode = config->data.enumd.enum_value;
+            while(enumNode->name != NULL) {
+                zfree(enumNode->name);
+                enumNode++;
+            }
+            zfree(config->data.enumd.enum_value);
+        } else if (config->type == SDS_CONFIG) {
+            if (config->data.sds.default_value) sdsfree((sds)config->data.sds.default_value);
+        }
+    }
+    dictDelete(configs, name);
+}
+
+/*-----------------------------------------------------------------------------
+ * Module Config
+ *----------------------------------------------------------------------------*/
+
+/* Create a bool/string/enum/numeric standardConfig for a module config in the configs dictionary */
+void addModuleBoolConfig(const char *module_name, const char *name, int flags, void *privdata, int default_val) {
+    sds config_name = sdscatfmt(sdsempty(), "%s.%s", module_name, name);
+    int config_dummy_address;
+    standardConfig module_config = createBoolConfig(config_name, NULL, flags | MODULE_CONFIG, config_dummy_address, default_val, NULL, NULL);
+    module_config.data.yesno.config = NULL;
+    module_config.privdata = privdata;
+    registerConfigValue(config_name, &module_config, 0);
+}
+
+void addModuleStringConfig(const char *module_name, const char *name, int flags, void *privdata, sds default_val) {
+    sds config_name = sdscatfmt(sdsempty(), "%s.%s", module_name, name);
+    sds config_dummy_address;
+    standardConfig module_config = createSDSConfig(config_name, NULL, flags | MODULE_CONFIG, 0, config_dummy_address, default_val, NULL, NULL);
+    module_config.data.sds.config = NULL;
+    module_config.privdata = privdata;
+    registerConfigValue(config_name, &module_config, 0);
+}
+
+void addModuleEnumConfig(const char *module_name, const char *name, int flags, void *privdata, int default_val, configEnum *enum_vals) {
+    sds config_name = sdscatfmt(sdsempty(), "%s.%s", module_name, name);
+    int config_dummy_address;
+    standardConfig module_config = createEnumConfig(config_name, NULL, flags | MODULE_CONFIG, enum_vals, config_dummy_address, default_val, NULL, NULL);
+    module_config.data.enumd.config = NULL;
+    module_config.privdata = privdata;
+    registerConfigValue(config_name, &module_config, 0);
+}
+
+void addModuleNumericConfig(const char *module_name, const char *name, int flags, void *privdata, long long default_val, int conf_flags, long long lower, long long upper) {
+    sds config_name = sdscatfmt(sdsempty(), "%s.%s", module_name, name);
+    long long config_dummy_address;
+    standardConfig module_config = createLongLongConfig(config_name, NULL, flags | MODULE_CONFIG, lower, upper, config_dummy_address, default_val, conf_flags, NULL, NULL);
+    module_config.data.numeric.config.ll = NULL;
+    module_config.privdata = privdata;
+    registerConfigValue(config_name, &module_config, 0);
+}
+
 /*-----------------------------------------------------------------------------
  * CONFIG HELP
  *----------------------------------------------------------------------------*/
diff --git a/src/module.c b/src/module.c
index 467e8c321..cc34c5bc8 100644
--- a/src/module.c
+++ b/src/module.c
@@ -396,6 +396,40 @@ typedef struct RedisModuleKeyOptCtx {
                                               as `copy2`, 'from_dbid' and 'to_dbid' are both valid. */
 } RedisModuleKeyOptCtx;
 
+/* Data structures related to redis module configurations */
+/* The function signatures for module config get callbacks. These are identical to the ones exposed in redismodule.h. */
+typedef RedisModuleString * (*RedisModuleConfigGetStringFunc)(const char *name, void *privdata);
+typedef long long (*RedisModuleConfigGetNumericFunc)(const char *name, void *privdata);
+typedef int (*RedisModuleConfigGetBoolFunc)(const char *name, void *privdata);
+typedef int (*RedisModuleConfigGetEnumFunc)(const char *name, void *privdata);
+/* The function signatures for module config set callbacks. These are identical to the ones exposed in redismodule.h. */
+typedef int (*RedisModuleConfigSetStringFunc)(const char *name, RedisModuleString *val, void *privdata, RedisModuleString **err);
+typedef int (*RedisModuleConfigSetNumericFunc)(const char *name, long long val, void *privdata, RedisModuleString **err);
+typedef int (*RedisModuleConfigSetBoolFunc)(const char *name, int val, void *privdata, RedisModuleString **err);
+typedef int (*RedisModuleConfigSetEnumFunc)(const char *name, int val, void *privdata, RedisModuleString **err);
+/* Apply signature, identical to redismodule.h */
+typedef int (*RedisModuleConfigApplyFunc)(RedisModuleCtx *ctx, void *privdata, RedisModuleString **err);
+
+/* Struct representing a module config. These are stored in a list in the module struct */
+struct ModuleConfig {
+    sds name; /* Name of config without the module name appended to the front */
+    void *privdata; /* Optional data passed into the module config callbacks */
+    union get_fn { /* The get callback specified by the module */
+        RedisModuleConfigGetStringFunc get_string;
+        RedisModuleConfigGetNumericFunc get_numeric;
+        RedisModuleConfigGetBoolFunc get_bool;
+        RedisModuleConfigGetEnumFunc get_enum;
+    } get_fn;
+    union set_fn { /* The set callback specified by the module */
+        RedisModuleConfigSetStringFunc set_string;
+        RedisModuleConfigSetNumericFunc set_numeric;
+        RedisModuleConfigSetBoolFunc set_bool;
+        RedisModuleConfigSetEnumFunc set_enum;
+    } set_fn;
+    RedisModuleConfigApplyFunc apply_fn;
+    RedisModule *module;
+};
+
 /* --------------------------------------------------------------------------
  * Prototypes
  * -------------------------------------------------------------------------- */
@@ -1941,6 +1975,16 @@ int moduleIsModuleCommand(void *module_handle, struct redisCommand *cmd) {
  * ## Module information and time measurement
  * -------------------------------------------------------------------------- */
 
+int moduleListConfigMatch(void *config, void *name) {
+    return strcasecmp(((ModuleConfig *) config)->name, (char *) name) == 0;
+}
+
+void moduleListFree(void *config) {
+    ModuleConfig *module_config = (ModuleConfig *) config;
+    sdsfree(module_config->name);
+    zfree(config);
+}
+
 void RM_SetModuleAttribs(RedisModuleCtx *ctx, const char *name, int ver, int apiver) {
     /* Called by RM_Init() to setup the `ctx->module` structure.
      *
@@ -1957,7 +2001,11 @@ void RM_SetModuleAttribs(RedisModuleCtx *ctx, const char *name, int ver, int api
     module->usedby = listCreate();
     module->using = listCreate();
     module->filters = listCreate();
+    module->module_configs = listCreate();
+    listSetMatchMethod(module->module_configs, moduleListConfigMatch);
+    listSetFreeMethod(module->module_configs, moduleListFree);
     module->in_call = 0;
+    module->configs_initialized = 0;
     module->in_hook = 0;
     module->options = 0;
     module->info_cb = 0;
@@ -10683,9 +10731,21 @@ void moduleRegisterCoreAPI(void);
 void moduleInitModulesSystemLast(void) {
 }
 
+
+dictType sdsKeyValueHashDictType = {
+    dictSdsCaseHash,            /* hash function */
+    NULL,                       /* key dup */
+    NULL,                       /* val dup */
+    dictSdsKeyCaseCompare,      /* key compare */
+    dictSdsDestructor,          /* key destructor */
+    dictSdsDestructor,          /* val destructor */
+    NULL                        /* allow to expand */
+};
+
 void moduleInitModulesSystem(void) {
     moduleUnblockedClients = listCreate();
     server.loadmodule_queue = listCreate();
+    server.module_configs_queue = dictCreate(&sdsKeyValueHashDictType);
     modules = dictCreate(&modulesDictType);
 
     /* Set up the keyspace notification subscriber list and static client */
@@ -10758,6 +10818,20 @@ void moduleLoadQueueEntryFree(struct moduleLoadQueueEntry *loadmod) {
     zfree(loadmod);
 }
 
+/* Remove Module Configs from standardConfig array in config.c */
+void moduleRemoveConfigs(RedisModule *module) {
+    listIter li;
+    listNode *ln;
+    listRewind(module->module_configs, &li);
+    while ((ln = listNext(&li))) {
+        ModuleConfig *config = listNodeValue(ln);
+        sds module_name = sdsnew(module->name);
+        sds full_name = sdscat(sdscat(module_name, "."), config->name); /* ModuleName.ModuleConfig */
+        removeConfig(full_name);
+        sdsfree(full_name);
+    }
+}
+
 /* Load all the modules in the server.loadmodule_queue list, which is
  * populated by `loadmodule` directives in the configuration file.
  * We can't load modules directly when processing the configuration file
@@ -10774,7 +10848,7 @@ void moduleLoadFromQueue(void) {
     listRewind(server.loadmodule_queue,&li);
     while((ln = listNext(&li))) {
         struct moduleLoadQueueEntry *loadmod = ln->value;
-        if (moduleLoad(loadmod->path,(void **)loadmod->argv,loadmod->argc)
+        if (moduleLoad(loadmod->path,(void **)loadmod->argv,loadmod->argc, 0)
             == C_ERR)
         {
             serverLog(LL_WARNING,
@@ -10785,6 +10859,10 @@ void moduleLoadFromQueue(void) {
         moduleLoadQueueEntryFree(loadmod);
         listDelNode(server.loadmodule_queue, ln);
     }
+    if (dictSize(server.module_configs_queue)) {
+        serverLog(LL_WARNING, "Module Configuration detected without loadmodule directive or no ApplyConfig call: aborting");
+        exit(1);
+    }
 }
 
 void moduleFreeModuleStructure(struct RedisModule *module) {
@@ -10792,6 +10870,7 @@ void moduleFreeModuleStructure(struct RedisModule *module) {
     listRelease(module->filters);
     listRelease(module->usedby);
     listRelease(module->using);
+    listRelease(module->module_configs);
     sdsfree(module->name);
     moduleLoadQueueEntryFree(module->loadmod);
     zfree(module);
@@ -10872,15 +10951,56 @@ void moduleUnregisterCommands(struct RedisModule *module) {
     dictReleaseIterator(di);
 }
 
+/* We parse argv to add sds "NAME VALUE" pairs to the server.module_configs_queue list of configs.
+ * We also increment the module_argv pointer to just after ARGS if there are args, otherwise
+ * we set it to NULL */
+int parseLoadexArguments(RedisModuleString ***module_argv, int *module_argc) {
+    int args_specified = 0;
+    RedisModuleString **argv = *module_argv;
+    int argc = *module_argc;
+    for (int i = 0; i < argc; i++) {
+        char *arg_val = argv[i]->ptr;
+        if (!strcasecmp(arg_val, "CONFIG")) {
+            if (i + 2 >= argc) {
+                serverLog(LL_NOTICE, "CONFIG specified without name value pair");
+                return REDISMODULE_ERR;
+            }
+            sds name = sdsdup(argv[i + 1]->ptr);
+            sds value = sdsdup(argv[i + 2]->ptr);
+            if (!dictReplace(server.module_configs_queue, name, value)) sdsfree(name);
+            i += 2;
+        } else if (!strcasecmp(arg_val, "ARGS")) {
+            args_specified = 1;
+            i++;
+            if (i >= argc) {
+                *module_argv = NULL;
+                *module_argc = 0;
+            } else {
+                *module_argv = argv + i;
+                *module_argc = argc - i;
+            }
+            break;
+        } else {
+            serverLog(LL_NOTICE, "Syntax Error from arguments to loadex around %s.", arg_val);
+            return REDISMODULE_ERR;
+        }
+    }
+    if (!args_specified) {
+        *module_argv = NULL;
+        *module_argc = 0;
+    }
+    return REDISMODULE_OK;
+}
+
 /* Load a module and initialize it. On success C_OK is returned, otherwise
  * C_ERR is returned. */
-int moduleLoad(const char *path, void **module_argv, int module_argc) {
+int moduleLoad(const char *path, void **module_argv, int module_argc, int is_loadex) {
     int (*onload)(void *, void **, int);
     void *handle;
 
     struct stat st;
-    if (stat(path, &st) == 0)
-    {   // this check is best effort
+    if (stat(path, &st) == 0) {
+        /* This check is best effort */
         if (!(st.st_mode & (S_IXUSR  | S_IXGRP | S_IXOTH))) {
             serverLog(LL_WARNING, "Module %s failed to load: It does not have execute permissions.", path);
             return C_ERR;
@@ -10904,16 +11024,17 @@ int moduleLoad(const char *path, void **module_argv, int module_argc) {
     moduleCreateContext(&ctx, NULL, REDISMODULE_CTX_TEMP_CLIENT); /* We pass NULL since we don't have a module yet. */
     selectDb(ctx.client, 0);
     if (onload((void*)&ctx,module_argv,module_argc) == REDISMODULE_ERR) {
+        serverLog(LL_WARNING,
+            "Module %s initialization failed. Module not loaded",path);
         if (ctx.module) {
             moduleUnregisterCommands(ctx.module);
             moduleUnregisterSharedAPI(ctx.module);
             moduleUnregisterUsedAPI(ctx.module);
+            moduleRemoveConfigs(ctx.module);
             moduleFreeModuleStructure(ctx.module);
         }
         moduleFreeContext(&ctx);
         dlclose(handle);
-        serverLog(LL_WARNING,
-            "Module %s initialization failed. Module not loaded",path);
         return C_ERR;
     }
 
@@ -10931,15 +11052,30 @@ int moduleLoad(const char *path, void **module_argv, int module_argc) {
     }
 
     serverLog(LL_NOTICE,"Module '%s' loaded from %s",ctx.module->name,path);
+
+    if (listLength(ctx.module->module_configs) && !ctx.module->configs_initialized) {
+        serverLogRaw(LL_WARNING, "Module Configurations were not set, likely a missing LoadConfigs call. Unloading the module.");
+        moduleUnload(ctx.module->name);
+        moduleFreeContext(&ctx);
+        return C_ERR;
+    }
+
+    if (is_loadex && dictSize(server.module_configs_queue)) {
+        serverLogRaw(LL_WARNING, "Loadex configurations were not applied, likely due to invalid arguments. Unloading the module.");
+        moduleUnload(ctx.module->name);
+        moduleFreeContext(&ctx);
+        return C_ERR;
+    }
+
     /* Fire the loaded modules event. */
     moduleFireServerEvent(REDISMODULE_EVENT_MODULE_CHANGE,
                           REDISMODULE_SUBEVENT_MODULE_LOADED,
                           ctx.module);
+
     moduleFreeContext(&ctx);
     return C_OK;
 }
 
-
 /* Unload the module registered with the specified name. On success
  * C_OK is returned, otherwise C_ERR is returned and errno is set
  * to the following values depending on the type of error:
@@ -10991,6 +11127,7 @@ int moduleUnload(sds name) {
     moduleUnregisterSharedAPI(module);
     moduleUnregisterUsedAPI(module);
     moduleUnregisterFilters(module);
+    moduleRemoveConfigs(module);
 
     /* Remove any notification subscribers this module might have */
     moduleUnsubscribeNotifications(module);
@@ -11119,10 +11256,433 @@ sds genModulesInfoString(sds info) {
     return info;
 }
 
+/* --------------------------------------------------------------------------
+ * Module Configurations API internals
+ * -------------------------------------------------------------------------- */
+	 
+/* Check if the configuration name is already registered */
+int isModuleConfigNameRegistered(RedisModule *module, sds name) {
+    listNode *match = listSearchKey(module->module_configs, (void *) name);
+    return match != NULL;
+}
+
+/* Assert that the flags passed into the RM_RegisterConfig Suite are valid */
+int moduleVerifyConfigFlags(unsigned int flags, configType type) {
+    if ((flags & ~(REDISMODULE_CONFIG_DEFAULT
+                    | REDISMODULE_CONFIG_IMMUTABLE
+                    | REDISMODULE_CONFIG_SENSITIVE
+                    | REDISMODULE_CONFIG_HIDDEN
+                    | REDISMODULE_CONFIG_PROTECTED
+                    | REDISMODULE_CONFIG_DENY_LOADING
+                    | REDISMODULE_CONFIG_MEMORY))) {
+        serverLogRaw(LL_WARNING, "Invalid flag(s) for configuration");
+        return REDISMODULE_ERR;
+    }
+    if (type != NUMERIC_CONFIG && flags & REDISMODULE_CONFIG_MEMORY) {
+        serverLogRaw(LL_WARNING, "Numeric flag provided for non-numeric configuration.");
+        return REDISMODULE_ERR;
+    }
+    return REDISMODULE_OK;
+}
+
+int moduleVerifyConfigName(sds name) {
+    if (sdslen(name) == 0) {
+        serverLogRaw(LL_WARNING, "Module config names cannot be an empty string.");
+        return REDISMODULE_ERR;
+    }
+    for (size_t i = 0 ; i < sdslen(name) ; ++i) {
+        char curr_char = name[i];
+        if ((curr_char >= 'a' && curr_char <= 'z') ||
+            (curr_char >= 'A' && curr_char <= 'Z') ||
+            (curr_char >= '0' && curr_char <= '9') ||
+            (curr_char == '_') || (curr_char == '-'))
+        {
+            continue;
+        }
+        serverLog(LL_WARNING, "Invalid character %c in Module Config name %s.", curr_char, name);
+        return REDISMODULE_ERR;
+    }
+    return REDISMODULE_OK;
+}
+
+/* This is a series of set functions for each type that act as dispatchers for 
+ * config.c to call module set callbacks. */
+#define CONFIG_ERR_SIZE 256
+static char configerr[CONFIG_ERR_SIZE];
+static void propagateErrorString(RedisModuleString *err_in, const char **err) {
+    if (err_in) {
+        strncpy(configerr, err_in->ptr, CONFIG_ERR_SIZE);
+        configerr[CONFIG_ERR_SIZE - 1] = '\0';
+        decrRefCount(err_in);
+        *err = configerr;
+    }
+}
+
+int setModuleBoolConfig(ModuleConfig *config, int val, const char **err) {
+    RedisModuleString *error = NULL;
+    int return_code = config->set_fn.set_bool(config->name, val, config->privdata, &error);
+    propagateErrorString(error, err);
+    return return_code == REDISMODULE_OK ? 1 : 0;
+}
+
+int setModuleStringConfig(ModuleConfig *config, sds strval, const char **err) {
+    RedisModuleString *error = NULL;
+    RedisModuleString *new = createStringObject(strval, sdslen(strval));
+    int return_code = config->set_fn.set_string(config->name, new, config->privdata, &error);
+    propagateErrorString(error, err);
+    decrRefCount(new);
+    return return_code == REDISMODULE_OK ? 1 : 0;
+}
+
+int setModuleEnumConfig(ModuleConfig *config, int val, const char **err) {
+    RedisModuleString *error = NULL;
+    int return_code = config->set_fn.set_enum(config->name, val, config->privdata, &error);
+    propagateErrorString(error, err);
+    return return_code == REDISMODULE_OK ? 1 : 0;
+}
+
+int setModuleNumericConfig(ModuleConfig *config, long long val, const char **err) {
+    RedisModuleString *error = NULL;
+    int return_code = config->set_fn.set_numeric(config->name, val, config->privdata, &error);
+    propagateErrorString(error, err);
+    return return_code == REDISMODULE_OK ? 1 : 0;
+}
+
+/* This is a series of get functions for each type that act as dispatchers for 
+ * config.c to call module set callbacks. */
+int getModuleBoolConfig(ModuleConfig *module_config) {
+    return module_config->get_fn.get_bool(module_config->name, module_config->privdata);
+}
+
+sds getModuleStringConfig(ModuleConfig *module_config) {
+    RedisModuleString *val = module_config->get_fn.get_string(module_config->name, module_config->privdata);
+    return val ? sdsdup(val->ptr) : NULL;
+}
+
+int getModuleEnumConfig(ModuleConfig *module_config) {
+    return module_config->get_fn.get_enum(module_config->name, module_config->privdata);
+}
+
+long long getModuleNumericConfig(ModuleConfig *module_config) {
+    return module_config->get_fn.get_numeric(module_config->name, module_config->privdata);
+}
+
+/* This function takes a module and a list of configs stored as sds NAME VALUE pairs.
+ * It attempts to call set on each of these configs. */
+int loadModuleConfigs(RedisModule *module) {
+    listIter li;
+    listNode *ln;
+    const char *err = NULL;
+    listRewind(module->module_configs, &li);
+    while ((ln = listNext(&li))) {
+        ModuleConfig *module_config = listNodeValue(ln);
+        sds config_name = sdscatfmt(sdsempty(), "%s.%s", module->name, module_config->name);
+        dictEntry *config_argument = dictFind(server.module_configs_queue, config_name);
+        if (config_argument) {
+            if (!performModuleConfigSetFromName(dictGetKey(config_argument), dictGetVal(config_argument), &err)) {
+                serverLog(LL_WARNING, "Issue during loading of configuration %s : %s", (sds) dictGetKey(config_argument), err);
+                sdsfree(config_name);
+                dictEmpty(server.module_configs_queue, NULL);
+                return REDISMODULE_ERR;
+            }
+        } else {
+            if (!performModuleConfigSetDefaultFromName(config_name, &err)) {
+                serverLog(LL_WARNING, "Issue attempting to set default value of configuration %s : %s", module_config->name, err);
+                sdsfree(config_name);
+                dictEmpty(server.module_configs_queue, NULL);
+                return REDISMODULE_ERR;
+            }
+        }
+        dictDelete(server.module_configs_queue, config_name);
+        sdsfree(config_name);
+    }
+    module->configs_initialized = 1;
+    return REDISMODULE_OK;
+}
+
+/* Add module_config to the list if the apply and privdata do not match one already in it. */
+void addModuleConfigApply(list *module_configs, ModuleConfig *module_config) {
+    if (!module_config->apply_fn) return;
+    listIter li;
+    listNode *ln;
+    ModuleConfig *pending_apply;
+    listRewind(module_configs, &li);
+    while ((ln = listNext(&li))) {
+        pending_apply = listNodeValue(ln);
+        if (pending_apply->apply_fn == module_config->apply_fn && pending_apply->privdata == module_config->privdata) {
+            return;
+        }
+    }
+    listAddNodeTail(module_configs, module_config);
+}
+
+/* Call apply on all module configs specified in set, if an apply function was specified at registration time. */
+int moduleConfigApplyConfig(list *module_configs, const char **err, const char **err_arg_name) {
+    if (!listLength(module_configs)) return 1;
+    listIter li;
+    listNode *ln;
+    ModuleConfig *module_config;
+    RedisModuleString *error = NULL;
+    RedisModuleCtx ctx;
+
+    listRewind(module_configs, &li);
+    while ((ln = listNext(&li))) {
+        module_config = listNodeValue(ln);
+        moduleCreateContext(&ctx, module_config->module, REDISMODULE_CTX_NONE);
+        if (module_config->apply_fn(&ctx, module_config->privdata, &error)) {
+            if (err_arg_name) *err_arg_name = module_config->name;
+            propagateErrorString(error, err);
+            moduleFreeContext(&ctx);
+            return 0;
+        }
+        moduleFreeContext(&ctx);
+    }
+    return 1;
+}
+
+/* --------------------------------------------------------------------------
+ * ## Module Configurations API
+ * -------------------------------------------------------------------------- */
+
+/* Create a module config object. */
+ModuleConfig *createModuleConfig(sds name, RedisModuleConfigApplyFunc apply_fn, void *privdata, RedisModule *module) {
+    ModuleConfig *new_config = zmalloc(sizeof(ModuleConfig));
+    new_config->name = sdsdup(name);
+    new_config->apply_fn = apply_fn;
+    new_config->privdata = privdata;
+    new_config->module = module;
+    return new_config;
+}
+
+int moduleConfigValidityCheck(RedisModule *module, sds name, unsigned int flags, configType type) {
+    if (moduleVerifyConfigFlags(flags, type) || moduleVerifyConfigName(name)) {
+        errno = EINVAL;
+        return REDISMODULE_ERR;
+    }
+    if (isModuleConfigNameRegistered(module, name)) {
+        serverLog(LL_WARNING, "Configuration by the name: %s already registered", name);
+        errno = EALREADY;
+        return REDISMODULE_ERR;
+    }
+    return REDISMODULE_OK;
+}
+
+unsigned int maskModuleConfigFlags(unsigned int flags) {
+    unsigned int new_flags = 0;
+    if (flags & REDISMODULE_CONFIG_DEFAULT) new_flags |= MODIFIABLE_CONFIG;
+    if (flags & REDISMODULE_CONFIG_IMMUTABLE) new_flags |= IMMUTABLE_CONFIG;
+    if (flags & REDISMODULE_CONFIG_HIDDEN) new_flags |= HIDDEN_CONFIG;
+    if (flags & REDISMODULE_CONFIG_PROTECTED) new_flags |= PROTECTED_CONFIG;
+    if (flags & REDISMODULE_CONFIG_DENY_LOADING) new_flags |= DENY_LOADING_CONFIG;
+    return new_flags;
+}
+
+unsigned int maskModuleNumericConfigFlags(unsigned int flags) {
+    unsigned int new_flags = 0;
+    if (flags & REDISMODULE_CONFIG_MEMORY) new_flags |= MEMORY_CONFIG;
+    return new_flags;
+}
+
+/* Create a string config that Redis users can interact with via the Redis config file,
+ * `CONFIG SET`, `CONFIG GET`, and `CONFIG REWRITE` commands.
+ *
+ * The actual config value is owned by the module, and the `getfn`, `setfn` and optional
+ * `applyfn` callbacks that are provided to Redis in order to access or manipulate the
+ * value. The `getfn` callback retrieves the value from the module, while the `setfn`
+ * callback provides a value to be stored into the module config.
+ * The optional `applyfn` callback is called after a `CONFIG SET` command modified one or
+ * more configs using the `setfn` callback and can be used to atomically apply a config
+ * after several configs were changed together.
+ * If there are multiple configs with `applyfn` callbacks set by a single `CONFIG SET`
+ * command, they will be deduplicated if their `applyfn` function and `privdata` pointers
+ * are identical, and the callback will only be run once.
+ * Both the `setfn` and `applyfn` can return an error if the provided value is invalid or
+ * cannot be used.
+ * The config also declares a type for the value that is validated by Redis and
+ * provided to the module. The config system provides the following types:
+ *
+ * * Redis String: Binary safe string data.
+ * * Enum: One of a finite number of string tokens, provided during registration.
+ * * Numeric: 64 bit signed integer, which also supports min and max values.
+ * * Bool: Yes or no value.
+ *
+ * The `setfn` callback is expected to return REDISMODULE_OK when the value is successfully
+ * applied. It can also return REDISMODULE_ERR if the value can't be applied, and the
+ * *err pointer can be set with a RedisModuleString error message to provide to the client.
+ * This RedisModuleString will be freed by redis after returning from the set callback.
+ *
+ * All configs are registered with a name, a type, a default value, private data that is made
+ * available in the callbacks, as well as several flags that modify the behavior of the config.
+ * The name must only contain alphanumeric characters or dashes. The supported flags are:
+ *
+ * * REDISMODULE_CONFIG_DEFAULT: The default flags for a config. This creates a config that can be modified after startup.
+ * * REDISMODULE_CONFIG_IMMUTABLE: This config can only be provided loading time.
+ * * REDISMODULE_CONFIG_SENSITIVE: The value stored in this config is redacted from all logging.
+ * * REDISMODULE_CONFIG_HIDDEN: The name is hidden from `CONFIG GET` with pattern matching.
+ * * REDISMODULE_CONFIG_PROTECTED: This config will be only be modifiable based off the value of enable-protected-configs.
+ * * REDISMODULE_CONFIG_DENY_LOADING: This config is not modifiable while the server is loading data.
+ * * REDISMODULE_CONFIG_MEMORY: For numeric configs, this config will convert data unit notations into their byte equivalent.
+ *
+ * Default values are used on startup to set the value if it is not provided via the config file
+ * or command line. Default values are also used to compare to on a config rewrite.
+ *
+ * Notes:
+ *
+ *  1. On string config sets that the string passed to the set callback will be freed after execution and the module must retain it.
+ *  2. On string config gets the string will not be consumed and will be valid after execution.
+ *
+ * Example implementation:
+ *
+ *     RedisModuleString *strval;
+ *     int adjustable = 1;
+ *     RedisModuleString *getStringConfigCommand(const char *name, void *privdata) {
+ *         return strval;
+ *     }
+ *
+ *     int setStringConfigCommand(const char *name, RedisModuleString *new, void *privdata, RedisModuleString **err) {
+ *        if (adjustable) {
+ *            RedisModule_Free(strval);
+ *            RedisModule_RetainString(NULL, new);
+ *            strval = new;
+ *            return REDISMODULE_OK;
+ *        }
+ *        *err = RedisModule_CreateString(NULL, "Not adjustable.", 15);
+ *        return REDISMODULE_ERR;
+ *     }
+ *     ...
+ *     RedisModule_RegisterStringConfig(ctx, "string", NULL, REDISMODULE_CONFIG_DEFAULT, getStringConfigCommand, setStringConfigCommand, NULL, NULL);
+ *
+ * If the registration fails, REDISMODULE_ERR is returned and one of the following
+ * errno is set:
+ * * EINVAL: The provided flags are invalid for the registration or the name of the config contains invalid characters.
+ * * EALREADY: The provided configuration name is already used. */
+int RM_RegisterStringConfig(RedisModuleCtx *ctx, const char *name, const char *default_val, unsigned int flags, RedisModuleConfigGetStringFunc getfn, RedisModuleConfigSetStringFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) {
+    RedisModule *module = ctx->module;
+    sds config_name = sdsnew(name);
+    if (moduleConfigValidityCheck(module, config_name, flags, NUMERIC_CONFIG)) {
+        sdsfree(config_name);
+        return REDISMODULE_ERR;
+    }
+    ModuleConfig *new_config = createModuleConfig(config_name, applyfn, privdata, module);
+    sdsfree(config_name);
+    new_config->get_fn.get_string = getfn;
+    new_config->set_fn.set_string = setfn;
+    listAddNodeTail(module->module_configs, new_config);
+    flags = maskModuleConfigFlags(flags);
+    addModuleStringConfig(module->name, name, flags, new_config, default_val ? sdsnew(default_val) : NULL);
+    return REDISMODULE_OK;
+}
+
+/* Create a bool config that server clients can interact with via the 
+ * `CONFIG SET`, `CONFIG GET`, and `CONFIG REWRITE` commands. See 
+ * RedisModule_RegisterStringConfig for detailed information about configs. */
+int RM_RegisterBoolConfig(RedisModuleCtx *ctx, const char *name, int default_val, unsigned int flags, RedisModuleConfigGetBoolFunc getfn, RedisModuleConfigSetBoolFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) {
+    RedisModule *module = ctx->module;
+    sds config_name = sdsnew(name);
+    if (moduleConfigValidityCheck(module, config_name, flags, BOOL_CONFIG)) {
+        sdsfree(config_name);
+        return REDISMODULE_ERR;
+    }
+    ModuleConfig *new_config = createModuleConfig(config_name, applyfn, privdata, module);
+    sdsfree(config_name);
+    new_config->get_fn.get_bool = getfn;
+    new_config->set_fn.set_bool = setfn;
+    listAddNodeTail(module->module_configs, new_config);
+    flags = maskModuleConfigFlags(flags);
+    addModuleBoolConfig(module->name, name, flags, new_config, default_val);
+    return REDISMODULE_OK;
+}
+
+/* 
+ * Create an enum config that server clients can interact with via the 
+ * `CONFIG SET`, `CONFIG GET`, and `CONFIG REWRITE` commands. 
+ * Enum configs are a set of string tokens to corresponding integer values, where 
+ * the string value is exposed to Redis clients but the value passed Redis and the
+ * module is the integer value. These values are defined in enum_values, an array
+ * of null-terminated c strings, and int_vals, an array of enum values who has an
+ * index partner in enum_values.
+ * Example Implementation:
+ *      const char *enum_vals[3] = {"first", "second", "third"};
+ *      const int int_vals[3] = {0, 2, 4};
+ *      int enum_val = 0;
+ *
+ *      int getEnumConfigCommand(const char *name, void *privdata) {
+ *          return enum_val;
+ *      }
+ *       
+ *      int setEnumConfigCommand(const char *name, int val, void *privdata, const char **err) {
+ *          enum_val = val;
+ *          return REDISMODULE_OK;
+ *      }
+ *      ...
+ *      RedisModule_RegisterEnumConfig(ctx, "enum", 0, REDISMODULE_CONFIG_DEFAULT, enum_vals, int_vals, 3, getEnumConfigCommand, setEnumConfigCommand, NULL, NULL);
+ * 
+ * See RedisModule_RegisterStringConfig for detailed general information about configs. */
+int RM_RegisterEnumConfig(RedisModuleCtx *ctx, const char *name, int default_val, unsigned int flags, const char **enum_values, const int *int_values, int num_enum_vals, RedisModuleConfigGetEnumFunc getfn, RedisModuleConfigSetEnumFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) {
+    RedisModule *module = ctx->module;
+    sds config_name = sdsnew(name);
+    if (moduleConfigValidityCheck(module, config_name, flags, ENUM_CONFIG)) {
+        sdsfree(config_name);
+        return REDISMODULE_ERR;
+    }
+    ModuleConfig *new_config = createModuleConfig(config_name, applyfn, privdata, module);
+    sdsfree(config_name);
+    new_config->get_fn.get_enum = getfn;
+    new_config->set_fn.set_enum = setfn;
+    configEnum *enum_vals = zmalloc((num_enum_vals + 1) * sizeof(configEnum));
+    for (int i = 0; i < num_enum_vals; i++) {
+        enum_vals[i].name = zstrdup(enum_values[i]);
+        enum_vals[i].val = int_values[i];
+    }
+    enum_vals[num_enum_vals].name = NULL;
+    enum_vals[num_enum_vals].val = 0;
+    listAddNodeTail(module->module_configs, new_config);
+    flags = maskModuleConfigFlags(flags);
+    addModuleEnumConfig(module->name, name, flags, new_config, default_val, enum_vals);
+    return REDISMODULE_OK;
+}
+
+/*
+ * Create an integer config that server clients can interact with via the 
+ * `CONFIG SET`, `CONFIG GET`, and `CONFIG REWRITE` commands. See 
+ * RedisModule_RegisterStringConfig for detailed information about configs. */
+int RM_RegisterNumericConfig(RedisModuleCtx *ctx, const char *name, long long default_val, unsigned int flags, long long min, long long max, RedisModuleConfigGetNumericFunc getfn, RedisModuleConfigSetNumericFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) {
+    RedisModule *module = ctx->module;
+    sds config_name = sdsnew(name);
+    if (moduleConfigValidityCheck(module, config_name, flags, NUMERIC_CONFIG)) {
+        sdsfree(config_name);
+        return REDISMODULE_ERR;
+    }
+    ModuleConfig *new_config = createModuleConfig(config_name, applyfn, privdata, module);
+    sdsfree(config_name);
+    new_config->get_fn.get_numeric = getfn;
+    new_config->set_fn.set_numeric = setfn;
+    listAddNodeTail(module->module_configs, new_config);
+    unsigned int numeric_flags = maskModuleNumericConfigFlags(flags);
+    flags = maskModuleConfigFlags(flags);
+    addModuleNumericConfig(module->name, name, flags, new_config, default_val, numeric_flags, min, max);
+    return REDISMODULE_OK;
+}
+
+/* Applies all pending configurations on the module load. This should be called
+ * after all of the configurations have been registered for the module inside of RedisModule_OnLoad.
+ * This API needs to be called when configurations are provided in either `MODULE LOADEX`
+ * or provided as startup arguments. */
+int RM_LoadConfigs(RedisModuleCtx *ctx) {
+    if (!ctx || !ctx->module) {
+        return REDISMODULE_ERR;
+    }
+    RedisModule *module = ctx->module;
+    /* Load configs from conf file or arguments from loadex */
+    if (loadModuleConfigs(module)) return REDISMODULE_ERR;
+    return REDISMODULE_OK;
+}
+
 /* Redis MODULE command.
  *
  * MODULE LIST
  * MODULE LOAD <path> [args...]
+ * MODULE LOADEX <path> [[CONFIG NAME VALUE] [CONFIG NAME VALUE]] [ARGS ...]
  * MODULE UNLOAD <name>
  */
 void moduleCommand(client *c) {
@@ -11134,6 +11694,8 @@ void moduleCommand(client *c) {
 "    Return a list of loaded modules.",
 "LOAD <path> [<arg> ...]",
 "    Load a module library from <path>, passing to it any optional arguments.",
+"LOADEX <path> [[CONFIG NAME VALUE] [CONFIG NAME VALUE]] [ARGS ...]",
+"    Load a module library from <path>, while passing it module configurations and optional arguments.",
 "UNLOAD <name>",
 "    Unload a module.",
 NULL
@@ -11148,11 +11710,30 @@ NULL
             argv = &c->argv[3];
         }
 
-        if (moduleLoad(c->argv[2]->ptr,(void **)argv,argc) == C_OK)
+        if (moduleLoad(c->argv[2]->ptr,(void **)argv,argc, 0) == C_OK)
             addReply(c,shared.ok);
         else
             addReplyError(c,
                 "Error loading the extension. Please check the server logs.");
+    } else if (!strcasecmp(subcmd,"loadex") && c->argc >= 3) {
+        robj **argv = NULL;
+        int argc = 0;
+
+        if (c->argc > 3) {
+            argc = c->argc - 3;
+            argv = &c->argv[3];
+        }
+        /* If this is a loadex command we want to populate server.module_configs_queue with 
+         * sds NAME VALUE pairs. We also want to increment argv to just after ARGS, if supplied. */
+        if (parseLoadexArguments((RedisModuleString ***) &argv, &argc) == REDISMODULE_OK &&
+            moduleLoad(c->argv[2]->ptr, (void **)argv, argc, 1) == C_OK)
+            addReply(c,shared.ok);
+        else {
+            dictEmpty(server.module_configs_queue, NULL);
+            addReplyError(c,
+                "Error loading the extension. Please check the server logs.");
+        }
+
     } else if (!strcasecmp(subcmd,"unload") && c->argc == 3) {
         if (moduleUnload(c->argv[2]->ptr) == C_OK)
             addReply(c,shared.ok);
@@ -11955,4 +12536,9 @@ void moduleRegisterCoreAPI(void) {
     REGISTER_API(EventLoopDel);
     REGISTER_API(EventLoopAddOneShot);
     REGISTER_API(Yield);
+    REGISTER_API(RegisterBoolConfig);
+    REGISTER_API(RegisterNumericConfig);
+    REGISTER_API(RegisterStringConfig);
+    REGISTER_API(RegisterEnumConfig);
+    REGISTER_API(LoadConfigs);
 }
diff --git a/src/redismodule.h b/src/redismodule.h
index a40d6eba0..f27c06b0e 100644
--- a/src/redismodule.h
+++ b/src/redismodule.h
@@ -80,6 +80,15 @@
 #define REDISMODULE_HASH_EXISTS     (1<<3)
 #define REDISMODULE_HASH_COUNT_ALL  (1<<4)
 
+#define REDISMODULE_CONFIG_DEFAULT 0 /* This is the default for a module config. */
+#define REDISMODULE_CONFIG_IMMUTABLE (1ULL<<0) /* Can this value only be set at startup? */
+#define REDISMODULE_CONFIG_SENSITIVE (1ULL<<1) /* Does this value contain sensitive information */
+#define REDISMODULE_CONFIG_HIDDEN (1ULL<<4) /* This config is hidden in `config get <pattern>` (used for tests/debugging) */
+#define REDISMODULE_CONFIG_PROTECTED (1ULL<<5) /* Becomes immutable if enable-protected-configs is enabled. */
+#define REDISMODULE_CONFIG_DENY_LOADING (1ULL<<6) /* This config is forbidden during loading. */
+
+#define REDISMODULE_CONFIG_MEMORY (1ULL<<7) /* Indicates if this value can be set as a memory value */
+
 /* StreamID type. */
 typedef struct RedisModuleStreamID {
     uint64_t ms;
@@ -807,6 +816,15 @@ typedef void (*RedisModuleScanCB)(RedisModuleCtx *ctx, RedisModuleString *keynam
 typedef void (*RedisModuleScanKeyCB)(RedisModuleKey *key, RedisModuleString *field, RedisModuleString *value, void *privdata);
 typedef void (*RedisModuleUserChangedFunc) (uint64_t client_id, void *privdata);
 typedef int (*RedisModuleDefragFunc)(RedisModuleDefragCtx *ctx);
+typedef RedisModuleString * (*RedisModuleConfigGetStringFunc)(const char *name, void *privdata);
+typedef long long (*RedisModuleConfigGetNumericFunc)(const char *name, void *privdata);
+typedef int (*RedisModuleConfigGetBoolFunc)(const char *name, void *privdata);
+typedef int (*RedisModuleConfigGetEnumFunc)(const char *name, void *privdata);
+typedef int (*RedisModuleConfigSetStringFunc)(const char *name, RedisModuleString *val, void *privdata, RedisModuleString **err);
+typedef int (*RedisModuleConfigSetNumericFunc)(const char *name, long long val, void *privdata, RedisModuleString **err);
+typedef int (*RedisModuleConfigSetBoolFunc)(const char *name, int val, void *privdata, RedisModuleString **err);
+typedef int (*RedisModuleConfigSetEnumFunc)(const char *name, int val, void *privdata, RedisModuleString **err);
+typedef int (*RedisModuleConfigApplyFunc)(RedisModuleCtx *ctx, void *privdata, RedisModuleString **err);
 
 typedef struct RedisModuleTypeMethods {
     uint64_t version;
@@ -1159,6 +1177,11 @@ REDISMODULE_API const RedisModuleString * (*RedisModule_GetKeyNameFromDefragCtx)
 REDISMODULE_API int (*RedisModule_EventLoopAdd)(int fd, int mask, RedisModuleEventLoopFunc func, void *user_data) REDISMODULE_ATTR;
 REDISMODULE_API int (*RedisModule_EventLoopDel)(int fd, int mask) REDISMODULE_ATTR;
 REDISMODULE_API int (*RedisModule_EventLoopAddOneShot)(RedisModuleEventLoopOneShotFunc func, void *user_data) REDISMODULE_ATTR;
+REDISMODULE_API int (*RedisModule_RegisterBoolConfig)(RedisModuleCtx *ctx, char *name, int default_val, unsigned int flags, RedisModuleConfigGetBoolFunc getfn, RedisModuleConfigSetBoolFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) REDISMODULE_ATTR;
+REDISMODULE_API int (*RedisModule_RegisterNumericConfig)(RedisModuleCtx *ctx, const char *name, long long default_val, unsigned int flags, long long min, long long max, RedisModuleConfigGetNumericFunc getfn, RedisModuleConfigSetNumericFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) REDISMODULE_ATTR;
+REDISMODULE_API int (*RedisModule_RegisterStringConfig)(RedisModuleCtx *ctx, const char *name, const char *default_val, unsigned int flags, RedisModuleConfigGetStringFunc getfn, RedisModuleConfigSetStringFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) REDISMODULE_ATTR;
+REDISMODULE_API int (*RedisModule_RegisterEnumConfig)(RedisModuleCtx *ctx, const char *name, int default_val, unsigned int flags, const char **enum_values, const int *int_values, int num_enum_vals, RedisModuleConfigGetEnumFunc getfn, RedisModuleConfigSetEnumFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) REDISMODULE_ATTR;
+REDISMODULE_API int (*RedisModule_LoadConfigs)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
 
 #define RedisModule_IsAOFClient(id) ((id) == UINT64_MAX)
 
@@ -1483,6 +1506,11 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
     REDISMODULE_GET_API(EventLoopAdd);
     REDISMODULE_GET_API(EventLoopDel);
     REDISMODULE_GET_API(EventLoopAddOneShot);
+    REDISMODULE_GET_API(RegisterBoolConfig);
+    REDISMODULE_GET_API(RegisterNumericConfig);
+    REDISMODULE_GET_API(RegisterStringConfig);
+    REDISMODULE_GET_API(RegisterEnumConfig);
+    REDISMODULE_GET_API(LoadConfigs);
 
     if (RedisModule_IsModuleNameBusy && RedisModule_IsModuleNameBusy(name)) return REDISMODULE_ERR;
     RedisModule_SetModuleAttribs(ctx,name,ver,apiver);
diff --git a/src/sentinel.c b/src/sentinel.c
index af1bfb729..3ad8f902b 100644
--- a/src/sentinel.c
+++ b/src/sentinel.c
@@ -391,7 +391,6 @@ void sentinelReceiveHelloMessages(redisAsyncContext *c, void *reply, void *privd
 sentinelRedisInstance *sentinelGetMasterByName(char *name);
 char *sentinelGetSubjectiveLeader(sentinelRedisInstance *master);
 char *sentinelGetObjectiveLeader(sentinelRedisInstance *master);
-int yesnotoi(char *s);
 void instanceLinkConnectionError(const redisAsyncContext *c);
 const char *sentinelRedisInstanceTypeStr(sentinelRedisInstance *ri);
 void sentinelAbortFailover(sentinelRedisInstance *ri);
diff --git a/src/server.h b/src/server.h
index ba072e7d3..edfea2226 100644
--- a/src/server.h
+++ b/src/server.h
@@ -760,6 +760,8 @@ struct RedisModule {
     list *usedby;   /* List of modules using APIs from this one. */
     list *using;    /* List of modules we use some APIs of. */
     list *filters;  /* List of filters the module has registered. */
+    list *module_configs; /* List of configurations the module has registered */
+    int configs_initialized; /* Have the module configurations been initialized? */
     int in_call;    /* RM_Call() nesting level */
     int in_hook;    /* Hooks callback nesting level for this module (0 or 1). */
     int options;    /* Module options and capabilities. */
@@ -1474,6 +1476,7 @@ struct redisServer {
     dict *moduleapi;            /* Exported core APIs dictionary for modules. */
     dict *sharedapi;            /* Like moduleapi but containing the APIs that
                                    modules share with each other. */
+    dict *module_configs_queue; /* Dict that stores module configurations from .conf file until after modules are loaded during startup or arguments to loadex. */
     list *loadmodule_queue;     /* List of modules to load at startup. */
     int module_pipe[2];         /* Pipe used to awake the event loop by module threads. */
     pid_t child_pid;            /* PID of current child */
@@ -2338,7 +2341,8 @@ int populateArgsStructure(struct redisCommandArg *args);
 void moduleInitModulesSystem(void);
 void moduleInitModulesSystemLast(void);
 void modulesCron(void);
-int moduleLoad(const char *path, void **argv, int argc);
+int moduleLoad(const char *path, void **argv, int argc, int is_loadex);
+int moduleUnload(sds name);
 void moduleLoadFromQueue(void);
 int moduleGetCommandKeysViaAPI(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
 int moduleGetCommandChannelsViaAPI(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
@@ -2970,6 +2974,40 @@ int keyspaceEventsStringToFlags(char *classes);
 sds keyspaceEventsFlagsToString(int flags);
 
 /* Configuration */
+/* Configuration Flags */
+#define MODIFIABLE_CONFIG 0 /* This is the implied default for a standard 
+                             * config, which is mutable. */
+#define IMMUTABLE_CONFIG (1ULL<<0) /* Can this value only be set at startup? */
+#define SENSITIVE_CONFIG (1ULL<<1) /* Does this value contain sensitive information */
+#define DEBUG_CONFIG (1ULL<<2) /* Values that are useful for debugging. */
+#define MULTI_ARG_CONFIG (1ULL<<3) /* This config receives multiple arguments. */
+#define HIDDEN_CONFIG (1ULL<<4) /* This config is hidden in `config get <pattern>` (used for tests/debugging) */
+#define PROTECTED_CONFIG (1ULL<<5) /* Becomes immutable if enable-protected-configs is enabled. */
+#define DENY_LOADING_CONFIG (1ULL<<6) /* This config is forbidden during loading. */
+#define ALIAS_CONFIG (1ULL<<7) /* For configs with multiple names, this flag is set on the alias. */
+#define MODULE_CONFIG (1ULL<<8) /* This config is a module config */
+
+#define INTEGER_CONFIG 0 /* No flags means a simple integer configuration */
+#define MEMORY_CONFIG (1<<0) /* Indicates if this value can be loaded as a memory value */
+#define PERCENT_CONFIG (1<<1) /* Indicates if this value can be loaded as a percent (and stored as a negative int) */
+#define OCTAL_CONFIG (1<<2) /* This value uses octal representation */
+
+/* Enum Configs contain an array of configEnum objects that match a string with an integer. */
+typedef struct configEnum {
+    char *name;
+    int val;
+} configEnum;
+
+/* Type of configuration. */
+typedef enum {
+    BOOL_CONFIG,
+    NUMERIC_CONFIG,
+    STRING_CONFIG,
+    SDS_CONFIG,
+    ENUM_CONFIG,
+    SPECIAL_CONFIG,
+} configType;
+
 void loadServerConfig(char *filename, char config_from_stdin, char *options);
 void appendServerSaveParams(time_t seconds, int changes);
 void resetServerSaveParams(void);
@@ -2978,9 +3016,29 @@ void rewriteConfigRewriteLine(struct rewriteConfigState *state, const char *opti
 void rewriteConfigMarkAsProcessed(struct rewriteConfigState *state, const char *option);
 int rewriteConfig(char *path, int force_write);
 void initConfigValues();
+void removeConfig(sds name);
 sds getConfigDebugInfo();
 int allowProtectedAction(int config, client *c);
 
+/* Module Configuration */
+typedef struct ModuleConfig ModuleConfig;
+int performModuleConfigSetFromName(sds name, sds value, const char **err);
+int performModuleConfigSetDefaultFromName(sds name, const char **err);
+void addModuleBoolConfig(const char *module_name, const char *name, int flags, void *privdata, int default_val);
+void addModuleStringConfig(const char *module_name, const char *name, int flags, void *privdata, sds default_val);
+void addModuleEnumConfig(const char *module_name, const char *name, int flags, void *privdata, int default_val, configEnum *enum_vals);
+void addModuleNumericConfig(const char *module_name, const char *name, int flags, void *privdata, long long default_val, int conf_flags, long long lower, long long upper);
+void addModuleConfigApply(list *module_configs, ModuleConfig *module_config);
+int moduleConfigApplyConfig(list *module_configs, const char **err, const char **err_arg_name);
+int getModuleBoolConfig(ModuleConfig *module_config);
+int setModuleBoolConfig(ModuleConfig *config, int val, const char **err);
+sds getModuleStringConfig(ModuleConfig *module_config);
+int setModuleStringConfig(ModuleConfig *config, sds strval, const char **err);
+int getModuleEnumConfig(ModuleConfig *module_config);
+int setModuleEnumConfig(ModuleConfig *config, int val, const char **err);
+long long getModuleNumericConfig(ModuleConfig *module_config);
+int setModuleNumericConfig(ModuleConfig *config, long long val, const char **err);
+
 /* db.c -- Keyspace access API */
 int removeExpire(redisDb *db, robj *key);
 void deleteExpiredKeyAndPropagate(redisDb *db, robj *keyobj);
diff --git a/src/util.h b/src/util.h
index d2780e00e..5ea71fecd 100644
--- a/src/util.h
+++ b/src/util.h
@@ -75,6 +75,7 @@ int string2d(const char *s, size_t slen, double *dp);
 int trimDoubleString(char *buf, size_t len);
 int d2string(char *buf, size_t len, double value);
 int ld2string(char *buf, size_t len, long double value, ld2string_mode mode);
+int yesnotoi(char *s);
 sds getAbsolutePath(char *filename);
 long getTimeZone(void);
 int pathIsBaseName(char *path);