diff --git a/Cargo.lock b/Cargo.lock index dbe6096c6..962dd05f6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2288,11 +2288,13 @@ dependencies = [ "datafusion-distributed", "datafusion-distributed-benchmarks", "futures", + "hyper-util", "ratatui", "structopt", "tokio", "tokio-stream", "tonic", + "tower", "url", ] @@ -5455,6 +5457,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ef86cd5876211988985292b91c96a8f2d298df24e75989a43a3c73f2d4d8168b" dependencies = [ "aws-lc-rs", + "log", "once_cell", "ring", "rustls-pki-types", @@ -6387,9 +6390,11 @@ dependencies = [ "hyper-util", "percent-encoding", "pin-project", + "rustls-native-certs", "socket2 0.6.3", "sync_wrapper", "tokio", + "tokio-rustls 0.26.4", "tokio-stream", "tower", "tower-layer", diff --git a/console/Cargo.toml b/console/Cargo.toml index 9f2512de9..1d1648589 100644 --- a/console/Cargo.toml +++ b/console/Cargo.toml @@ -11,7 +11,12 @@ crossterm = "0.29.0" futures = "0.3.31" ratatui = "0.30.0" tokio = { version = "1.49.0", features = ["full"] } -tonic = "0.14.2" +# `tls-ring` turns on the TLS stack; `tls-native-roots` lets `--worker-origin` reach a worker +# whose certificate chains to a CA the machine already trusts, without `--ca-cert`. +tonic = { version = "0.14.2", features = ["tls-ring", "tls-native-roots"] } +# Dialing an address the endpoint URI does not name needs a custom connector. +tower = { version = "0.5.2", features = ["util"] } +hyper-util = { version = "0.1.16", features = ["tokio"] } datafusion-distributed = { path = "..", features = ["integration", "system-metrics"] } url = "2.5.7" tokio-stream = "0.1.18" diff --git a/console/src/app.rs b/console/src/app.rs index 6f0f2cb2a..27d4cc446 100644 --- a/console/src/app.rs +++ b/console/src/app.rs @@ -78,11 +78,12 @@ impl App { self.maybe_discover_workers().await; // Attempt connection for workers in Connecting or Disconnected state - for worker in &mut self.workers { - if worker.should_retry_connection() { - worker.try_connect().await; - } - } + let reconnect_workers = self + .workers + .iter_mut() + .filter(|worker| worker.should_retry_connection()) + .map(|worker| worker.try_connect()); + futures::future::join_all(reconnect_workers).await; // Poll all connected workers in parallel with timeout let poll_workers: Vec<_> = self diff --git a/console/src/connector.rs b/console/src/connector.rs index 006944bf9..c5ab55822 100644 --- a/console/src/connector.rs +++ b/console/src/connector.rs @@ -1,6 +1,61 @@ use datafusion_distributed::grpc::ObservabilityServiceClient; -use tonic::transport::{Channel, Error}; -use url::Url; +use hyper_util::rt::TokioIo; +use std::time::Duration; +use tokio::net::TcpStream; +use tonic::transport::{Certificate, Channel, ClientTlsConfig, Endpoint}; +use url::{Host, ParseError, Url}; + +/// Anything that can go wrong opening a worker connection. +/// +/// Dialing now spans URL handling, TLS setup and the transport itself, and `tonic` does not let +/// callers build its transport error, so the failures are boxed behind their `Display`, which is +/// all the console does with them. +pub(crate) type ConnectError = Box; + +/// Budget for opening a worker connection: the TCP connect plus, when the worker is behind TLS, +/// the handshake. +/// +/// The console dials on the UI thread's tick, so an unreachable worker must fail rather than +/// stall the whole cluster view; a second is long enough for a handshake across a region and +/// short enough to notice. +pub(crate) const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_millis(1000); + +/// The name a cluster answers to, as opposed to the address the console dials. +/// +/// A console may reach workers through addresses that are not present on their certificates. This +/// holds the name half: the scheme, host and port used for TLS, SNI and the gRPC `:authority`, plus +/// the CA that signs the workers if it is not a public one. Every physical worker address must +/// still be reachable from the console process. +#[derive(Clone, Debug)] +pub(crate) struct LogicalOrigin { + url: Url, + ca_certificate: Option, +} + +impl LogicalOrigin { + /// Builds an origin, rejecting URLs that cannot name a gRPC endpoint. + /// + /// The origin is only ever used for its scheme, host and port, so a URL missing any of them + /// would fail later at dial time with a much less obvious message. + pub(crate) fn new(url: Url, ca_certificate: Option) -> Result { + if !matches!(url.scheme(), "http" | "https") { + return Err(format!( + "worker origin must be http or https, got `{}` in {url}", + url.scheme() + )); + } + if !url.has_host() { + return Err(format!("worker origin {url} has no host")); + } + if url.scheme() != "https" && ca_certificate.is_some() { + return Err("a CA certificate requires an https worker origin".to_string()); + } + Ok(Self { + url, + ca_certificate, + }) + } +} /// Opens gRPC channels to the workers the console monitors. /// @@ -8,39 +63,183 @@ use url::Url; /// ([`crate::worker::discover_cluster_workers`]) both dial workers, and both must agree on how a /// worker URL becomes a connection. Holding that decision in one cheap, cloneable value keeps the /// two call sites from drifting apart when the transport changes. -#[derive(Clone, Debug, Default)] -pub(crate) struct WorkerConnector; +#[derive(Clone, Debug)] +pub(crate) struct WorkerConnector { + /// `None` dials whatever address the worker reports, in the clear. + origin: Option, + connect_timeout: Duration, +} + +#[cfg(test)] +impl Default for WorkerConnector { + fn default() -> Self { + Self::new(DEFAULT_CONNECT_TIMEOUT) + } +} impl WorkerConnector { - /// Opens a channel to a worker. + /// Builds a connector that talks plaintext to the addresses workers report. + pub(crate) fn new(connect_timeout: Duration) -> Self { + Self { + origin: None, + connect_timeout, + } + } + + /// Makes every connection present `origin` on the wire while still dialing the physical + /// address it was handed. + pub(crate) fn with_origin(self, origin: LogicalOrigin) -> Self { + Self { + origin: Some(origin), + ..self + } + } + + /// Opens a channel to a worker at `target`. + /// + /// `target` is the address to dial. Where an origin is configured, it — not `target` — decides + /// the scheme, the certificate name and the `:authority` header, so the console can reach a + /// worker through a tunnel or a rewritten address without the worker's certificate having to + /// mention it. /// /// Connecting alone does not prove the worker serves the observability API, so callers that /// care about liveness follow up with their own `Ping`. pub(crate) async fn connect( &self, - url: &Url, - ) -> Result, Error> { - ObservabilityServiceClient::connect(url.to_string()).await + target: &Url, + ) -> Result, ConnectError> { + let (host, port) = dial_target(target) + .ok_or_else(|| format!("worker address {target} has no host and port to dial"))?; + + let channel = self + .endpoint(target)? + .connect_with_connector(tower::service_fn(move |_| { + // The endpoint URI names the worker; this closure decides where the bytes go. The + // URI it is handed is deliberately ignored so the two can disagree. + let address = (host.clone(), port); + async move { + let stream = TcpStream::connect(address).await?; + Ok::<_, std::io::Error>(TokioIo::new(stream)) + } + })) + .await?; + + Ok(ObservabilityServiceClient::new(channel)) + } + + /// Describes the worker to `tonic`: the URI it reports as `:authority` and the TLS it expects. + fn endpoint(&self, target: &Url) -> Result { + let named = self.origin.as_ref().map_or(target, |origin| &origin.url); + + let endpoint = + Endpoint::from_shared(named.to_string())?.connect_timeout(self.connect_timeout); + + // `tonic` only wraps the connector in TLS when the endpoint URI is https, so the origin's + // scheme is what turns TLS on. + if named.scheme() != "https" { + return Ok(endpoint); + } + + let mut tls = ClientTlsConfig::new(); + // SNI needs a DNS name. An IP-literal origin has none, and `tonic` already derives the + // right `ServerName` from the endpoint URI in that case. + if let Some(Host::Domain(domain)) = named.host() { + tls = tls.domain_name(domain); + } + if let Some(ca) = self.origin.as_ref().and_then(|o| o.ca_certificate.clone()) { + tls = tls.ca_certificate(ca); + } else { + tls = tls.with_enabled_roots(); + } + + Ok(endpoint.tls_config(tls)?) } /// Turns a worker URL reported by `GetClusterWorkers` into a URL this connector can dial. /// /// Workers describe themselves through their own `WorkerResolver`, so the reported form is /// whatever that implementation chose and is not guaranteed to be directly dialable. - pub(crate) fn worker_url(&self, reported: &str) -> Result { - Url::parse(reported) + /// + /// With an origin configured the report is moved onto the origin's scheme, because a worker + /// that only knows its in-cluster plaintext address cannot report the https front door the + /// console reaches it through. A port the worker states explicitly is kept — it is the one + /// piece of the report that is per-worker and cannot be reconstructed — and otherwise the + /// origin's port applies. A reported port that matches its scheme's default (`:80` on http) + /// is indistinguishable from no port at all and takes the origin's port. + pub(crate) fn worker_url(&self, reported: &str) -> Result { + let reported = parse_reported(reported, self.scheme())?; + + let Some(origin) = &self.origin else { + return Ok(reported); + }; + + // Rebuilt as text rather than mutated in place: `host_str` already brackets IPv6 + // literals, which is exactly the form a URL wants them in. + let host = reported.host_str().ok_or(ParseError::EmptyHost)?; + let port = reported + .port() + .or_else(|| origin.url.port_or_known_default()); + + let mut rewritten = format!("{}://{host}", origin.url.scheme()); + if let Some(port) = port { + rewritten.push_str(&format!(":{port}")); + } + Url::parse(&rewritten) + } + + /// Scheme reported URLs are assumed to use when they do not say. + fn scheme(&self) -> &str { + self.origin + .as_ref() + .map_or("http", |origin| origin.url.scheme()) + } +} + +/// Parses a worker's self-report, which may be a full URL or a bare authority. +/// +/// `Url::parse` reads `localhost:9001` as the scheme `localhost` with the path `9001`, and rejects +/// `10.0.0.1:9001` and `[::1]:9001` outright, so anything that comes back without a host is +/// retried as an authority under `scheme`. +fn parse_reported(reported: &str, scheme: &str) -> Result { + match Url::parse(reported) { + Ok(url) if url.has_host() => Ok(url), + Ok(_) | Err(ParseError::RelativeUrlWithoutBase) => { + Url::parse(&format!("{scheme}://{reported}")) + } + Err(e) => Err(e), } } +/// Splits a URL into the host and port to hand to the TCP stack. +/// +/// The host is unbracketed: URLs write IPv6 literals as `[::1]` but the socket layer wants `::1`. +fn dial_target(url: &Url) -> Option<(String, u16)> { + let host = match url.host()? { + Host::Domain(domain) => domain.to_string(), + Host::Ipv4(addr) => addr.to_string(), + Host::Ipv6(addr) => addr.to_string(), + }; + Some((host, url.port_or_known_default()?)) +} + #[cfg(test)] mod tests { use super::*; + use datafusion::common::runtime::SpawnedTask; use std::error::Error; use tokio::net::TcpListener; + fn origin(url: &str) -> LogicalOrigin { + LogicalOrigin::new(Url::parse(url).expect("valid origin URL"), None).expect("usable origin") + } + + fn logical(url: &str) -> WorkerConnector { + WorkerConnector::default().with_origin(origin(url)) + } + #[test] fn worker_url_keeps_a_reported_url_as_is() -> Result<(), Box> { - let connector = WorkerConnector; + let connector = WorkerConnector::default(); let url = connector.worker_url("http://10.0.0.1:9001")?; assert_eq!(url.as_str(), "http://10.0.0.1:9001/"); Ok(()) @@ -48,10 +247,135 @@ mod tests { #[test] fn worker_url_rejects_a_value_that_is_not_a_url() { - let connector = WorkerConnector; + let connector = WorkerConnector::default(); assert!(connector.worker_url("not a url").is_err()); } + #[test] + fn worker_url_reads_a_bare_authority_as_a_host_and_port() -> Result<(), Box> { + let connector = WorkerConnector::default(); + // Without the authority fallback `Url::parse` reads this as the scheme `localhost`. + assert_eq!( + connector.worker_url("localhost:9001")?.as_str(), + "http://localhost:9001/" + ); + assert_eq!( + connector.worker_url("10.0.0.1:9001")?.as_str(), + "http://10.0.0.1:9001/" + ); + Ok(()) + } + + #[test] + fn worker_url_moves_a_report_onto_the_logical_scheme() -> Result<(), Box> { + let connector = logical("https://workers.example.com"); + assert_eq!( + connector + .worker_url("http://worker-3.internal:9001")? + .as_str(), + "https://worker-3.internal:9001/" + ); + Ok(()) + } + + #[test] + fn worker_url_takes_the_logical_port_when_the_worker_states_none() -> Result<(), Box> + { + let connector = logical("https://workers.example.com:8443"); + assert_eq!( + connector.worker_url("worker-3.internal")?.as_str(), + "https://worker-3.internal:8443/" + ); + // A default port for the reported scheme is erased by `Url` and cannot be told apart + // from no port, so it takes the logical port too. + assert_eq!( + connector + .worker_url("http://worker-3.internal:80")? + .as_str(), + "https://worker-3.internal:8443/" + ); + Ok(()) + } + + #[test] + fn worker_url_keeps_a_port_the_worker_states_explicitly() -> Result<(), Box> { + let connector = logical("https://workers.example.com:8443"); + assert_eq!( + connector + .worker_url("http://worker-3.internal:9001")? + .as_str(), + "https://worker-3.internal:9001/" + ); + Ok(()) + } + + #[test] + fn worker_url_defaults_to_the_logical_schemes_port() -> Result<(), Box> { + // The origin states no port, so https' own default is what the workers get. `Url` writes + // a scheme's default port as no port at all, so the dial target is what to assert on. + let connector = logical("https://workers.example.com"); + let url = connector.worker_url("worker-3.internal")?; + assert_eq!(url.as_str(), "https://worker-3.internal/"); + assert_eq!( + dial_target(&url).ok_or("expected a dialable address")?, + ("worker-3.internal".to_string(), 443) + ); + Ok(()) + } + + #[test] + fn worker_url_handles_ipv4_and_ipv6_literals() -> Result<(), Box> { + let connector = logical("https://workers.example.com:8443"); + + assert_eq!( + connector.worker_url("http://10.0.0.1:9001")?.as_str(), + "https://10.0.0.1:9001/" + ); + assert_eq!( + connector.worker_url("10.0.0.1")?.as_str(), + "https://10.0.0.1:8443/" + ); + // IPv6 literals stay bracketed in URL form, whether reported as a URL or an authority. + assert_eq!( + connector.worker_url("http://[2001:db8::1]:9001")?.as_str(), + "https://[2001:db8::1]:9001/" + ); + assert_eq!( + connector.worker_url("[::1]:9001")?.as_str(), + "https://[::1]:9001/" + ); + assert_eq!( + connector.worker_url("[::1]")?.as_str(), + "https://[::1]:8443/" + ); + Ok(()) + } + + #[test] + fn logical_origin_rejects_a_scheme_it_cannot_dial() { + let url = Url::parse("grpc://workers.example.com").expect("valid URL"); + assert!(LogicalOrigin::new(url, None).is_err()); + } + + #[test] + fn logical_origin_rejects_ca_for_plaintext() { + let url = Url::parse("http://workers.example.com").expect("valid URL"); + let ca = Certificate::from_pem(b"not parsed until a TLS connection is attempted"); + assert!(LogicalOrigin::new(url, Some(ca)).is_err()); + } + + #[test] + fn dial_target_unbrackets_ipv6_and_fills_in_the_default_port() -> Result<(), Box> { + let (host, port) = dial_target(&Url::parse("https://[2001:db8::1]:9001")?) + .ok_or("expected a dialable address")?; + assert_eq!((host.as_str(), port), ("2001:db8::1", 9001)); + + let (host, port) = dial_target(&Url::parse("https://workers.example.com")?) + .ok_or("expected an address")?; + assert_eq!((host.as_str(), port), ("workers.example.com", 443)); + Ok(()) + } + #[tokio::test] async fn connect_fails_when_nothing_is_listening() -> Result<(), Box> { // Binding and dropping a listener yields a port that is very unlikely to be reused. @@ -59,9 +383,26 @@ mod tests { let port = listener.local_addr()?.port(); drop(listener); - let connector = WorkerConnector; + let connector = WorkerConnector::default(); let url = Url::parse(&format!("http://127.0.0.1:{port}"))?; assert!(connector.connect(&url).await.is_err()); Ok(()) } + + #[tokio::test] + async fn connect_dials_the_physical_address_and_not_the_origin() -> Result<(), Box> { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let port = listener.local_addr()?.port(); + // Dropping the accepted socket ends the handshake, so `connect` fails fast either way; + // the connection arriving at all is what proves the dial ignored the origin's host. + let accepted = SpawnedTask::spawn(async move { listener.accept().await.map(|_| ()) }); + + // `.invalid` never resolves (RFC 2606), so nothing can reach this listener by name. + let connector = logical(&format!("http://workers.invalid:{port}")); + let target = Url::parse(&format!("http://127.0.0.1:{port}"))?; + let _ = connector.connect(&target).await; + + accepted.await??; + Ok(()) + } } diff --git a/console/src/main.rs b/console/src/main.rs index 706a4ff5e..6790a4f1d 100644 --- a/console/src/main.rs +++ b/console/src/main.rs @@ -6,11 +6,14 @@ mod ui; mod worker; use app::App; -use connector::WorkerConnector; +use color_eyre::eyre::{Report, bail, eyre}; +use connector::{DEFAULT_CONNECT_TIMEOUT, LogicalOrigin, WorkerConnector}; use crossterm::event::{self, Event}; use ratatui::DefaultTerminal; +use std::path::PathBuf; use std::time::{Duration, Instant}; use structopt::StructOpt; +use tonic::transport::Certificate; use url::Url; #[derive(StructOpt)] @@ -19,13 +22,31 @@ use url::Url; about = "Console for monitoring DataFusion distributed workers" )] struct Args { - /// Port of a worker to connect to for auto-discovery. + /// Address of a worker to connect to for auto-discovery: a port (`9001`), a `host:port`, or + /// a full URL. This is the address the console dials, which with --worker-origin need not be + /// the name the worker answers to. /// The console calls GetClusterWorkers on this worker to discover the full cluster. - port: u16, + seed: String, + + /// Origin the workers answer to, e.g. `https://workers.example.com`. TLS, SNI and the gRPC + /// authority come from this URL, and the workers discovered through the seed are moved onto + /// its scheme and port. Omit it to talk plaintext to the addresses workers report. + #[structopt(long = "worker-origin")] + worker_origin: Option, + + /// Path to a PEM-encoded CA certificate that signs the worker certificates, for a cluster + /// whose CA is not a public one. Requires --worker-origin. + #[structopt(long = "ca-cert", parse(from_os_str))] + ca_cert: Option, /// Polling interval in milliseconds - #[structopt(long = "poll-interval", default_value = "100")] + #[structopt(long = "poll-interval", default_value = "1000")] poll_interval: u64, + + /// Budget for opening a worker connection, TCP connect and TLS handshake together, in + /// milliseconds [default: 1000] + #[structopt(long = "connect-timeout")] + connect_timeout: Option, } #[tokio::main] @@ -34,10 +55,13 @@ async fn main() -> color_eyre::Result<()> { let args = Args::from_args(); - let seed_url = Url::parse(&format!("http://localhost:{}", args.port)).expect("valid URL"); + let connector = build_connector(&args)?; + // The seed goes through the same normalization as a discovered worker, so the address the + // console is started with and the ones the cluster reports are dialed the same way. + let seed_url = connector.worker_url(&seed_address(&args.seed))?; let poll_interval = Duration::from_millis(args.poll_interval); - let mut app = App::new(seed_url, WorkerConnector); + let mut app = App::new(seed_url, connector); let mut terminal = ratatui::init(); terminal.clear()?; @@ -49,6 +73,41 @@ async fn main() -> color_eyre::Result<()> { result } +/// Expands a bare port, which is how the console has always been started, into an authority. +fn seed_address(seed: &str) -> String { + match seed.parse::() { + Ok(port) => format!("localhost:{port}"), + Err(_) => seed.to_string(), + } +} + +/// Builds the transport policy shared by discovery and every per-worker connection. +fn build_connector(args: &Args) -> color_eyre::Result { + let connect_timeout = args + .connect_timeout + .map_or(DEFAULT_CONNECT_TIMEOUT, Duration::from_millis); + let connector = WorkerConnector::new(connect_timeout); + + let Some(origin) = args.worker_origin.clone() else { + // Accepting a CA and then ignoring it would leave the console talking plaintext to a + // cluster the operator believes it is verifying. + if args.ca_cert.is_some() { + bail!("--ca-cert only applies together with --worker-origin"); + } + return Ok(connector); + }; + + let ca_certificate = match &args.ca_cert { + Some(path) => Some(Certificate::from_pem(std::fs::read(path).map_err(|e| { + eyre!("failed to read CA certificate {}: {e}", path.display()) + })?)), + None => None, + }; + + let origin = LogicalOrigin::new(origin, ca_certificate).map_err(Report::msg)?; + Ok(connector.with_origin(origin)) +} + async fn run_app( terminal: &mut DefaultTerminal, app: &mut App, @@ -78,3 +137,31 @@ async fn run_app( Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + fn args(worker_origin: Option, ca_cert: Option) -> Args { + Args { + seed: "9001".to_string(), + worker_origin, + ca_cert, + poll_interval: 1000, + connect_timeout: None, + } + } + + #[test] + fn bare_seed_port_keeps_legacy_cli_compatibility() { + assert_eq!(seed_address("9001"), "localhost:9001"); + assert_eq!(seed_address("127.0.0.1:9001"), "127.0.0.1:9001"); + } + + #[test] + fn ca_certificate_requires_worker_origin() { + let args = args(None, Some(PathBuf::from("unused.pem"))); + let error = build_connector(&args).expect_err("CA without origin must be rejected"); + assert!(error.to_string().contains("--worker-origin")); + } +}