diff --git a/.github/scripts/common.sh b/.github/scripts/common.sh index f0732f1c..dd59f768 100644 --- a/.github/scripts/common.sh +++ b/.github/scripts/common.sh @@ -474,16 +474,18 @@ function bootstrap_eloqkv() { # [extra...] function run_scenario() { local store_type=$1 log_name=$2 build_type=$3 evicted=$4 wal=$5 data_store=$6 + local admin_port=6380 shift 6 - launch_eloqkv "${store_type}" "${log_name}" 6379 "${wal}" "${data_store}" "$@" + launch_eloqkv "${store_type}" "${log_name}" 6379 "${wal}" "${data_store}" \ + --admin_port="${admin_port}" "$@" wait_until_ready echo "Redis server is ready!" # See run_cluster_scenario: tear down from a trap so a failing suite still # releases the port. trap stop_single_node RETURN - run_tcl_tests all "${build_type}" false "${evicted}" + run_tcl_tests all "${build_type}" false "${evicted}" "${admin_port}" } # Stop the node started by the enclosing run_scenario. @@ -579,6 +581,7 @@ function run_tcl_tests() { fault_inject="" fi local evicted=${4:-false} + local admin_port=${5:-0} local no_evicted="--tags -needs:no_evicted" if [[ $evicted = "false" ]]; then no_evicted="" @@ -593,6 +596,7 @@ function run_tcl_tests() { tclsh tests/test_helper.tcl \ --host 127.0.0.1 \ --port 6379 \ + --admin-port ${admin_port} \ --tags -needs:repl \ --tags -needs:config-maxmemory \ --tags -needs:debug \ diff --git a/docs/02-command-processing.md b/docs/02-command-processing.md index 71e4afdb..7512951c 100644 --- a/docs/02-command-processing.md +++ b/docs/02-command-processing.md @@ -48,8 +48,12 @@ service layer only; engine internals are in `data_substrate/docs/` (esp. `02-thr mode it exits the process right here after table creation (`src/redis_service.cpp:704-721`). 6. brpc `Server::Start()` with `server_options.redis_service = redis_service_impl` (ownership transfers to the server), the public listener declared Redis-only, `redis_max_connections` set - from `maxclients`, and optional force-SSL settings; then `RunUntilAskedToQuit()` - (`src/redis_server.cpp:502-548`). + from `maxclients`, and optional force-SSL settings. When `admin_port` is nonzero, a second + Redis-only `brpc::Server` starts on the same bind address with an independent + `admin_maxclients` limit. Its non-owning service proxy forwards connection-context creation and + every command to the primary `RedisServiceImpl`, so command, authentication, namespace, and TLS + behavior remain shared. The primary server then waits in `RunUntilAskedToQuit()` + (`src/redis_server.cpp`). ## 2. Request path end-to-end @@ -111,6 +115,16 @@ changes it for subsequent accepts. Lowering the limit does not disconnect establ The runtime value is not written back to disk; a restart loads `maxclients` from the `[local]` section of `eloqkv.ini` (or its gflag override) again. +An optional administrative listener (`admin_port`, disabled by default) has its own Acceptor and +fixed `admin_maxclients` admission limit. Reaching the primary `maxclients` therefore does not +consume administrative connection slots. `CONFIG SET maxclients` deliberately continues to +update only the primary listener, preserving the independent escape path. Both listeners share +the same process file-descriptor limit and bthread/engine resources: deployments must leave FD +headroom above `maxclients + admin_maxclients`, and the second listener does not guarantee access +after process-wide FD, memory, CPU, or scheduler exhaustion. The administrative port inherits the +primary bind address, authentication, and force-TLS configuration and exposes the same command +dispatcher; protect it with host/network access controls. + One `RedisConnectionContext` per admitted socket, created by `NewConnectionContext` (`src/redis_service.cpp:5917`) and owned by brpc's per-socket parsing context. Key fields (`include/redis_connection_context.h:61-170`): diff --git a/eloqkv.ini b/eloqkv.ini index d0214373..ecbe3f4a 100644 --- a/eloqkv.ini +++ b/eloqkv.ini @@ -30,6 +30,17 @@ port = 6379 # again. maxclients = 500000 +# Optional second Redis listener for administrative access when the primary +# listener has reached maxclients. It accepts the same commands and inherits +# the primary listener's bind address, authentication, and TLS configuration. +# Keep this port protected by the host firewall. A value of 0 disables it. +admin_port = 0 + +# Independent connection limit for the administrative listener. Ensure the +# process open-file limit leaves room for these connections and internal file +# descriptors after the primary listener reaches maxclients. +admin_maxclients = 16 + # EloqKV data directory. By default, the transaction service data, log service data, and local key-value storage data # are all stored in the eloq_data_path directory. eloq_data_path = eloq_data diff --git a/src/redis_server.cpp b/src/redis_server.cpp index 4d4f3428..eb73a0cb 100644 --- a/src/redis_server.cpp +++ b/src/redis_server.cpp @@ -25,6 +25,8 @@ #include #include +#include +#include #include #include @@ -43,6 +45,12 @@ constexpr char VERSION[] = "1.3.2"; // EloqKV flags - these are converted to tx flags DEFINE_string(ip, "127.0.0.1", "Redis IP"); DEFINE_int32(port, 6379, "Redis Port"); +DEFINE_int32(admin_port, + 0, + "Administrative Redis port; 0 disables the admin listener"); +DEFINE_uint32(admin_maxclients, + 16, + "Maximum connections accepted by the admin Redis listener"); DEFINE_string(ip_port_list, "", "redis server cluster ip port list"); DEFINE_string(standby_ip_port_list, "", @@ -63,6 +71,80 @@ DEFINE_string(voter_ip_port_list, extern brpc::Acceptor *EloqKV::server_acceptor; extern std::string EloqKV::redis_ip_port; +namespace +{ +/** + * A non-owning Redis service view for the administrative listener. + * + * brpc::Server owns and deletes ServerOptions::redis_service, so the primary + * RedisServiceImpl cannot be installed in two Server instances directly. The + * primary Server owns the implementation; the administrative Server owns this + * proxy and is always stopped and destroyed first. + */ +class RedisServiceProxy final : public brpc::RedisService +{ +public: + explicit RedisServiceProxy(EloqKV::RedisServiceImpl *service) + : service_(service) + { + } + + std::unique_ptr NewConnectionContext( + brpc::Socket *socket) const override + { + return service_->NewConnectionContext(socket); + } + + brpc::RedisCommandHandlerResult DispatchCommand( + brpc::ConnectionContext *ctx, + const std::vector &args, + brpc::RedisReply *output, + bool flush_batched) const override + { + return service_->DispatchCommand(ctx, args, output, flush_batched); + } + +private: + EloqKV::RedisServiceImpl *service_; +}; + +void ConfigureRedisListener(brpc::ServerOptions *options, + EloqKV::RedisServiceImpl *redis_service, + size_t max_connections, + const char *listener_name) +{ + std::string n_bthreads; + GFLAGS_NAMESPACE::GetCommandLineOption("bthread_concurrency", &n_bthreads); + options->num_threads = std::stoi(n_bthreads); + options->has_builtin_services = false; + options->enabled_protocols = "redis"; + options->redis_max_connections = max_connections; + + if (!redis_service->IsTlsEnabled()) + { + return; + } + + options->force_ssl = true; + brpc::ServerSSLOptions *ssl_options = options->mutable_ssl_options(); + ssl_options->default_cert.certificate = redis_service->GetTlsCertFile(); + ssl_options->default_cert.private_key = redis_service->GetTlsKeyFile(); + + LOG(INFO) << "TLS enabled for " << listener_name + << " Redis listener. Certificate: " + << redis_service->GetTlsCertFile() + << ", Key: " << redis_service->GetTlsKeyFile(); +} + +std::string RedisListenAddress(uint32_t port) +{ + const auto &network_config = DataSubstrate::Instance().GetNetworkConfig(); + const std::string ip = + network_config.bind_all ? "0.0.0.0" : network_config.local_ip; + return ip + ":" + std::to_string(port); +} +} // namespace + void PrintHelloText() { std::cout << EloqKV::asscii_logo << std::endl; @@ -450,6 +532,33 @@ int main(int argc, char *argv[]) // Convert eloqkv flags to tx flags ConvertEloqkvFlagsToTxFlags(&config_reader); + const int64_t configured_admin_port = + IsEloqkvFlagSet("admin_port") + ? FLAGS_admin_port + : config_reader.GetInteger("local", "admin_port", FLAGS_admin_port); + const int64_t configured_admin_maxclients = + IsEloqkvFlagSet("admin_maxclients") + ? FLAGS_admin_maxclients + : config_reader.GetInteger( + "local", "admin_maxclients", FLAGS_admin_maxclients); + if (configured_admin_port < 0 || + configured_admin_port > std::numeric_limits::max()) + { + LOG(ERROR) << "admin_port must be between 0 and " + << std::numeric_limits::max(); + return -1; + } + if (configured_admin_maxclients <= 0 || + configured_admin_maxclients > std::numeric_limits::max()) + { + LOG(ERROR) << "admin_maxclients must be between 1 and " + << std::numeric_limits::max(); + return -1; + } + const uint32_t admin_port = static_cast(configured_admin_port); + const uint32_t admin_maxclients = + static_cast(configured_admin_maxclients); + // Step 1: Initialize DataSubstrate if (!DataSubstrate::Instance().Init(config_file)) { @@ -461,6 +570,9 @@ int main(int argc, char *argv[]) LOG(INFO) << "Starting EloqKV Server ..."; DataSubstrate::Instance().EnableEngine(txservice::TableEngine::EloqKv); brpc::Server server; + // Declared after the primary Server so that its non-owning Redis service + // proxy is destroyed before the primary Server deletes RedisServiceImpl. + brpc::Server admin_server; brpc::ServerOptions server_options; auto redis_service_impl = std::make_unique(config_file, VERSION); @@ -499,37 +611,26 @@ int main(int argc, char *argv[]) #endif return -1; } - std::string n_bthreads; - GFLAGS_NAMESPACE::GetCommandLineOption("bthread_concurrency", &n_bthreads); - server_options.num_threads = std::stoi(n_bthreads); + if (admin_port != 0 && admin_port == redis_service_ptr->GetRedisPort()) + { + LOG(ERROR) << "admin_port must differ from the primary Redis port"; + redis_service_ptr->Stop(); + DataSubstrate::Instance().Shutdown(); +#if BRPC_WITH_GLOG + google::ShutdownGoogleLogging(); +#endif + return -1; + } + // Notice: redis_service_impl will be deleted in server's destructor. server_options.redis_service = redis_service_impl.release(); - server_options.has_builtin_services = false; // This listener is exclusively RESP. Declaring it as such lets brpc // enforce maxclients immediately after accept, before any optional TLS // handshake, without applying the limit to EloqKV's other RPC servers. - server_options.enabled_protocols = "redis"; - server_options.redis_max_connections = - redis_service_ptr->MaxConnectionCount(); - - // Configure TLS if enabled - if (redis_service_ptr->IsTlsEnabled()) - { - server_options.force_ssl = true; - brpc::ServerSSLOptions *ssl_options = - server_options.mutable_ssl_options(); - - // Set server certificate and key (required when TLS is enabled) - // Validation in Init() ensures both files are provided - ssl_options->default_cert.certificate = - redis_service_ptr->GetTlsCertFile(); - ssl_options->default_cert.private_key = - redis_service_ptr->GetTlsKeyFile(); - - LOG(INFO) << "TLS enabled for brpc server. Certificate: " - << redis_service_ptr->GetTlsCertFile() - << ", Key: " << redis_service_ptr->GetTlsKeyFile(); - } + ConfigureRedisListener(&server_options, + redis_service_ptr, + redis_service_ptr->MaxConnectionCount(), + "primary"); if (server.Start(redis_ip_port.c_str(), &server_options) != 0) { @@ -541,6 +642,37 @@ int main(int argc, char *argv[]) #endif return -1; } + EloqKV::server_acceptor = server.GetAcceptor(); + + std::string admin_ip_port; + if (admin_port != 0) + { + admin_ip_port = RedisListenAddress(admin_port); + brpc::ServerOptions admin_server_options; + // The proxy exposes exactly the same service behavior while keeping + // ownership and connection admission independent between listeners. + admin_server_options.redis_service = + new RedisServiceProxy(redis_service_ptr); + ConfigureRedisListener(&admin_server_options, + redis_service_ptr, + admin_maxclients, + "administrative"); + if (admin_server.Start(admin_ip_port.c_str(), &admin_server_options) != + 0) + { + LOG(ERROR) << "Failed to start the administrative Redis listener " + << "on " << admin_ip_port; + server.Stop(0); + server.Join(); + EloqKV::server_acceptor = nullptr; + redis_service_ptr->Stop(); + DataSubstrate::Instance().Shutdown(); +#if BRPC_WITH_GLOG + google::ShutdownGoogleLogging(); +#endif + return -1; + } + } if (!FLAGS_alsologtostderr) { @@ -549,11 +681,28 @@ int main(int argc, char *argv[]) } LOG(INFO) << "==== EloqKV Server Started, listening on " << redis_ip_port << "===="; - - EloqKV::server_acceptor = server.GetAcceptor(); + if (admin_port != 0) + { + if (!FLAGS_alsologtostderr) + { + std::cout << "Administrative Redis listener started on " + << admin_ip_port << std::endl; + } + LOG(INFO) << "==== Administrative Redis listener started on " + << admin_ip_port << ", maxclients=" << admin_maxclients + << " ===="; + } server.RunUntilAskedToQuit(); + // Stop the proxy listener before stopping the shared RedisServiceImpl. + if (admin_server.IsRunning()) + { + admin_server.Stop(0); + admin_server.Join(); + } + EloqKV::server_acceptor = nullptr; + if (!FLAGS_alsologtostderr) { std::cout << "\nEloqKV Server Stopping..." << std::endl; diff --git a/tests/test_helper.tcl b/tests/test_helper.tcl index 4d4e5c77..e15e90c1 100644 --- a/tests/test_helper.tcl +++ b/tests/test_helper.tcl @@ -124,6 +124,7 @@ set ::next_test 0 set ::host 127.0.0.1 set ::port 6379; # port for external server +set ::admin_port 0; # optional administrative port for external server tests set ::baseport 21111; # initial port for spawned redis servers set ::portcount 8000; # we don't wanna use more than 10000 to avoid collision with cluster bus ports set ::traceleaks 0 @@ -645,6 +646,7 @@ proc print_help_screen {} { "--tls-module Run tests in TLS mode with Redis module." "--host Run tests against an external host." "--port TCP port to use against external host." + "--admin-port Administrative port of the external host, if enabled." "--baseport Initial port number for spawned redis servers." "--portcount Port range for spawned redis servers." "--singledb Use a single database, avoid SELECT." @@ -712,6 +714,9 @@ for {set j 0} {$j < [llength $argv]} {incr j} { } elseif {$opt eq {--port}} { set ::port $arg incr j + } elseif {$opt eq {--admin-port}} { + set ::admin_port $arg + incr j } elseif {$opt eq {--baseport}} { set ::baseport $arg incr j diff --git a/tests/unit/eloq/maxclients.tcl b/tests/unit/eloq/maxclients.tcl index 76fe5aa4..a132a166 100644 --- a/tests/unit/eloq/maxclients.tcl +++ b/tests/unit/eloq/maxclients.tcl @@ -28,4 +28,26 @@ start_server {tags {"maxclients network"}} { r config set maxclients $original } + + if {$::admin_port != 0} { + test {Admin listener remains available when primary maxclients is reached} { + set original [lindex [r config get maxclients] 1] + assert_equal {OK} [r config set maxclients 1] + + if {$::tls} { + set expected_rejection {*I/O error*} + } else { + set expected_rejection {*ERR max*reached*} + } + set rejected [catch {redis_deferring_client} rejection] + assert_equal {1} $rejected + assert_match $expected_rejection $rejection + + set admin [redis $::host $::admin_port 0 $::tls] + assert_equal {PONG} [$admin ping] + $admin close + + r config set maxclients $original + } + } }