From 2aa0b70035d461f272644a7e3f6b14c98da9d66a Mon Sep 17 00:00:00 2001 From: Shahar Mike Date: Sun, 4 Aug 2024 12:35:14 +0300 Subject: [PATCH] feat(server): Support `replica-announce-ip`/`port` (#3421) * feat: Support `replica-announce-ip`/`port` Before this PR, we only supported `cluster_announce_ip`. It's basically the same feature, but used for cluster announcements instead of replication. This PR adds support for `replica-announce-ip` and `replica-announce-port`, which can be set via new flags `--announce_ip=` and `--announce_port=`. These flags apply to both cluster and replica announcements. Tested via running Sentinel, and making sure it is able to connect to announced ip+port, while it can't connect to announced false / unavailable ip+port. Note: this PR deprecates `--cluster_announce_ip`, but continues to support it. We will remove it in a future version. Fixes #3380 * fix failing test * destructure --- README.ja-JP.md | 2 +- README.ko-KR.md | 2 +- README.md | 3 ++- README.zh-CN.md | 2 +- src/server/cluster/cluster_family.cc | 25 ++++++++++++++++++----- src/server/cluster/cluster_family_test.cc | 2 +- src/server/conn_context.h | 1 + src/server/dflycmd.cc | 2 +- src/server/main_service.cc | 5 +++++ src/server/replica.cc | 13 +++++++++++- src/server/server_family.cc | 7 +++++++ tests/dragonfly/cluster_test.py | 8 +++++--- tests/dragonfly/replication_test.py | 21 +++++++++++++++++++ 13 files changed, 78 insertions(+), 15 deletions(-) diff --git a/README.ja-JP.md b/README.ja-JP.md index fe03b0e1a..cd68b0b73 100644 --- a/README.ja-JP.md +++ b/README.ja-JP.md @@ -113,7 +113,7 @@ Dragonfly 特有の議論もある: * `admin_bind`: 管理コンソールの TCP 接続を指定されたアドレスにバインドする(`default: any`)。HTTP と RESP の両方のプロトコルをサポートする。 * `admin_nopass`: 割り当てられたポートで、認証トークンなしでコンソールへのオープン管理アクセスを有効にする (`default: false`)。HTTP と RESP の両方のプロトコルをサポートする。 * `cluster_mode`: サポートするクラスターモード (`default: ""`)。現在は `emulated` のみをサポートしている。 - * `cluster_announce_ip`: クラスタコマンドがクライアントにアナウンスする IP。 + * `announce_ip`: クラスタコマンドがクライアントにアナウンスする IP。 ### 一般的なオプションを使用した開始スクリプトの例: diff --git a/README.ko-KR.md b/README.ko-KR.md index 4b88806c0..dda61fc9d 100644 --- a/README.ko-KR.md +++ b/README.ko-KR.md @@ -111,7 +111,7 @@ Dragonfly는 현재 아래와 같은 Redis 인수들을 지원합니다 : * `admin_bind`: 주어진 주소에 관리자 콘솔 TCP 연결을 바인딩합니다. (`기본값: any`). HTTP와 RESP 프로토콜 모두를 지원합니다. * `admin_nopass`: 할당된 포트에 대해서 인증 토큰 없이 관리자 콘솔 접근을 활성화합니다. (`default: false`). HTTP와 RESP 프로토콜 모두를 지원합니다. * `cluster_mode`: 클러스터 모드가 지원됩니다. (`기본값: ""`). 현재는`emulated` 만 지원합니다. - * `cluster_announce_ip`: 클러스터 명령을 클라이언트에게 알리는 IP 주소. + * `announce_ip`: 클러스터 명령을 클라이언트에게 알리는 IP 주소. ### 주요 옵션을 활용한 실행 스크립트 예시: diff --git a/README.md b/README.md index 84f68a6ee..f5844606f 100644 --- a/README.md +++ b/README.md @@ -166,7 +166,8 @@ There are also some Dragonfly-specific arguments: * `admin_bind`: To bind the admin console TCP connection to a given address (`default: any`). Supports both HTTP and RESP protocols. * `admin_nopass`: To enable open admin access to console on the assigned port, without auth token needed (`default: false`). Supports both HTTP and RESP protocols. * `cluster_mode`: Cluster mode supported (`default: ""`). Currently supports only `emulated`. - * `cluster_announce_ip`: The IP that cluster commands announce to the client. + * `announce_ip`: The IP that cluster commands announce to the client, and to replication master. + * `announce_port`: The port that cluster commands announce to the client, and to replication master. ### Example start script with popular options: diff --git a/README.zh-CN.md b/README.zh-CN.md index 05e2b3073..fdfea228b 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -135,7 +135,7 @@ Dragonfly 支持 Redis 的常见参数。 * `cluster_mode`：支持集群模式。目前仅支持 `emulated`。默认为空 `""`。 -* `cluster_announce_ip`：集群模式下向客户端公开的 IP。 +* `announce_ip`：集群模式下向客户端公开的 IP。 ### 启动脚本示例，包含常用选项： diff --git a/src/server/cluster/cluster_family.cc b/src/server/cluster/cluster_family.cc index 6d58f8398..616a637b3 100644 --- a/src/server/cluster/cluster_family.cc +++ b/src/server/cluster/cluster_family.cc @@ -25,12 +25,15 @@ #include "server/server_family.h" #include "server/server_state.h" -ABSL_FLAG(std::string, cluster_announce_ip, "", "ip that cluster commands announce to the client"); +ABSL_FLAG(std::string, cluster_announce_ip, "", "DEPRECATED: use --announce_ip"); + ABSL_FLAG(std::string, cluster_node_id, "", "ID within a cluster, used for slot assignment. MUST be unique. If empty, uses master " "replication ID (random string)"); ABSL_DECLARE_FLAG(int32_t, port); +ABSL_DECLARE_FLAG(std::string, announce_ip); +ABSL_DECLARE_FLAG(uint16_t, announce_port); namespace dfly { namespace acl { @@ -66,6 +69,16 @@ ClusterFamily::ClusterFamily(ServerFamily* server_family) : server_family_(serve InitializeCluster(); + // TODO: Remove flag cluster_announce_ip in v1.23+ + if (!absl::GetFlag(FLAGS_cluster_announce_ip).empty()) { + CHECK(absl::GetFlag(FLAGS_announce_ip).empty()) + << "Can't use both --cluster_announce_ip and --announce_ip"; + + LOG(WARNING) << "WARNING: Flag --cluster_announce_ip is deprecated in favor of --announce_ip. " + "Use the latter, as the former will be removed in a future release."; + absl::SetFlag(&FLAGS_announce_ip, absl::GetFlag(FLAGS_cluster_announce_ip)); + } + id_ = absl::GetFlag(FLAGS_cluster_node_id); if (id_.empty()) { id_ = server_family_->master_replid(); @@ -104,13 +117,15 @@ ClusterShardInfo ClusterFamily::GetEmulatedShardInfo(ConnectionContext* cntx) co ServerState& etl = *ServerState::tlocal(); if (!replication_info.has_value()) { DCHECK(etl.is_master); - std::string cluster_announce_ip = absl::GetFlag(FLAGS_cluster_announce_ip); + std::string cluster_announce_ip = absl::GetFlag(FLAGS_announce_ip); std::string preferred_endpoint = cluster_announce_ip.empty() ? cntx->conn()->LocalBindAddress() : cluster_announce_ip; + uint16_t cluster_announce_port = absl::GetFlag(FLAGS_announce_port); + uint16_t preferred_port = cluster_announce_port == 0 + ? static_cast(absl::GetFlag(FLAGS_port)) + : cluster_announce_port; - info.master = {.id = id_, - .ip = preferred_endpoint, - .port = static_cast(absl::GetFlag(FLAGS_port))}; + info.master = {.id = id_, .ip = preferred_endpoint, .port = preferred_port}; for (const auto& replica : server_family_->GetDflyCmd()->GetReplicasRoleInfo()) { info.replicas.push_back({.id = replica.id, diff --git a/src/server/cluster/cluster_family_test.cc b/src/server/cluster/cluster_family_test.cc index d523bb5ae..feaa1d6d5 100644 --- a/src/server/cluster/cluster_family_test.cc +++ b/src/server/cluster/cluster_family_test.cc @@ -696,7 +696,7 @@ class ClusterFamilyEmulatedTest : public ClusterFamilyTest { public: ClusterFamilyEmulatedTest() { SetTestFlag("cluster_mode", "emulated"); - SetTestFlag("cluster_announce_ip", "fake-host"); + SetTestFlag("announce_ip", "fake-host"); } }; diff --git a/src/server/conn_context.h b/src/server/conn_context.h index a61c513fd..11acf9852 100644 --- a/src/server/conn_context.h +++ b/src/server/conn_context.h @@ -131,6 +131,7 @@ struct ConnectionState { // then it holds positive sync session id. uint32_t repl_session_id = 0; uint32_t repl_flow_id = UINT32_MAX; + std::string repl_ip_address; uint32_t repl_listening_port = 0; DflyVersion repl_version = DflyVersion::VER0; }; diff --git a/src/server/dflycmd.cc b/src/server/dflycmd.cc index 968901b25..a973736c5 100644 --- a/src/server/dflycmd.cc +++ b/src/server/dflycmd.cc @@ -563,7 +563,7 @@ auto DflyCmd::CreateSyncSession(ConnectionContext* cntx) fb2::Fiber("stop_replication", &DflyCmd::StopReplication, this, sync_id).Detach(); }; - string address = cntx->conn()->RemoteEndpointAddress(); + string address = cntx->conn_state.replication_info.repl_ip_address; uint32_t port = cntx->conn_state.replication_info.repl_listening_port; LOG(INFO) << "Registered replica " << address << ":" << port; diff --git a/src/server/main_service.cc b/src/server/main_service.cc index 72e59fa5c..f335d24f8 100644 --- a/src/server/main_service.cc +++ b/src/server/main_service.cc @@ -67,6 +67,11 @@ using facade::ErrorReply; ABSL_FLAG(int32_t, port, 6379, "Redis port. 0 disables the port, -1 will bind on a random available port."); +ABSL_FLAG(std::string, announce_ip, "", + "IP address that Dragonfly announces to cluster clients and replication master"); +ABSL_FLAG(uint16_t, announce_port, 0, + "Port that Dragonfly announces to cluster clients and replication master"); + ABSL_FLAG(uint32_t, memcached_port, 0, "Memcached port"); ABSL_FLAG(uint32_t, num_shards, 0, "Number of database shards, 0 - to choose automatically"); diff --git a/src/server/replica.cc b/src/server/replica.cc index f0df153ed..4e13d077f 100644 --- a/src/server/replica.cc +++ b/src/server/replica.cc @@ -42,6 +42,8 @@ ABSL_FLAG(bool, break_replication_on_master_restart, false, "When in replica mode, and master restarts, break replication from master to avoid " "flushing the replica's data."); ABSL_DECLARE_FLAG(int32_t, port); +ABSL_DECLARE_FLAG(uint16_t, announce_port); +ABSL_DECLARE_FLAG(std::string, announce_ip); ABSL_FLAG( int, replica_priority, 100, "Published by info command for sentinel to pick replica based on score during a failover"); @@ -266,10 +268,19 @@ error_code Replica::Greet() { PC_RETURN_ON_BAD_RESPONSE(CheckRespIsSimpleReply("PONG")); // Corresponds to server.repl_state == REPL_STATE_SEND_HANDSHAKE condition in replication.c - auto port = absl::GetFlag(FLAGS_port); + uint16_t port = absl::GetFlag(FLAGS_announce_port); + if (port == 0) { + port = static_cast(absl::GetFlag(FLAGS_port)); + } RETURN_ON_ERR(SendCommandAndReadResponse(StrCat("REPLCONF listening-port ", port))); PC_RETURN_ON_BAD_RESPONSE(CheckRespIsSimpleReply("OK")); + auto announce_ip = absl::GetFlag(FLAGS_announce_ip); + if (!announce_ip.empty()) { + RETURN_ON_ERR(SendCommandAndReadResponse(StrCat("REPLCONF ip-address ", announce_ip))); + PC_RETURN_ON_BAD_RESPONSE(CheckRespIsSimpleReply("OK")); + } + // Corresponds to server.repl_state == REPL_STATE_SEND_CAPA RETURN_ON_ERR(SendCommandAndReadResponse("REPLCONF capa eof capa psync2")); PC_RETURN_ON_BAD_RESPONSE(CheckRespIsSimpleReply("OK")); diff --git a/src/server/server_family.cc b/src/server/server_family.cc index 97d17ec87..d511ff869 100644 --- a/src/server/server_family.cc +++ b/src/server/server_family.cc @@ -2695,6 +2695,13 @@ void ServerFamily::ReplConf(CmdArgList args, ConnectionContext* cntx) { return; } cntx->conn_state.replication_info.repl_listening_port = replica_listening_port; + // We set a default value of ip_address here, because LISTENING-PORT is a mandatory field + // but IP-ADDRESS is optional + if (cntx->conn_state.replication_info.repl_ip_address.empty()) { + cntx->conn_state.replication_info.repl_ip_address = cntx->conn()->RemoteEndpointAddress(); + } + } else if (cmd == "IP-ADDRESS") { + cntx->conn_state.replication_info.repl_ip_address = arg; } else if (cmd == "CLIENT-ID" && args.size() == 2) { auto info = dfly_cmd_->GetReplicaInfo(cntx); DCHECK(info != nullptr); diff --git a/tests/dragonfly/cluster_test.py b/tests/dragonfly/cluster_test.py index 22febc1f0..082b3fb57 100644 --- a/tests/dragonfly/cluster_test.py +++ b/tests/dragonfly/cluster_test.py @@ -203,7 +203,9 @@ class TestEmulated: assert val == [True, "bar"] -@dfly_args({"cluster_mode": "emulated", "cluster_announce_ip": "127.0.0.2"}) +# Unfortunately we can't test --announce_port here because that causes the Python Cluster client to +# throw if it can't access the port in `CLUSTER SLOTS` :| +@dfly_args({"cluster_mode": "emulated", "announce_ip": "127.0.0.2"}) class TestEmulatedWithAnnounceIp: def test_cluster_slots_command(self, df_server, cluster_client: redis.RedisCluster): expected = {(0, 16383): {"primary": ("127.0.0.2", df_server.port), "replicas": []}} @@ -327,7 +329,7 @@ async def test_emulated_cluster_with_replicas(df_factory): await close_clients(c_master, *c_replicas) -@dfly_args({"cluster_mode": "emulated", "cluster_announce_ip": "127.0.0.2"}) +@dfly_args({"cluster_mode": "emulated"}) async def test_cluster_info(async_client): res = await async_client.execute_command("CLUSTER INFO") assert len(res) == 16 @@ -351,7 +353,7 @@ async def test_cluster_info(async_client): } -@dfly_args({"cluster_mode": "emulated", "cluster_announce_ip": "127.0.0.2"}) +@dfly_args({"cluster_mode": "emulated", "announce_ip": "127.0.0.2"}) @pytest.mark.asyncio async def test_cluster_nodes(df_server, async_client): res = await async_client.execute_command("CLUSTER NODES") diff --git a/tests/dragonfly/replication_test.py b/tests/dragonfly/replication_test.py index 5f9bb1191..7fd977077 100644 --- a/tests/dragonfly/replication_test.py +++ b/tests/dragonfly/replication_test.py @@ -2213,3 +2213,24 @@ async def test_replica_reconnect(df_factory, break_conn): assert await c_replica.execute_command("get k") == "6789" await disconnect_clients(c_master, c_replica) + + +@pytest.mark.asyncio +async def test_announce_ip_port(df_factory): + master = df_factory.create() + replica = df_factory.create(announce_ip="overrode-host", announce_port="1337") + + master.start() + replica.start() + + # Connect clients, connect replica to master + c_master = master.client() + c_replica = replica.client() + await c_replica.execute_command(f"REPLICAOF localhost {master.port}") + await wait_available_async(c_replica) + + role, node = await c_master.execute_command("role") + assert role == "master" + host, port, _ = node[0] + assert host == "overrode-host" + assert port == "1337"