diff --git a/Cargo.toml b/Cargo.toml index 02c1e2f6..50666bc8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ members = [ "registry/zookeeper", "registry/nacos", "dubbo", + "dubbo-rs-core", "examples/echo", "examples/greeter", "dubbo-build", diff --git a/dubbo-rs-core/Cargo.toml b/dubbo-rs-core/Cargo.toml new file mode 100644 index 00000000..a9197d0e --- /dev/null +++ b/dubbo-rs-core/Cargo.toml @@ -0,0 +1,30 @@ +[package] +name = "dubbo-rs-core" +version = "0.1.0" +edition = "2021" +license = "Apache-2.0" +description = "Stable core facade for embedding Dubbo Rust in other runtimes" +repository = "https://github.com/apache/dubbo-rust.git" +readme = "README.md" +keywords = ["dubbo", "triple", "rpc", "napi"] +categories = ["network-programming", "api-bindings"] + +[dependencies] +bytes.workspace = true +dubbo = { path = "../dubbo", version = "0.4.0" } +dubbo-registry-nacos = { path = "../registry/nacos", version = "0.4.0", optional = true } +dubbo-registry-zookeeper = { path = "../registry/zookeeper", version = "0.4.0", optional = true } +http = "0.2" +tokio = { workspace = true, features = ["time"] } + +[features] +default = [] +registry-nacos = ["dep:dubbo-registry-nacos"] +registry-zookeeper = ["dep:dubbo-registry-zookeeper"] +registry = ["registry-nacos", "registry-zookeeper"] + +[dev-dependencies] +http-body = "0.4.4" +hyper = { version = "0.14.26", features = ["full"] } +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] } +tower-service.workspace = true diff --git a/dubbo-rs-core/README.md b/dubbo-rs-core/README.md new file mode 100644 index 00000000..4ba8aa4f --- /dev/null +++ b/dubbo-rs-core/README.md @@ -0,0 +1,89 @@ +# dubbo-rs-core + +`dubbo-rs-core` is a small, byte-oriented facade for embedding Dubbo Rust in +other runtimes. + +The crate is intentionally lower level than the generated Rust client APIs. It +keeps the host language in control of service descriptors, protobuf +serialization, and public API shape, while Rust owns Dubbo transport and +governance behavior such as Triple calls, registry discovery, load balancing, +cluster strategy, routing, and timeouts. + +This boundary is useful for packages such as `dubbo-js`, where JavaScript should +keep the user-facing client API while a N-API addon reuses the Rust core. + +## Current Surface + +- Raw unary Triple requests and responses. +- Static provider endpoints. +- Registry-backed clients behind optional features. +- Request-level and client-default unary timeouts. +- Load balancing: `random`, `round_robin`, and `p2c`; `random` and + `round_robin` honor provider `weight` URL parameters. +- Cluster strategy: `failfast` and `failover`, with configurable failover + retries and retry delay. +- Compression: `gzip` or `identity`. +- Metadata routing for tag, group, and version. +- Stable status code and message access for host-language error mapping. + +## Features + +- `registry-nacos`: enable Nacos registry support. +- `registry-zookeeper`: enable Zookeeper registry support. +- `registry`: enable all registry backends currently wired into this crate. + +## Publishing And Host Usage + +This crate is intended to be consumed by host-language bindings, such as a +Node.js N-API package. During local development those bindings can depend on the +crate by path. After publishing, switch them to a normal version dependency and +enable only the registry features they need: + +```toml +dubbo-rs-core = { version = "0.1", features = ["registry"] } +``` + +The crate keeps protobuf encoding and generated service shape outside the Rust +boundary. Host runtimes pass raw request bytes, metadata, and endpoint or +registry options into Rust, then decode raw response bytes themselves. + +## Example + +```rust +use bytes::Bytes; +use dubbo_rs_core::{ + RawMetadata, RawTripleClient, RawTripleClientOptions, RawUnaryRequest, +}; + +# async fn call() -> Result<(), Box> { +let mut client = RawTripleClient::from_static_endpoints_with_options( + ["http://127.0.0.1:50051?interface=example.EchoService"], + RawTripleClientOptions { + timeout_ms: Some(3_000), + load_balance: Some("round_robin".to_string()), + cluster: Some("failfast".to_string()), + failover_retries: None, + failover_retry_delay_ms: None, + compression: Some("gzip".to_string()), + ..RawTripleClientOptions::default() + }, +)?; + +let response = client + .unary(RawUnaryRequest { + service: "example.EchoService".to_string(), + method: "Echo".to_string(), + path: "/example.EchoService/Echo".to_string(), + metadata: RawMetadata::new().insert("tri-service-group", "default"), + body: Bytes::from_static(b"\x0a\x05hello"), + timeout_ms: None, + }) + .await?; + +let _bytes = response.body; +# Ok(()) +# } +``` + +The request and response bodies are protobuf bytes. Host runtimes should encode +and decode messages using their own protobuf toolchain. diff --git a/dubbo-rs-core/src/lib.rs b/dubbo-rs-core/src/lib.rs new file mode 100644 index 00000000..9f989931 --- /dev/null +++ b/dubbo-rs-core/src/lib.rs @@ -0,0 +1,400 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use std::time::Duration; + +use bytes::Bytes; +use dubbo::{ + cluster::ClusterStrategy, + codegen::{ClientBuilder, Request, RpcInvocation, TripleClient}, + invocation::Metadata, + loadbalancer::LoadBalanceStrategy, + triple::compression::CompressionEncoding, + Url, +}; +#[cfg(feature = "registry-nacos")] +use dubbo_registry_nacos::NacosRegistry; +#[cfg(feature = "registry-zookeeper")] +use dubbo_registry_zookeeper::ZookeeperRegistry; + +pub use dubbo::status::{Code, Status}; + +/// A stable raw Triple client facade for embedding Dubbo Rust in other runtimes. +/// +/// This is intentionally byte-oriented. Hosts such as Node.js should own their +/// protobuf message shape and use this facade for Dubbo transport and governance. +pub struct RawTripleClient { + inner: TripleClient, + default_timeout_ms: Option, + default_metadata: RawMetadata, +} + +impl RawTripleClient { + pub fn from_static(endpoint: &str) -> Self { + Self::from_static_endpoints([endpoint]).expect("static endpoint must be a valid Dubbo URL") + } + + pub fn from_static_endpoints<'a, I>(endpoints: I) -> Result + where + I: IntoIterator, + { + Self::from_static_endpoints_with_options(endpoints, RawTripleClientOptions::default()) + } + + pub fn from_static_endpoints_with_options<'a, I>( + endpoints: I, + options: RawTripleClientOptions, + ) -> Result + where + I: IntoIterator, + { + let endpoints = endpoints + .into_iter() + .map(|endpoint| { + endpoint.parse::().map_err(|err| { + Status::new( + Code::InvalidArgument, + format!("invalid static endpoint {endpoint}: {err}"), + ) + }) + }) + .collect::, _>>()?; + if endpoints.is_empty() { + return Err(Status::new( + Code::InvalidArgument, + "at least one static endpoint is required".to_string(), + )); + } + let builder = + options.apply(ClientBuilder::from_static_urls(endpoints).with_direct(true))?; + Ok(Self { + inner: TripleClient::new(builder), + default_timeout_ms: options.timeout_ms, + default_metadata: options.default_metadata, + }) + } + + pub async fn from_registry(registry_url: &str) -> Result { + Self::from_registry_with_options(registry_url, RawTripleClientOptions::default()).await + } + + pub async fn from_registry_with_options( + registry_url: &str, + options: RawTripleClientOptions, + ) -> Result { + let registry_url = registry_url.parse::().map_err(|err| { + Status::new( + Code::InvalidArgument, + format!("invalid registry URL {registry_url}: {err}"), + ) + })?; + + register_registry_extension(registry_url.protocol()).await?; + + let builder = options.apply(ClientBuilder::new().with_registry(registry_url))?; + Ok(Self { + inner: TripleClient::new(builder), + default_timeout_ms: options.timeout_ms, + default_metadata: options.default_metadata, + }) + } + + pub async fn unary( + &mut self, + request: RawUnaryRequest, + ) -> Result { + let path = request.path.parse().map_err(|err| { + dubbo::status::Status::new( + dubbo::status::Code::InvalidArgument, + format!("invalid request path: {err}"), + ) + })?; + let invocation = RpcInvocation::default() + .with_service_unique_name(request.service) + .with_method_name(request.method); + let timeout_ms = request.timeout_ms.or(self.default_timeout_ms); + let call = self.inner.raw_unary_with_trailers( + Request::from_parts( + self.default_metadata.clone().merge(request.metadata).into(), + request.body, + ), + path, + invocation, + ); + let response = if let Some(timeout_ms) = timeout_ms { + tokio::time::timeout(Duration::from_millis(timeout_ms), call) + .await + .map_err(|_| { + Status::new( + Code::DeadlineExceeded, + format!("request timed out after {timeout_ms}ms"), + ) + })?? + } else { + call.await? + }; + let (response, trailers) = response; + let (metadata, body) = response.into_parts(); + + Ok(RawUnaryResponse { + metadata: metadata.into(), + trailers: trailers.map(Into::into).unwrap_or_default(), + body, + }) + } +} + +#[derive(Debug, Clone, Default)] +pub struct RawTripleClientOptions { + pub timeout_ms: Option, + pub load_balance: Option, + pub cluster: Option, + pub failover_retries: Option, + pub failover_retry_delay_ms: Option, + pub compression: Option, + pub default_metadata: RawMetadata, +} + +impl RawTripleClientOptions { + fn apply(&self, builder: ClientBuilder) -> Result { + let builder = match self.load_balance.as_deref() { + Some(load_balance) => { + let strategy = LoadBalanceStrategy::parse(load_balance).ok_or_else(|| { + Status::new( + Code::InvalidArgument, + format!( + "unsupported load balance {load_balance}; expected random, round_robin, or p2c" + ), + ) + })?; + builder.with_load_balance(strategy) + } + None => builder, + }; + + let builder = match self.cluster.as_deref() { + Some(cluster) => { + let strategy = ClusterStrategy::parse(cluster).ok_or_else(|| { + Status::new( + Code::InvalidArgument, + format!("unsupported cluster {cluster}; expected failfast or failover"), + ) + })?; + builder.with_cluster(strategy) + } + None => builder, + }; + + let builder = match self.failover_retries { + Some(retries) => { + let attempts = usize::try_from(retries) + .ok() + .and_then(|retries| retries.checked_add(1)) + .ok_or_else(|| { + Status::new( + Code::InvalidArgument, + format!("unsupported failover retries {retries}"), + ) + })?; + builder.with_failover_attempts(attempts) + } + None => builder, + }; + + let builder = match self.failover_retry_delay_ms { + Some(delay_ms) => builder.with_failover_retry_delay(Duration::from_millis(delay_ms)), + None => builder, + }; + + let builder = match self.compression.as_deref() { + Some("gzip") => builder.with_compression(Some(CompressionEncoding::Gzip)), + Some("identity") | Some("none") => builder.with_compression(None), + Some(compression) => { + return Err(Status::new( + Code::InvalidArgument, + format!("unsupported compression {compression}; expected gzip or identity"), + )); + } + None => builder, + }; + + Ok(builder) + } +} + +async fn register_registry_extension(protocol: &str) -> Result<(), Status> { + match protocol { + #[cfg(feature = "registry-nacos")] + "nacos" => dubbo::extension::EXTENSIONS + .register::>() + .await + .map_err(registry_extension_error), + #[cfg(feature = "registry-zookeeper")] + "zookeeper" => dubbo::extension::EXTENSIONS + .register::>( + ) + .await + .map_err(registry_extension_error), + protocol => Err(Status::new( + Code::Unimplemented, + format!("registry protocol {protocol} is not enabled in dubbo-rs-core"), + )), + } +} + +#[cfg(any(feature = "registry-nacos", feature = "registry-zookeeper"))] +fn registry_extension_error(err: dubbo::StdError) -> Status { + Status::new( + Code::Internal, + format!("failed to register registry extension: {err}"), + ) +} + +#[derive(Debug, Clone)] +pub struct RawUnaryRequest { + pub service: String, + pub method: String, + pub path: String, + pub metadata: RawMetadata, + pub body: Bytes, + pub timeout_ms: Option, +} + +#[derive(Debug, Clone)] +pub struct RawUnaryResponse { + pub metadata: RawMetadata, + pub trailers: RawMetadata, + pub body: Bytes, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct RawMetadata { + pub entries: Vec<(String, String)>, +} + +impl RawMetadata { + pub fn new() -> Self { + Self::default() + } + + pub fn insert(mut self, key: impl Into, value: impl Into) -> Self { + self.entries.push((key.into(), value.into())); + self + } + + pub fn merge(mut self, other: RawMetadata) -> Self { + self.entries.extend(other.entries); + self + } + + pub fn get(&self, key: &str) -> Option<&String> { + self.entries + .iter() + .rev() + .find_map(|(entry_key, value)| (entry_key == key).then_some(value)) + } +} + +impl From for Metadata { + fn from(value: RawMetadata) -> Self { + value + .entries + .into_iter() + .fold(Metadata::default(), |metadata, (key, value)| { + metadata.insert(key, value) + }) + } +} + +impl From for RawMetadata { + fn from(value: Metadata) -> Self { + let entries = value + .into_headers() + .iter() + .filter_map(|(key, value)| { + value + .to_str() + .ok() + .map(|value| (key.to_string(), value.to_string())) + }) + .collect(); + + Self { entries } + } +} + +#[cfg(all( + test, + not(any(feature = "registry-nacos", feature = "registry-zookeeper")) +))] +mod tests { + use super::*; + + #[test] + fn static_client_rejects_unknown_load_balance() { + let result = RawTripleClient::from_static_endpoints_with_options( + ["http://127.0.0.1:50051?interface=example.Echo"], + RawTripleClientOptions { + load_balance: Some("least_active".to_string()), + ..RawTripleClientOptions::default() + }, + ); + match result { + Ok(_) => panic!("client should reject unsupported load balance strategy"), + Err(err) => assert_eq!(err.code(), Code::InvalidArgument), + } + } + + #[test] + fn static_client_rejects_unknown_cluster() { + let result = RawTripleClient::from_static_endpoints_with_options( + ["http://127.0.0.1:50051?interface=example.Echo"], + RawTripleClientOptions { + cluster: Some("failsafe".to_string()), + ..RawTripleClientOptions::default() + }, + ); + match result { + Ok(_) => panic!("client should reject unsupported cluster strategy"), + Err(err) => assert_eq!(err.code(), Code::InvalidArgument), + } + } + + #[test] + fn static_client_rejects_unknown_compression() { + let result = RawTripleClient::from_static_endpoints_with_options( + ["http://127.0.0.1:50051?interface=example.Echo"], + RawTripleClientOptions { + compression: Some("brotli".to_string()), + ..RawTripleClientOptions::default() + }, + ); + match result { + Ok(_) => panic!("client should reject unsupported compression"), + Err(err) => assert_eq!(err.code(), Code::InvalidArgument), + } + } + + #[tokio::test] + async fn from_registry_reports_disabled_backend() { + let result = RawTripleClient::from_registry("nacos://127.0.0.1:8848").await; + match result { + Ok(_) => panic!("registry client should not be created without registry features"), + Err(err) => assert_eq!(err.code(), Code::Unimplemented), + } + } +} diff --git a/dubbo-rs-core/tests/raw_unary.rs b/dubbo-rs-core/tests/raw_unary.rs new file mode 100644 index 00000000..17a91da8 --- /dev/null +++ b/dubbo-rs-core/tests/raw_unary.rs @@ -0,0 +1,841 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use std::{ + convert::Infallible, + future::Future, + net::SocketAddr, + pin::Pin, + task::{Context, Poll}, + time::Duration, +}; + +use bytes::{BufMut, Bytes, BytesMut}; +use dubbo::{status::Code, triple::transport::DubboServer, BoxBody}; +use dubbo_rs_core::{RawMetadata, RawTripleClient, RawTripleClientOptions, RawUnaryRequest}; +use http_body::Body; +use tower_service::Service; + +#[derive(Clone)] +struct SlowRawUnaryService { + delay: Duration, + response_payload: Bytes, +} + +#[derive(Clone)] +struct MetadataEchoService; + +#[derive(Clone)] +struct CompressionEchoService; + +#[derive(Clone)] +struct HeaderTrailerEchoService; + +impl Service> for SlowRawUnaryService { + type Response = http::Response; + type Error = Infallible; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, _req: http::Request) -> Self::Future { + let delay = self.delay; + let response_payload = self.response_payload.clone(); + Box::pin(async move { + tokio::time::sleep(delay).await; + let body = http_body::combinators::UnsyncBoxBody::new( + hyper::Body::from(grpc_frame(response_payload)).map_err(|err| { + dubbo::status::Status::new(dubbo::status::Code::Internal, err.to_string()) + }), + ); + + Ok(http::Response::builder() + .status(http::StatusCode::OK) + .header("content-type", "application/grpc+proto") + .body(body) + .unwrap()) + }) + } +} + +impl Service> for MetadataEchoService { + type Response = http::Response; + type Error = Infallible; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, req: http::Request) -> Self::Future { + let default_header = req + .headers() + .get("x-client-app") + .and_then(|value| value.to_str().ok()) + .unwrap_or("") + .to_string(); + let call_header = req + .headers() + .get("x-call-id") + .and_then(|value| value.to_str().ok()) + .unwrap_or("") + .to_string(); + + Box::pin(async move { + let body = http_body::combinators::UnsyncBoxBody::new( + hyper::Body::from(grpc_frame(protobuf_string(format!( + "{default_header}:{call_header}" + )))) + .map_err(|err| { + dubbo::status::Status::new(dubbo::status::Code::Internal, err.to_string()) + }), + ); + + Ok(http::Response::builder() + .status(http::StatusCode::OK) + .header("content-type", "application/grpc+proto") + .body(body) + .unwrap()) + }) + } +} + +impl Service> for CompressionEchoService { + type Response = http::Response; + type Error = Infallible; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, req: http::Request) -> Self::Future { + let compression = req + .headers() + .get("grpc-encoding") + .and_then(|value| value.to_str().ok()) + .unwrap_or("identity") + .to_string(); + + Box::pin(async move { + let body = http_body::combinators::UnsyncBoxBody::new( + hyper::Body::from(grpc_frame(protobuf_string(compression))).map_err(|err| { + dubbo::status::Status::new(dubbo::status::Code::Internal, err.to_string()) + }), + ); + + Ok(http::Response::builder() + .status(http::StatusCode::OK) + .header("content-type", "application/grpc+proto") + .body(body) + .unwrap()) + }) + } +} + +impl Service> for HeaderTrailerEchoService { + type Response = http::Response; + type Error = Infallible; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, _req: http::Request) -> Self::Future { + Box::pin(async move { + let (mut sender, body) = hyper::Body::channel(); + tokio::spawn(async move { + let _ = sender + .send_data(grpc_frame(Bytes::from_static(b"\x0a\x0draw response"))) + .await; + let mut trailers = http::HeaderMap::new(); + trailers.insert("x-trailer", http::HeaderValue::from_static("done")); + let _ = sender.send_trailers(trailers).await; + }); + + let body = http_body::combinators::UnsyncBoxBody::new(body.map_err(|err| { + dubbo::status::Status::new(dubbo::status::Code::Internal, err.to_string()) + })); + + Ok(http::Response::builder() + .status(http::StatusCode::OK) + .header("content-type", "application/grpc+proto") + .header("x-response-header", "ready") + .body(body) + .unwrap()) + }) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn raw_unary_core_integration_suite() { + raw_unary_uses_static_endpoint_list().await; + raw_unary_timeout_returns_deadline_exceeded().await; + raw_unary_applies_default_metadata_and_allows_request_override().await; + raw_unary_separates_response_headers_and_trailers().await; + raw_unary_configures_compression().await; + raw_unary_routes_static_endpoints_by_tag().await; + raw_unary_random_load_balance_honors_provider_weight().await; + raw_unary_round_robin_load_balance_honors_provider_weight().await; + raw_unary_routes_static_endpoints_by_group_and_version().await; + raw_unary_errors_when_routing_metadata_matches_no_provider().await; + raw_unary_uses_client_default_timeout().await; + raw_unary_request_timeout_overrides_client_default_timeout().await; +} + +async fn raw_unary_uses_static_endpoint_list() { + const SERVICE: &str = "grpc.examples.echo.StaticEndpointEcho"; + let addr = unused_local_addr(); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + + let server_task = spawn_server( + addr, + shutdown_rx, + SERVICE.to_string(), + Duration::from_millis(0), + ); + + wait_for_server(addr).await; + + let first_endpoint = format!("http://{addr}?interface={SERVICE}&instance=one"); + let second_endpoint = format!("http://{addr}?interface={SERVICE}&instance=two"); + let mut client = + RawTripleClient::from_static_endpoints([first_endpoint.as_str(), second_endpoint.as_str()]) + .unwrap(); + let response = client + .unary(RawUnaryRequest { + service: SERVICE.to_string(), + method: "UnaryEcho".to_string(), + path: format!("/{SERVICE}/UnaryEcho"), + metadata: RawMetadata::new(), + body: Bytes::from_static(b"\x0a\x08dubbo-js"), + timeout_ms: Some(10_000), + }) + .await + .unwrap(); + + assert_eq!(response.body, Bytes::from_static(b"\x0a\x0draw response")); + + let _ = shutdown_tx.send(()); + server_task.await.unwrap(); +} + +async fn raw_unary_separates_response_headers_and_trailers() { + const SERVICE: &str = "grpc.examples.echo.HeaderTrailerEcho"; + let addr = unused_local_addr(); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + + let server_task = tokio::spawn(async move { + DubboServer::new() + .with_listener("tcp".to_string()) + .add_service(SERVICE.to_string(), HeaderTrailerEchoService) + .serve_with_graceful(addr, async move { + let _ = shutdown_rx.await; + }) + .await + .unwrap(); + }); + + wait_for_server(addr).await; + + let endpoint = format!("http://{addr}?interface={SERVICE}"); + let mut client = RawTripleClient::from_static(&endpoint); + let response = client + .unary(RawUnaryRequest { + service: SERVICE.to_string(), + method: "UnaryEcho".to_string(), + path: format!("/{SERVICE}/UnaryEcho"), + metadata: RawMetadata::new(), + body: Bytes::from_static(b"\x0a\x08dubbo-js"), + timeout_ms: Some(10_000), + }) + .await + .unwrap(); + + assert_eq!( + response + .metadata + .get("x-response-header") + .map(String::as_str), + Some("ready") + ); + assert_eq!( + response.trailers.get("x-trailer").map(String::as_str), + Some("done") + ); + + let _ = shutdown_tx.send(()); + server_task.await.unwrap(); +} + +async fn raw_unary_applies_default_metadata_and_allows_request_override() { + const SERVICE: &str = "grpc.examples.echo.DefaultMetadataEcho"; + let addr = unused_local_addr(); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + + let server_task = tokio::spawn(async move { + DubboServer::new() + .with_listener("tcp".to_string()) + .add_service(SERVICE.to_string(), MetadataEchoService) + .serve_with_graceful(addr, async move { + let _ = shutdown_rx.await; + }) + .await + .unwrap(); + }); + + wait_for_server(addr).await; + + let endpoint = format!("http://{addr}?interface={SERVICE}"); + let mut client = RawTripleClient::from_static_endpoints_with_options( + [endpoint.as_str()], + RawTripleClientOptions { + default_metadata: RawMetadata::new() + .insert("x-client-app", "default-app") + .insert("x-call-id", "default-call"), + ..RawTripleClientOptions::default() + }, + ) + .unwrap(); + let response = client + .unary(RawUnaryRequest { + service: SERVICE.to_string(), + method: "UnaryEcho".to_string(), + path: format!("/{SERVICE}/UnaryEcho"), + metadata: RawMetadata::new().insert("x-call-id", "request-call"), + body: Bytes::from_static(b"\x0a\x08dubbo-js"), + timeout_ms: Some(10_000), + }) + .await + .unwrap(); + + assert_eq!(response.body, protobuf_string("default-app:request-call")); + + let _ = shutdown_tx.send(()); + server_task.await.unwrap(); +} + +async fn raw_unary_configures_compression() { + const SERVICE: &str = "grpc.examples.echo.CompressionEcho"; + let addr = unused_local_addr(); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + + let server_task = tokio::spawn(async move { + DubboServer::new() + .with_listener("tcp".to_string()) + .add_service(SERVICE.to_string(), CompressionEchoService) + .serve_with_graceful(addr, async move { + let _ = shutdown_rx.await; + }) + .await + .unwrap(); + }); + + wait_for_server(addr).await; + + let endpoint = format!("http://{addr}?interface={SERVICE}"); + let mut gzip_client = RawTripleClient::from_static(&endpoint); + let gzip_response = gzip_client + .unary(RawUnaryRequest { + service: SERVICE.to_string(), + method: "UnaryEcho".to_string(), + path: format!("/{SERVICE}/UnaryEcho"), + metadata: RawMetadata::new(), + body: Bytes::from_static(b"\x0a\x08dubbo-js"), + timeout_ms: Some(10_000), + }) + .await + .unwrap(); + + assert_eq!(gzip_response.body, protobuf_string("gzip")); + + let mut identity_client = RawTripleClient::from_static_endpoints_with_options( + [endpoint.as_str()], + RawTripleClientOptions { + compression: Some("identity".to_string()), + ..RawTripleClientOptions::default() + }, + ) + .unwrap(); + let identity_response = identity_client + .unary(RawUnaryRequest { + service: SERVICE.to_string(), + method: "UnaryEcho".to_string(), + path: format!("/{SERVICE}/UnaryEcho"), + metadata: RawMetadata::new(), + body: Bytes::from_static(b"\x0a\x08dubbo-js"), + timeout_ms: Some(10_000), + }) + .await + .unwrap(); + + assert_eq!(identity_response.body, protobuf_string("identity")); + + let _ = shutdown_tx.send(()); + server_task.await.unwrap(); +} + +async fn raw_unary_routes_static_endpoints_by_tag() { + const SERVICE: &str = "grpc.examples.echo.TaggedEndpointEcho"; + let blue_addr = unused_local_addr(); + let green_addr = unused_local_addr(); + let (blue_shutdown_tx, blue_shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + let (green_shutdown_tx, green_shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + + let blue_server_task = spawn_server_with_payload( + blue_addr, + blue_shutdown_rx, + SERVICE.to_string(), + Duration::from_millis(0), + Bytes::from_static(b"\x0a\x04blue"), + ); + let green_server_task = spawn_server_with_payload( + green_addr, + green_shutdown_rx, + SERVICE.to_string(), + Duration::from_millis(0), + Bytes::from_static(b"\x0a\x05green"), + ); + + wait_for_server(blue_addr).await; + wait_for_server(green_addr).await; + + let blue_endpoint = format!("http://{blue_addr}?interface={SERVICE}&tag=blue"); + let green_endpoint = format!("http://{green_addr}?interface={SERVICE}&tag=green"); + let mut client = + RawTripleClient::from_static_endpoints([green_endpoint.as_str(), blue_endpoint.as_str()]) + .unwrap(); + let response = client + .unary(RawUnaryRequest { + service: SERVICE.to_string(), + method: "UnaryEcho".to_string(), + path: format!("/{SERVICE}/UnaryEcho"), + metadata: RawMetadata::new().insert("dubbo.tag", "blue"), + body: Bytes::from_static(b"\x0a\x08dubbo-js"), + timeout_ms: Some(10_000), + }) + .await + .unwrap(); + + assert_eq!(response.body, Bytes::from_static(b"\x0a\x04blue")); + + let _ = blue_shutdown_tx.send(()); + let _ = green_shutdown_tx.send(()); + blue_server_task.await.unwrap(); + green_server_task.await.unwrap(); +} + +async fn raw_unary_random_load_balance_honors_provider_weight() { + const SERVICE: &str = "grpc.examples.echo.WeightedEndpointEcho"; + let zero_addr = unused_local_addr(); + let weighted_addr = unused_local_addr(); + let (zero_shutdown_tx, zero_shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + let (weighted_shutdown_tx, weighted_shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + + let zero_server_task = spawn_server_with_payload( + zero_addr, + zero_shutdown_rx, + SERVICE.to_string(), + Duration::from_millis(0), + Bytes::from_static(b"\x0a\x04zero"), + ); + let weighted_server_task = spawn_server_with_payload( + weighted_addr, + weighted_shutdown_rx, + SERVICE.to_string(), + Duration::from_millis(0), + Bytes::from_static(b"\x0a\x08weighted"), + ); + + wait_for_server(zero_addr).await; + wait_for_server(weighted_addr).await; + + let zero_endpoint = format!("http://{zero_addr}?interface={SERVICE}&weight=0"); + let weighted_endpoint = format!("http://{weighted_addr}?interface={SERVICE}&weight=100"); + let mut client = RawTripleClient::from_static_endpoints_with_options( + [zero_endpoint.as_str(), weighted_endpoint.as_str()], + RawTripleClientOptions { + load_balance: Some("random".to_string()), + ..RawTripleClientOptions::default() + }, + ) + .unwrap(); + + for _ in 0..5 { + let response = client + .unary(RawUnaryRequest { + service: SERVICE.to_string(), + method: "UnaryEcho".to_string(), + path: format!("/{SERVICE}/UnaryEcho"), + metadata: RawMetadata::new(), + body: Bytes::from_static(b"\x0a\x08dubbo-js"), + timeout_ms: Some(10_000), + }) + .await + .unwrap(); + + assert_eq!(response.body, Bytes::from_static(b"\x0a\x08weighted")); + } + + let _ = zero_shutdown_tx.send(()); + let _ = weighted_shutdown_tx.send(()); + zero_server_task.await.unwrap(); + weighted_server_task.await.unwrap(); +} + +async fn raw_unary_round_robin_load_balance_honors_provider_weight() { + const SERVICE: &str = "grpc.examples.echo.WeightedRoundRobinEndpointEcho"; + let zero_addr = unused_local_addr(); + let weighted_addr = unused_local_addr(); + let (zero_shutdown_tx, zero_shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + let (weighted_shutdown_tx, weighted_shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + + let zero_server_task = spawn_server_with_payload( + zero_addr, + zero_shutdown_rx, + SERVICE.to_string(), + Duration::from_millis(0), + Bytes::from_static(b"\x0a\x04zero"), + ); + let weighted_server_task = spawn_server_with_payload( + weighted_addr, + weighted_shutdown_rx, + SERVICE.to_string(), + Duration::from_millis(0), + Bytes::from_static(b"\x0a\x08weighted"), + ); + + wait_for_server(zero_addr).await; + wait_for_server(weighted_addr).await; + + let zero_endpoint = format!("http://{zero_addr}?interface={SERVICE}&weight=0"); + let weighted_endpoint = format!("http://{weighted_addr}?interface={SERVICE}&weight=100"); + let mut client = RawTripleClient::from_static_endpoints_with_options( + [zero_endpoint.as_str(), weighted_endpoint.as_str()], + RawTripleClientOptions { + load_balance: Some("round_robin".to_string()), + ..RawTripleClientOptions::default() + }, + ) + .unwrap(); + + for _ in 0..5 { + let response = client + .unary(RawUnaryRequest { + service: SERVICE.to_string(), + method: "UnaryEcho".to_string(), + path: format!("/{SERVICE}/UnaryEcho"), + metadata: RawMetadata::new(), + body: Bytes::from_static(b"\x0a\x08dubbo-js"), + timeout_ms: Some(10_000), + }) + .await + .unwrap(); + + assert_eq!(response.body, Bytes::from_static(b"\x0a\x08weighted")); + } + + let _ = zero_shutdown_tx.send(()); + let _ = weighted_shutdown_tx.send(()); + zero_server_task.await.unwrap(); + weighted_server_task.await.unwrap(); +} + +async fn raw_unary_routes_static_endpoints_by_group_and_version() { + const SERVICE: &str = "grpc.examples.echo.GroupVersionEndpointEcho"; + let blue_addr = unused_local_addr(); + let green_addr = unused_local_addr(); + let (blue_shutdown_tx, blue_shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + let (green_shutdown_tx, green_shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + + let blue_server_task = spawn_server_with_payload( + blue_addr, + blue_shutdown_rx, + SERVICE.to_string(), + Duration::from_millis(0), + Bytes::from_static(b"\x0a\x04blue"), + ); + let green_server_task = spawn_server_with_payload( + green_addr, + green_shutdown_rx, + SERVICE.to_string(), + Duration::from_millis(0), + Bytes::from_static(b"\x0a\x05green"), + ); + + wait_for_server(blue_addr).await; + wait_for_server(green_addr).await; + + let blue_endpoint = format!("http://{blue_addr}?interface={SERVICE}&group=blue&version=1.0.0"); + let green_endpoint = + format!("http://{green_addr}?interface={SERVICE}&group=green&version=1.0.0"); + let mut client = + RawTripleClient::from_static_endpoints([green_endpoint.as_str(), blue_endpoint.as_str()]) + .unwrap(); + let response = client + .unary(RawUnaryRequest { + service: SERVICE.to_string(), + method: "UnaryEcho".to_string(), + path: format!("/{SERVICE}/UnaryEcho"), + metadata: RawMetadata::new() + .insert("tri-service-group", "blue") + .insert("tri-service-version", "1.0.0"), + body: Bytes::from_static(b"\x0a\x08dubbo-js"), + timeout_ms: Some(10_000), + }) + .await + .unwrap(); + + assert_eq!(response.body, Bytes::from_static(b"\x0a\x04blue")); + + let _ = blue_shutdown_tx.send(()); + let _ = green_shutdown_tx.send(()); + blue_server_task.await.unwrap(); + green_server_task.await.unwrap(); +} + +async fn raw_unary_errors_when_routing_metadata_matches_no_provider() { + const SERVICE: &str = "grpc.examples.echo.NoMatchedEndpointEcho"; + let addr = unused_local_addr(); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + + let server_task = spawn_server( + addr, + shutdown_rx, + SERVICE.to_string(), + Duration::from_millis(0), + ); + + wait_for_server(addr).await; + + let endpoint = format!("http://{addr}?interface={SERVICE}&group=green&version=1.0.0"); + let mut client = RawTripleClient::from_static(&endpoint); + let err = client + .unary(RawUnaryRequest { + service: SERVICE.to_string(), + method: "UnaryEcho".to_string(), + path: format!("/{SERVICE}/UnaryEcho"), + metadata: RawMetadata::new() + .insert("tri-service-group", "blue") + .insert("tri-service-version", "1.0.0"), + body: Bytes::from_static(b"\x0a\x08dubbo-js"), + timeout_ms: Some(10_000), + }) + .await + .unwrap_err(); + + assert_eq!(err.code(), Code::Unavailable); + + let _ = shutdown_tx.send(()); + server_task.await.unwrap(); +} + +async fn raw_unary_timeout_returns_deadline_exceeded() { + const SERVICE: &str = "grpc.examples.echo.TimeoutEcho"; + let addr = unused_local_addr(); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + + let server_task = spawn_server( + addr, + shutdown_rx, + SERVICE.to_string(), + Duration::from_millis(500), + ); + + wait_for_server(addr).await; + + let endpoint = format!("http://{addr}?interface={SERVICE}"); + let mut client = RawTripleClient::from_static(&endpoint); + let err = client + .unary(RawUnaryRequest { + service: SERVICE.to_string(), + method: "UnaryEcho".to_string(), + path: format!("/{SERVICE}/UnaryEcho"), + metadata: RawMetadata::new(), + body: Bytes::from_static(b"\x0a\x08dubbo-js"), + timeout_ms: Some(50), + }) + .await + .unwrap_err(); + + assert_eq!(err.code(), Code::DeadlineExceeded); + + let _ = shutdown_tx.send(()); + server_task.await.unwrap(); +} + +async fn raw_unary_uses_client_default_timeout() { + const SERVICE: &str = "grpc.examples.echo.DefaultTimeoutEcho"; + let addr = unused_local_addr(); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + + let server_task = spawn_server( + addr, + shutdown_rx, + SERVICE.to_string(), + Duration::from_millis(100), + ); + + wait_for_server(addr).await; + + let endpoint = format!("http://{addr}?interface={SERVICE}"); + let mut client = RawTripleClient::from_static_endpoints_with_options( + [endpoint.as_str()], + RawTripleClientOptions { + timeout_ms: Some(10), + ..RawTripleClientOptions::default() + }, + ) + .unwrap(); + let err = client + .unary(RawUnaryRequest { + service: SERVICE.to_string(), + method: "UnaryEcho".to_string(), + path: format!("/{SERVICE}/UnaryEcho"), + metadata: RawMetadata::new(), + body: Bytes::from_static(b"\x0a\x08dubbo-js"), + timeout_ms: None, + }) + .await + .unwrap_err(); + + assert_eq!(err.code(), Code::DeadlineExceeded); + + let _ = shutdown_tx.send(()); + server_task.await.unwrap(); +} + +async fn raw_unary_request_timeout_overrides_client_default_timeout() { + const SERVICE: &str = "grpc.examples.echo.OverrideTimeoutEcho"; + let addr = unused_local_addr(); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + + let server_task = spawn_server( + addr, + shutdown_rx, + SERVICE.to_string(), + Duration::from_millis(20), + ); + + wait_for_server(addr).await; + + let endpoint = format!("http://{addr}?interface={SERVICE}"); + let mut client = RawTripleClient::from_static_endpoints_with_options( + [endpoint.as_str()], + RawTripleClientOptions { + timeout_ms: Some(1), + ..RawTripleClientOptions::default() + }, + ) + .unwrap(); + let response = client + .unary(RawUnaryRequest { + service: SERVICE.to_string(), + method: "UnaryEcho".to_string(), + path: format!("/{SERVICE}/UnaryEcho"), + metadata: RawMetadata::new(), + body: Bytes::from_static(b"\x0a\x08dubbo-js"), + timeout_ms: Some(10_000), + }) + .await + .unwrap(); + + assert_eq!(response.body, Bytes::from_static(b"\x0a\x0draw response")); + + let _ = shutdown_tx.send(()); + server_task.await.unwrap(); +} + +fn spawn_server( + addr: SocketAddr, + shutdown_rx: tokio::sync::oneshot::Receiver<()>, + service_name: String, + delay: Duration, +) -> tokio::task::JoinHandle<()> { + spawn_server_with_payload( + addr, + shutdown_rx, + service_name, + delay, + Bytes::from_static(b"\x0a\x0draw response"), + ) +} + +fn spawn_server_with_payload( + addr: SocketAddr, + shutdown_rx: tokio::sync::oneshot::Receiver<()>, + service_name: String, + delay: Duration, + response_payload: Bytes, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + DubboServer::new() + .with_listener("tcp".to_string()) + .add_service( + service_name, + SlowRawUnaryService { + delay, + response_payload, + }, + ) + .serve_with_graceful(addr, async move { + let _ = shutdown_rx.await; + }) + .await + .unwrap(); + }) +} + +async fn wait_for_server(addr: SocketAddr) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(1); + loop { + if tokio::net::TcpStream::connect(addr).await.is_ok() { + return; + } + if tokio::time::Instant::now() >= deadline { + panic!("server did not start listening on {addr}"); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } +} + +fn grpc_frame(payload: Bytes) -> Bytes { + let mut frame = BytesMut::with_capacity(5 + payload.len()); + frame.put_u8(0); + frame.put_u32(payload.len() as u32); + frame.extend_from_slice(&payload); + frame.freeze() +} + +fn protobuf_string(value: impl AsRef) -> Bytes { + let value = value.as_ref().as_bytes(); + let mut message = BytesMut::with_capacity(2 + value.len()); + message.put_u8(0x0a); + message.put_u8(value.len() as u8); + message.extend_from_slice(value); + message.freeze() +} + +fn unused_local_addr() -> SocketAddr { + std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() +} diff --git a/dubbo/src/cluster/mod.rs b/dubbo/src/cluster/mod.rs index 1a20c160..27d53dc9 100644 --- a/dubbo/src/cluster/mod.rs +++ b/dubbo/src/cluster/mod.rs @@ -15,30 +15,32 @@ * limitations under the License. */ +use futures_core::future::BoxFuture; use http::Request; +use std::time::Duration; +use tower::ServiceExt; use tower_service::Service; use crate::{ codegen::RpcInvocation, invoker::clone_body::CloneBody, param::Param, svc::NewService, }; -use self::failover::Failover; - -mod failover; - pub struct NewCluster { inner: N, // new loadbalancer service + strategy: ClusterStrategy, } pub struct Cluster { - inner: S, // failover service + inner: S, // loadbalancer service + strategy: ClusterStrategy, } impl NewCluster { - pub fn layer() -> impl tower_layer::Layer { - tower_layer::layer_fn(|inner: N| { + pub fn layer(strategy: ClusterStrategy) -> impl tower_layer::Layer { + tower_layer::layer_fn(move |inner: N| { NewCluster { inner, // new loadbalancer service + strategy: strategy.clone(), } }) } @@ -50,24 +52,78 @@ where // new loadbalancer service S: NewService, { - type Service = Cluster>; + type Service = Cluster; fn new_service(&self, target: T) -> Self::Service { Cluster { - inner: Failover::new(self.inner.new_service(target)), + inner: self.inner.new_service(target), + strategy: self.strategy.clone(), + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum ClusterStrategy { + Failfast, + Failover { + attempts: usize, + retry_delay: Duration, + }, +} + +impl Default for ClusterStrategy { + fn default() -> Self { + Self::Failover { + attempts: Self::DEFAULT_FAILOVER_ATTEMPTS, + retry_delay: Duration::from_millis(0), + } + } +} + +impl ClusterStrategy { + pub const DEFAULT_FAILOVER_ATTEMPTS: usize = 2; + + pub fn parse(strategy: &str) -> Option { + match strategy { + "failfast" => Some(Self::Failfast), + "failover" => Some(Self::default()), + _ => None, + } + } + + pub fn with_failover_attempts(self, attempts: usize) -> Self { + match self { + Self::Failover { retry_delay, .. } => Self::Failover { + attempts: attempts.max(1), + retry_delay, + }, + Self::Failfast => Self::Failfast, + } + } + + pub fn with_failover_retry_delay(self, retry_delay: Duration) -> Self { + match self { + Self::Failover { attempts, .. } => Self::Failover { + attempts, + retry_delay, + }, + Self::Failfast => Self::Failfast, } } } impl Service> for Cluster where - S: Service>, + S: Service> + Clone + Send + 'static, + S::Future: Send + 'static, + S::Error: Send + 'static, + S::Response: Send + 'static, { type Response = S::Response; type Error = S::Error; - type Future = S::Future; + type Future = BoxFuture<'static, Result>; fn poll_ready( &mut self, @@ -79,7 +135,172 @@ where fn call(&mut self, req: Request) -> Self::Future { let (parts, body) = req.into_parts(); let clone_body = CloneBody::new(body); - let req = Request::from_parts(parts, clone_body); - self.inner.call(req) + let replay_body = clone_body.clone(); + let method = parts.method; + let uri = parts.uri; + let version = parts.version; + let headers = parts.headers; + let extensions = parts.extensions; + let inner = self.inner.clone(); + let strategy = self.strategy.clone(); + + Box::pin(async move { + let make_request = |body, extensions| { + let mut req = Request::new(body); + *req.method_mut() = method.clone(); + *req.uri_mut() = uri.clone(); + *req.version_mut() = version; + *req.headers_mut() = headers.clone(); + if let Some(extensions) = extensions { + *req.extensions_mut() = extensions; + } + req + }; + + match strategy { + ClusterStrategy::Failfast => { + inner + .oneshot(make_request(clone_body, Some(extensions))) + .await + } + ClusterStrategy::Failover { + attempts, + retry_delay, + } => { + let mut last_error = None; + let mut first_body = Some(clone_body); + let mut first_extensions = Some(extensions); + for attempt in 0..attempts { + let body = first_body.take().unwrap_or_else(|| replay_body.clone()); + let extensions = first_extensions.take(); + match inner.clone().oneshot(make_request(body, extensions)).await { + Ok(response) => return Ok(response), + Err(err) => { + last_error = Some(err); + if attempt + 1 < attempts && !retry_delay.is_zero() { + tokio::time::sleep(retry_delay).await; + } + } + } + } + + Err(last_error.expect("failover attempts must be greater than zero")) + } + } + }) + } +} + +#[cfg(test)] +mod tests { + use std::{ + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }, + task::{Context, Poll}, + time::Duration, + }; + + use crate::invoker::clone_body::CloneBody; + use http::Request; + use tower_service::Service; + + use super::{Cluster, ClusterStrategy}; + + #[derive(Clone)] + struct FailsBeforeSuccess { + calls: Arc, + failures: usize, + } + + impl Service> for FailsBeforeSuccess { + type Response = http::Response<()>; + type Error = crate::Error; + type Future = std::future::Ready>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, _req: Request) -> Self::Future { + let call = self.calls.fetch_add(1, Ordering::SeqCst); + if call < self.failures { + std::future::ready(Err(Box::new(std::io::Error::new( + std::io::ErrorKind::ConnectionRefused, + "provider unavailable", + )))) + } else { + std::future::ready(Ok(http::Response::new(()))) + } + } + } + + #[tokio::test] + async fn failover_honors_configured_attempts() { + let calls = Arc::new(AtomicUsize::new(0)); + let mut cluster = Cluster { + inner: FailsBeforeSuccess { + calls: Arc::clone(&calls), + failures: 1, + }, + strategy: ClusterStrategy::Failover { + attempts: 2, + retry_delay: Duration::from_millis(0), + }, + }; + + cluster + .call(Request::new(hyper::Body::empty())) + .await + .unwrap(); + + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn failover_returns_last_error_after_attempts_are_exhausted() { + let calls = Arc::new(AtomicUsize::new(0)); + let mut cluster = Cluster { + inner: FailsBeforeSuccess { + calls: Arc::clone(&calls), + failures: 2, + }, + strategy: ClusterStrategy::Failover { + attempts: 1, + retry_delay: Duration::from_millis(0), + }, + }; + + cluster + .call(Request::new(hyper::Body::empty())) + .await + .unwrap_err(); + + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn failover_waits_between_attempts() { + let calls = Arc::new(AtomicUsize::new(0)); + let mut cluster = Cluster { + inner: FailsBeforeSuccess { + calls: Arc::clone(&calls), + failures: 1, + }, + strategy: ClusterStrategy::Failover { + attempts: 2, + retry_delay: Duration::from_millis(20), + }, + }; + + let started = std::time::Instant::now(); + cluster + .call(Request::new(hyper::Body::empty())) + .await + .unwrap(); + + assert!(started.elapsed() >= Duration::from_millis(20)); + assert_eq!(calls.load(Ordering::SeqCst), 2); } } diff --git a/dubbo/src/extension/registry_extension.rs b/dubbo/src/extension/registry_extension.rs index 2e9291b3..4c132bab 100644 --- a/dubbo/src/extension/registry_extension.rs +++ b/dubbo/src/extension/registry_extension.rs @@ -141,9 +141,7 @@ impl RegistryExtensionFactory { &mut self, url: Url, ) -> Result, StdError> { - let registry_url = url.query::().unwrap(); - let registry_url = registry_url.value(); - let url_str = registry_url.as_str().to_string(); + let url_str = url.as_str().to_string(); match self.instances.get(&url_str) { Some(proxy) => { let proxy = proxy.clone(); diff --git a/dubbo/src/invoker/clone_invoker.rs b/dubbo/src/invoker/clone_invoker.rs index 019d0ee9..708cd24b 100644 --- a/dubbo/src/invoker/clone_invoker.rs +++ b/dubbo/src/invoker/clone_invoker.rs @@ -16,7 +16,7 @@ */ use std::{mem, pin::Pin, task::Poll}; -use crate::{logger::tracing::debug, StdError}; +use crate::{logger::tracing::debug, StdError, Url}; use futures_core::{future::BoxFuture, ready, Future, TryFuture}; use futures_util::FutureExt; use pin_project::pin_project; @@ -173,6 +173,7 @@ where Inv::Future: Send, { inner: Buffer, http::Request>, + url: Option, rx: Receiver, poll: ReusableBoxFuture<'static, ObserveState>, polling: bool, @@ -194,11 +195,23 @@ where Self { inner: buffer, + url: None, rx, polling: false, poll: ReusableBoxFuture::new(futures::future::pending()), } } + + pub fn new_with_url(invoker: Inv, url: Url) -> Self { + Self { + url: Some(url), + ..Self::new(invoker) + } + } + + pub fn url(&self) -> Option<&Url> { + self.url.as_ref() + } } impl Service> for CloneInvoker @@ -262,6 +275,7 @@ where fn clone(&self) -> Self { Self { inner: self.inner.clone(), + url: self.url.clone(), rx: self.rx.clone(), polling: false, poll: ReusableBoxFuture::new(futures::future::pending()), diff --git a/dubbo/src/invoker/mod.rs b/dubbo/src/invoker/mod.rs index 1c87c0ec..9745d595 100644 --- a/dubbo/src/invoker/mod.rs +++ b/dubbo/src/invoker/mod.rs @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -use crate::{codegen::TripleInvoker, invoker::clone_invoker::CloneInvoker, svc::NewService}; +use crate::{codegen::TripleInvoker, invoker::clone_invoker::CloneInvoker, svc::NewService, Url}; pub mod clone_body; pub mod clone_invoker; @@ -27,7 +27,7 @@ impl NewService for NewInvoker { fn new_service(&self, url: String) -> Self::Service { // todo create another invoker by url protocol - let url = url.parse().unwrap(); - CloneInvoker::new(TripleInvoker::new(url)) + let url: Url = url.parse().unwrap(); + CloneInvoker::new_with_url(TripleInvoker::new(url.clone()), url) } } diff --git a/dubbo/src/loadbalancer/mod.rs b/dubbo/src/loadbalancer/mod.rs index d49ec5df..78d3a7b6 100644 --- a/dubbo/src/loadbalancer/mod.rs +++ b/dubbo/src/loadbalancer/mod.rs @@ -18,7 +18,13 @@ pub mod random; use futures_core::future::BoxFuture; -use std::error::Error; +use std::{ + error::Error, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }, +}; use tokio::time::Duration; use tower::{discover::ServiceList, ServiceExt}; use tower_service::Service; @@ -31,24 +37,32 @@ use crate::{ loadbalancer::random::RandomLoadBalancer, param::Param, protocol::triple::triple_invoker::TripleInvoker, + status::{Code, Status}, svc::NewService, - StdError, + StdError, Url, }; +const DEFAULT_PROVIDER_WEIGHT: u32 = 100; +const PROVIDER_WEIGHT_KEY: &str = "weight"; + pub struct NewLoadBalancer { inner: N, + strategy: LoadBalanceStrategy, } #[derive(Clone)] pub struct LoadBalancerSvc { inner: S, // Routes service + strategy: LoadBalanceStrategy, + round_robin_next: Arc, } impl NewLoadBalancer { - pub fn layer() -> impl tower_layer::Layer { - tower_layer::layer_fn(|inner| { + pub fn layer(strategy: LoadBalanceStrategy) -> impl tower_layer::Layer { + tower_layer::layer_fn(move |inner| { NewLoadBalancer { inner, // NewRoutes + strategy: strategy.clone(), } }) } @@ -66,7 +80,11 @@ where // Routes service let svc = self.inner.new_service(target); - LoadBalancerSvc { inner: svc } + LoadBalancerSvc { + inner: svc, + strategy: self.strategy.clone(), + round_robin_next: Arc::new(AtomicUsize::new(0)), + } } } @@ -92,6 +110,8 @@ where fn call(&mut self, req: http::Request) -> Self::Future { let routes = self.inner.call(()); + let strategy = self.strategy.clone(); + let round_robin_next = Arc::clone(&self.round_robin_next); let fut = async move { let routes = routes.await; @@ -112,9 +132,17 @@ where // invks.oneshot(req).await // let service_list = ServiceList::new(service_list); + if routes.is_empty() { + return Err(Status::new( + Code::Unavailable, + "no provider available for request".to_string(), + ) + .into()); + } + // let p2c = tower::balance::p2c::Balance::new(service_list); // let p: Box, http::Response>, Box>> + std::marker::Send + std::marker::Sync> = get_loadbalancer("p2c").into(); - let p = get_loadbalancer("p2c"); + let p = get_loadbalancer(&strategy, round_robin_next); // let ivk = p.select_invokers(invokers, metadata); let ivk = p.select_invokers(routes, metadata); @@ -141,16 +169,33 @@ pub trait LoadBalancer { ) -> Self::Invoker; } +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub enum LoadBalanceStrategy { + Random, + RoundRobin, + #[default] + P2c, +} + +impl LoadBalanceStrategy { + pub fn parse(strategy: &str) -> Option { + match strategy { + "random" => Some(Self::Random), + "round_robin" | "roundrobin" => Some(Self::RoundRobin), + "p2c" => Some(Self::P2c), + _ => None, + } + } +} + fn get_loadbalancer( - loadbalancer: &str, + loadbalancer: &LoadBalanceStrategy, + round_robin_next: Arc, ) -> Box + Send + Sync + 'static> { match loadbalancer { - "random" => { - println!("random!"); - Box::new(RandomLoadBalancer::default()) - } - "p2c" => Box::new(P2cBalancer::default()), - _ => Box::new(P2cBalancer::default()), + LoadBalanceStrategy::Random => Box::new(RandomLoadBalancer::default()), + LoadBalanceStrategy::RoundRobin => Box::new(RoundRobinLoadBalancer::new(round_robin_next)), + LoadBalanceStrategy::P2c => Box::new(P2cBalancer::default()), } } const DEFAULT_RTT: Duration = Duration::from_millis(30); @@ -185,3 +230,116 @@ impl LoadBalancer for P2cBalancer { svc } } + +#[derive(Debug)] +pub struct RoundRobinLoadBalancer { + next: Arc, +} + +impl RoundRobinLoadBalancer { + pub fn new(next: Arc) -> Self { + Self { next } + } +} + +impl LoadBalancer for RoundRobinLoadBalancer { + type Invoker = DubboBoxService; + + fn select_invokers( + &self, + invokers: Vec>, + _metadata: Metadata, + ) -> Self::Invoker { + debug!("round-robin load balancer"); + let next = self.next.fetch_add(1, Ordering::Relaxed); + let index = + weighted_round_robin_index(&invokers, next).unwrap_or_else(|| next % invokers.len()); + DubboBoxService::new(invokers[index].clone()) + } +} + +fn weighted_round_robin_index( + invokers: &[CloneInvoker], + next: usize, +) -> Option { + let weights = invokers + .iter() + .map(|invoker| provider_weight(invoker.url())) + .collect::>(); + + weighted_round_robin_index_for_weights(&weights, next) +} + +fn weighted_round_robin_index_for_weights(weights: &[u32], next: usize) -> Option { + let total_weight = weights.iter().try_fold(0usize, |total, weight| { + total.checked_add(usize::try_from(*weight).ok()?) + })?; + if total_weight == 0 { + return None; + } + + let selected_weight = next % total_weight; + let mut cumulative_weight = 0usize; + for (index, weight) in weights.iter().enumerate() { + cumulative_weight += usize::try_from(*weight).ok()?; + if selected_weight < cumulative_weight { + return Some(index); + } + } + + None +} + +fn provider_weight(url: Option<&Url>) -> u32 { + url.and_then(|url| url.query_param_by_key(PROVIDER_WEIGHT_KEY)) + .and_then(|weight| weight.parse::().ok()) + .unwrap_or(DEFAULT_PROVIDER_WEIGHT) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn provider_weight_defaults_invalid_or_missing_weight() { + let no_weight = "http://127.0.0.1:50051?interface=example.Echo" + .parse::() + .unwrap(); + let invalid_weight = "http://127.0.0.1:50051?interface=example.Echo&weight=bad" + .parse::() + .unwrap(); + let explicit_weight = "http://127.0.0.1:50051?interface=example.Echo&weight=25" + .parse::() + .unwrap(); + + assert_eq!(provider_weight(Some(&no_weight)), DEFAULT_PROVIDER_WEIGHT); + assert_eq!( + provider_weight(Some(&invalid_weight)), + DEFAULT_PROVIDER_WEIGHT + ); + assert_eq!(provider_weight(Some(&explicit_weight)), 25); + } + + #[test] + fn weighted_round_robin_index_skips_zero_weight_when_positive_weight_exists() { + for next in 0..10 { + assert_eq!( + weighted_round_robin_index_for_weights(&[0, 100], next), + Some(1) + ); + } + } + + #[test] + fn weighted_round_robin_index_uses_weighted_slots() { + assert_eq!(weighted_round_robin_index_for_weights(&[1, 2], 0), Some(0)); + assert_eq!(weighted_round_robin_index_for_weights(&[1, 2], 1), Some(1)); + assert_eq!(weighted_round_robin_index_for_weights(&[1, 2], 2), Some(1)); + assert_eq!(weighted_round_robin_index_for_weights(&[1, 2], 3), Some(0)); + } + + #[test] + fn weighted_round_robin_index_returns_none_when_all_weights_are_zero() { + assert_eq!(weighted_round_robin_index_for_weights(&[0, 0], 0), None); + } +} diff --git a/dubbo/src/loadbalancer/random.rs b/dubbo/src/loadbalancer/random.rs index 7739cba4..af9cf817 100644 --- a/dubbo/src/loadbalancer/random.rs +++ b/dubbo/src/loadbalancer/random.rs @@ -15,10 +15,13 @@ * limitations under the License. */ -use rand::prelude::SliceRandom; +use rand::{ + distributions::{Distribution, WeightedIndex}, + Rng, +}; use tracing::debug; -use super::{DubboBoxService, LoadBalancer}; +use super::{provider_weight, DubboBoxService, LoadBalancer}; use crate::{ invocation::Metadata, loadbalancer::CloneInvoker, protocol::triple::triple_invoker::TripleInvoker, @@ -36,7 +39,51 @@ impl LoadBalancer for RandomLoadBalancer { metadata: Metadata, ) -> Self::Invoker { debug!("random loadbalance {:?}", metadata); - let ivk = invokers.choose(&mut rand::thread_rng()).unwrap().clone(); + let mut rng = rand::thread_rng(); + let index = weighted_random_index(&invokers, &mut rng) + .unwrap_or_else(|| rng.gen_range(0..invokers.len())); + let ivk = invokers[index].clone(); DubboBoxService::new(ivk) } } + +fn weighted_random_index( + invokers: &[CloneInvoker], + rng: &mut R, +) -> Option { + let weights = invokers + .iter() + .map(|invoker| provider_weight(invoker.url())) + .collect::>(); + + weighted_index(&weights, rng) +} + +fn weighted_index(weights: &[u32], rng: &mut R) -> Option { + WeightedIndex::new(weights) + .ok() + .map(|dist| dist.sample(rng)) +} + +#[cfg(test)] +mod tests { + use rand::{rngs::StdRng, SeedableRng}; + + use super::*; + + #[test] + fn weighted_index_ignores_zero_weight_when_positive_weight_exists() { + let mut rng = StdRng::seed_from_u64(7); + + for _ in 0..100 { + assert_eq!(weighted_index(&[0, 100], &mut rng), Some(1)); + } + } + + #[test] + fn weighted_index_returns_none_when_all_weights_are_zero() { + let mut rng = StdRng::seed_from_u64(7); + + assert_eq!(weighted_index(&[0, 0], &mut rng), None); + } +} diff --git a/dubbo/src/protocol/triple/triple_invoker.rs b/dubbo/src/protocol/triple/triple_invoker.rs index 516dab2e..f21e62cf 100644 --- a/dubbo/src/protocol/triple/triple_invoker.rs +++ b/dubbo/src/protocol/triple/triple_invoker.rs @@ -88,36 +88,36 @@ impl TripleInvoker { "authority", HeaderValue::from_str(uri.authority().unwrap().as_str()).unwrap(), ); - req.headers_mut().insert( + insert_default_header( + req.headers_mut(), "content-type", HeaderValue::from_static("application/grpc+proto"), ); - req.headers_mut() - .insert("user-agent", HeaderValue::from_static("dubbo-rust/0.1.0")); - req.headers_mut() - .insert("te", HeaderValue::from_static("trailers")); - req.headers_mut().insert( + insert_default_header( + req.headers_mut(), + "user-agent", + HeaderValue::from_static("dubbo-rust/0.1.0"), + ); + insert_default_header( + req.headers_mut(), + "te", + HeaderValue::from_static("trailers"), + ); + insert_default_header( + req.headers_mut(), "tri-service-version", HeaderValue::from_static("dubbo-rust/0.1.0"), ); - req.headers_mut() - .insert("tri-service-group", HeaderValue::from_static("cluster")); - req.headers_mut().insert( + insert_default_header( + req.headers_mut(), + "tri-service-group", + HeaderValue::from_static("cluster"), + ); + insert_default_header( + req.headers_mut(), "tri-unit-info", HeaderValue::from_static("dubbo-rust/0.1.0"), ); - // if let Some(_encoding) = self.send_compression_encoding { - - // } - - req.headers_mut() - .insert("grpc-encoding", http::HeaderValue::from_static("gzip")); - - req.headers_mut().insert( - "grpc-accept-encoding", - http::HeaderValue::from_static("gzip"), - ); - // // const ( // // TripleContentType = "application/grpc+proto" // // TripleUserAgent = "grpc-go/1.35.0-dev" @@ -134,6 +134,12 @@ impl TripleInvoker { } } +fn insert_default_header(headers: &mut http::HeaderMap, name: &'static str, value: HeaderValue) { + if !headers.contains_key(name) { + headers.insert(name, value); + } +} + impl Service> for TripleInvoker { type Response = http::Response; diff --git a/dubbo/src/route/mod.rs b/dubbo/src/route/mod.rs index b369d83a..d6a4ec7b 100644 --- a/dubbo/src/route/mod.rs +++ b/dubbo/src/route/mod.rs @@ -25,11 +25,22 @@ use tower_service::Service; use crate::{ codegen::{RpcInvocation, TripleInvoker}, + invocation::Metadata, invoker::clone_invoker::CloneInvoker, param::Param, + status::{Code, Status}, svc::NewService, + Url, }; +const DUBBO_TAG_KEY: &str = "dubbo.tag"; +const TRIPLE_SERVICE_TAG_KEY: &str = "tri-service-tag"; +const TAG_KEY: &str = "tag"; +const TRIPLE_SERVICE_GROUP_KEY: &str = "tri-service-group"; +const GROUP_KEY: &str = "group"; +const TRIPLE_SERVICE_VERSION_KEY: &str = "tri-service-version"; +const VERSION_KEY: &str = "version"; + pub struct NewRoutes { inner: N, } @@ -161,8 +172,128 @@ where } fn call(&mut self, _: ()) -> Self::Future { - // some router operator - // if new_invokers changed, send new invokers to routes_rx after router operator - futures_util::future::ok(self.invokers.clone()) + futures_util::future::ready(route_invokers( + self.invokers.clone(), + self.target.param().get_metadata(), + )) + } +} + +fn route_invokers( + invokers: Vec>, + metadata: Metadata, +) -> Result>, StdError> { + let invokers = route_by_service_metadata(invokers, metadata.clone())?; + Ok(route_by_tag(invokers, metadata)) +} + +fn route_by_service_metadata( + invokers: Vec>, + metadata: Metadata, +) -> Result>, StdError> { + let group = metadata_value(metadata.clone(), &[TRIPLE_SERVICE_GROUP_KEY, GROUP_KEY]); + let version = metadata_value(metadata, &[TRIPLE_SERVICE_VERSION_KEY, VERSION_KEY]); + + if group.is_none() && version.is_none() { + return Ok(invokers); } + + let has_candidates = !invokers.is_empty(); + let filtered = invokers + .into_iter() + .filter(|invoker| { + invoker.url().is_some_and(|url| { + matches_provider_param( + url, + &[TRIPLE_SERVICE_GROUP_KEY, GROUP_KEY], + group.as_deref(), + ) && matches_provider_param( + url, + &[TRIPLE_SERVICE_VERSION_KEY, VERSION_KEY], + version.as_deref(), + ) + }) + }) + .collect::>(); + + if has_candidates && filtered.is_empty() { + return Err(Status::new( + Code::Unavailable, + "no provider matched request routing metadata".to_string(), + ) + .into()); + } + + Ok(filtered) +} + +fn route_by_tag( + invokers: Vec>, + metadata: Metadata, +) -> Vec> { + let Some(request_tag) = request_tag(metadata) else { + return prefer_untagged(invokers); + }; + + let mut tagged = Vec::new(); + let mut untagged = Vec::new(); + for invoker in invokers.iter() { + match invoker.url().and_then(provider_tag) { + Some(provider_tag) if provider_tag == request_tag => tagged.push(invoker.clone()), + None => untagged.push(invoker.clone()), + _ => {} + } + } + + if !tagged.is_empty() { + return tagged; + } + if !untagged.is_empty() { + return untagged; + } + + invokers +} + +fn prefer_untagged(invokers: Vec>) -> Vec> { + let untagged = invokers + .iter() + .filter(|invoker| invoker.url().and_then(provider_tag).is_none()) + .cloned() + .collect::>(); + + if untagged.is_empty() { + invokers + } else { + untagged + } +} + +fn request_tag(metadata: Metadata) -> Option { + metadata_value(metadata, &[DUBBO_TAG_KEY, TRIPLE_SERVICE_TAG_KEY, TAG_KEY]) +} + +fn provider_tag(url: &Url) -> Option { + [DUBBO_TAG_KEY, TRIPLE_SERVICE_TAG_KEY, TAG_KEY] + .into_iter() + .find_map(|key| url.query_param_by_key(key)) + .filter(|tag| !tag.is_empty()) +} + +fn metadata_value(metadata: Metadata, keys: &[&str]) -> Option { + let headers = metadata.into_headers(); + keys.iter() + .find_map(|key| headers.get(*key).and_then(|value| value.to_str().ok())) + .filter(|value| !value.is_empty()) + .map(ToOwned::to_owned) +} + +fn matches_provider_param(url: &Url, keys: &[&str], expected: Option<&str>) -> bool { + let Some(expected) = expected else { + return true; + }; + + keys.iter() + .find_map(|key| url.query_param_by_key(key)) + .is_some_and(|actual| actual == expected) } diff --git a/dubbo/src/status.rs b/dubbo/src/status.rs index 7258b481..b6e1c6c0 100644 --- a/dubbo/src/status.rs +++ b/dubbo/src/status.rs @@ -274,13 +274,20 @@ impl Status { } pub fn from_error(err: crate::Error) -> Self { - Status::new(Code::Internal, err.to_string()) + match err.downcast::() { + Ok(status) => *status, + Err(err) => Status::new(Code::Internal, err.to_string()), + } } pub fn code(&self) -> Code { self.code } + pub fn message(&self) -> &str { + &self.message + } + pub fn to_http(&self) -> http::Response { let (mut parts, _) = http::Response::new(()).into_parts(); diff --git a/dubbo/src/triple/client/builder.rs b/dubbo/src/triple/client/builder.rs index ce0fa661..810b48ac 100644 --- a/dubbo/src/triple/client/builder.rs +++ b/dubbo/src/triple/client/builder.rs @@ -15,11 +15,16 @@ * limitations under the License. */ -use std::sync::Arc; +use std::{sync::Arc, time::Duration}; use crate::{ - cluster::NewCluster, directory::NewCachedDirectory, extension, loadbalancer::NewLoadBalancer, - route::NewRoutes, utils::boxed_clone::BoxCloneService, + cluster::{ClusterStrategy, NewCluster}, + directory::NewCachedDirectory, + extension, + loadbalancer::{LoadBalanceStrategy, NewLoadBalancer}, + route::NewRoutes, + triple::compression::CompressionEncoding, + utils::boxed_clone::BoxCloneService, }; use crate::{ @@ -41,6 +46,9 @@ pub struct ClientBuilder { pub connector: &'static str, registry_extension_url: Option, pub direct: bool, + pub load_balance: LoadBalanceStrategy, + pub cluster: ClusterStrategy, + pub send_compression_encoding: Option, } impl ClientBuilder { @@ -50,16 +58,38 @@ impl ClientBuilder { connector: "", registry_extension_url: None, direct: false, + load_balance: LoadBalanceStrategy::default(), + cluster: ClusterStrategy::default(), + send_compression_encoding: Some(CompressionEncoding::Gzip), } } pub fn from_static(host: &str) -> ClientBuilder { - let registry_extension_url = StaticRegistry::to_extension_url(vec![host.parse().unwrap()]); + Self::from_static_hosts([host]) + } + + pub fn from_static_hosts<'a, I>(hosts: I) -> ClientBuilder + where + I: IntoIterator, + { + Self::from_static_urls( + hosts + .into_iter() + .map(|host| host.parse().unwrap()) + .collect(), + ) + } + + pub fn from_static_urls(urls: Vec) -> ClientBuilder { + let registry_extension_url = StaticRegistry::to_extension_url(urls); Self { timeout: None, connector: "", registry_extension_url: Some(registry_extension_url), direct: true, + load_balance: LoadBalanceStrategy::default(), + cluster: ClusterStrategy::default(), + send_compression_encoding: Some(CompressionEncoding::Gzip), } } @@ -95,6 +125,38 @@ impl ClientBuilder { Self { direct, ..self } } + pub fn with_load_balance(self, load_balance: LoadBalanceStrategy) -> Self { + Self { + load_balance, + ..self + } + } + + pub fn with_cluster(self, cluster: ClusterStrategy) -> Self { + Self { cluster, ..self } + } + + pub fn with_failover_attempts(self, attempts: usize) -> Self { + Self { + cluster: self.cluster.with_failover_attempts(attempts), + ..self + } + } + + pub fn with_failover_retry_delay(self, retry_delay: Duration) -> Self { + Self { + cluster: self.cluster.with_failover_retry_delay(retry_delay), + ..self + } + } + + pub fn with_compression(self, compression: Option) -> Self { + Self { + send_compression_encoding: compression, + ..self + } + } + pub fn build(mut self) -> ServiceMK { let registry = self .registry_extension_url @@ -102,8 +164,8 @@ impl ClientBuilder { .expect("registry must not be empty"); let mk_service = ServiceBuilder::new() - .layer(NewCluster::layer()) - .layer(NewLoadBalancer::layer()) + .layer(NewCluster::layer(self.cluster)) + .layer(NewLoadBalancer::layer(self.load_balance)) .layer(NewRoutes::layer()) .layer(NewCachedDirectory::layer()) .service(MkRegistryService::new(registry)); diff --git a/dubbo/src/triple/client/triple.rs b/dubbo/src/triple/client/triple.rs index 2948dec5..399ff2a8 100644 --- a/dubbo/src/triple/client/triple.rs +++ b/dubbo/src/triple/client/triple.rs @@ -16,6 +16,7 @@ */ use aws_smithy_http::body::SdkBody; +use bytes::Bytes; use futures_util::{future, stream, StreamExt, TryStreamExt}; use http::HeaderValue; use prost::Message; @@ -29,7 +30,7 @@ use crate::{ status::Status, svc::NewService, triple::{ - codec::{Codec, Decoder, Encoder}, + codec::{bytes::BytesCodec, Codec, Decoder, Encoder}, compression::CompressionEncoding, decode::Decoding, encode::encode, @@ -56,8 +57,9 @@ impl TripleClient { } pub fn new(builder: ClientBuilder) -> Self { + let send_compression_encoding = builder.send_compression_encoding; TripleClient { - send_compression_encoding: Some(CompressionEncoding::Gzip), + send_compression_encoding, mk: builder.build(), } } @@ -92,28 +94,44 @@ impl TripleClient { "authority", HeaderValue::from_str(uri.authority().unwrap().as_str()).unwrap(), ); - req.headers_mut().insert( + insert_default_header( + req.headers_mut(), "content-type", HeaderValue::from_static("application/grpc+proto"), ); - req.headers_mut() - .insert("user-agent", HeaderValue::from_static("dubbo-rust/0.1.0")); - req.headers_mut() - .insert("te", HeaderValue::from_static("trailers")); - req.headers_mut().insert( + insert_default_header( + req.headers_mut(), + "user-agent", + HeaderValue::from_static("dubbo-rust/0.1.0"), + ); + insert_default_header( + req.headers_mut(), + "te", + HeaderValue::from_static("trailers"), + ); + insert_default_header( + req.headers_mut(), "tri-service-version", HeaderValue::from_static("dubbo-rust/0.1.0"), ); - req.headers_mut() - .insert("tri-service-group", HeaderValue::from_static("cluster")); - req.headers_mut().insert( + insert_default_header( + req.headers_mut(), + "tri-service-group", + HeaderValue::from_static("cluster"), + ); + insert_default_header( + req.headers_mut(), "tri-unit-info", HeaderValue::from_static("dubbo-rust/0.1.0"), ); if let Some(_encoding) = self.send_compression_encoding { - req.headers_mut() - .insert("grpc-encoding", http::HeaderValue::from_static("gzip")); - req.headers_mut().insert( + insert_default_header( + req.headers_mut(), + "grpc-encoding", + http::HeaderValue::from_static("gzip"), + ); + insert_default_header( + req.headers_mut(), "grpc-accept-encoding", http::HeaderValue::from_static("gzip"), ); @@ -167,6 +185,7 @@ impl TripleClient { .header("path", path.to_string()) .body(body) .unwrap(); + self.set_compression_headers(request.headers_mut()); for (k, v) in mt.into_headers().iter() { request.headers_mut().insert(k, v.to_owned()); @@ -204,6 +223,89 @@ impl TripleClient { } } + pub async fn raw_unary( + &mut self, + req: Request, + path: http::uri::PathAndQuery, + invocation: RpcInvocation, + ) -> Result, crate::status::Status> { + let (response, trailers) = self.raw_unary_with_trailers(req, path, invocation).await?; + let (mut parts, message) = response.into_parts(); + + if let Some(trailers) = trailers { + let mut h = parts.into_headers(); + h.extend(trailers.into_headers()); + parts = Metadata::from_headers(h); + } + + Ok(Response::from_parts(parts, message)) + } + + pub async fn raw_unary_with_trailers( + &mut self, + req: Request, + path: http::uri::PathAndQuery, + mut invocation: RpcInvocation, + ) -> Result<(Response, Option), crate::status::Status> { + let mut codec = BytesCodec::default(); + let decoder: Box + Send + 'static> = + Box::new(codec.decoder()); + let encoder: Box + Send + 'static> = + Box::new(codec.encoder()); + + let mt = req.metadata.clone(); + + let req = req.map(|m| stream::once(future::ready(m))); + let body_stream = encode( + encoder, + req.into_inner().map(Ok), + self.send_compression_encoding, + true, + ) + .into_stream(); + let body = hyper::Body::wrap_stream(body_stream); + + invocation = invocation.with_metadata(mt.clone()); + let mut invoker = self.mk.new_service(invocation); + + let mut request = http::Request::builder() + .header("path", path.to_string()) + .body(body) + .unwrap(); + self.set_compression_headers(request.headers_mut()); + + for (k, v) in mt.into_headers().iter() { + request.headers_mut().insert(k, v.to_owned()); + } + + let response = invoker + .call(request) + .await + .map_err(|err| crate::status::Status::from_error(err.into())); + + match response { + Ok(v) => { + let resp = v + .map(|body| Decoding::new(body, decoder, self.send_compression_encoding, true)); + let (parts, body) = Response::from_http(resp).into_parts(); + + futures_util::pin_mut!(body); + + let message = body.try_next().await?.ok_or_else(|| { + crate::status::Status::new( + crate::status::Code::Internal, + "Missing response message.".to_string(), + ) + })?; + + let trailers = body.trailer().await?; + + Ok((Response::from_parts(parts, message), trailers)) + } + Err(err) => Err(err), + } + } + pub async fn bidi_streaming( &mut self, req: impl IntoStreamingRequest, @@ -238,6 +340,7 @@ impl TripleClient { .header("path", path.to_string()) .body(body) .unwrap(); + self.set_compression_headers(request.headers_mut()); for (k, v) in mt.into_headers().iter() { request.headers_mut().insert(k, v.to_owned()); @@ -292,6 +395,7 @@ impl TripleClient { .header("path", path.to_string()) .body(body) .unwrap(); + self.set_compression_headers(request.headers_mut()); for (k, v) in mt.into_headers().iter() { request.headers_mut().insert(k, v.to_owned()); @@ -363,6 +467,7 @@ impl TripleClient { .header("path", path.to_string()) .body(body) .unwrap(); + self.set_compression_headers(request.headers_mut()); for (k, v) in mt.into_headers().iter() { request.headers_mut().insert(k, v.to_owned()); @@ -383,6 +488,19 @@ impl TripleClient { Err(err) => Err(err), } } + + fn set_compression_headers(&self, headers: &mut http::HeaderMap) { + if let Some(encoding) = self.send_compression_encoding { + headers.insert("grpc-encoding", encoding.into_header_value()); + headers.insert("grpc-accept-encoding", encoding.into_header_value()); + } + } +} + +fn insert_default_header(headers: &mut http::HeaderMap, name: &'static str, value: HeaderValue) { + if !headers.contains_key(name) { + headers.insert(name, value); + } } pub fn get_codec( diff --git a/dubbo/src/triple/codec/bytes.rs b/dubbo/src/triple/codec/bytes.rs new file mode 100644 index 00000000..1fff6c85 --- /dev/null +++ b/dubbo/src/triple/codec/bytes.rs @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use bytes::{Buf, BufMut, Bytes}; + +use super::{Codec, DecodeBuf, Decoder, EncodeBuf, Encoder}; + +/// A codec that treats each Triple message as an opaque protobuf payload. +/// +/// This is the boundary needed by higher-level runtimes such as JavaScript: +/// JS can own protobuf encoding/decoding while Rust owns Triple transport, +/// registry, routing, and connection state. +#[derive(Debug, Clone, Default)] +pub struct BytesCodec; + +impl Codec for BytesCodec { + type Encode = Bytes; + type Decode = Bytes; + + type Encoder = BytesEncoder; + type Decoder = BytesDecoder; + + fn encoder(&mut self) -> Self::Encoder { + BytesEncoder + } + + fn decoder(&mut self) -> Self::Decoder { + BytesDecoder + } +} + +#[derive(Debug, Clone, Default)] +pub struct BytesEncoder; + +impl Encoder for BytesEncoder { + type Item = Bytes; + type Error = crate::status::Status; + + fn encode(&mut self, item: Self::Item, buf: &mut EncodeBuf<'_>) -> Result<(), Self::Error> { + buf.reserve(item.len()); + buf.put_slice(&item); + Ok(()) + } +} + +#[derive(Debug, Clone, Default)] +pub struct BytesDecoder; + +impl Decoder for BytesDecoder { + type Item = Bytes; + type Error = crate::status::Status; + + fn decode(&mut self, buf: &mut DecodeBuf<'_>) -> Result, Self::Error> { + let len = buf.remaining(); + Ok(Some(buf.copy_to_bytes(len))) + } +} + +#[cfg(test)] +mod tests { + use bytes::BytesMut; + + use super::*; + + #[test] + fn bytes_codec_preserves_payload() { + let payload = Bytes::from_static(b"\x08\x96\x01"); + let mut encoded = BytesMut::new(); + let mut encoder = BytesEncoder; + + encoder + .encode(payload.clone(), &mut EncodeBuf::new(&mut encoded)) + .unwrap(); + + let len = encoded.len(); + let mut decoder = BytesDecoder; + let decoded = decoder + .decode(&mut DecodeBuf::new(&mut encoded, len)) + .unwrap() + .unwrap(); + + assert_eq!(decoded, payload); + } +} diff --git a/dubbo/src/triple/codec/mod.rs b/dubbo/src/triple/codec/mod.rs index 2ba79865..9b891de8 100644 --- a/dubbo/src/triple/codec/mod.rs +++ b/dubbo/src/triple/codec/mod.rs @@ -16,6 +16,7 @@ */ pub mod buffer; +pub mod bytes; pub mod prost; pub mod serde_codec; diff --git a/dubbo/src/triple/transport/connection.rs b/dubbo/src/triple/transport/connection.rs index 060a24ee..c1a8cc57 100644 --- a/dubbo/src/triple/transport/connection.rs +++ b/dubbo/src/triple/transport/connection.rs @@ -101,7 +101,9 @@ impl Service> for Connection { let uri = self.host.clone(); let call_fut = connect.call(uri); let fut = async move { - let mut con = call_fut.await.unwrap(); + let mut con = call_fut + .await + .map_err(|err| -> crate::Error { Box::new(err) })?; con.call(req) .await .map_err(|err| err.into()) diff --git a/dubbo/tests/raw_unary.rs b/dubbo/tests/raw_unary.rs new file mode 100644 index 00000000..a514f637 --- /dev/null +++ b/dubbo/tests/raw_unary.rs @@ -0,0 +1,133 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use std::{ + convert::Infallible, + future::Future, + net::SocketAddr, + pin::Pin, + task::{Context, Poll}, +}; + +use bytes::{BufMut, Bytes, BytesMut}; +use dubbo::{ + codegen::{ClientBuilder, Request, RpcInvocation, TripleClient}, + triple::transport::DubboServer, + BoxBody, +}; +use http_body::Body; +use tower_service::Service; + +#[derive(Clone)] +struct RawUnaryService; + +impl Service> for RawUnaryService { + type Response = http::Response; + type Error = Infallible; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, req: http::Request) -> Self::Future { + Box::pin(async move { + assert_eq!(req.uri().path(), "/grpc.examples.echo.Echo/UnaryEcho"); + assert_eq!( + req.headers() + .get("tri-service-group") + .and_then(|v| v.to_str().ok()), + Some("test") + ); + assert_eq!( + req.headers() + .get("tri-service-version") + .and_then(|v| v.to_str().ok()), + Some("1.0.0") + ); + + let response_payload = Bytes::from_static(b"\x0a\x0draw response"); + let body = http_body::combinators::UnsyncBoxBody::new( + hyper::Body::from(grpc_frame(response_payload)).map_err(|err| { + dubbo::status::Status::new(dubbo::status::Code::Internal, err.to_string()) + }), + ); + + Ok(http::Response::builder() + .status(http::StatusCode::OK) + .header("content-type", "application/grpc+proto") + .body(body) + .unwrap()) + }) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn raw_unary_round_trips_protobuf_bytes() { + let addr = unused_local_addr(); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + + let server_task = tokio::spawn(async move { + DubboServer::new() + .with_listener("tcp".to_string()) + .add_service("grpc.examples.echo.Echo".to_string(), RawUnaryService) + .serve_with_graceful(addr, async move { + let _ = shutdown_rx.await; + }) + .await + .unwrap(); + }); + + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + let endpoint = format!("http://{addr}?interface=grpc.examples.echo.Echo"); + let builder = ClientBuilder::from_static(&endpoint).with_direct(true); + let mut client = TripleClient::new(builder); + let path = http::uri::PathAndQuery::from_static("/grpc.examples.echo.Echo/UnaryEcho"); + let invocation = RpcInvocation::default() + .with_service_unique_name("grpc.examples.echo.Echo".to_string()) + .with_method_name("UnaryEcho".to_string()); + let request = Request::from_parts( + dubbo::invocation::Metadata::default() + .insert("tri-service-group".to_string(), "test".to_string()) + .insert("tri-service-version".to_string(), "1.0.0".to_string()), + Bytes::from_static(b"\x0a\x08dubbo-js"), + ); + + let response = client.raw_unary(request, path, invocation).await.unwrap(); + + let (_, response_bytes) = response.into_parts(); + assert_eq!(response_bytes, Bytes::from_static(b"\x0a\x0draw response")); + + let _ = shutdown_tx.send(()); + server_task.await.unwrap(); +} + +fn grpc_frame(payload: Bytes) -> Bytes { + let mut frame = BytesMut::with_capacity(5 + payload.len()); + frame.put_u8(0); + frame.put_u32(payload.len() as u32); + frame.extend_from_slice(&payload); + frame.freeze() +} + +fn unused_local_addr() -> SocketAddr { + std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() +} diff --git a/registry/zookeeper/src/lib.rs b/registry/zookeeper/src/lib.rs index 8de2bd16..1812aaef 100644 --- a/registry/zookeeper/src/lib.rs +++ b/registry/zookeeper/src/lib.rs @@ -31,8 +31,11 @@ use tokio::{select, sync::mpsc}; use zookeeper::{Acl, CreateMode, WatchedEvent, WatchedEventType, Watcher, ZooKeeper}; use dubbo::{ - extension::registry_extension::{DiscoverStream, Registry, ServiceChange}, - params::registry_param::InterfaceName, + extension::{ + registry_extension::{DiscoverStream, Registry, ServiceChange}, + Extension, + }, + params::registry_param::{InterfaceName, RegistryUrl}, }; // Get metadata of a service registration from a URL @@ -260,6 +263,27 @@ impl Default for ZookeeperRegistry { } } +#[async_trait] +impl Extension for ZookeeperRegistry { + type Target = Box; + + fn name() -> String { + "zookeeper".to_string() + } + + async fn create(url: Url) -> Result { + // url example: + // extension://0.0.0.0?extension-type=registry&extension-name=zookeeper®istry=zookeeper://127.0.0.1:2181 + let registry_url = url.query::().unwrap(); + let registry_url = registry_url.value(); + let host = registry_url.host().unwrap_or("127.0.0.1"); + let port = registry_url.port().unwrap_or(2181); + let connect_string = format!("{host}:{port}"); + + Ok(Box::new(ZookeeperRegistry::new(&connect_string))) + } +} + #[async_trait] impl Registry for ZookeeperRegistry { async fn register(&self, url: Url) -> Result<(), StdError> {