refactor:熔断器流程调整

Author: chuntaojun

examples/discover.rs

@@ -17,12 +17,11 @@ use std::{collections::HashMap, sync::Arc, time::Duration};
 
 use polaris_rust::{core::{
     context::SDKContext,
-    model::{error::PolarisError, naming::Location},
+    model::{error::PolarisError, loadbalance::Criteria, naming::Location, router::RouteInfo},
 }, discovery::{
     api::{new_consumer_api_by_context, new_provider_api_by_context, ConsumerAPI, ProviderAPI},
     req::{
-        GetAllInstanceRequest, InstanceDeregisterRequest, InstanceRegisterRequest,
-        WatchInstanceRequest,
+        GetAllInstanceRequest, GetOneInstanceRequest, InstanceDeregisterRequest, InstanceRegisterRequest, WatchInstanceRequest
     },
 }, error, info};
 use tracing::level_filters::LevelFilter;
@@ -152,6 +151,30 @@ async fn main() -> Result<(), PolarisError> {
         }
     }
 
+    // 执行路由以及负载均衡能力
+    let mut route_info = RouteInfo::default();
+
+    let ret = consumer.get_one_instance(GetOneInstanceRequest{
+        flow_id: uuid::Uuid::new_v4().to_string(),
+        timeout: Duration::from_secs(10),
+        namespace: "rust-demo".to_string(),
+        service: "polaris-rust-provider".to_string(),
+        criteria: Criteria{
+            policy: "random".to_string(),
+            hash_key: "".to_string(),
+        },
+        route_info: route_info,
+    }).await;
+
+    match ret {
+        Err(err) => {
+            tracing::error!("get one instance fail: {}", err.to_string());
+        }
+        Ok(instance) => {
+            tracing::info!("get one instance: {:?}", instance);
+        }
+    }
+
     for _ in 0..120 {
         std::thread::sleep(Duration::from_secs(1));
     }

src/circuitbreaker/api.rs

@@ -13,18 +13,19 @@
 // CONDITIONS OF ANY KIND, either express or implied. See the License for the
 // specific language governing permissions and limitations under the License.
 
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 
 use crate::core::{
     flow::CircuitBreakerFlow,
     model::{
-        circuitbreaker::{CallAbortedError, CheckResult, Resource, ResourceStat},
+        circuitbreaker::{CallAbortedError, CheckResult, MethodResource, Resource, ResourceStat, RetStatus, ServiceResource},
         error::PolarisError,
     },
 };
 
 use super::req::{RequestContext, ResponseContext};
 
+/// CircuitBreakerAPI .
 #[async_trait::async_trait]
 pub trait CircuitBreakerAPI
 where
@@ -41,8 +42,11 @@ where
     ) -> Result<Arc<InvokeHandler>, PolarisError>;
 }
 
+/// InvokeHandler .
 pub struct InvokeHandler {
+    // req_ctx: 请求上下文
     req_ctx: RequestContext,
+    // flow: 熔断器流程
     flow: Arc<CircuitBreakerFlow>,
 }
 
@@ -53,14 +57,87 @@ impl InvokeHandler {
 
     /// acquire_permission 检查当前请求是否可放通
     async fn acquire_permission(&self) -> Result<(), CallAbortedError> {
-        Ok(())
+        let svc_res = ServiceResource::new_waith_caller(
+            self.req_ctx.caller_service.clone(), 
+        self.req_ctx.callee_service.clone());
+
+        match self.flow.check_resource(Resource::ServiceResource(svc_res)).await {
+            Ok(ret) => {
+                if ret.pass {
+                    Ok(())
+                } else {
+                    Err(CallAbortedError::new(ret.rule_name, ret.fallback_info))
+                }
+            },
+            Err(e) => {
+                // 内部异常，不触发垄断，但是需要记录
+                crate::error!("[circuitbreaker][invoke] check resource failed: {:?}", e);
+                Ok(())
+            },
+        }
     }
 
     async fn on_success(&self, rsp: ResponseContext) -> Result<(), PolarisError> {
-        Ok(())
+        let cost = rsp.duration.clone();
+        let mut code = -1 as i32;
+        let mut status = RetStatus::RetSuccess;
+
+        if let Some(r) = &self.req_ctx.result_to_code {
+            code = r.on_success(rsp.result.unwrap());
+        }
+        if let Some(e) = rsp.error {
+            let ret = e.downcast::<CallAbortedError>();
+            if ret.is_ok() {
+                status = RetStatus::RetReject;
+            }
+        }
+        self.common_report(cost, code, status).await
     }
 
     async fn on_error(&self, rsp: ResponseContext) -> Result<(), PolarisError> {
-        Ok(())
+        let cost = rsp.duration.clone();
+        let mut code = 0 as i32;
+        let mut status = RetStatus::RetUnknown;
+
+        if let Some(r) = &self.req_ctx.result_to_code {
+            code = r.on_success(rsp.result.unwrap());
+        }
+        self.common_report(cost, code, status).await
+    }
+
+    async fn common_report(&self, cost: Duration, code: i32, status: RetStatus) -> Result<(), PolarisError> {
+        let stat = ResourceStat {
+            resource: Resource::ServiceResource(ServiceResource::new_waith_caller(
+                self.req_ctx.caller_service.clone(), 
+            self.req_ctx.callee_service.clone())),
+            ret_code: code.to_string(),
+            delay: cost,
+            status: status.clone(),
+        };
+
+        let ret = self.flow.report_stat(stat).await;
+        if ret.is_err() {
+            crate::error!("[circuitbreaker][invoke] report stat failed");
+            return ret;
+        }
+
+        if self.req_ctx.path.is_empty() {
+            return Ok(());
+        }
+
+        // 补充一个接口级别的数据上报
+        let stat = ResourceStat {
+            resource: Resource::MethodResource(MethodResource::new_waith_caller(
+                self.req_ctx.caller_service.clone(),
+                self.req_ctx.callee_service.clone(),
+                self.req_ctx.protocol.clone(),
+                self.req_ctx.method.clone(),
+                self.req_ctx.path.clone(),
+            )),
+            ret_code: code.to_string(),
+            delay: cost,
+            status,
+        };
+        self.flow.report_stat(stat).await
     }
 }

src/circuitbreaker/req.rs

@@ -21,8 +21,8 @@ pub trait ResultToErrorCode
 where
     Self: Send + Sync,
 {
-    fn on_success(&self, ret: dyn Any) -> i32;
-    fn on_error(&self, err: dyn Any) -> i32;
+    fn on_success(&self, ret: Box<dyn Any>) -> i32;
+    fn on_error(&self, err: Box<dyn Any>) -> i32;
 }
 
 pub struct RequestContext {
@@ -31,7 +31,7 @@ pub struct RequestContext {
     pub protocol: String,
     pub method: String,
     pub path: String,
-    pub result_to_code: Box<dyn ResultToErrorCode>,
+    pub result_to_code: Option<Box<dyn ResultToErrorCode>>,
 }
 
 pub struct ResponseContext {

src/core/context.rs

@@ -14,10 +14,8 @@
 // specific language governing permissions and limitations under the License.
 
 use std::sync::Arc;
-use tracing::log::{log, logger};
 use crate::core::config::config::{load_default, Configuration};
 use crate::core::engine::Engine;
-use crate::core::logger;
 use crate::core::model::error::{ErrorCode, PolarisError};
 use crate::info;
 

src/core/engine.rs

@@ -21,13 +21,10 @@ use tokio::runtime::{Builder, Runtime};
 use super::flow::{CircuitBreakerFlow, ClientFlow, RouterFlow};
 use super::model::config::{ConfigFile, ConfigGroup};
 use super::model::naming::{ServiceContractRequest, ServiceInstances};
-use super::model::router::RouteInfo;
 use super::model::ClientContext;
 use super::plugin::cache::{Filter, ResourceCache, ResourceListener};
 use super::plugin::connector::Connector;
-use super::plugin::loadbalance::LoadBalancer;
 use super::plugin::location::{LocationProvider, LocationSupplier};
-use super::plugin::router::RouteContext;
 use crate::config::req::{
     CreateConfigFileRequest, GetConfigFileRequest, GetConfigGroupRequest, PublishConfigFileRequest,
     UpdateConfigFileRequest, UpsertAndPublishConfigFileRequest,

src/core/flow.rs

@@ -144,6 +144,7 @@ impl CircuitBreakerFlow {
     }
 }
 
+/// RouterFlow 路由流程
 pub struct RouterFlow {
     load_balancer: Arc<RwLock<HashMap<String, Arc<Box<dyn LoadBalancer>>>>>,
     extensions: Arc<Extensions>,
@@ -230,6 +231,15 @@ impl RouterFlow {
     }
 }
 
-pub struct ReatelimitFlow {
+/// RatelimitFlow 限流流程
+pub struct RatelimitFlow {
     extensions: Arc<Extensions>,
 }
+
+impl RatelimitFlow {
+    pub fn new(extensions: Arc<Extensions>) -> Self {
+        Self {
+            extensions,
+        }
+    }
+}

src/core/model/circuitbreaker.rs

@@ -16,10 +16,10 @@
 use std::{
     fmt::{self, format, Display},
     iter::Map,
-    str,
+    str, time::Duration,
 };
 
-use super::error::{ErrorCode, PolarisError};
+use super::{error::{ErrorCode, PolarisError}, naming::ServiceKey};
 
 #[derive(Debug, PartialEq, Eq, Clone)]
 pub enum Status {
@@ -43,6 +43,7 @@ pub struct CircuitBreakerStatus {
     pub destroy: bool,
 }
 
+#[derive(Debug, PartialEq, Eq, Clone)]
 pub enum RetStatus {
     RetUnknown,
     RetSuccess,
@@ -55,7 +56,7 @@ pub enum RetStatus {
 pub struct ResourceStat {
     pub resource: Resource,
     pub ret_code: String,
-    pub delay: u32,
+    pub delay: Duration,
     pub status: RetStatus,
 }
 
@@ -65,13 +66,44 @@ pub enum Resource {
     InstanceResource(InstanceResource),
 }
 
-pub struct ServiceResource {}
+/// ServiceResource 服务资源
+pub struct ServiceResource {
+    pub callee: ServiceKey,
+    pub caller: Option<ServiceKey>,
+}
+
+impl ServiceResource {
+
+    pub fn new(callee: ServiceKey) -> Self {
+        ServiceResource { caller: None, callee }
+    }
 
-impl ServiceResource {}
+    pub fn new_waith_caller(caller: ServiceKey, callee: ServiceKey) -> Self {
+        ServiceResource { caller: Some(caller), callee }
+    }
 
-pub struct MethodResource {}
+}
 
-impl MethodResource {}
+/// MethodResource 方法资源
+pub struct MethodResource {
+    pub callee: ServiceKey,
+    pub caller: Option<ServiceKey>,
+    pub protocol: String,
+    pub method: String,
+    pub path: String,
+}
+
+impl MethodResource {
+
+    pub fn new(callee: ServiceKey, protocol: String, method: String, path: String) -> Self {
+        MethodResource { caller: None, callee, protocol, method, path }
+    }
+
+    pub fn new_waith_caller(caller: ServiceKey, callee: ServiceKey, protocol: String, method: String, path: String) -> Self {
+        MethodResource { caller: Some(caller), callee, protocol, method, path }
+    }
+
+}
 
 pub struct InstanceResource {}
 

src/discovery/api.rs

@@ -122,6 +122,7 @@ where
     async fn report_service_call(&self, req: ServiceCallResult);
 }
 
+/// new_lossless_api 创建优雅上下线客户端实例
 pub(crate) fn new_lossless_api() -> Result<impl LosslessAPI, PolarisError> {
     let context_ret = SDKContext::default();
     if context_ret.is_err() {
@@ -131,12 +132,14 @@ pub(crate) fn new_lossless_api() -> Result<impl LosslessAPI, PolarisError> {
     Ok(DefaultLosslessAPI::new(context_ret.unwrap()))
 }
 
+/// new_lossless_api_by_context 创建优雅上下线客户端实例
 pub(crate) fn new_lossless_api_by_context(
     context: SDKContext,
 ) -> Result<Arc<dyn LosslessAPI>, PolarisError> {
     Ok(Arc::new(DefaultLosslessAPI::new(context)))
 }
 
+/// LosslessAPI 负责优雅上下线客户端的生命周期管理
 pub(crate) trait LosslessAPI {
     fn set_action_provider(
         &self,

src/discovery/default.rs

@@ -22,7 +22,7 @@ use tokio::task::JoinHandle;
 
 use crate::core::context::SDKContext;
 use crate::core::model::error::PolarisError;
-use crate::core::model::naming::ServiceInstancesChangeEvent;
+use crate::core::model::naming::{ServiceInstancesChangeEvent, ServiceKey};
 use crate::core::plugin::cache::ResourceListener;
 use crate::discovery::api::{ConsumerAPI, LosslessAPI, ProviderAPI};
 use crate::discovery::req::{
@@ -176,6 +176,13 @@ impl ConsumerAPI for DefaultConsumerAPI {
             )
             .await;
 
+        // 重新设置被调服务数据信息
+        let mut route_info = req.route_info;
+        route_info.callee  = ServiceKey{
+            namespace: req.namespace.clone(),
+            name: req.service.clone(),
+        };
+
         match rsp {
             Ok(rsp) => {
                 let instances = rsp.instances;
@@ -186,7 +193,7 @@ impl ConsumerAPI for DefaultConsumerAPI {
                     .router_api
                     .router(ProcessRouteRequest {
                         service_instances: instances,
-                        route_info: req.route_info,
+                        route_info: route_info,
                     })
                     .await;