feat: refine meta election logic (risingwavelabs#7669)

This PR refine a series of meta election related codes

1. meta's addr supports multiple addresses
2. meta client adds a meta address mode parameter to distinguish the behavior of election members found in loadbalance (kubernetes environment) and list (normal environment)

Approved-By: yezizp2012

proto/meta.proto

@@ -255,17 +255,6 @@ service NotificationService {
   rpc Subscribe(SubscribeRequest) returns (stream SubscribeResponse);
 }
 
-message MetaLeaderInfo {
-  string node_address = 1;
-  uint64 lease_id = 2;
-}
-
-message MetaLeaseInfo {
-  MetaLeaderInfo leader = 1;
-  uint64 lease_register_time = 2;
-  uint64 lease_expire_time = 3;
-}
-
 message PauseRequest {}
 
 message PauseResponse {}
@@ -303,3 +292,18 @@ service ScaleService {
   rpc GetClusterInfo(GetClusterInfoRequest) returns (GetClusterInfoResponse);
   rpc Reschedule(RescheduleRequest) returns (RescheduleResponse);
 }
+
+message MembersRequest {}
+
+message MetaMember {
+  common.HostAddress address = 1;
+  bool is_leader = 2;
+}
+
+message MembersResponse {
+  repeated MetaMember members = 1;
+}
+
+service MetaMemberService {
+  rpc Members(MembersRequest) returns (MembersResponse);
+}

src/common/src/util/addr.rs

@@ -16,16 +16,9 @@ use std::net::SocketAddr;
 use std::str::FromStr;
 
 use risingwave_pb::common::HostAddress as ProstHostAddress;
-use risingwave_pb::meta::MetaLeaderInfo;
 
 use crate::error::{internal_error, Result};
 
-pub fn leader_info_to_host_addr(mli: MetaLeaderInfo) -> HostAddr {
-    mli.node_address
-        .parse::<HostAddr>()
-        .expect("invalid leader addr")
-}
-
 /// General host address and port.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct HostAddr {

src/meta/src/hummock/manager/mod.rs

@@ -46,7 +46,6 @@ use risingwave_pb::hummock::{
     HummockVersionDelta, HummockVersionDeltas, HummockVersionStats, IntraLevelDelta, LevelType,
 };
 use risingwave_pb::meta::subscribe_response::{Info, Operation};
-use risingwave_pb::meta::MetaLeaderInfo;
 use tokio::sync::oneshot::Sender;
 use tokio::sync::{Notify, RwLockReadGuard, RwLockWriteGuard};
 use tokio::task::JoinHandle;
@@ -128,7 +127,7 @@ macro_rules! commit_multi_var {
                     $val_txn.apply_to_txn(&mut trx)?;
                 )*
                 // Commit to state store
-                $hummock_mgr.commit_trx($hummock_mgr.env.meta_store(), trx, $context_id, $hummock_mgr.env.get_leader_info())
+                $hummock_mgr.commit_trx($hummock_mgr.env.meta_store(), trx, $context_id)
                 .await?;
                 // Upon successful commit, commit the change to local in-mem state
                 $(
@@ -491,7 +490,6 @@ where
         meta_store: &S,
         trx: Transaction,
         context_id: Option<HummockContextId>,
-        _info: MetaLeaderInfo,
     ) -> Result<()> {
         if let Some(context_id) = context_id {
             if context_id == META_NODE_ID {