Add support for using Redis MONITOR inputs files as the commands used during memtier execution

Author: fcostaoliveira

README.md

@@ -95,6 +95,27 @@ $ memtier_benchmark --help
 
 for command line options.
 
+### Using monitor input files
+
+You can replay real command streams by pointing memtier_benchmark to a monitor log file with the `--monitor-input=/path/to/file` option. Special commands such as `__monitor_c1__` pick a specific entry from the file, while `__monitor_c@__` selects commands at runtime (optionally combined with `--monitor-pattern` and `--command-ratio`). For example, the following command replays the first command from the file on each request:
+
+```
+$ memtier_benchmark --monitor-input=monitor.txt --command=__monitor_c1__
+```
+
+This lets you mix synthetic workloads with realistic captured traffic in both standalone and Redis Cluster deployments.
+
+To generate monitor logs, you can use the Redis `MONITOR` command from `redis-cli`, which prints all commands received by the server. For example:
+
+```
+$ redis-cli MONITOR
+OK
+1460100081.165665 [0 127.0.0.1:51706] "set" "shipment:8000736522714:status" "sorting"
+1460100083.053365 [0 127.0.0.1:51707] "get" "shipment:8000736522714:status"
+```
+
+You can pipe this output and filter specific patterns with tools such as `grep`, then save it to a file and use it as a `--monitor-input` source. For more details, see the official Redis documentation on [monitoring commands executed in Redis](https://redis.io/docs/latest/develop/tools/cli/#monitor-commands-executed-in-redis).
+
 ## Crash Reporting
 
 memtier_benchmark includes built-in crash handling that automatically generates detailed bug reports when the program crashes. If you encounter a crash, the tool will print a comprehensive report including:

client.cpp

@@ -279,6 +279,64 @@ bool client::create_arbitrary_request(unsigned int command_index, struct timeval
 
     benchmark_debug_log("%s: %s:\n", m_connections[conn_id]->get_readable_id(), cmd.command.c_str());
 
+	    // Check if this is a monitor command placeholder - handle it specially
+	    if (cmd.command_args.size() == 1 && cmd.command_args[0].type == monitor_random_type) {
+	        // Select a command from the monitor file at runtime based on the monitor pattern
+	        size_t selected_index = 0;
+	        const std::string* monitor_cmd_ptr = NULL;
+	        if (m_config->monitor_pattern == 'R') {
+	            monitor_cmd_ptr = &m_config->monitor_commands->get_random_command(&selected_index);
+	            benchmark_debug_log("%s: random monitor command selected (q%zu): %s\n",
+	                              m_connections[conn_id]->get_readable_id(),
+	                              selected_index + 1,  // 1-based index for user display
+	                              monitor_cmd_ptr->c_str());
+	        } else {
+	            monitor_cmd_ptr = &m_config->monitor_commands->get_next_sequential_command(&selected_index);
+	            benchmark_debug_log("%s: sequential monitor command selected (q%zu): %s\n",
+	                              m_connections[conn_id]->get_readable_id(),
+	                              selected_index + 1,  // 1-based index for user display
+	                              monitor_cmd_ptr->c_str());
+	        }
+
+	        const std::string& monitor_cmd = *monitor_cmd_ptr;
+
+        // Parse and format the monitor command into a temporary arbitrary_command
+        arbitrary_command temp_cmd(monitor_cmd.c_str());
+        if (!temp_cmd.split_command_to_args()) {
+            fprintf(stderr, "error: failed to parse random monitor command at runtime: %s\n", monitor_cmd.c_str());
+            return false;
+        }
+
+        // Format the command for the protocol (adds RESP headers)
+        if (!m_connections[conn_id]->get_protocol()->format_arbitrary_command(temp_cmd)) {
+            fprintf(stderr, "error: failed to format random monitor command at runtime: %s\n", monitor_cmd.c_str());
+            return false;
+        }
+
+        // Send the randomly selected command
+        for (unsigned int i = 0; i < temp_cmd.command_args.size(); i++) {
+            const command_arg* arg = &temp_cmd.command_args[i];
+            if (arg->type == const_type) {
+                cmd_size += m_connections[conn_id]->send_arbitrary_command(arg);
+            } else if (arg->type == key_type) {
+                unsigned long long key_index;
+                get_key_response res = get_key_for_conn(command_index, conn_id, &key_index);
+                assert(res == available_for_conn);
+                cmd_size += m_connections[conn_id]->send_arbitrary_command(arg, m_obj_gen->get_key(), m_obj_gen->get_key_len());
+            } else if (arg->type == data_type) {
+                unsigned int value_len;
+                const char *value = m_obj_gen->get_value(0, &value_len);
+                assert(value != NULL);
+                assert(value_len > 0);
+                cmd_size += m_connections[conn_id]->send_arbitrary_command(arg, value, value_len);
+            }
+        }
+
+        m_connections[conn_id]->send_arbitrary_command_end(command_index, &timestamp, cmd_size);
+        return true;
+    }
+
+    // Normal arbitrary command handling
     for (unsigned int i = 0; i < cmd.command_args.size(); i++) {
         const command_arg* arg = &cmd.command_args[i];
         if (arg->type == const_type) {

config_types.cpp

@@ -489,3 +489,88 @@ bool arbitrary_command::split_command_to_args() {
     err:
     return false;
 }
+
+// Monitor command list implementation
+bool monitor_command_list::load_from_file(const char* filename) {
+    FILE* file = fopen(filename, "r");
+    if (!file) {
+        fprintf(stderr, "error: failed to open monitor input file: %s\n", filename);
+        return false;
+    }
+
+    char line[65536]; // Large buffer for monitor lines
+    size_t total_lines = 0;
+    while (fgets(line, sizeof(line), file)) {
+        total_lines++;
+        // Find the first quote - this is where the command starts
+        char* first_quote = strchr(line, '"');
+        if (!first_quote) {
+            continue; // Skip lines without commands
+        }
+
+        // Extract everything from first quote to end of line
+        // We keep the quotes as-is to avoid re-parsing
+        std::string command_str(first_quote);
+
+        // Remove trailing newline if present
+        if (!command_str.empty() && command_str[command_str.length() - 1] == '\n') {
+            command_str.erase(command_str.length() - 1);
+        }
+        if (!command_str.empty() && command_str[command_str.length() - 1] == '\r') {
+            command_str.erase(command_str.length() - 1);
+        }
+
+        commands.push_back(command_str);
+    }
+
+    fclose(file);
+
+    if (commands.empty()) {
+        fprintf(stderr, "error: no commands found in monitor input file: %s\n", filename);
+        return false;
+    }
+
+    fprintf(stderr, "Loaded %zu monitor commands from %zu total lines\n", commands.size(), total_lines);
+    return true;
+}
+
+const std::string& monitor_command_list::get_command(size_t index) const {
+    if (index >= commands.size()) {
+        static std::string empty;
+        return empty;
+    }
+    return commands[index];
+}
+
+const std::string& monitor_command_list::get_random_command() const {
+    if (commands.empty()) {
+        static std::string empty;
+        return empty;
+    }
+    size_t random_index = rand() % commands.size();
+    return commands[random_index];
+}
+
+const std::string& monitor_command_list::get_random_command(size_t* out_index) const {
+    if (commands.empty()) {
+        static std::string empty;
+        if (out_index) *out_index = 0;
+        return empty;
+    }
+    size_t random_index = rand() % commands.size();
+    if (out_index) *out_index = random_index;
+    return commands[random_index];
+}
+	
+	const std::string& monitor_command_list::get_next_sequential_command(size_t* out_index) {
+	    if (commands.empty()) {
+	        static std::string empty;
+	        if (out_index) *out_index = 0;
+	        return empty;
+	    }
+	    // Use a global sequential index across all clients/threads.
+	    size_t index = next_index.fetch_add(1, std::memory_order_relaxed);
+	    index = index % commands.size();
+	    if (out_index) *out_index = index;
+	    return commands[index];
+	}

config_types.h

@@ -106,18 +106,24 @@ struct server_addr {
 
 #define KEY_PLACEHOLDER "__key__"
 #define DATA_PLACEHOLDER "__data__"
+#define MONITOR_PLACEHOLDER_PREFIX "__monitor_c"
+#define MONITOR_RANDOM_PLACEHOLDER "__monitor_c@__"
 
 enum command_arg_type {
     const_type      = 0,
     key_type        = 1,
     data_type       = 2,
-    undefined_type  = 3
+    monitor_type    = 3,
+    monitor_random_type = 4,
+    undefined_type  = 5
 };
 
 struct command_arg {
-    command_arg(const char* arg, unsigned int arg_len) : type(undefined_type), data(arg, arg_len), has_key_affixes(false) {;}
+    command_arg(const char* arg, unsigned int arg_len) : type(undefined_type), data(arg, arg_len), monitor_index(0), has_key_affixes(false) {;}
     command_arg_type type;
     std::string data;
+    // For monitor_type, stores the index (1-based)
+    size_t monitor_index;
     // the prefix and suffix strings are used for mixed key placeholder storing of substrings
     std::string data_prefix;
     std::string data_suffix;
@@ -183,4 +189,20 @@ struct arbitrary_command_list {
     }
 };
 
+struct monitor_command_list {
+private:
+    std::vector<std::string> commands;
+	    std::atomic<size_t> next_index;
+
+public:
+	    monitor_command_list() : next_index(0) {;}
+
+    bool load_from_file(const char* filename);
+    const std::string& get_command(size_t index) const;
+    const std::string& get_random_command() const;
+    const std::string& get_random_command(size_t* out_index) const;
+	    const std::string& get_next_sequential_command(size_t* out_index);
+    size_t size() const { return commands.size(); }
+};
+
 #endif /* _CONFIG_TYPES_H */