diff --git a/seaweed-volume/src/main.rs b/seaweed-volume/src/main.rs index 844b5c15b..fb9e97106 100644 --- a/seaweed-volume/src/main.rs +++ b/seaweed-volume/src/main.rs @@ -12,7 +12,10 @@ use seaweed_volume::security::tls::{ use seaweed_volume::security::{Guard, SigningKey}; #[cfg(unix)] use seaweed_volume::server::debug::build_debug_router; -use seaweed_volume::server::grpc_client::load_outgoing_grpc_tls; +use seaweed_volume::server::grpc_client::{ + GRPC_INITIAL_WINDOW_SIZE, GRPC_KEEPALIVE_INTERVAL, GRPC_KEEPALIVE_TIMEOUT, + GRPC_MAX_MESSAGE_SIZE, load_outgoing_grpc_tls, +}; use seaweed_volume::server::grpc_server::VolumeGrpcService; #[cfg(unix)] use seaweed_volume::server::profiling::CpuProfileSession; @@ -31,10 +34,10 @@ type CpuProfileParam = Option; #[cfg(not(unix))] type CpuProfileParam = Option<()>; -const GRPC_MAX_MESSAGE_SIZE: usize = 1 << 30; -const GRPC_KEEPALIVE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60); -const GRPC_KEEPALIVE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20); -const GRPC_INITIAL_WINDOW_SIZE: u32 = 16 * 1024 * 1024; +// The two settings that only make sense for the inbound server. The rest of +// this server's HTTP/2 tuning — keepalive, window sizes, message size — is +// imported from `server::grpc_client` above, which is also what the outgoing +// clients dial with, so the two directions cannot drift apart. const GRPC_MAX_HEADER_LIST_SIZE: u32 = 8 * 1024 * 1024; const GRPC_MAX_CONCURRENT_STREAMS: u32 = 1000; diff --git a/seaweed-volume/src/server/grpc_client.rs b/seaweed-volume/src/server/grpc_client.rs index 8a0404f4d..e8c35f4d9 100644 --- a/seaweed-volume/src/server/grpc_client.rs +++ b/seaweed-volume/src/server/grpc_client.rs @@ -1,16 +1,42 @@ +//! Construction of the volume server's *outgoing* gRPC clients: TLS material, +//! endpoint tuning, dial bounds, and the three client constructors every call +//! site goes through. +//! +//! The keepalive, window-size and message-size constants below are shared with +//! the *inbound* server built in `main.rs`, which imports them from here rather +//! than declaring its own. Changing one therefore changes both directions at +//! once, which is deliberate: a volume server talks to its peers with the same +//! HTTP/2 settings it offers them. + use std::error::Error; use std::fmt; use std::time::Duration; use hyper::http::Uri; +use tonic::service::interceptor::InterceptedService; use tonic::transport::{Certificate, Channel, ClientTlsConfig, Endpoint, Identity}; +use tonic::{Request, Status}; use crate::config::VolumeServerConfig; +use crate::pb::filer_pb::seaweed_filer_client::SeaweedFilerClient; +use crate::pb::master_pb::seaweed_client::SeaweedClient; +use crate::pb::volume_server_pb::volume_server_client::VolumeServerClient; +use crate::server::request_id::outgoing_request_id_interceptor; pub const GRPC_MAX_MESSAGE_SIZE: usize = 1 << 30; -const GRPC_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(60); -const GRPC_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(20); -const GRPC_INITIAL_WINDOW_SIZE: u32 = 16 * 1024 * 1024; +pub const GRPC_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(60); +pub const GRPC_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(20); +pub const GRPC_INITIAL_WINDOW_SIZE: u32 = 16 * 1024 * 1024; + +/// Bound on the TCP connect of every outgoing dial. `build_grpc_endpoint` is +/// private and `connect_channel` is the only way out of this module, so every +/// call site picks this up whether it thinks about timeouts or not. +/// +/// It bounds the TCP handshake only — tonic hands it to +/// `HttpConnector::set_connect_timeout`. A peer that completes the handshake +/// and then stalls in the TLS or HTTP/2 exchange is not covered; callers that +/// need that bound wrap the whole dial (see `connect_ping_target`). +const GRPC_CONNECT_TIMEOUT: Duration = Duration::from_secs(5); #[derive(Clone, Debug)] pub struct OutgoingGrpcTlsConfig { @@ -81,7 +107,7 @@ pub fn grpc_endpoint_uri(grpc_host_port: &str, tls: Option<&OutgoingGrpcTlsConfi format!("{}://{}", scheme, grpc_host_port) } -pub fn build_grpc_endpoint( +fn build_grpc_endpoint( grpc_host_port: &str, tls: Option<&OutgoingGrpcTlsConfig>, ) -> Result { @@ -149,6 +175,152 @@ pub async fn connect_guarded( .map_err(|e| GrpcClientError(format!("connect {} failed: {}", target, e))) } +/// How a dial is bounded. +/// +/// `connect_timeout` is handed to the TCP connector. `request_timeout` becomes +/// [`Endpoint::timeout`], which tonic installs as a `GrpcTimeout` layer in +/// front of *every* request the resulting channel carries — it is not a +/// property of one call. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct GrpcDialOptions { + /// Bound on establishing the connection to the peer. + pub connect_timeout: Duration, + /// Deadline applied to each RPC on the channel, or `None` to leave them + /// unbounded. + pub request_timeout: Option, +} + +impl GrpcDialOptions { + /// A short request/response call: connect within 5 s, answer within 10 s. + pub fn unary() -> Self { + Self { + connect_timeout: GRPC_CONNECT_TIMEOUT, + request_timeout: Some(Duration::from_secs(10)), + } + } + + /// A call the peer may take a while to answer: connect within 5 s, answer + /// within 30 s. + pub fn long() -> Self { + Self { + connect_timeout: GRPC_CONNECT_TIMEOUT, + request_timeout: Some(Duration::from_secs(30)), + } + } + + /// A bounded connect with no deadline on the RPCs themselves. + /// + /// `request_timeout` must stay `None` here. [`Endpoint::timeout`] is not a + /// transfer budget: tonic layers it as a `GrpcTimeout` around the + /// response future, which resolves when the server's *first response + /// headers* arrive, so it bounds how long the peer may take to start + /// answering — per request, for every request the channel carries. A 10 s + /// value picked to suit one short call would therefore also be the header + /// deadline for the `VolumeCopy` that shares the dial, and a busy source + /// that takes longer than that to open its file would lose the whole copy. + /// `VolumeCopy`, `VolumeTailSender` and `VolumeEcShardsCopy` have never + /// carried one. + pub fn stream() -> Self { + Self { + connect_timeout: GRPC_CONNECT_TIMEOUT, + request_timeout: None, + } + } +} + +/// Dial a peer and return a connected channel. +/// +/// The error carries only the transport failure: every caller already wraps it +/// with the address and the operation it was attempting. +pub async fn connect_channel( + grpc_host_port: &str, + tls: Option<&OutgoingGrpcTlsConfig>, + opts: GrpcDialOptions, +) -> Result { + let mut endpoint = + build_grpc_endpoint(grpc_host_port, tls)?.connect_timeout(opts.connect_timeout); + if let Some(request_timeout) = opts.request_timeout { + endpoint = endpoint.timeout(request_timeout); + } + endpoint + .connect() + .await + .map_err(|e| GrpcClientError(e.to_string())) +} + +/// Dial a copy/tail source and return a connected channel, re-validating every +/// resolved address at connect time. +/// +/// The guarded equivalent of [`connect_channel`]: same `opts` bounds, but the +/// dial goes through [`connect_guarded`] so a source address that passed +/// validation cannot be re-pointed by DNS between the check and the connect. +/// The bounds are applied to the endpoint *before* delegating, so the +/// `allow_untrusted` opt-out is timed too. +/// +/// `target` is the caller-facing source address (the unparsed +/// `"ip:port.grpcPort"` form), which is what the guard pins against; the error +/// carries only the transport failure, as every caller already wraps it with +/// the address and the operation it was attempting. +pub async fn connect_channel_guarded( + grpc_host_port: &str, + target: &str, + tls: Option<&OutgoingGrpcTlsConfig>, + opts: GrpcDialOptions, + allow_untrusted: bool, +) -> Result { + let mut endpoint = + build_grpc_endpoint(grpc_host_port, tls)?.connect_timeout(opts.connect_timeout); + if let Some(request_timeout) = opts.request_timeout { + endpoint = endpoint.timeout(request_timeout); + } + connect_guarded(endpoint, target, allow_untrusted).await +} + +/// The outgoing request-id interceptor as a concrete type, so the client +/// aliases below can name it. +pub type RequestIdInterceptor = fn(Request<()>) -> Result, Status>; + +/// A volume-server client with the request-id interceptor attached. +pub type VolumeServerGrpcClient = + VolumeServerClient>; +/// A master client with the request-id interceptor attached. +pub type MasterGrpcClient = SeaweedClient>; +/// A filer client with the request-id interceptor attached. +pub type FilerGrpcClient = SeaweedFilerClient>; + +/// Wrap a connected channel in a volume-server client that forwards the +/// current request id and lifts both message-size limits. +pub fn volume_server_client(channel: Channel) -> VolumeServerGrpcClient { + VolumeServerClient::with_interceptor( + channel, + outgoing_request_id_interceptor as RequestIdInterceptor, + ) + .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) + .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE) +} + +/// Wrap a connected channel in a master client that forwards the current +/// request id and lifts both message-size limits. +pub fn master_client(channel: Channel) -> MasterGrpcClient { + SeaweedClient::with_interceptor( + channel, + outgoing_request_id_interceptor as RequestIdInterceptor, + ) + .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) + .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE) +} + +/// Wrap a connected channel in a filer client that forwards the current +/// request id and lifts both message-size limits. +pub fn filer_client(channel: Channel) -> FilerGrpcClient { + SeaweedFilerClient::with_interceptor( + channel, + outgoing_request_id_interceptor as RequestIdInterceptor, + ) + .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) + .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE) +} + /// Parse a SeaweedFS server address (`"ip:port.grpcPort"` or /// `"ip:port"`) into the `host:grpcPort` form `build_grpc_endpoint` /// expects. With the trailing `.grpcPort` segment, that segment IS @@ -169,9 +341,16 @@ pub fn parse_grpc_address(source: &str) -> Result { #[cfg(test)] mod tests { - use super::{build_grpc_endpoint, grpc_endpoint_uri, load_outgoing_grpc_tls}; + use super::{ + GrpcDialOptions, build_grpc_endpoint, connect_channel, grpc_endpoint_uri, + load_outgoing_grpc_tls, volume_server_client, + }; use crate::config::{NeedleMapKind, ReadMode, VolumeServerConfig}; + use crate::pb::volume_server_pb; use crate::security::tls::TlsPolicy; + use crate::server::request_id::scope_request_id; + use std::sync::{Arc, Mutex}; + use std::time::Duration; const TEST_CERT_PEM: &str = "-----BEGIN CERTIFICATE-----\nMIIBPDCB76ADAgECAhRuRPQgeAu43BT/M7EfAWSdapVdYDAFBgMrZXAwFDESMBAG\nA1UEAwwJbG9jYWxob3N0MB4XDTI2MDcwNTE2MTUyOVoXDTM2MDcwMjE2MTUyOVow\nFDESMBAGA1UEAwwJbG9jYWxob3N0MCowBQYDK2VwAyEAr/3bNIFI+8V32oCiY6y+\nXRFmZpdNQ2g//VtRkT+nQg+jUzBRMB0GA1UdDgQWBBTsy9tLf1zPiXCQfgci6zNi\ndEzRSjAfBgNVHSMEGDAWgBTsy9tLf1zPiXCQfgci6zNidEzRSjAPBgNVHRMBAf8E\nBTADAQH/MAUGAytlcANBAIvsdw0IbvOBBkb9cd7BfMJfIP9pQQrAL03pCRWJFnFh\nSysaLVgFXI4T078IiaM874oO+iB+5vNbWEpc7CkGow4=\n-----END CERTIFICATE-----\n"; const TEST_KEY_PEM: &str = "-----BEGIN PRIVATE KEY-----\nMC4CAQAwBQYDK2VwBCIEIHbyn71Kk+Y7KT3sBctit7uZpErpoH6qDbFj6P8qGaZH\n-----END PRIVATE KEY-----\n"; @@ -385,4 +564,112 @@ mod tests { let endpoint = build_grpc_endpoint(&parse_grpc_address("::1:9333").unwrap(), None).unwrap(); assert_eq!(endpoint.uri().port_u16(), Some(19333)); } + /// A minimal HTTP/2 server that records the gRPC request headers it is + /// sent and answers every call with a trailers-only `unimplemented`. It is + /// enough to prove what a helper-built client puts on the wire, without + /// standing up the whole `VolumeServer` service behind a tonic server. + async fn serve_header_capture() -> (u16, Arc>>) { + use hyper::service::service_fn; + use hyper_util::rt::{TokioExecutor, TokioIo}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let seen: Arc>> = Arc::new(Mutex::new(None)); + let captured = Arc::clone(&seen); + + tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let captured = Arc::clone(&captured); + tokio::spawn(async move { + let _ = hyper::server::conn::http2::Builder::new(TokioExecutor::new()) + .serve_connection( + TokioIo::new(stream), + service_fn(move |req: hyper::Request| { + let captured = Arc::clone(&captured); + async move { + let value = req + .headers() + .get("x-amz-request-id") + .and_then(|v| v.to_str().ok()) + .map(str::to_string); + *captured.lock().unwrap() = value; + Ok::<_, std::convert::Infallible>( + hyper::http::Response::builder() + .status(200) + .header("content-type", "application/grpc") + .header("grpc-status", "12") + .body(tonic::body::Body::empty()) + .unwrap(), + ) + } + }), + ) + .await; + }); + } + }); + + (port, seen) + } + + #[tokio::test] + async fn test_helper_built_client_sends_the_scoped_request_id() { + let (port, seen) = serve_header_capture().await; + + let channel = connect_channel( + &format!("127.0.0.1:{}", port), + None, + GrpcDialOptions::unary(), + ) + .await + .expect("dial the header-capturing server"); + + let mut client = volume_server_client(channel); + // The interceptor has a request id to forward only inside a scope, so + // the call has to run inside one for this to test anything. + let _ = scope_request_id("REQUEST-ID-ON-THE-WIRE".to_string(), async move { + client + .ping(volume_server_pb::PingRequest { + target: String::new(), + target_type: String::new(), + }) + .await + }) + .await; + + assert_eq!( + seen.lock().unwrap().as_deref(), + Some("REQUEST-ID-ON-THE-WIRE"), + "a client built by volume_server_client must carry the outgoing request id" + ); + } + + #[test] + fn test_dial_presets_match_the_call_sites_they_replace() { + assert_eq!( + GrpcDialOptions::unary().connect_timeout, + Duration::from_secs(5) + ); + assert_eq!( + GrpcDialOptions::unary().request_timeout, + Some(Duration::from_secs(10)) + ); + assert_eq!( + GrpcDialOptions::long().connect_timeout, + Duration::from_secs(5) + ); + assert_eq!( + GrpcDialOptions::long().request_timeout, + Some(Duration::from_secs(30)) + ); + assert_eq!( + GrpcDialOptions::stream().connect_timeout, + Duration::from_secs(5) + ); + assert_eq!( + GrpcDialOptions::stream().request_timeout, + None, + "a streaming dial must not put a per-request deadline on the channel" + ); + } } diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index ef0403540..1b298e6fb 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -14,7 +14,6 @@ use tonic::{Request, Response, Status, Streaming}; use crate::pb::filer_pb; use crate::pb::master_pb; -use crate::pb::master_pb::seaweed_client::SeaweedClient; use crate::pb::volume_server_pb; use crate::pb::volume_server_pb::volume_server_server::VolumeServer; use crate::storage::erasure_coding::ec_shard::{DATA_SHARDS_COUNT, ShardId, shard_id_try_from}; @@ -22,7 +21,10 @@ use crate::storage::needle::needle::{self, Needle}; use crate::storage::types::*; use crate::storage::volume::VolumeSpec; -use super::grpc_client::{GRPC_MAX_MESSAGE_SIZE, build_grpc_endpoint}; +use super::grpc_client::{ + GrpcDialOptions, connect_channel, connect_channel_guarded, filer_client, master_client, + volume_server_client, +}; use super::volume_server::VolumeServerState; type BoxStream = Pin> + Send + 'static>>; @@ -337,20 +339,14 @@ impl VolumeGrpcService { let grpc_addr = parse_grpc_address(&master_url).map_err(|e| { Status::internal(format!("invalid master address {}: {}", master_url, e)) })?; - let endpoint = build_grpc_endpoint(&grpc_addr, self.state.outgoing_grpc_tls.as_ref()) - .map_err(|e| Status::internal(format!("master address {}: {}", master_url, e)))? - .connect_timeout(std::time::Duration::from_secs(5)) - .timeout(std::time::Duration::from_secs(30)); - let channel = endpoint - .connect() - .await - .map_err(|e| Status::internal(format!("connect to master {}: {}", master_url, e)))?; - let mut client = SeaweedClient::with_interceptor( - channel, - super::request_id::outgoing_request_id_interceptor, + let channel = connect_channel( + &grpc_addr, + self.state.outgoing_grpc_tls.as_ref(), + GrpcDialOptions::long(), ) - .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) - .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + .await + .map_err(|e| Status::internal(format!("connect to master {}: {}", master_url, e)))?; + let mut client = master_client(channel); client .volume_mark_readonly(master_pb::VolumeMarkReadonlyRequest { ip: info.ip.clone(), @@ -1844,13 +1840,11 @@ impl VolumeServer for VolumeGrpcService { )) })?; - let endpoint = build_grpc_endpoint(&grpc_addr, self.state.outgoing_grpc_tls.as_ref()) - .map_err(|e| { - Status::internal(format!("VolumeCopy volume {} parse source: {}", vid, e)) - })?; - let channel = super::grpc_client::connect_guarded( - endpoint, + let channel = connect_channel_guarded( + &grpc_addr, source, + self.state.outgoing_grpc_tls.as_ref(), + GrpcDialOptions::stream(), self.state.allow_untrusted_remote_endpoints, ) .await @@ -1861,13 +1855,7 @@ impl VolumeServer for VolumeGrpcService { )) })?; - let mut client = - volume_server_pb::volume_server_client::VolumeServerClient::with_interceptor( - channel, - super::request_id::outgoing_request_id_interceptor, - ) - .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) - .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + let mut client = volume_server_client(channel); // Get file status from source let vol_info = client @@ -2932,23 +2920,17 @@ impl VolumeServer for VolumeGrpcService { let grpc_addr = parse_grpc_address(source) .map_err(|e| Status::internal(format!("invalid source address {}: {}", source, e)))?; - let endpoint = build_grpc_endpoint(&grpc_addr, self.state.outgoing_grpc_tls.as_ref()) - .map_err(|e| Status::internal(format!("parse source: {}", e)))?; - let channel = super::grpc_client::connect_guarded( - endpoint, + let channel = connect_channel_guarded( + &grpc_addr, source, + self.state.outgoing_grpc_tls.as_ref(), + GrpcDialOptions::stream(), self.state.allow_untrusted_remote_endpoints, ) .await .map_err(|e| Status::internal(format!("connect to {}: {}", grpc_addr, e)))?; - let mut client = - volume_server_pb::volume_server_client::VolumeServerClient::with_interceptor( - channel, - super::request_id::outgoing_request_id_interceptor, - ) - .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) - .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + let mut client = volume_server_client(channel); // Call VolumeTailSender on source let mut stream = client @@ -3449,16 +3431,11 @@ impl VolumeServer for VolumeGrpcService { )) })?; - let endpoint = build_grpc_endpoint(&grpc_addr, self.state.outgoing_grpc_tls.as_ref()) - .map_err(|e| { - Status::internal(format!( - "VolumeEcShardsCopy volume {} parse source: {}", - vid, e - )) - })?; - let channel = super::grpc_client::connect_guarded( - endpoint, + let channel = connect_channel_guarded( + &grpc_addr, source, + self.state.outgoing_grpc_tls.as_ref(), + GrpcDialOptions::stream(), self.state.allow_untrusted_remote_endpoints, ) .await @@ -3469,13 +3446,7 @@ impl VolumeServer for VolumeGrpcService { )) })?; - let mut client = - volume_server_pb::volume_server_client::VolumeServerClient::with_interceptor( - channel, - super::request_id::outgoing_request_id_interceptor, - ) - .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) - .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + let mut client = volume_server_client(channel); // Copy each shard for &shard_id in &shard_ids { @@ -5414,13 +5385,26 @@ impl VolumeServer for VolumeGrpcService { } } -/// Build a gRPC endpoint from a SeaweedFS server address. -fn to_grpc_endpoint( +/// Dial a ping target, bounding the whole connect at 5s. +/// +/// The outer timeout is not redundant with `GrpcDialOptions`' connect timeout: +/// tonic hands that one to the HTTP connector, so it bounds the TCP dial only. +/// A ping to a TLS peer that accepts the connection and then stalls in the +/// handshake needs this wrapper to come back at all. +async fn connect_ping_target( target: &str, tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>, -) -> Result { +) -> Result { let grpc_host_port = parse_grpc_address(target)?; - build_grpc_endpoint(&grpc_host_port, tls).map_err(|e| e.to_string()) + // Ping is unary, but `stream()` is still right: these three have no + // per-request deadline today, and `unary()` would add a 10 s one. + tokio::time::timeout( + std::time::Duration::from_secs(5), + connect_channel(&grpc_host_port, tls, GrpcDialOptions::stream()), + ) + .await + .map_err(|_| "connection timeout".to_string())? + .map_err(|e| e.to_string()) } /// Ping a remote volume server target by actually calling its Ping RPC (matches Go behavior). @@ -5428,18 +5412,7 @@ async fn ping_volume_server_target( target: &str, tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>, ) -> Result { - let endpoint = to_grpc_endpoint(target, tls)?; - let channel = tokio::time::timeout(std::time::Duration::from_secs(5), endpoint.connect()) - .await - .map_err(|_| "connection timeout".to_string())? - .map_err(|e| e.to_string())?; - - let mut client = volume_server_pb::volume_server_client::VolumeServerClient::with_interceptor( - channel, - super::request_id::outgoing_request_id_interceptor, - ) - .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) - .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + let mut client = volume_server_client(connect_ping_target(target, tls).await?); let resp = client .ping(volume_server_pb::PingRequest { target: String::new(), @@ -5455,18 +5428,7 @@ async fn ping_master_target( target: &str, tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>, ) -> Result { - let endpoint = to_grpc_endpoint(target, tls)?; - let channel = tokio::time::timeout(std::time::Duration::from_secs(5), endpoint.connect()) - .await - .map_err(|_| "connection timeout".to_string())? - .map_err(|e| e.to_string())?; - - let mut client = master_pb::seaweed_client::SeaweedClient::with_interceptor( - channel, - super::request_id::outgoing_request_id_interceptor, - ) - .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) - .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + let mut client = master_client(connect_ping_target(target, tls).await?); let resp = client .ping(master_pb::PingRequest { target: String::new(), @@ -5482,18 +5444,7 @@ async fn ping_filer_target( target: &str, tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>, ) -> Result { - let endpoint = to_grpc_endpoint(target, tls)?; - let channel = tokio::time::timeout(std::time::Duration::from_secs(5), endpoint.connect()) - .await - .map_err(|_| "connection timeout".to_string())? - .map_err(|e| e.to_string())?; - - let mut client = filer_pb::seaweed_filer_client::SeaweedFilerClient::with_interceptor( - channel, - super::request_id::outgoing_request_id_interceptor, - ) - .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) - .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + let mut client = filer_client(connect_ping_target(target, tls).await?); let resp = client .ping(filer_pb::PingRequest::default()) .await @@ -5997,6 +5948,7 @@ mod tests { use crate::config::MinFreeSpace; use crate::remote_storage::s3_tier::{S3TierBackend, S3TierConfig, global_s3_tier_registry}; use crate::security::{Guard, SigningKey}; + use crate::server::grpc_client::GRPC_MAX_MESSAGE_SIZE; use crate::storage::needle_map::NeedleMapKind; use crate::storage::store::Store; use std::sync::RwLock; diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index f209faf5b..929248719 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -16,7 +16,7 @@ use axum::response::{IntoResponse, Response}; use serde::{Deserialize, Serialize}; use super::absolute_display_path; -use super::grpc_client::{GRPC_MAX_MESSAGE_SIZE, build_grpc_endpoint}; +use super::grpc_client::{GrpcDialOptions, connect_channel, volume_server_client}; use super::volume_server::{VolumeServerState, normalize_outgoing_http_url, to_http_address}; use crate::config::ReadMode; use crate::metrics; @@ -541,19 +541,16 @@ async fn batch_delete_file_ids( } for (grpc_addr, batch) in server_to_file_ids { - let endpoint = build_grpc_endpoint(&grpc_addr, state.outgoing_grpc_tls.as_ref()) - .map_err(|e| format!("batch delete {}: {}", grpc_addr, e))?; - let channel = endpoint - .connect() - .await - .map_err(|e| format!("batch delete {}: {}", grpc_addr, e))?; - let mut client = - volume_server_pb::volume_server_client::VolumeServerClient::with_interceptor( - channel, - super::request_id::outgoing_request_id_interceptor, - ) - .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) - .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + // BatchDelete is unary, but `stream()` is still right: this fan-out has + // no per-request deadline today, and `unary()` would add a 10 s one. + let channel = connect_channel( + &grpc_addr, + state.outgoing_grpc_tls.as_ref(), + GrpcDialOptions::stream(), + ) + .await + .map_err(|e| format!("batch delete {}: {}", grpc_addr, e))?; + let mut client = volume_server_client(channel); let response = client .batch_delete(volume_server_pb::BatchDeleteRequest { diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index eb19ec78e..0f8363f3a 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -12,10 +12,9 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::broadcast; use tracing::{error, info, warn}; -use super::grpc_client::{GRPC_MAX_MESSAGE_SIZE, build_grpc_endpoint}; +use super::grpc_client::{GrpcDialOptions, connect_channel, master_client}; use super::volume_server::VolumeServerState; use crate::pb::master_pb; -use crate::pb::master_pb::seaweed_client::SeaweedClient; use crate::pb::volume_server_pb; use crate::remote_storage::s3_tier::{S3TierBackend, S3TierConfig}; use crate::storage::store::Store; @@ -245,17 +244,8 @@ pub async fn try_get_master_configuration( grpc_addr: &str, tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>, ) -> Result> { - let channel = build_grpc_endpoint(grpc_addr, tls)? - .connect_timeout(Duration::from_secs(5)) - .timeout(Duration::from_secs(10)) - .connect() - .await?; - let mut client = SeaweedClient::with_interceptor( - channel, - super::request_id::outgoing_request_id_interceptor, - ) - .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) - .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + let channel = connect_channel(grpc_addr, tls, GrpcDialOptions::unary()).await?; + let mut client = master_client(channel); let resp = client .get_master_configuration(master_pb::GetMasterConfigurationRequest {}) .await?; @@ -390,18 +380,14 @@ async fn do_heartbeat( pulse: Duration, shutdown_rx: &mut broadcast::Receiver<()>, ) -> Result, Box> { - let channel = build_grpc_endpoint(grpc_addr, state.outgoing_grpc_tls.as_ref())? - .connect_timeout(Duration::from_secs(5)) - .timeout(Duration::from_secs(30)) - .connect() - .await?; - - let mut client = SeaweedClient::with_interceptor( - channel, - super::request_id::outgoing_request_id_interceptor, + let channel = connect_channel( + grpc_addr, + state.outgoing_grpc_tls.as_ref(), + GrpcDialOptions::long(), ) - .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) - .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + .await?; + + let mut client = master_client(channel); let (tx, rx) = tokio::sync::mpsc::channel::(32); diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index 0d48b9e51..0c60d01b8 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -38,12 +38,11 @@ use reed_solomon_erasure::galois_8::ReedSolomon; use tokio::sync::Semaphore; use tonic::Request; -use crate::pb::master_pb::{self, LookupEcVolumeRequest, seaweed_client::SeaweedClient}; -use crate::pb::volume_server_pb::{ - CopyFileRequest, VolumeEcShardReadRequest, volume_server_client::VolumeServerClient, +use crate::pb::master_pb::{self, LookupEcVolumeRequest}; +use crate::pb::volume_server_pb::{CopyFileRequest, VolumeEcShardReadRequest}; +use crate::server::grpc_client::{ + GrpcDialOptions, connect_channel, master_client, parse_grpc_address, volume_server_client, }; -use crate::server::grpc_client::{GRPC_MAX_MESSAGE_SIZE, build_grpc_endpoint, parse_grpc_address}; -use crate::server::request_id::outgoing_request_id_interceptor; use crate::server::volume_server::{VolumeServerState, to_http_address}; use crate::storage::erasure_coding::ec_shard::{ShardId, shard_id_try_from}; use crate::storage::needle::needle::{Needle, NeedleError, get_actual_size}; @@ -868,18 +867,15 @@ async fn cached_lookup_ec_shard_locations( let grpc_addr = parse_grpc_address(&master).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; - let endpoint = build_grpc_endpoint(&grpc_addr, state.outgoing_grpc_tls.as_ref()) - .map_err(|e| io::Error::other(e.to_string()))?; - let channel = endpoint - .connect_timeout(Duration::from_secs(5)) - .timeout(Duration::from_secs(10)) - .connect() - .await - .map_err(|e| io::Error::other(format!("master connect: {}", e)))?; + let channel = connect_channel( + &grpc_addr, + state.outgoing_grpc_tls.as_ref(), + GrpcDialOptions::unary(), + ) + .await + .map_err(|e| io::Error::other(format!("master connect: {}", e)))?; - let mut client = SeaweedClient::with_interceptor(channel, outgoing_request_id_interceptor) - .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) - .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + let mut client = master_client(channel); let resp = client .lookup_ec_volume(Request::new(LookupEcVolumeRequest { volume_id: vid.0 })) @@ -1060,14 +1056,13 @@ async fn do_read_remote_ec_shard_interval( } = iv; let grpc_addr = parse_grpc_address(source).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; - let endpoint = build_grpc_endpoint(&grpc_addr, state.outgoing_grpc_tls.as_ref()) - .map_err(|e| io::Error::other(e.to_string()))?; - let channel = endpoint - .connect_timeout(Duration::from_secs(5)) - .timeout(Duration::from_secs(30)) - .connect() - .await - .map_err(|e| io::Error::other(format!("connect to {}: {}", source, e)))?; + let channel = connect_channel( + &grpc_addr, + state.outgoing_grpc_tls.as_ref(), + GrpcDialOptions::long(), + ) + .await + .map_err(|e| io::Error::other(format!("connect to {}: {}", source, e)))?; // TODO(grpc-jwt): clusters with `jwt.signing.key` configured will // reject peer-to-peer VolumeEcShardRead calls until the Rust @@ -1077,9 +1072,7 @@ async fn do_read_remote_ec_shard_interval( // here in isolation would split the credential plumbing across // call sites. Re-visit when outgoing JWT signing lands as a // server-wide helper. - let mut client = VolumeServerClient::with_interceptor(channel, outgoing_request_id_interceptor) - .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) - .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + let mut client = volume_server_client(channel); let req = VolumeEcShardReadRequest { volume_id: vid.0, @@ -1467,16 +1460,14 @@ async fn fetch_ec_index_from_one_peer( ) -> io::Result<()> { let grpc_addr = parse_grpc_address(peer).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; - let channel = build_grpc_endpoint(&grpc_addr, state.outgoing_grpc_tls.as_ref()) - .map_err(|e| io::Error::other(e.to_string()))? - .connect_timeout(Duration::from_secs(5)) - .timeout(Duration::from_secs(30)) - .connect() - .await - .map_err(|e| io::Error::other(format!("connect {}: {}", peer, e)))?; - let mut client = VolumeServerClient::with_interceptor(channel, outgoing_request_id_interceptor) - .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) - .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + let channel = connect_channel( + &grpc_addr, + state.outgoing_grpc_tls.as_ref(), + GrpcDialOptions::long(), + ) + .await + .map_err(|e| io::Error::other(format!("connect {}: {}", peer, e)))?; + let mut client = volume_server_client(channel); let copy_req = |ext: &str, ignore_not_found: bool| CopyFileRequest { volume_id: m.vid.0,