1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275
//! Provides the builder and implementation of [`GrpcService`] that enables
//! periodic service discovery.
use crate::{
service_probe::{GrpcServiceProbe, GrpcServiceProbeConfig},
DnsResolver, LookupService, ServiceDefinition,
};
use anyhow::Context as _;
use http::Request;
use std::{
convert::TryInto,
task::{Context, Poll},
};
use tokio::time::Duration;
use tonic::client::GrpcService;
use tonic::transport::channel::Channel;
use tonic::{body::BoxBody, transport::ClientTlsConfig};
use tower::Service;
// Determines the channel size of the channel we use
// to report endpoint changes to tonic.
// This is effectively how many changes we can report in one go.
// We set the number high to avoid any blocking on our side.
static GRPC_REPORT_ENDPOINTS_CHANNEL_SIZE: usize = 1024;
/// Implements tonic [`GrpcService`] for a client-side load balanced [`Channel`] (using `The Power of
/// Two Choices`).
///
/// [`GrpcService`]
///
/// ```rust
/// #[tokio::main]
/// async fn main() {
/// use ginepro::LoadBalancedChannel;
/// use shared_proto::pb::tester_client::TesterClient;
/// use std::convert::TryInto;
///
/// let load_balanced_channel = LoadBalancedChannel::builder(("my.hostname", 5000))
/// .channel()
/// .await
/// .expect("failed to construct LoadBalancedChannel");
///
/// let client = TesterClient::new(load_balanced_channel);
/// }
/// ```
///
#[derive(Debug, Clone)]
pub struct LoadBalancedChannel(Channel);
impl From<LoadBalancedChannel> for Channel {
fn from(channel: LoadBalancedChannel) -> Self {
channel.0
}
}
impl LoadBalancedChannel {
/// Start configuring a `LoadBalancedChannel` by passing in the [`ServiceDefinition`]
/// for the gRPC server service you want to call - e.g. `my.service.uri` and `5000`.
///
/// All the service endpoints of a [`ServiceDefinition`] will be
/// constructed by resolving IPs for [`ServiceDefinition::hostname`], and
/// using the port number [`ServiceDefinition::port`].
pub fn builder<S>(service_definition: S) -> LoadBalancedChannelBuilder<DnsResolver, S>
where
S: TryInto<ServiceDefinition> + Send + Sync + 'static,
S::Error: Into<Box<dyn std::error::Error + Send + Sync>> + Send + Sync,
{
LoadBalancedChannelBuilder::new_with_service(service_definition)
}
}
impl Service<http::Request<BoxBody>> for LoadBalancedChannel {
type Response = http::Response<<Channel as GrpcService<BoxBody>>::ResponseBody>;
type Error = <Channel as GrpcService<BoxBody>>::Error;
type Future = <Channel as GrpcService<BoxBody>>::Future;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
GrpcService::poll_ready(&mut self.0, cx)
}
fn call(&mut self, request: Request<BoxBody>) -> Self::Future {
GrpcService::call(&mut self.0, request)
}
}
/// Enumerates the different domain name resolution strategies that
/// the [`LoadBalancedChannelBuilder`] supports.
pub enum ResolutionStrategy {
/// Creates the channel without attempting to resolve
/// a set of initial IPs.
Lazy,
/// Tries to resolve the domain name before creating the channel
/// in order to start with a non-empty set of IPs.
Eager { timeout: Duration },
}
/// Builder to configure and create a [`LoadBalancedChannel`].
pub struct LoadBalancedChannelBuilder<T, S> {
service_definition: S,
probe_interval: Option<Duration>,
resolution_strategy: ResolutionStrategy,
timeout: Option<Duration>,
connect_timeout: Option<Duration>,
tls_config: Option<ClientTlsConfig>,
lookup_service: Option<T>,
}
impl<S> LoadBalancedChannelBuilder<DnsResolver, S>
where
S: TryInto<ServiceDefinition> + 'static,
S::Error: Into<Box<dyn std::error::Error + Send + Sync>> + Send + Sync,
{
/// Set the [`ServiceDefinition`] of the gRPC server service
/// - e.g. `my.service.uri` and `5000`.
///
/// All the service endpoints of a [`ServiceDefinition`] will be
/// constructed by resolving all ips from [`ServiceDefinition::hostname`], and
/// using the portnumber [`ServiceDefinition::port`].
pub fn new_with_service(service_definition: S) -> LoadBalancedChannelBuilder<DnsResolver, S> {
Self {
service_definition,
probe_interval: None,
timeout: None,
connect_timeout: None,
tls_config: None,
lookup_service: None,
resolution_strategy: ResolutionStrategy::Lazy,
}
}
/// Set a custom [`LookupService`].
pub fn lookup_service<T: LookupService + Send + Sync + 'static>(
self,
lookup_service: T,
) -> LoadBalancedChannelBuilder<T, S> {
LoadBalancedChannelBuilder {
lookup_service: Some(lookup_service),
service_definition: self.service_definition,
probe_interval: self.probe_interval,
tls_config: self.tls_config,
timeout: self.timeout,
connect_timeout: self.connect_timeout,
resolution_strategy: self.resolution_strategy,
}
}
}
impl<T: LookupService + Send + Sync + 'static + Sized, S> LoadBalancedChannelBuilder<T, S>
where
S: TryInto<ServiceDefinition> + 'static,
S::Error: Into<Box<dyn std::error::Error + Send + Sync>> + Send + Sync,
{
/// Set the how often, the client should probe for changes to gRPC server endpoints.
/// Default interval in seconds is 10.
pub fn dns_probe_interval(self, interval: Duration) -> LoadBalancedChannelBuilder<T, S> {
Self {
probe_interval: Some(interval),
..self
}
}
/// Set a request timeout that will be applied to every new `Endpoint`.
pub fn timeout(self, timeout: Duration) -> LoadBalancedChannelBuilder<T, S> {
Self {
timeout: Some(timeout),
..self
}
}
/// Set a connection timeout that will be applied to every new `Endpoint`.
///
/// Defaults to the overall request `timeout` if not set.
pub fn connect_timeout(self, connection_timeout: Duration) -> LoadBalancedChannelBuilder<T, S> {
Self {
connect_timeout: Some(connection_timeout),
..self
}
}
/// Set the [`ResolutionStrategy`].
///
/// Default set to [`ResolutionStrategy::Lazy`].
///
/// If [`ResolutionStrategy::Lazy`] the domain name will be resolved after-the-fact.
///
/// Instead, if [`ResolutionStrategy::Eager`] is set the domain name will be attempted resolved
/// once before the [`LoadBalancedChannel`] is created, which ensures that the channel
/// will have a non-empty of IPs on startup. If it fails the channel creation will also fail.
pub fn resolution_strategy(
self,
resolution_strategy: ResolutionStrategy,
) -> LoadBalancedChannelBuilder<T, S> {
Self {
resolution_strategy,
..self
}
}
/// Configure the channel to use tls.
/// A `tls_config` MUST be specified to use the `HTTPS` scheme.
pub fn with_tls(self, tls_config: ClientTlsConfig) -> LoadBalancedChannelBuilder<T, S> {
Self {
tls_config: Some(tls_config),
..self
}
}
/// Construct a [`LoadBalancedChannel`] from the [`LoadBalancedChannelBuilder`] instance.
pub async fn channel(mut self) -> Result<LoadBalancedChannel, anyhow::Error> {
match self.lookup_service.take() {
Some(lookup_service) => self.channel_inner(lookup_service).await,
None => {
self.channel_inner(DnsResolver::from_system_config().await?)
.await
}
}
}
async fn channel_inner<U: LookupService>(
self,
lookup_service: U,
) -> Result<LoadBalancedChannel, anyhow::Error>
where
U: LookupService + Send + Sync + 'static + Sized,
{
let (channel, sender) = Channel::balance_channel(GRPC_REPORT_ENDPOINTS_CHANNEL_SIZE);
let config = GrpcServiceProbeConfig {
service_definition: self
.service_definition
.try_into()
.map_err(Into::into)
.map_err(|err| anyhow::anyhow!(err))?,
dns_lookup: lookup_service,
endpoint_timeout: self.timeout,
endpoint_connect_timeout: self.connect_timeout.or(self.timeout),
probe_interval: self
.probe_interval
.unwrap_or_else(|| Duration::from_secs(10)),
};
let tls_config = self.tls_config.map(|mut tls_config| {
// Since we resolve the hostname to an IP, which is not a valid DNS name,
// we have to set the hostname explicitly on the tls config,
// otherwise the IP will be set as the domain name and tls handshake will fail.
tls_config = tls_config.domain_name(config.service_definition.hostname());
tls_config
});
let mut service_probe = GrpcServiceProbe::new_with_reporter(config, sender);
if let Some(tls_config) = tls_config {
service_probe = service_probe.with_tls(tls_config);
}
if let ResolutionStrategy::Eager { timeout } = self.resolution_strategy {
// Make sure we resolve the hostname once before we create the channel.
tokio::time::timeout(timeout, service_probe.probe_once())
.await
.context("timeout out while attempting to resolve IPs")?
.context("failed to resolve IPs")?;
}
tokio::spawn(service_probe.probe());
Ok(LoadBalancedChannel(channel))
}
}
const _: () = {
const fn assert_is_send<T: Send>() {}
assert_is_send::<LoadBalancedChannelBuilder<DnsResolver, ServiceDefinition>>();
assert_is_send::<LoadBalancedChannel>();
};