feat:控制台注册中心服务订阅页面集群场景支持查询指定节点下的订阅者信息; #231

Author: heqingpan

src/common/model/privilege.rs

@@ -178,7 +178,7 @@ where
     }
 }
 
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct NamespacePrivilegeGroup(pub PrivilegeGroup<Arc<String>>);
 
 impl NamespacePrivilegeGroup {

src/console/api.rs

@@ -284,6 +284,10 @@ pub fn console_api_config_v2(config: &mut web::ServiceConfig) {
                 web::resource("/service/list")
                     .route(web::get().to(v2::naming_api::query_service_list)),
             )
+            .service(
+                web::resource("/service/subscriber/list")
+                    .route(web::get().to(v2::naming_api::query_subscribers_list)),
+            )
             .service(
                 web::resource("/service/add").route(web::post().to(v2::naming_api::add_service)),
             )

src/console/model/naming_model.rs

@@ -23,14 +23,16 @@ pub struct ServiceKeyParam{
 }
 */
 
-#[derive(Debug, Serialize, Deserialize, Default)]
+#[derive(Debug, Serialize, Deserialize, Default, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct ServiceQueryListRequest {
     pub page_no: Option<usize>,
     pub page_size: Option<usize>,
     pub namespace_id: Option<String>,
     pub group_name_param: Option<String>,
     pub service_name_param: Option<String>,
+    /// 节点id,在查询指定节点信息时用到
+    pub node_id: Option<u64>,
 }
 
 impl ServiceQueryListRequest {

src/console/v2/naming_api.rs

@@ -4,9 +4,14 @@ use crate::console::model::naming_model::{
     InstanceParams, ServiceDto, ServiceParam, ServiceQueryListRequest,
 };
 use crate::console::v2::ERROR_CODE_SYSTEM_ERROR;
+use crate::grpc::handler::NAMING_ROUTE_REQUEST;
+use crate::grpc::PayloadUtils;
 use crate::naming::api_model::InstanceVO;
+use crate::naming::cluster::model::{NamingRouteRequest, NamingRouterResponse};
 use crate::naming::core::{NamingActor, NamingCmd, NamingResult};
 use crate::naming::model::{InstanceUpdateTag, ServiceDetailDto};
+use crate::naming::service::SubscriberInfoDto;
+use crate::naming::service_index::ServiceQueryParam;
 use crate::naming::NamingUtils;
 use crate::{user_namespace_privilege, user_no_namespace_permission};
 use actix::Addr;
@@ -295,3 +300,59 @@ pub async fn remove_instance(
         )),
     }
 }
+
+pub async fn query_subscribers_list(
+    app: Data<Arc<AppShareData>>,
+    req: HttpRequest,
+    param: web::Query<ServiceQueryListRequest>,
+) -> impl Responder {
+    let node_id = param.node_id.unwrap_or(0);
+    let service_param = param.0.to_param(&req).unwrap();
+    if !service_param
+        .namespace_privilege
+        .check_option_value_permission(&service_param.namespace_id, true)
+    {
+        user_no_namespace_permission!(&service_param.namespace_id);
+    }
+    match do_query_subscribers_list(app, service_param, node_id).await {
+        Ok(res) => HttpResponse::Ok().json(ApiResult::success(Some(res))),
+        Err(err) => HttpResponse::Ok().json(ApiResult::<()>::error(
+            ERROR_CODE_SYSTEM_ERROR.to_string(),
+            Some(err.to_string()),
+        )),
+    }
+}
+
+async fn do_query_subscribers_list(
+    app: Data<Arc<AppShareData>>,
+    param: ServiceQueryParam,
+    node_id: u64,
+) -> anyhow::Result<PageResult<SubscriberInfoDto>> {
+    if node_id == 0 || node_id == app.sys_config.raft_node_id {
+        if let NamingResult::ServiceSubscribersPage((total_count, list)) = app
+            .naming_addr
+            .send(NamingCmd::QueryServiceSubscribersPageV2(param))
+            .await??
+        {
+            Ok(PageResult { total_count, list })
+        } else {
+            Err(anyhow::anyhow!("query subscribers error"))
+        }
+    } else {
+        let addr = app.naming_node_manage.get_node_addr(node_id).await?;
+        let req = NamingRouteRequest::QueryServiceSubscriberPage(param);
+        let request = serde_json::to_string(&req).unwrap_or_default();
+        let payload = PayloadUtils::build_payload(NAMING_ROUTE_REQUEST, request);
+        let resp_payload = app
+            .cluster_sender
+            .send_request(addr.clone(), payload)
+            .await?;
+        let body_vec = resp_payload.body.unwrap_or_default().value;
+        let resp: NamingRouterResponse = serde_json::from_slice(&body_vec)?;
+        if let NamingRouterResponse::ServiceSubscribersPage((total_count, list)) = resp {
+            Ok(PageResult { total_count, list })
+        } else {
+            Err(anyhow::anyhow!("query subscribers error"))
+        }
+    }
+}

src/grpc/handler/naming_route.rs

@@ -186,6 +186,7 @@ impl NamingRouteRequestHandler {
                 }
                 args.merge_args(tmp_args);
             }
+            NamingRouteRequest::QueryServiceSubscriberPage(param) => {}
         }
         Ok(args.to_string())
     }

src/naming/cluster/mod.rs

@@ -224,6 +224,15 @@ pub async fn handle_naming_route(
                 }
             }
         }
+        NamingRouteRequest::QueryServiceSubscriberPage(param) => {
+            let resp = app
+                .naming_addr
+                .send(NamingCmd::QueryServiceSubscribersPageV2(param))
+                .await??;
+            if let NamingResult::ServiceSubscribersPage((total, list)) = resp {
+                return Ok(NamingRouterResponse::ServiceSubscribersPage((total, list)));
+            }
+        }
     };
     Ok(NamingRouterResponse::None)
 }

src/naming/cluster/model.rs

@@ -1,5 +1,7 @@
 use crate::metrics::timeline::model::{TimelineQueryParam, TimelineQueryResponse};
 use crate::naming::model::{Instance, InstanceKey, InstanceUpdateTag, ServiceDetailDto};
+use crate::naming::service::SubscriberInfoDto;
+use crate::naming::service_index::ServiceQueryParam;
 use actix::prelude::*;
 use serde::{Deserialize, Serialize};
 use std::collections::{HashMap, HashSet};
@@ -47,6 +49,7 @@ pub enum NamingRouteRequest {
     MetricsTimelineQuery(TimelineQueryParam),
     SyncDistroClientInstances(HashMap<Arc<String>, HashSet<InstanceKey>>),
     QueryDistroInstanceSnapshot(Vec<InstanceKey>),
+    QueryServiceSubscriberPage(ServiceQueryParam),
 }
 
 impl NamingRouteRequest {
@@ -65,6 +68,7 @@ impl NamingRouteRequest {
             NamingRouteRequest::MetricsTimelineQuery(_) => "MetricsTimelineQuery",
             NamingRouteRequest::SyncDistroClientInstances(_) => "SyncDistroClientInstances",
             NamingRouteRequest::QueryDistroInstanceSnapshot(_) => "QueryDistroInstanceSnapshot",
+            NamingRouteRequest::QueryServiceSubscriberPage(_) => "QueryServiceSubscriberPage",
         }
     }
 }
@@ -73,6 +77,7 @@ impl NamingRouteRequest {
 pub enum NamingRouterResponse {
     None,
     MetricsTimeLineResponse(TimelineQueryResponse),
+    ServiceSubscribersPage((usize, Vec<SubscriberInfoDto>)),
 }
 
 #[derive(Message, Debug, Clone)]

src/naming/core.rs

@@ -50,11 +50,13 @@ use std::collections::BTreeSet;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::collections::LinkedList;
+use std::default::Default;
 use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::Duration;
 
 use crate::common::constant::EMPTY_ARC_STRING;
+use crate::common::model::privilege::NamespacePrivilegeGroup;
 use crate::metrics::metrics_key::MetricsKey;
 use crate::metrics::model::{MetricsItem, MetricsQuery, MetricsRecord};
 use crate::namespace::NamespaceActor;
@@ -689,6 +691,8 @@ impl NamingActor {
         (size, service_names)
     }
 
+    ///过程生成的不必要的临时对象过多,使用v2版本
+    #[deprecated]
     pub fn get_subscribers_list(
         &self,
         page_size: usize,
@@ -740,6 +744,35 @@ impl NamingActor {
         (total, paginated_result)
     }
 
+    pub fn get_subscribers_list_v2(
+        &self,
+        page_size: usize,
+        page_index: usize,
+        key: &ServiceKey,
+    ) -> (usize, Vec<SubscriberInfoDto>) {
+        let offset = if page_index == 0 {
+            0
+        } else {
+            page_size * (page_index - 1)
+        };
+        let param = ServiceQueryParam {
+            offset,
+            limit: page_size,
+            namespace_id: Some(key.namespace_id.clone()),
+            like_group: Some(key.group_name.as_ref().clone()),
+            like_service: Some(key.service_name.as_ref().clone()),
+            ..Default::default()
+        };
+        self.subscriber.query_service_listener_page(&param)
+    }
+
+    pub fn get_subscribers_list_by_param(
+        &self,
+        param: ServiceQueryParam,
+    ) -> (usize, Vec<SubscriberInfoDto>) {
+        self.subscriber.query_service_listener_page(&param)
+    }
+
     pub fn get_service_info_page(&self, param: ServiceQueryParam) -> (usize, Vec<ServiceInfoDto>) {
         let (size, list) = self.namespace_index.query_service_page(&param);
 
@@ -1026,6 +1059,7 @@ pub enum NamingCmd {
     QueryServiceInfo(ServiceKey, String, bool),
     QueryServicePage(ServiceKey, usize, usize),
     QueryServiceSubscribersPage(ServiceKey, usize, usize),
+    QueryServiceSubscribersPageV2(ServiceQueryParam),
     //查询服务实际信息列表
     QueryServiceInfoPage(ServiceQueryParam),
     //CreateService(ServiceDetailDto),
@@ -1056,7 +1090,7 @@ pub enum NamingResult {
     InstanceListString(String),
     ServiceInfo(ServiceInfo),
     ServicePage((usize, Vec<Arc<String>>)),
-    ServiceSubscribersPage((usize, Vec<Arc<SubscriberInfoDto>>)),
+    ServiceSubscribersPage((usize, Vec<SubscriberInfoDto>)),
     ServiceInfoPage((usize, Vec<ServiceInfoDto>)),
     ClientInstanceCount(Vec<(Arc<String>, usize)>),
     RewriteToCluster(u64, Instance),
@@ -1158,12 +1192,15 @@ impl Handler<NamingCmd> for NamingActor {
             }
             NamingCmd::QueryServiceSubscribersPage(service_key, page_size, page_index) => {
                 Ok(NamingResult::ServiceSubscribersPage(
-                    self.get_subscribers_list(page_size, page_index, &service_key),
+                    self.get_subscribers_list_v2(page_size, page_index, &service_key),
                 ))
             }
             NamingCmd::QueryServiceInfoPage(param) => Ok(NamingResult::ServiceInfoPage(
                 self.get_service_info_page(param),
             )),
+            NamingCmd::QueryServiceSubscribersPageV2(param) => Ok(
+                NamingResult::ServiceSubscribersPage(self.get_subscribers_list_by_param(param)),
+            ),
             NamingCmd::PeekListenerTimeout => {
                 self.time_check();
                 //self.notify_check();

src/naming/model.rs

@@ -202,7 +202,7 @@ impl Default for InstanceUpdateTag {
     }
 }
 
-#[derive(Debug, Clone, Default, Hash, PartialEq, Eq)]
+#[derive(Debug, Clone, Default, Hash, PartialEq, Eq, PartialOrd, Ord)]
 pub struct ServiceKey {
     pub namespace_id: Arc<String>,
     pub group_name: Arc<String>,

src/naming/naming_subscriber.rs

@@ -1,16 +1,18 @@
 #![allow(unused_imports)]
 
-use std::{
-    collections::{HashMap, HashSet},
-    sync::Arc,
-};
-
-use actix::prelude::*;
-
 use super::{
     model::ServiceKey,
     naming_delay_nofity::{DelayNotifyActor, DelayNotifyCmd},
 };
+use crate::common::constant::EMPTY_ARC_STRING;
+use crate::naming::service::SubscriberInfoDto;
+use crate::naming::service_index::ServiceQueryParam;
+use actix::prelude::*;
+use std::collections::BTreeMap;
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
 
 #[derive(Debug, Clone, Hash, PartialEq, Eq)]
 pub enum ListenerClusterType {
@@ -40,7 +42,7 @@ pub struct NamingListenerItem {
 
 #[derive(Default)]
 pub struct Subscriber {
-    listener: HashMap<ServiceKey, HashMap<Arc<String>, Option<HashSet<String>>>>,
+    listener: BTreeMap<ServiceKey, BTreeMap<Arc<String>, Option<HashSet<String>>>>,
     client_keys: HashMap<Arc<String>, HashSet<ServiceKey>>,
     notify_addr: Option<Addr<DelayNotifyActor>>,
 }
@@ -79,7 +81,7 @@ impl Subscriber {
                     set.insert(client_id.clone(), item.clusters);
                 }
                 None => {
-                    let mut set = HashMap::new();
+                    let mut set = BTreeMap::new();
                     set.insert(client_id.clone(), item.clusters);
                     self.listener.insert(item.service_key, set);
                 }
@@ -191,7 +193,7 @@ impl Subscriber {
         group_name: &str,
         service_name: &str,
         namespace_id: &str,
-    ) -> HashMap<ServiceKey, HashMap<Arc<String>, Option<HashSet<String>>>> {
+    ) -> BTreeMap<ServiceKey, BTreeMap<Arc<String>, Option<HashSet<String>>>> {
         self.listener
             .iter()
             .filter(|(key, _)| {
@@ -202,4 +204,47 @@ impl Subscriber {
             .map(|(key, value)| (key.clone(), value.clone()))
             .collect()
     }
+
+    pub fn query_service_listener_page(
+        &self,
+        param: &ServiceQueryParam,
+    ) -> (usize, Vec<SubscriberInfoDto>) {
+        let mut rlist = vec![];
+        let end_index = param.offset + param.limit;
+        let mut index = 0;
+        for (key, value) in self.listener.iter() {
+            if param.match_namespace_id(&key.namespace_id)
+                && param.match_group(&key.group_name)
+                && param.match_service(&key.service_name)
+            {
+                for (ip_port, _) in value.iter() {
+                    if index >= param.offset && index < end_index {
+                        let (ip, port) = Self::parse_ip_port(ip_port.as_ref());
+                        rlist.push(SubscriberInfoDto {
+                            service_name: key.service_name.clone(),
+                            group_name: key.group_name.clone(),
+                            namespace_id: key.namespace_id.clone(),
+                            ip,
+                            port,
+                        });
+                    }
+                    index += 1;
+                }
+            }
+        }
+        (index, rlist)
+    }
+
+    fn parse_ip_port(ip_port: &str) -> (Arc<String>, u16) {
+        let parts: Vec<&str> = ip_port.split(':').collect();
+        let mut ip = EMPTY_ARC_STRING.clone();
+        let mut port = 0;
+        if parts.len() == 2 {
+            ip = Arc::new(parts[0].to_string());
+            if let Ok(p) = parts[1].parse::<u16>() {
+                port = p;
+            }
+        }
+        (ip, port)
+    }
 }