Merge pull request apache#3222 from wayslog/feat/redis-cluster-channel

feat(redis): add native Redis Cluster channel support

Author: Huixxi

docs/cn/redis_client.md

@@ -161,6 +161,55 @@ response中的所有reply的ownership属于response。当response析构时，rep
 
 或者你可以沿用常见的[twemproxy](https://github.com/twitter/twemproxy)方案。这个方案虽然需要额外部署proxy，还增加了延时，但client端仍可以像访问单点一样的访问它。
 
+如果你要直接访问原生 Redis Cluster（按 slot 路由、自动处理 MOVED/ASK，并通过 `CLUSTER SLOTS`/`CLUSTER NODES` 刷新拓扑），可以使用 `brpc::RedisClusterChannel`：
+
+```c++
+#include <brpc/redis_cluster.h>
+
+brpc::RedisClusterChannel channel;
+brpc::RedisClusterChannelOptions options;
+options.max_redirect = 5;
+if (channel.Init("127.0.0.1:7000,127.0.0.1:7001", &options) != 0) {
+    LOG(ERROR) << "Fail to init redis cluster channel";
+}
+```
+
+`RedisClusterChannel` 支持同步/异步 `CallMethod`、自动重定向重试和周期性拓扑刷新。多 key 支持 `MGET/MSET/DEL/EXISTS/UNLINK/EVAL/EVALSHA`，暂不支持 `MULTI/EXEC`。
+
+## RedisClusterChannel 示例
+
+`example/redis_c++/redis_cluster_client.cpp` 覆盖了以下常见场景：
+
+- 通过多个 seed 节点初始化；
+- 自动处理 MOVED/ASK 重定向与重试；
+- 先走 `CLUSTER SLOTS`，失败后回退 `CLUSTER NODES`；
+- 同一个 channel 同时执行同步 pipeline 和异步请求。
+
+构建并运行：
+
+```bash
+cd example/redis_c++
+make redis_cluster_client
+./redis_cluster_client \
+  --seeds=127.0.0.1:7000,127.0.0.1:7001 \
+  --max_redirect=5 \
+  --timeout_ms=1000
+```
+
+常用选项：
+
+- `RedisClusterChannelOptions::max_redirect`：单个命令的最大重定向次数；
+- `RedisClusterChannelOptions::refresh_interval_s`：周期刷新拓扑的间隔；
+- `RedisClusterChannelOptions::topology_refresh_timeout_ms`：拓扑命令超时；
+- `RedisClusterChannelOptions::channel_options`：每个 redis 节点的通用 brpc 参数；
+- `RedisClusterChannelOptions::enable_periodic_refresh`：是否启用后台周期刷新。
+
+说明：
+
+- `MGET/MSET/DEL/EXISTS/UNLINK` 会按 key 分发后按请求顺序合并返回；
+- `EVAL/EVALSHA` 要求声明的 key 位于同一 slot；
+- `MULTI/EXEC` 当前会直接返回错误 reply。
+
 # 查看发出的请求和收到的回复
 
  打开[-redis_verbose](http://brpc.baidu.com:8765/flags/redis_verbose)即看到所有的redis request和response，注意这应该只用于线下调试，而不是线上程序。
@@ -242,6 +291,8 @@ TRACE: 02-13 18:07:42:   * 0 client.cpp:180] Accessing redis server at qps=75238
 
 [example/redis_c++/redis_cli](https://github.com/apache/brpc/blob/master/example/redis_c%2B%2B/redis_cli.cpp)是一个类似于官方CLI的命令行工具，以展示brpc对redis协议的处理能力。当使用brpc访问redis-server出现不符合预期的行为时，也可以使用这个CLI进行交互式的调试。
 
+如果是原生 Redis Cluster 场景，可直接参考 [example/redis_c++/redis_cluster_client.cpp](https://github.com/apache/brpc/blob/master/example/redis_c%2B%2B/redis_cluster_client.cpp)。
+
 和官方CLI类似，`redis_cli <command>`也可以直接运行命令，-server参数可以指定redis-server的地址。
 
 ```

docs/en/redis_client.md

@@ -162,6 +162,55 @@ Create a `Channel` using the consistent hashing as the load balancing algorithm(
 
 Another choice is to use the common [twemproxy](https://github.com/twitter/twemproxy) solution, which makes clients access the cluster just like accessing a single server, although the solution needs to deploy proxies and adds more latency.
 
+For native Redis Cluster (slot based routing, MOVED/ASK redirection and topology refresh from `CLUSTER SLOTS`/`CLUSTER NODES`), use `brpc::RedisClusterChannel`:
+
+```c++
+#include <brpc/redis_cluster.h>
+
+brpc::RedisClusterChannel channel;
+brpc::RedisClusterChannelOptions options;
+options.max_redirect = 5;
+if (channel.Init("127.0.0.1:7000,127.0.0.1:7001", &options) != 0) {
+    LOG(ERROR) << "Fail to init redis cluster channel";
+}
+```
+
+`RedisClusterChannel` supports synchronous/asynchronous `CallMethod`, automatic redirection retries and periodic topology refresh. Multi-key support includes `MGET/MSET/DEL/EXISTS/UNLINK/EVAL/EVALSHA`. `MULTI/EXEC` is currently not supported.
+
+## RedisClusterChannel example
+
+`example/redis_c++/redis_cluster_client.cpp` demonstrates:
+
+- bootstrap from multiple seed nodes.
+- MOVED/ASK auto-redirection and retry.
+- topology refresh from `CLUSTER SLOTS` with `CLUSTER NODES` fallback.
+- sync pipeline and async calls using one channel.
+
+Build and run:
+
+```bash
+cd example/redis_c++
+make redis_cluster_client
+./redis_cluster_client \
+  --seeds=127.0.0.1:7000,127.0.0.1:7001 \
+  --max_redirect=5 \
+  --timeout_ms=1000
+```
+
+Frequently used options:
+
+- `RedisClusterChannelOptions::max_redirect`: max redirects per command.
+- `RedisClusterChannelOptions::refresh_interval_s`: interval of periodic topology refresh.
+- `RedisClusterChannelOptions::topology_refresh_timeout_ms`: timeout for topology commands.
+- `RedisClusterChannelOptions::channel_options`: normal brpc channel options for each redis node.
+- `RedisClusterChannelOptions::enable_periodic_refresh`: disable this when your app controls refresh explicitly.
+
+Notes:
+
+- `MGET/MSET/DEL/EXISTS/UNLINK` are executed per key and merged in request order.
+- `EVAL/EVALSHA` requires all declared keys to be in one slot.
+- `MULTI/EXEC` returns an error reply by design.
+
 # Debug
 
 Turn on [-redis_verbose](http://brpc.baidu.com:8765/flags/redis_verbose) to print contents of all redis requests and responses. Note that this should only be used for debugging rather than online services.
@@ -243,6 +292,8 @@ We can see a tremendous drop of QPS compared to the one using single connection
 
 [example/redis_c++/redis_cli](https://github.com/apache/brpc/blob/master/example/redis_c%2B%2B/redis_cli.cpp) is a command line tool similar to the official CLI, demostrating brpc's capability to talk with redis servers. When unexpected results are got from a redis-server using a brpc client, you can debug with this tool interactively as well.
 
+For native Redis Cluster, you can start from [example/redis_c++/redis_cluster_client.cpp](https://github.com/apache/brpc/blob/master/example/redis_c%2B%2B/redis_cluster_client.cpp).
+
 Like the official CLI, `redis_cli <command>` runs the command directly, and `-server` which is address of the redis-server can be specified.
 
 ```

example/redis_c++/CMakeLists.txt

@@ -138,9 +138,11 @@ endif()
 add_executable(redis_cli redis_cli.cpp)
 add_executable(redis_press redis_press.cpp)
 add_executable(redis_server redis_server.cpp)
+add_executable(redis_cluster_client redis_cluster_client.cpp)
 
 set(AUX_LIB readline ncurses)
 
 target_link_libraries(redis_cli ${BRPC_LIB} ${DYNAMIC_LIB} ${AUX_LIB})
 target_link_libraries(redis_press ${BRPC_LIB} ${DYNAMIC_LIB})
 target_link_libraries(redis_server ${BRPC_LIB} ${DYNAMIC_LIB})
+target_link_libraries(redis_cluster_client ${BRPC_LIB} ${DYNAMIC_LIB})

example/redis_c++/Makefile

@@ -29,10 +29,12 @@ DYNAMIC_LINKINGS += -lreadline -lncurses
 PRESS_SOURCES = redis_press.cpp
 CLI_SOURCES = redis_cli.cpp
 SERVER_SOURCES = redis_server.cpp
+CLUSTER_SOURCES = redis_cluster_client.cpp
 
 PRESS_OBJS = $(addsuffix .o, $(basename $(PRESS_SOURCES))) 
 CLI_OBJS = $(addsuffix .o, $(basename $(CLI_SOURCES))) 
 SERVER_OBJS = $(addsuffix .o, $(basename $(SERVER_SOURCES))) 
+CLUSTER_OBJS = $(addsuffix .o, $(basename $(CLUSTER_SOURCES)))
 
 ifeq ($(SYSTEM),Darwin)
  ifneq ("$(LINK_SO)", "")
@@ -50,12 +52,13 @@ else ifeq ($(SYSTEM),Linux)
 endif
 
 .PHONY:all
-all: redis_press redis_cli redis_server
+all: redis_press redis_cli redis_server redis_cluster_client
 
 .PHONY:clean
 clean:
 	@echo "> Cleaning"
-	rm -rf redis_press redis_cli $(PRESS_OBJS) $(CLI_OBJS) $(SERVER_OBJS)
+	rm -rf redis_press redis_cli redis_server redis_cluster_client \
+        $(PRESS_OBJS) $(CLI_OBJS) $(SERVER_OBJS) $(CLUSTER_OBJS)
 
 redis_press:$(PRESS_OBJS)
 	@echo "> Linking $@"
@@ -81,6 +84,14 @@ else
 	$(CXX) $(LIBPATHS) $(LINK_OPTIONS) -o $@
 endif
 
+redis_cluster_client:$(CLUSTER_OBJS)
+	@echo "> Linking $@"
+ifneq ("$(LINK_SO)", "")
+	$(CXX) $(LIBPATHS) $(SOPATHS) $(LINK_OPTIONS_SO) -o $@
+else
+	$(CXX) $(LIBPATHS) $(LINK_OPTIONS) -o $@
+endif
+
 %.o:%.cpp
 	@echo "> Compiling $@"
 	$(CXX) -c $(HDRPATHS) $(CXXFLAGS) $< -o $@

example/redis_c++/redis_cluster_client.cpp

@@ -0,0 +1,119 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// A basic client for native Redis Cluster using brpc::RedisClusterChannel.
+
+#include <gflags/gflags.h>
+#include <bthread/countdown_event.h>
+#include <butil/logging.h>
+#include <brpc/controller.h>
+#include <brpc/redis.h>
+#include <brpc/redis_cluster.h>
+
+DEFINE_string(seeds, "127.0.0.1:7000,127.0.0.1:7001",
+              "Comma-separated redis cluster seed endpoints");
+DEFINE_string(key_prefix, "brpc_cluster_demo", "Prefix for demo keys");
+DEFINE_int32(timeout_ms, 1000, "RPC timeout in milliseconds");
+DEFINE_int32(rpc_max_retry, 1, "Max retries for a single sub RPC");
+DEFINE_int32(max_redirect, 5, "Max MOVED/ASK redirect retries");
+DEFINE_int32(refresh_interval_s, 30, "Periodic topology refresh interval");
+DEFINE_int32(topology_refresh_timeout_ms, 1000,
+             "Timeout of CLUSTER SLOTS/NODES request");
+DEFINE_bool(disable_periodic_refresh, false, "Disable periodic topology refresh");
+
+namespace {
+
+class Done : public google::protobuf::Closure {
+public:
+    explicit Done(bthread::CountdownEvent* event) : _event(event) {}
+    void Run() override { _event->signal(); }
+
+private:
+    bthread::CountdownEvent* _event;
+};
+
+int PrintResponse(const brpc::RedisResponse& response) {
+    for (int i = 0; i < response.reply_size(); ++i) {
+        const brpc::RedisReply& reply = response.reply(i);
+        if (reply.is_error()) {
+            LOG(ERROR) << "reply[" << i << "] error=" << reply.error_message();
+            return -1;
+        }
+        LOG(INFO) << "reply[" << i << "] " << reply;
+    }
+    return 0;
+}
+
+}  // namespace
+
+int main(int argc, char* argv[]) {
+    GFLAGS_NAMESPACE::ParseCommandLineFlags(&argc, &argv, true);
+
+    brpc::RedisClusterChannelOptions options;
+    options.max_redirect = FLAGS_max_redirect;
+    options.refresh_interval_s = FLAGS_refresh_interval_s;
+    options.enable_periodic_refresh = !FLAGS_disable_periodic_refresh;
+    options.topology_refresh_timeout_ms = FLAGS_topology_refresh_timeout_ms;
+    options.channel_options.timeout_ms = FLAGS_timeout_ms;
+    options.channel_options.max_retry = FLAGS_rpc_max_retry;
+
+    brpc::RedisClusterChannel channel;
+    if (channel.Init(FLAGS_seeds, &options) != 0) {
+        LOG(ERROR) << "Fail to init redis cluster channel, seeds=" << FLAGS_seeds;
+        return -1;
+    }
+
+    const std::string key1 = FLAGS_key_prefix + "_1";
+    const std::string key2 = FLAGS_key_prefix + "_2";
+
+    // Sync pipeline.
+    brpc::RedisRequest request;
+    brpc::RedisResponse response;
+    brpc::Controller cntl;
+    CHECK(request.AddCommand("set %s v1", key1.c_str()));
+    CHECK(request.AddCommand("set %s v2", key2.c_str()));
+    CHECK(request.AddCommand("mget %s %s", key1.c_str(), key2.c_str()));
+    channel.CallMethod(NULL, &cntl, &request, &response, NULL);
+    if (cntl.Failed()) {
+        LOG(ERROR) << "Sync call failed: " << cntl.ErrorText();
+        return -1;
+    }
+    if (PrintResponse(response) != 0) {
+        return -1;
+    }
+
+    // Async single request.
+    brpc::RedisRequest async_request;
+    brpc::RedisResponse async_response;
+    brpc::Controller async_cntl;
+    CHECK(async_request.AddCommand("get %s", key1.c_str()));
+
+    bthread::CountdownEvent event(1);
+    Done done(&event);
+    channel.CallMethod(NULL, &async_cntl, &async_request, &async_response, &done);
+    event.wait();
+    if (async_cntl.Failed()) {
+        LOG(ERROR) << "Async call failed: " << async_cntl.ErrorText();
+        return -1;
+    }
+    if (PrintResponse(async_response) != 0) {
+        return -1;
+    }
+
+    LOG(INFO) << "Redis cluster demo finished";
+    return 0;
+}