Add pre-computed key routing for monitor commands in cluster mode

- Pre-compute keys for all monitor commands at load time (second argument = key)
- FT.SEARCH and FT.AGGREGATE treated as keyless (any connection can send)
- Sequential mode: peek at next command's key to route to correct slot owner
- Random mode: send from current connection, rely on MOVED/ASK
- Add literal_key_type for specific monitor placeholders (__monitor_line1__, etc.)
- Add protocol validation: monitor input only works with Redis protocols
- Remove debug print statements from tests

Author: fcostaoliveira

client.cpp

@@ -327,6 +327,13 @@ bool client::create_arbitrary_request(unsigned int command_index, struct timeval
             return false;
         }
 
+        // Mark the first argument as a literal key for cluster routing
+        // Most Redis commands have the key as the first argument after the command name
+        if (temp_cmd.command_args.size() > 0) {
+            temp_cmd.command_args[0].type = literal_key_type;
+            temp_cmd.keys_count = 1;
+        }
+
         // 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());
@@ -344,6 +351,9 @@ bool client::create_arbitrary_request(unsigned int command_index, struct timeval
                 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 == literal_key_type) {
+                // Send the literal key value from the monitor command
+                cmd_size += m_connections[conn_id]->send_arbitrary_command(arg, arg->data.c_str(), arg->data.length());
             } else if (arg->type == data_type) {
                 unsigned int value_len;
                 const char *value = m_obj_gen->get_value(0, &value_len);
@@ -409,6 +419,9 @@ bool client::create_arbitrary_request(unsigned int command_index, struct timeval
                 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 == literal_key_type) {
+            // Send the literal key value from the monitor command
+            cmd_size += m_connections[conn_id]->send_arbitrary_command(arg, arg->data.c_str(), arg->data.length());
         } else if (arg->type == data_type) {
             unsigned int value_len;
             const char *value = m_obj_gen->get_value(0, &value_len);

cluster_client.cpp

@@ -349,12 +349,91 @@ bool cluster_client::create_arbitrary_request(unsigned int command_index, struct
      * if the generated key belongs to this connection before starting to send it */
     assert(m_key_index_pools[conn_id]->empty());
 
+    const arbitrary_command &cmd = get_arbitrary_command(command_index);
+
+    /* Check if this command has a literal key (from monitor input) */
+    for (unsigned int i = 0; i < cmd.command_args.size(); i++) {
+        if (cmd.command_args[i].type == literal_key_type) {
+            /* For literal keys, calculate slot directly from the key value */
+            const std::string &literal_key = cmd.command_args[i].data;
+            unsigned int hslot = calc_hslot_crc16_cluster(literal_key.c_str(), literal_key.length());
+            unsigned int target_conn_id = m_slot_to_shard[hslot];
+
+            /* If this connection owns the slot, send the request */
+            if (target_conn_id == conn_id) {
+                client::create_arbitrary_request(command_index, timestamp, conn_id);
+                return true;
+            }
+
+            /* If the target connection is disconnected, trigger slot refresh */
+            if (m_connections[target_conn_id]->get_connection_state() == conn_disconnected) {
+                m_connections[conn_id]->set_cluster_slots();
+                return false;
+            }
+
+            /* If target connection is refreshing slots, skip */
+            if (m_connections[target_conn_id]->get_cluster_slots_state() != setup_done) {
+                return false;
+            }
+
+            /* Queue is full, skip for now */
+            key_index_pool *key_idx_pool = m_key_index_pools[target_conn_id];
+            if (key_idx_pool->size() >= KEY_INDEX_QUEUE_MAX_SIZE) {
+                return false;
+            }
+
+            /* Store command index for the target connection (no key_index needed for literal keys) */
+            key_idx_pool->push(command_index);
+            key_idx_pool->push(0); /* Dummy key_index - not used for literal keys */
+            return true;
+        }
+    }
+
+    /* Handle monitor_random_type: use pre-computed keys for cluster routing.
+     * For sequential mode, we peek at the index before advancing to check ownership.
+     * For random mode, we send from any connection and rely on MOVED/ASK because
+     * we can't predict which random command will be selected. */
+    if (cmd.command_args.size() == 1 && cmd.command_args[0].type == monitor_random_type) {
+        bool is_random = (m_config->monitor_pattern == 'R');
+
+        if (is_random) {
+            // For random mode, just send from this connection.
+            // MOVED/ASK will redirect if needed. We can't pre-route because
+            // the random command selection happens inside client::create_arbitrary_request()
+            // and would differ from any check we do here.
+            client::create_arbitrary_request(command_index, timestamp, conn_id);
+            return true;
+        }
+
+        // For sequential mode, peek at the next index without advancing
+        size_t monitor_index = m_config->monitor_commands->peek_next_sequential_index();
+
+        // Check if this command has a key (using pre-computed key)
+        if (m_config->monitor_commands->has_key(monitor_index)) {
+            const std::string &key = m_config->monitor_commands->get_key(monitor_index);
+            unsigned int hslot = calc_hslot_crc16_cluster(key.c_str(), key.length());
+            unsigned int target_conn_id = m_slot_to_shard[hslot];
+
+            // If this connection doesn't own the slot, skip this request
+            // The connection that owns this slot will pick it up later
+            if (target_conn_id != conn_id) {
+                return false;
+            }
+        }
+        // If no key (keyless command like FT.SEARCH/FT.AGGREGATE), any connection can send it
+
+        // This connection owns the slot (or command is keyless), proceed with sending
+        client::create_arbitrary_request(command_index, timestamp, conn_id);
+        return true;
+    }
+
     /* keyless command can be used by any connection */
-    if (get_arbitrary_command(command_index).keys_count == 0) {
+    if (cmd.keys_count == 0) {
         client::create_arbitrary_request(command_index, timestamp, conn_id);
         return true;
     }
 
+    /* Normal key placeholder handling */
     unsigned long long key_index;
     get_key_response res = get_key_for_conn(command_index, conn_id, &key_index);
 

config_types.cpp

@@ -478,6 +478,83 @@ bool arbitrary_command::split_command_to_args()
 }
 
 // Monitor command list implementation
+
+// Extract key from a command string (second argument after command name)
+// Returns empty string for keyless commands (FT.SEARCH, FT.AGGREGATE, or commands with no args)
+std::string monitor_command_list::extract_key_from_command(const std::string &command)
+{
+    const char *p = command.c_str();
+    std::string first_word;
+    std::string second_word;
+    int word_count = 0;
+
+    while (*p && word_count < 2) {
+        // Skip blanks
+        while (*p && isspace(*p)) {
+            p++;
+        }
+        if (!*p) break;
+
+        std::string current_word;
+        bool in_quotes = false;
+        bool in_single_quotes = false;
+
+        while (*p) {
+            if (in_quotes) {
+                if (*p == '"') {
+                    in_quotes = false;
+                    p++;
+                    break;
+                }
+                current_word += *p++;
+            } else if (in_single_quotes) {
+                if (*p == '\'') {
+                    in_single_quotes = false;
+                    p++;
+                    break;
+                }
+                current_word += *p++;
+            } else {
+                if (*p == '"') {
+                    in_quotes = true;
+                    p++;
+                } else if (*p == '\'') {
+                    in_single_quotes = true;
+                    p++;
+                } else if (isspace(*p)) {
+                    break;
+                } else {
+                    current_word += *p++;
+                }
+            }
+        }
+
+        if (!current_word.empty()) {
+            if (word_count == 0) {
+                first_word = current_word;
+            } else {
+                second_word = current_word;
+            }
+            word_count++;
+        }
+    }
+
+    // No second argument - keyless command
+    if (second_word.empty()) {
+        return "";
+    }
+
+    // Check for FT.SEARCH and FT.AGGREGATE - these are keyless (can be sent to any connection)
+    std::string upper_cmd = first_word;
+    std::transform(upper_cmd.begin(), upper_cmd.end(), upper_cmd.begin(), ::toupper);
+    if (upper_cmd == "FT.SEARCH" || upper_cmd == "FT.AGGREGATE") {
+        return "";
+    }
+
+    // Return the second argument as the key
+    return second_word;
+}
+
 bool monitor_command_list::load_from_file(const char *filename)
 {
     FILE *file = fopen(filename, "r");
@@ -509,6 +586,10 @@ bool monitor_command_list::load_from_file(const char *filename)
         }
 
         commands.push_back(command_str);
+
+        // Extract and store the key for cluster routing
+        std::string key = extract_key_from_command(command_str);
+        keys.push_back(key);
     }
 
     fclose(file);
@@ -557,3 +638,28 @@ const std::string &monitor_command_list::get_next_sequential_command(size_t *out
     if (out_index) *out_index = index;
     return commands[index];
 }
+
+const std::string &monitor_command_list::get_key(size_t index) const
+{
+    static std::string empty;
+    if (index >= keys.size()) {
+        return empty;
+    }
+    return keys[index];
+}
+
+bool monitor_command_list::has_key(size_t index) const
+{
+    if (index >= keys.size()) {
+        return false;
+    }
+    return !keys[index].empty();
+}
+
+size_t monitor_command_list::peek_next_sequential_index() const
+{
+    if (commands.empty()) {
+        return 0;
+    }
+    return next_index.load(std::memory_order_relaxed) % commands.size();
+}

config_types.h

@@ -127,7 +127,8 @@ enum command_arg_type
     data_type = 2,
     monitor_type = 3,
     monitor_random_type = 4,
-    undefined_type = 5
+    literal_key_type = 5, // Key is a literal value (from monitor commands) - used for cluster routing
+    undefined_type = 6
 };
 
 struct command_arg
@@ -205,15 +206,27 @@ struct monitor_command_list
 {
 private:
     std::vector<std::string> commands;
+    std::vector<std::string> keys; // Pre-computed key for each command (empty if keyless)
     std::atomic<size_t> next_index;
 
+    // Extract key from a command string (second argument, or empty for keyless commands)
+    static std::string extract_key_from_command(const std::string &command);
+
 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(object_generator *obj_gen, size_t *out_index) const;
     const std::string &get_next_sequential_command(size_t *out_index);
+
+    // Get the pre-computed key for a command (empty string if keyless)
+    const std::string &get_key(size_t index) const;
+    bool has_key(size_t index) const;
+
+    // Peek at next sequential index without advancing (for cluster routing)
+    size_t peek_next_sequential_index() const;
+
     size_t size() const { return commands.size(); }
 };
 

memtier_benchmark.cpp

@@ -1985,6 +1985,12 @@ int main(int argc, char *argv[])
 
     // Load monitor input file if specified
     if (cfg.monitor_input) {
+        // Monitor input only works with Redis protocols
+        if (!is_redis_protocol(cfg.protocol)) {
+            fprintf(stderr, "error: --monitor-input is only supported with Redis protocols (redis, resp2, resp3).\n");
+            exit(1);
+        }
+
         if (!cfg.monitor_commands->load_from_file(cfg.monitor_input)) {
             exit(1);
         }
@@ -2050,6 +2056,13 @@ int main(int argc, char *argv[])
                     exit(1);
                 }
 
+                // Mark the first argument as a literal key for cluster routing
+                // Most Redis commands have the key as the first argument after the command name
+                if (cmd.command_args.size() > 0) {
+                    cmd.command_args[0].type = literal_key_type;
+                    cmd.keys_count = 1;
+                }
+
                 // Update command name (first word of the command)
                 size_t pos = cmd.command.find(" ");
                 if (pos == std::string::npos) {

tests/test_monitor_input.py

@@ -588,16 +588,13 @@ def test_command_stats_breakdown_by_command(env):
     # Check stdout for aggregated output
     with open("{0}/mb.stdout".format(config.results_dir)) as stdout:
         stdout_content = stdout.read()
-        env.debugPrint("stdout content:\n{}".format(stdout_content), True)
 
         # Count occurrences of "Sets" and "Gets" in the output
         # With aggregation, we should see exactly one "Sets" row and one "Gets" row
         lines = stdout_content.split("\n")
         sets_count = sum(1 for line in lines if line.strip().startswith("Sets"))
         gets_count = sum(1 for line in lines if line.strip().startswith("Gets"))
 
-        env.debugPrint("Sets rows: {}, Gets rows: {}".format(sets_count, gets_count), True)
-
         # Should have exactly 1 Sets row and 1 Gets row (aggregated)
         env.assertEqual(sets_count, 1)
         env.assertEqual(gets_count, 1)
@@ -648,16 +645,13 @@ def test_command_stats_breakdown_by_line(env):
     # Check stdout for per-command output
     with open("{0}/mb.stdout".format(config.results_dir)) as stdout:
         stdout_content = stdout.read()
-        env.debugPrint("stdout content:\n{}".format(stdout_content), True)
 
         # Count occurrences of "Sets" and "Gets" in the output
         # Without aggregation, we should see 2 "Sets" rows and 2 "Gets" rows
         lines = stdout_content.split("\n")
         sets_count = sum(1 for line in lines if line.strip().startswith("Sets"))
         gets_count = sum(1 for line in lines if line.strip().startswith("Gets"))
 
-        env.debugPrint("Sets rows: {}, Gets rows: {}".format(sets_count, gets_count), True)
-
         # Should have 2 Sets rows and 2 Gets rows (one per command)
         env.assertEqual(sets_count, 2)
         env.assertEqual(gets_count, 2)