Bump redis-rs + Route Function Stats to all nodes (#2117)

--------- Signed-off-by: Shoham Elias <[email protected]> Signed-off-by: Yury-Fridlyand <[email protected]> Signed-off-by: Shoham Elias <[email protected]> Co-authored-by: Yury-Fridlyand <[email protected]> Co-authored-by: Andrew Carbonetto <[email protected]>
valkey-io · Aug 20, 2024 · 45001ed · 45001ed
1 parent d91fbab
commit 45001ed
Show file tree

Hide file tree

Showing 25 changed files with 328 additions and 223 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -84,13 +84,14 @@
 * Node: Added PUBSUB * commands ([#2090](https:/valkey-io/valkey-glide/pull/2090))
 * Python: Added PUBSUB * commands ([#2043](https:/valkey-io/valkey-glide/pull/2043))
 * Node: Added XGROUP CREATE & XGROUP DESTROY commands ([#2084](https:/valkey-io/valkey-glide/pull/2084))
-* Node: Added BZPOPMAX & BZPOPMIN command ([#2077]((https:/valkey-io/valkey-glide/pull/2077))
+* Node: Added BZPOPMAX & BZPOPMIN command ([#2077](https:/valkey-io/valkey-glide/pull/2077))
 * Node: Added XGROUP CREATECONSUMER & XGROUP DELCONSUMER commands ([#2088](https:/valkey-io/valkey-glide/pull/2088))
 * Node: Added GETEX command ([#2107]((https:/valkey-io/valkey-glide/pull/2107))
 * Node: Added ZINTER and ZUNION commands ([#2146](https:/aws/glide-for-redis/pull/2146))
 
 #### Breaking Changes
 * Node: (Refactor) Convert classes to types ([#2005](https:/valkey-io/valkey-glide/pull/2005))
+* Core: Change FUNCTION STATS command to return multi node response for standalone mode ([#2117](https:/valkey-io/valkey-glide/pull/2117))
 
 #### Fixes
 * Java: Add overloads for XADD to allow duplicate entry keys ([#1970](https:/valkey-io/valkey-glide/pull/1970))

diff --git a/glide-core/src/client/reconnecting_connection.rs b/glide-core/src/client/reconnecting_connection.rs
@@ -165,7 +165,7 @@ impl ReconnectingConnection {
  create_connection(backend, connection_retry_strategy, push_sender).await
  }
 
- fn node_address(&self) -> String {
+ pub(crate) fn node_address(&self) -> String {
  self.inner
  .backend
  .connection_info

diff --git a/glide-core/src/client/standalone_client.rs b/glide-core/src/client/standalone_client.rs
@@ -312,7 +312,22 @@ impl StandaloneClient {
  Some(ResponsePolicy::CombineMaps) => future::try_join_all(requests)
  .await
  .and_then(cluster_routing::combine_map_results),
- Some(ResponsePolicy::Special) | None => {
+ Some(ResponsePolicy::Special) => {
+ // Await all futures and collect results
+ let results = future::try_join_all(requests).await?;
+ // Create key-value pairs where the key is the node address and the value is the corresponding result
+ let node_result_pairs = self
+ .inner
+ .nodes
+ .iter()
+ .zip(results)
+ .map(|(node, result)| (Value::BulkString(node.node_address().into()), result))
+ .collect();
+
+ Ok(Value::Map(node_result_pairs))
+ }
+
+ None => {
  // This is our assumption - if there's no coherent way to aggregate the responses, we just collect them in an array, and pass it to the user.
  // TODO - once Value::Error is merged, we can use join_all and report separate errors and also pass successes.
  future::try_join_all(requests).await.map(Value::Array)

diff --git a/java/client/src/main/java/glide/api/BaseClient.java b/java/client/src/main/java/glide/api/BaseClient.java
@@ -208,6 +208,7 @@
 import glide.api.commands.StreamBaseCommands;
 import glide.api.commands.StringBaseCommands;
 import glide.api.commands.TransactionsBaseCommands;
+import glide.api.models.ClusterValue;
 import glide.api.models.GlideString;
 import glide.api.models.PubSubMessage;
 import glide.api.models.Script;
@@ -696,7 +697,7 @@ protected Map<GlideString, Object>[] handleFunctionListResponseBinary(Object[] r
  return data;
  }
 
- /** Process a <code>FUNCTION STATS</code> standalone response. */
+ /** Process a <code>FUNCTION STATS</code> response from one node. */
  protected Map<String, Map<String, Object>> handleFunctionStatsResponse(
  Map<String, Map<String, Object>> response) {
  Map<String, Object> runningScriptInfo = response.get("running_script");
@@ -707,7 +708,7 @@ protected Map<String, Map<String, Object>> handleFunctionStatsResponse(
  return response;
  }
 
- /** Process a <code>FUNCTION STATS</code> standalone response. */
+ /** Process a <code>FUNCTION STATS</code> response from one node. */
  protected Map<GlideString, Map<GlideString, Object>> handleFunctionStatsBinaryResponse(
  Map<GlideString, Map<GlideString, Object>> response) {
  Map<GlideString, Object> runningScriptInfo = response.get(gs("running_script"));
@@ -718,6 +719,36 @@ protected Map<GlideString, Map<GlideString, Object>> handleFunctionStatsBinaryRe
  return response;
  }
 
+ /** Process a <code>FUNCTION STATS</code> cluster response. */
+ protected ClusterValue<Map<String, Map<String, Object>>> handleFunctionStatsResponse(
+ Response response, boolean isSingleValue) {
+ if (isSingleValue) {
+ return ClusterValue.ofSingleValue(handleFunctionStatsResponse(handleMapResponse(response)));
+ } else {
+ Map<String, Map<String, Map<String, Object>>> data = handleMapResponse(response);
+ for (var nodeInfo : data.entrySet()) {
+ nodeInfo.setValue(handleFunctionStatsResponse(nodeInfo.getValue()));
+ }
+ return ClusterValue.ofMultiValue(data);
+ }
+ }
+
+ /** Process a <code>FUNCTION STATS</code> cluster response. */
+ protected ClusterValue<Map<GlideString, Map<GlideString, Object>>>
+ handleFunctionStatsBinaryResponse(Response response, boolean isSingleValue) {
+ if (isSingleValue) {
+ return ClusterValue.ofSingleValue(
+ handleFunctionStatsBinaryResponse(handleBinaryStringMapResponse(response)));
+ } else {
+ Map<GlideString, Map<GlideString, Map<GlideString, Object>>> data =
+ handleBinaryStringMapResponse(response);
+ for (var nodeInfo : data.entrySet()) {
+ nodeInfo.setValue(handleFunctionStatsBinaryResponse(nodeInfo.getValue()));
+ }
+ return ClusterValue.ofMultiValueBinary(data);
+ }
+ }
+
  /** Process a <code>LCS key1 key2 IDX</code> response */
  protected Map<String, Object> handleLcsIdxResponse(Map<String, Object> response)
  throws GlideException {

diff --git a/java/client/src/main/java/glide/api/GlideClient.java b/java/client/src/main/java/glide/api/GlideClient.java
@@ -444,19 +444,20 @@ public CompletableFuture<String> functionKill() {
  }
 
  @Override
- public CompletableFuture<Map<String, Map<String, Object>>> functionStats() {
+ public CompletableFuture<Map<String, Map<String, Map<String, Object>>>> functionStats() {
  return commandManager.submitNewCommand(
  FunctionStats,
  new String[0],
- response -> handleFunctionStatsResponse(handleMapResponse(response)));
+ response -> handleFunctionStatsResponse(response, false).getMultiValue());
  }
 
  @Override
- public CompletableFuture<Map<GlideString, Map<GlideString, Object>>> functionStatsBinary() {
+ public CompletableFuture<Map<String, Map<GlideString, Map<GlideString, Object>>>>
+ functionStatsBinary() {
  return commandManager.submitNewCommand(
  FunctionStats,
  new GlideString[0],
- response -> handleFunctionStatsBinaryResponse(handleBinaryStringMapResponse(response)));
+ response -> handleFunctionStatsBinaryResponse(response, false).getMultiValue());
  }
 
  @Override

diff --git a/java/client/src/main/java/glide/api/GlideClusterClient.java b/java/client/src/main/java/glide/api/GlideClusterClient.java
@@ -946,36 +946,6 @@ public CompletableFuture<String> functionKill(@NonNull Route route) {
  FunctionKill, new String[0], route, this::handleStringResponse);
  }
 
- /** Process a <code>FUNCTION STATS</code> cluster response. */
- protected ClusterValue<Map<String, Map<String, Object>>> handleFunctionStatsResponse(
- Response response, boolean isSingleValue) {
- if (isSingleValue) {
- return ClusterValue.ofSingleValue(handleFunctionStatsResponse(handleMapResponse(response)));
- } else {
- Map<String, Map<String, Map<String, Object>>> data = handleMapResponse(response);
- for (var nodeInfo : data.entrySet()) {
- nodeInfo.setValue(handleFunctionStatsResponse(nodeInfo.getValue()));
- }
- return ClusterValue.ofMultiValue(data);
- }
- }
-
- /** Process a <code>FUNCTION STATS</code> cluster response. */
- protected ClusterValue<Map<GlideString, Map<GlideString, Object>>>
- handleFunctionStatsBinaryResponse(Response response, boolean isSingleValue) {
- if (isSingleValue) {
- return ClusterValue.ofSingleValue(
- handleFunctionStatsBinaryResponse(handleBinaryStringMapResponse(response)));
- } else {
- Map<GlideString, Map<GlideString, Map<GlideString, Object>>> data =
- handleBinaryStringMapResponse(response);
- for (var nodeInfo : data.entrySet()) {
- nodeInfo.setValue(handleFunctionStatsBinaryResponse(nodeInfo.getValue()));
- }
- return ClusterValue.ofMultiValueBinary(data);
- }
- }
-
  @Override
  public CompletableFuture<ClusterValue<Map<String, Map<String, Object>>>> functionStats() {
  return commandManager.submitNewCommand(

diff --git a/java/client/src/main/java/glide/api/commands/ScriptingAndFunctionsClusterCommands.java b/java/client/src/main/java/glide/api/commands/ScriptingAndFunctionsClusterCommands.java
@@ -865,7 +865,7 @@ CompletableFuture<ClusterValue<Object>> fcallReadOnly(
  /**
  * Kills a function that is currently executing.<br>
  * <code>FUNCTION KILL</code> terminates read-only functions only.<br>
- * The command will be routed to all primary nodes.
+ * The command will be routed to all nodes.
  *
  * @since Valkey 7.0 and above.
  * @see <a href="https://valkey.io/commands/function-kill/">valkey.io</a> for details.

diff --git a/java/client/src/main/java/glide/api/commands/ScriptingAndFunctionsCommands.java b/java/client/src/main/java/glide/api/commands/ScriptingAndFunctionsCommands.java
@@ -328,7 +328,8 @@ CompletableFuture<Map<GlideString, Object>[]> functionListBinary(
 
  /**
  * Kills a function that is currently executing.<br>
- * <code>FUNCTION KILL</code> terminates read-only functions only.
+ * <code>FUNCTION KILL</code> terminates read-only functions only. <code>FUNCTION KILL</code> runs
+ * on all nodes of the server, including primary and replicas.
  *
  * @since Valkey 7.0 and above.
  * @see <a href="https://valkey.io/commands/function-kill/">valkey.io</a> for details.
@@ -343,63 +344,73 @@ CompletableFuture<Map<GlideString, Object>[]> functionListBinary(
 
  /**
  * Returns information about the function that's currently running and information about the
- * available execution engines.
+ * available execution engines.<br>
+ * <code>FUNCTION STATS</code> runs on all nodes of the server, including primary and replicas.
+ * The response includes a mapping from node address to the command response for that node.
  *
  * @since Valkey 7.0 and above.
  * @see <a href="https://valkey.io/commands/function-stats/">valkey.io</a> for details.
- * @return A <code>Map</code> with two keys:
+ * @return A <code>Map</code> from node address to the command response for that node, where the
+ * command contains a <code>Map</code> with two keys:
  * <ul>
  * <li><code>running_script</code> with information about the running script.
  * <li><code>engines</code> with information about available engines and their stats.
  * </ul>
  * See example for more details.
  * @example
  * <pre>{@code
- * Map<String, Map<String, Object>> response = client.functionStats().get();
- * Map<String, Object> runningScriptInfo = response.get("running_script");
- * if (runningScriptInfo != null) {
- * String[] commandLine = (String[]) runningScriptInfo.get("command");
- * System.out.printf("Server is currently running function '%s' with command line '%s', which has been running for %d ms%n",
- * runningScriptInfo.get("name"), String.join(" ", commandLine), (long)runningScriptInfo.get("duration_ms"));
- * }
- * Map<String, Object> enginesInfo = response.get("engines");
- * for (String engineName : enginesInfo.keySet()) {
- * Map<String, Long> engine = (Map<String, Long>) enginesInfo.get(engineName);
- * System.out.printf("Server supports engine '%s', which has %d libraries and %d functions in total%n",
- * engineName, engine.get("libraries_count"), engine.get("functions_count"));
+ * Map<String, Map<String, Map<String, Object>>> response = client.functionStats().get();
+ * for (String node : response.keySet()) {
+ * Map<String, Object> runningScriptInfo = response.get(node).get("running_script");
+ * if (runningScriptInfo != null) {
+ * String[] commandLine = (String[]) runningScriptInfo.get("command");
+ * System.out.printf("Node '%s' is currently running function '%s' with command line '%s', which has been running for %d ms%n",
+ * node, runningScriptInfo.get("name"), String.join(" ", commandLine), (long)runningScriptInfo.get("duration_ms"));
+ * }
+ * Map<String, Object> enginesInfo = response.get(node).get("engines");
+ * for (String engineName : enginesInfo.keySet()) {
+ * Map<String, Long> engine = (Map<String, Long>) enginesInfo.get(engineName);
+ * System.out.printf("Node '%s' supports engine '%s', which has %d libraries and %d functions in total%n",
+ * node, engineName, engine.get("libraries_count"), engine.get("functions_count"));
+ * }
  * }
  * }</pre>
  */
- CompletableFuture<Map<String, Map<String, Object>>> functionStats();
+ CompletableFuture<Map<String, Map<String, Map<String, Object>>>> functionStats();
 
  /**
  * Returns information about the function that's currently running and information about the
- * available execution engines.
+ * available execution engines.<br>
+ * <code>FUNCTION STATS</code> runs on all nodes of the server, including primary and replicas.
+ * The response includes a mapping from node address to the command response for that node.
  *
  * @since Valkey 7.0 and above.
  * @see <a href="https://valkey.io/commands/function-stats/">valkey.io</a> for details.
- * @return A <code>Map</code> with two keys:
+ * @return A <code>Map</code> from node address to the command response for that node, where the
+ * command contains a <code>Map</code> with two keys:
  * <ul>
  * <li><code>running_script</code> with information about the running script.
  * <li><code>engines</code> with information about available engines and their stats.
  * </ul>
  * See example for more details.
  * @example
  * <pre>{@code
- * Map<GlideString, Map<GlideString, Object>> response = client.functionStats().get();
- * Map<GlideString, Object> runningScriptInfo = response.get(gs("running_script"));
- * if (runningScriptInfo != null) {
- * GlideString[] commandLine = (GlideString[]) runningScriptInfo.get(gs("command"));
- * System.out.printf("Server is currently running function '%s' with command line '%s', which has been running for %d ms%n",
- * runningScriptInfo.get(gs("name")), String.join(" ", Arrays.toString(commandLine)), (long)runningScriptInfo.get(gs("duration_ms")));
- * }
- * Map<GlideString, Object> enginesInfo = response.get(gs("engines"));
- * for (GlideString engineName : enginesInfo.keySet()) {
- * Map<GlideString, Long> engine = (Map<GlideString, Long>) enginesInfo.get(gs(engineName));
- * System.out.printf("Server supports engine '%s', which has %d libraries and %d functions in total%n",
- * engineName, engine.get(gs("libraries_count")), engine.get(gs("functions_count")));
+ * Map<GlideString, Map<GlideString, Map<GlideString, Object>>> response = client.functionStats().get();
+ * for (String node : response.keySet()) {
+ * Map<GlideString, Object> runningScriptInfo = response.get(gs(node)).get(gs("running_script"));
+ * if (runningScriptInfo != null) {
+ * GlideString[] commandLine = (GlideString[]) runningScriptInfo.get(gs("command"));
+ * System.out.printf("Node '%s' is currently running function '%s' with command line '%s', which has been running for %d ms%n",
+ * node, runningScriptInfo.get(gs("name")), String.join(" ", Arrays.toString(commandLine)), (long)runningScriptInfo.get(gs("duration_ms")));
+ * }
+ * Map<GlideString, Object> enginesInfo = response.get(gs(node)).get(gs("engines"));
+ * for (GlideString engineName : enginesInfo.keySet()) {
+ * Map<GlideString, Long> engine = (Map<GlideString, Long>) enginesInfo.get(gs(engineName));
+ * System.out.printf("Node '%s' supports engine '%s', which has %d libraries and %d functions in total%n",
+ * node, engineName, engine.get(gs("libraries_count")), engine.get(gs("functions_count")));
+ * }
  * }
  * }</pre>
  */
- CompletableFuture<Map<GlideString, Map<GlideString, Object>>> functionStatsBinary();
+ CompletableFuture<Map<String, Map<GlideString, Map<GlideString, Object>>>> functionStatsBinary();
 }
diff --git a/java/client/src/test/java/glide/api/GlideClientTest.java b/java/client/src/test/java/glide/api/GlideClientTest.java
@@ -11485,18 +11485,21 @@ public void functionKill_returns_success() {
  public void functionStats_returns_success() {
  // setup
  String[] args = new String[0];
- Map<String, Map<String, Object>> value = Map.of("1", Map.of("2", 2));
- CompletableFuture<Map<String, Map<String, Object>>> testResponse = new CompletableFuture<>();
+ Map<String, Map<String, Map<String, Object>>> value =
+ Map.of("::1", Map.of("1", Map.of("2", 2)));
+ CompletableFuture<Map<String, Map<String, Map<String, Object>>>> testResponse =
+ new CompletableFuture<>();
  testResponse.complete(value);
 
  // match on protobuf request
- when(commandManager.<Map<String, Map<String, Object>>>submitNewCommand(
+ when(commandManager.<Map<String, Map<String, Map<String, Object>>>>submitNewCommand(
  eq(FunctionStats), eq(args), any()))
  .thenReturn(testResponse);
 
  // exercise
- CompletableFuture<Map<String, Map<String, Object>>> response = service.functionStats();
- Map<String, Map<String, Object>> payload = response.get();
+ CompletableFuture<Map<String, Map<String, Map<String, Object>>>> response =
+ service.functionStats();
+ Map<String, Map<String, Map<String, Object>>> payload = response.get();
 
  // verify
  assertEquals(testResponse, response);
@@ -11508,20 +11511,21 @@ public void functionStats_returns_success() {
  public void functionStatsBinary_returns_success() {
  // setup
  GlideString[] args = new GlideString[0];
- Map<GlideString, Map<GlideString, Object>> value = Map.of(gs("1"), Map.of(gs("2"), 2));
- CompletableFuture<Map<GlideString, Map<GlideString, Object>>> testResponse =
+ Map<String, Map<GlideString, Map<GlideString, Object>>> value =
+ Map.of("::1", Map.of(gs("1"), Map.of(gs("2"), 2)));
+ CompletableFuture<Map<String, Map<GlideString, Map<GlideString, Object>>>> testResponse =
  new CompletableFuture<>();
  testResponse.complete(value);
 
  // match on protobuf request
- when(commandManager.<Map<GlideString, Map<GlideString, Object>>>submitNewCommand(
+ when(commandManager.<Map<String, Map<GlideString, Map<GlideString, Object>>>>submitNewCommand(
  eq(FunctionStats), eq(args), any()))
  .thenReturn(testResponse);
 
  // exercise
- CompletableFuture<Map<GlideString, Map<GlideString, Object>>> response =
+ CompletableFuture<Map<String, Map<GlideString, Map<GlideString, Object>>>> response =
  service.functionStatsBinary();
- Map<GlideString, Map<GlideString, Object>> payload = response.get();
+ Map<String, Map<GlideString, Map<GlideString, Object>>> payload = response.get();
 
  // verify
  assertEquals(testResponse, response);