andygrove commented on code in PR #2577: URL: https://github.com/apache/datafusion-ballista/pull/2577#discussion_r4243061471
########## ballista/result-service/src/main.rs: ########## @@ -0,0 +1,251 @@ +// 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. + +//! Ballista Result Service: serves query results to clients by forwarding each +//! partition fetch to the executor that holds the partition. Point the +//! scheduler's `--advertise-flight-endpoint` at it. + +mod http; +mod metered; +mod metrics; + +use std::future::Future; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use arrow_flight::flight_service_server::FlightServiceServer; +use ballista_core::serving::{ForwardingBackend, ServingFlightService}; +use ballista_core::utils::{GrpcServerConfig, create_grpc_server}; +use clap::Parser; +use log::{error, info, warn}; +use tonic::service::Routes; +use tonic::transport::server::Router; + +use crate::metered::MeteredBackend; +use crate::metrics::{ResultMetricsCollector, default_metrics_collector}; + +/// Command-line configuration for the result service. +#[derive(Debug, Parser)] +#[command(name = "ballista-result-service", version, about)] +struct Config { + /// Host or IP address the Flight service binds to. + #[arg(long, default_value = "0.0.0.0")] + bind_host: String, + + /// Port the Flight service, health probes (/healthz and /readyz) and + /// metrics (/api/metrics) bind to. + #[arg(long, default_value_t = 50055)] + bind_port: u16, + + /// Use TLS when connecting to executors. + #[arg(long, default_value_t = false)] + use_tls: bool, Review Comment: I think `--use-tls` quietly falls back to plaintext at the moment. It switches the executor URL to `https://`, but this crate builds tonic without any of its TLS features, and without them tonic ignores the scheme and opens a plain h2c connection. I checked with a small program on tonic 0.14.6 with the same feature set this crate resolves to, and an `https://` endpoint connected to a plaintext gRPC server and got a normal response. So against stock executors the flag looks like it works but nothing is encrypted, and against executors that do serve TLS it would fail with a confusing protocol error. Would it make sense to either drop the flag until TLS to executors is supported, or enable `tonic/tls-ring` and `tls-native-roots` here so `https://` really does TLS against the system roots? The ca-certificates comment in `ballista-result-service.Dockerfile` reads like the second one was the intent. A test that fails if `--use-tls` ever reaches a plaintext server would keep this from coming back. ########## chaos-testing/src/bin/chaos-result-service.rs: ########## @@ -0,0 +1,59 @@ +// 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. + +//! result service for the chaos harness. It is configured from the environment +//! because `TestCluster` spawns it as a child process, like `chaos-scheduler` +//! and `chaos-executor`. + +use arrow_flight::flight_service_server::FlightServiceServer; +use ballista_core::serving::{ForwardingBackend, ServingFlightService}; +use ballista_core::utils::{GrpcServerConfig, create_grpc_server}; +use std::net::SocketAddr; + +const MAX_MESSAGE_SIZE: usize = 16 * 1024 * 1024; + +#[tokio::main] +async fn main() -> ballista_core::error::Result<()> { + env_logger::init(); + + let bind_host = + std::env::var("CHAOS_BIND_HOST").unwrap_or_else(|_| "127.0.0.1".into()); + let bind_port: u16 = std::env::var("CHAOS_RESULT_SERVICE_PORT") + .expect("CHAOS_RESULT_SERVICE_PORT must be set") + .parse() + .expect("CHAOS_RESULT_SERVICE_PORT must be a u16"); + + let addr: SocketAddr = format!("{bind_host}:{bind_port}") + .parse() + .expect("result service address must parse"); + + // Plaintext: the harness runs everything on loopback. + let backend = ForwardingBackend::new(MAX_MESSAGE_SIZE, MAX_MESSAGE_SIZE, false, None); Review Comment: The chaos scenarios run this stand-in rather than the shipped `ballista-result-service`, so they don't cover what's specific to the binary: gRPC and HTTP/1.1 sharing a port, the metering wrapper, the drain on SIGTERM, and the `/healthz` and `/readyz` probes the Kubernetes docs rely on (the chaos k8s manifest uses a `tcpSocket` probe instead). Would it be worth running the real thing? One way to keep the "cargo test builds everything" property is to give `ballista-result-service` a small lib target with `server()` and have this binary call it. That would also help with the endpoint override question. ########## ballista/result-service/src/main.rs: ########## @@ -0,0 +1,251 @@ +// 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. + +//! Ballista Result Service: serves query results to clients by forwarding each +//! partition fetch to the executor that holds the partition. Point the +//! scheduler's `--advertise-flight-endpoint` at it. + +mod http; +mod metered; +mod metrics; + +use std::future::Future; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use arrow_flight::flight_service_server::FlightServiceServer; +use ballista_core::serving::{ForwardingBackend, ServingFlightService}; +use ballista_core::utils::{GrpcServerConfig, create_grpc_server}; +use clap::Parser; +use log::{error, info, warn}; +use tonic::service::Routes; +use tonic::transport::server::Router; + +use crate::metered::MeteredBackend; +use crate::metrics::{ResultMetricsCollector, default_metrics_collector}; + +/// Command-line configuration for the result service. +#[derive(Debug, Parser)] +#[command(name = "ballista-result-service", version, about)] +struct Config { + /// Host or IP address the Flight service binds to. + #[arg(long, default_value = "0.0.0.0")] + bind_host: String, + + /// Port the Flight service, health probes (/healthz and /readyz) and + /// metrics (/api/metrics) bind to. + #[arg(long, default_value_t = 50055)] + bind_port: u16, + + /// Use TLS when connecting to executors. + #[arg(long, default_value_t = false)] + use_tls: bool, + + /// Maximum gRPC message size, in bytes, this service will decode. + #[arg(long, default_value_t = 16777216)] + grpc_max_decoding_message_size: usize, + + /// Maximum gRPC message size, in bytes, this service will encode. + #[arg(long, default_value_t = 16777216)] + grpc_max_encoding_message_size: usize, + + /// Time in seconds to let in-flight result streams finish after a + /// shutdown signal (SIGTERM or Ctrl-C) before exiting anyway. Keep it below + /// your orchestrator's termination grace period (Kubernetes' + /// terminationGracePeriodSeconds, 30 by default), so this bound is reached + /// before a SIGKILL is. + #[arg(long, default_value_t = 10)] + graceful_shutdown_timeout_seconds: u64, +} + +#[tokio::main] +async fn main() -> Result<(), Box<dyn std::error::Error>> { + env_logger::init(); + let config = Config::parse(); + + let addr: SocketAddr = format!("{}:{}", config.bind_host, config.bind_port) + .parse() + .map_err(|e| { + format!( + "invalid bind address {}:{}: {e}", + config.bind_host, config.bind_port + ) + })?; + + let server = server(&config, default_metrics_collector()?); + + info!("Ballista Result Service listening on {addr}"); + + let (stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>(); + let server = server.serve_with_shutdown(addr, async { + let _ = stop_rx.await; + }); + tokio::pin!(server); + + // Serve until a shutdown signal arrives, unless the server fails first. + tokio::select! { + result = &mut server => { + result?; + info!("Ballista Result Service stopped"); + return Ok(()); + } + () = shutdown_signal() => {} + } + + // Stop accepting connections, then give in-flight streams a bounded time to + // finish. + let _ = stop_tx.send(()); + drain_or_time_out( + server, + Duration::from_secs(config.graceful_shutdown_timeout_seconds), + ) + .await?; + + info!("Ballista Result Service stopped"); + Ok(()) +} + +/// The Flight service, health probes and metrics, served on one port. +fn server(config: &Config, metrics: Arc<dyn ResultMetricsCollector>) -> Router { + let backend = MeteredBackend::new( + ForwardingBackend::new( + config.grpc_max_decoding_message_size, + config.grpc_max_encoding_message_size, + config.use_tls, + None, Review Comment: Related to the TLS flag. The embedded proxy uses `SchedulerConfig::override_create_grpc_client_endpoint` for its executor connections, which is how embedders set up mTLS or a private CA today. Here the hook is always `None`, and since the crate is binary-only there's no way to pass one in. So deployments that rely on the hook don't have a migration path yet. Is the plan to close that before the proxy goes away in 57.0.0? Flags for a CA bundle and a client cert and key would cover most cases, or a small lib target that exposes `server()` with an endpoint override. In the meantime it might be worth a sentence in the 56.0.0 upgrade guide, maybe pointing out that embedders can build their own service from `ForwardingBackend` and `ServingFlightService` in `ballista_core::serving`. ########## docs/source/user-guide/deployment/kubernetes.md: ########## @@ -392,6 +400,217 @@ Use the following command to set up port-forwarding. kubectl port-forward service/ballista-scheduler 50050:50050 ``` +## Serving Results to Clients Outside the Cluster + +Port forwarding the scheduler is enough to submit queries, but by default clients fetch query +results directly from the executors, and executor pods are not reachable from outside the cluster. +Deploy a result service that clients can reach, and have the scheduler point them at it. See +[Fetching Query Results](../scheduler.md#fetching-query-results) for how it works. + +Save the following as `result-service.yaml`: + +```yaml +apiVersion: v1 +kind: Service +metadata: + name: ballista-result-service +spec: + ports: + - port: 50055 + name: flight + # Plaintext HTTP/2, so a Gateway API route speaks gRPC to the pods. + appProtocol: kubernetes.io/h2c + selector: + app: ballista-result-service +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: ballista-result-service +spec: + # Replicas are interchangeable: each forwards any fetch to the executor that + # holds the partition and keeps no state, so scale with result traffic. + replicas: 2 + # Start a replacement before stopping an old pod, so rollouts keep capacity. + strategy: + type: RollingUpdate + rollingUpdate: + maxSurge: 1 + maxUnavailable: 0 + selector: + matchLabels: + app: ballista-result-service + template: + metadata: + labels: + app: ballista-result-service + ballista-cluster: ballista + spec: + # Prefer placing replicas on different nodes, so they don't share a + # node's network bandwidth. + topologySpreadConstraints: + - maxSkew: 1 + topologyKey: kubernetes.io/hostname + whenUnsatisfiable: ScheduleAnyway + labelSelector: + matchLabels: + app: ballista-result-service + # Leaves 10s of headroom after the preStop sleep (5s) and the drain + # bound (45s). + terminationGracePeriodSeconds: 60 + containers: + - name: ballista-result-service + image: <your-repo>/datafusion-ballista-result-service:latest + args: + - "--bind-port=50055" + # After SIGTERM, stop accepting connections and give in-flight + # result streams up to 45s to finish. + - "--graceful-shutdown-timeout-seconds=45" + lifecycle: + # Kubernetes removes a terminating pod from the Service's endpoints + # in parallel with shutting it down, and kube-proxy and gateways + # learn of the removal asynchronously, so new connections can still + # arrive for a few seconds. Sleeping keeps the pod accepting them + # until routing catches up; SIGTERM follows once the hook returns. + preStop: + sleep: + seconds: 5 + ports: + - containerPort: 50055 + name: flight + livenessProbe: + httpGet: + path: /healthz + port: 50055 + failureThreshold: 3 + periodSeconds: 10 + readinessProbe: + httpGet: + path: /readyz + port: 50055 + failureThreshold: 3 + periodSeconds: 5 +--- +apiVersion: policy/v1 +kind: PodDisruptionBudget +metadata: + name: ballista-result-service +spec: + # Node drains evict at most one replica at a time. + minAvailable: 1 + selector: + matchLabels: + app: ballista-result-service +``` + +When a pod is replaced or moved to another node, Kubernetes removes it from the Service's +endpoints at the same time as it starts shutting the pod down. kube-proxy and any gateway pick up +that removal asynchronously, so for a few seconds new connections can still be routed to the pod. +The `preStop` sleep keeps the pod accepting connections during that window. On SIGTERM, which +follows the sleep, it stops accepting connections and lets in-flight result streams finish, for up +to `--graceful-shutdown-timeout-seconds` (10 by default), before exiting. The rolling-update +strategy and the PodDisruptionBudget keep at least one replica serving throughout. + +Then add `--advertise-flight-endpoint` to the scheduler container's `args` in `cluster.yaml`. Which +address to advertise depends on how clients reach the cluster. + +### Through port forwarding + +For a client on your own machine, forward a local port to the result service as well as to the +scheduler, and advertise the local address: + +```yaml +args: ["--bind-port=50050", "--advertise-flight-endpoint=127.0.0.1:50055"] +``` + +```bash +kubectl apply -f result-service.yaml +kubectl apply -f cluster.yaml +kubectl port-forward service/ballista-scheduler 50050:50050 & +kubectl port-forward service/ballista-result-service 50055:50055 & +``` + +`kubectl port-forward` sends all traffic to a single pod, so use this for testing only. + +### Through a TLS-terminating gateway + +In production, expose the result service through the [Gateway API] with a Gateway that terminates Review Comment: Since the result service checks no credentials, would it be worth saying right here that the gateway should only be reachable by trusted clients, for example through an internal load balancer or a source-range allowlist? "In production" with a public-looking hostname reads like an internet-facing setup, and the trusted-network warning doesn't come until the next section. ########## ballista/scheduler/src/config.rs: ########## @@ -364,6 +367,11 @@ pub struct SchedulerConfig { /// *told*. When a non-empty endpoint is also set it takes precedence for /// the advertisement, which is how you put a load balancer in front of an /// embedded proxy. + /// + /// Deprecated, and to be removed in 57.0.0: the proxy puts result traffic + /// on the scheduler. Run a result service and advertise it with + /// [`Self::advertise_flight_endpoint`] instead. The scheduler logs a warning + /// at startup while this is set. pub enable_embedded_flight_proxy: bool, Review Comment: Nit: the upgrade guide lists `SchedulerConfig::enable_embedded_flight_proxy` as deprecated, but only the builder method has `#[deprecated]`. Embedders that set the field in a struct literal, like `chaos-scheduler.rs` does, won't get a compiler warning before it goes away in 57.0.0. Putting `#[deprecated]` on the field too would cover them, at the cost of a few `#[allow(deprecated)]` on the internal uses. ########## ballista/result-service/src/metrics.rs: ########## @@ -0,0 +1,228 @@ +// 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. + +//! Metrics for result fetches. + +use std::sync::Arc; +use std::time::Duration; + +use ballista_core::error::Result; + +/// How a result fetch ended. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum FetchOutcome { + /// The whole stream was delivered. + Completed, + /// The executor or the stream returned an error. + Failed, + /// The client went away before the stream ended. + Cancelled, +} + +/// Records result fetches. +pub trait ResultMetricsCollector: Send + Sync { + /// A fetch started. + fn record_started(&self); + + /// A fetch relayed `bytes` of Flight data to its client. + fn record_bytes(&self, bytes: u64); + + /// A fetch ended after `elapsed`. + fn record_finished(&self, outcome: FetchOutcome, elapsed: Duration); + + /// The metrics and their content type, or `None` if this collector exports + /// none. + fn gather_metrics(&self) -> Result<Option<(Vec<u8>, String)>>; +} + +/// A collector that records nothing. +#[cfg(any(test, not(feature = "prometheus-metrics")))] +#[derive(Default)] +pub struct NoopMetricsCollector; + +#[cfg(any(test, not(feature = "prometheus-metrics")))] +impl ResultMetricsCollector for NoopMetricsCollector { + fn record_started(&self) {} + + fn record_bytes(&self, _bytes: u64) {} + + fn record_finished(&self, _outcome: FetchOutcome, _elapsed: Duration) {} + + fn gather_metrics(&self) -> Result<Option<(Vec<u8>, String)>> { + Ok(None) + } +} + +/// The Prometheus collector with the `prometheus-metrics` feature, otherwise +/// one that records nothing. +pub fn default_metrics_collector() -> Result<Arc<dyn ResultMetricsCollector>> { + #[cfg(feature = "prometheus-metrics")] + { + Ok(Arc::new(PrometheusMetricsCollector::new( + ::prometheus::default_registry(), + )?)) + } + #[cfg(not(feature = "prometheus-metrics"))] + { + Ok(Arc::new(NoopMetricsCollector)) + } +} + +#[cfg(feature = "prometheus-metrics")] +pub use prometheus_metrics::PrometheusMetricsCollector; + +#[cfg(feature = "prometheus-metrics")] +mod prometheus_metrics { + use super::{FetchOutcome, ResultMetricsCollector}; + use ballista_core::error::{BallistaError, Result}; + use prometheus::{ + Counter, Encoder, Gauge, Histogram, Registry, TextEncoder, + register_counter_with_registry, register_gauge_with_registry, + register_histogram_with_registry, + }; + use std::time::Duration; + + /// A [`ResultMetricsCollector`] that exports Prometheus metrics: + /// + /// - `fetch_started_total`, `fetch_completed_total`, `fetch_failed_total` + /// and `fetch_cancelled_total` count fetches by how they ended + /// - `fetch_in_flight` is the number of fetches currently streaming + /// - `fetch_time_seconds` is a histogram of fetch durations + /// - `fetch_bytes_total` counts the Flight data relayed to clients + pub struct PrometheusMetricsCollector { + registry: Registry, + started: Counter, + completed: Counter, + failed: Counter, + cancelled: Counter, + in_flight: Gauge, + time: Histogram, + bytes: Counter, + } + + fn registration_error(e: prometheus::Error) -> BallistaError { + BallistaError::Internal(format!("Error registering metric: {e:?}")) + } + + impl PrometheusMetricsCollector { + /// Creates the collector, registering its metrics in `registry`. + pub fn new(registry: &Registry) -> Result<Self> { + let counter = |name: &str, help: &str| { + register_counter_with_registry!(name, help, registry) + .map_err(registration_error) + }; + Ok(Self { + registry: registry.clone(), + started: counter( + "fetch_started_total", Review Comment: These names will end up in people's dashboards and alerts, so now is probably the cheapest time to settle them. What do you think about a `ballista_result_service_` prefix? The newest scheduler metric, `ballista_scheduler_rejected_by_protocol_version_total`, already goes that way, and plain `fetch_*` could get confusing if executors ever export shuffle fetch metrics. A single `fetches_total` counter with an `outcome` label would be a bit more idiomatic too. Small related thing: the Prometheus tests here and in `http.rs` only build with `prometheus-metrics`, and none of the CI test jobs enable it. Clippy builds them with `--all-features` but doesn't run them. Adding `ballista-result-service/prometheus-metrics` to the `cargo test` features in `rust.yml` would cover them. ########## ballista/result-service/src/main.rs: ########## @@ -0,0 +1,251 @@ +// 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. + +//! Ballista Result Service: serves query results to clients by forwarding each +//! partition fetch to the executor that holds the partition. Point the +//! scheduler's `--advertise-flight-endpoint` at it. + +mod http; +mod metered; +mod metrics; + +use std::future::Future; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use arrow_flight::flight_service_server::FlightServiceServer; +use ballista_core::serving::{ForwardingBackend, ServingFlightService}; +use ballista_core::utils::{GrpcServerConfig, create_grpc_server}; +use clap::Parser; +use log::{error, info, warn}; +use tonic::service::Routes; +use tonic::transport::server::Router; + +use crate::metered::MeteredBackend; +use crate::metrics::{ResultMetricsCollector, default_metrics_collector}; + +/// Command-line configuration for the result service. +#[derive(Debug, Parser)] +#[command(name = "ballista-result-service", version, about)] +struct Config { + /// Host or IP address the Flight service binds to. + #[arg(long, default_value = "0.0.0.0")] + bind_host: String, + + /// Port the Flight service, health probes (/healthz and /readyz) and + /// metrics (/api/metrics) bind to. + #[arg(long, default_value_t = 50055)] + bind_port: u16, + + /// Use TLS when connecting to executors. + #[arg(long, default_value_t = false)] + use_tls: bool, + + /// Maximum gRPC message size, in bytes, this service will decode. + #[arg(long, default_value_t = 16777216)] + grpc_max_decoding_message_size: usize, + + /// Maximum gRPC message size, in bytes, this service will encode. + #[arg(long, default_value_t = 16777216)] + grpc_max_encoding_message_size: usize, + + /// Time in seconds to let in-flight result streams finish after a + /// shutdown signal (SIGTERM or Ctrl-C) before exiting anyway. Keep it below + /// your orchestrator's termination grace period (Kubernetes' + /// terminationGracePeriodSeconds, 30 by default), so this bound is reached + /// before a SIGKILL is. + #[arg(long, default_value_t = 10)] + graceful_shutdown_timeout_seconds: u64, +} + +#[tokio::main] +async fn main() -> Result<(), Box<dyn std::error::Error>> { + env_logger::init(); + let config = Config::parse(); + + let addr: SocketAddr = format!("{}:{}", config.bind_host, config.bind_port) + .parse() + .map_err(|e| { + format!( + "invalid bind address {}:{}: {e}", + config.bind_host, config.bind_port + ) + })?; + + let server = server(&config, default_metrics_collector()?); + + info!("Ballista Result Service listening on {addr}"); + + let (stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>(); + let server = server.serve_with_shutdown(addr, async { + let _ = stop_rx.await; + }); + tokio::pin!(server); + + // Serve until a shutdown signal arrives, unless the server fails first. + tokio::select! { + result = &mut server => { + result?; + info!("Ballista Result Service stopped"); + return Ok(()); + } + () = shutdown_signal() => {} + } + + // Stop accepting connections, then give in-flight streams a bounded time to + // finish. + let _ = stop_tx.send(()); Review Comment: I tried tonic's drain in a small program and noticed one thing. After the signal, tonic stops calling `accept` but keeps the listening socket open until every connection has closed. So a connection that arrives during the drain completes the TCP handshake and then hangs, rather than being refused. Probes still fail (by timeout), and with the `preStop` sleep very few new connections should show up, so this is low priority. Outside Kubernetes, though, a client routed to a draining replica would wait out its request timeout instead of failing over right away. If you want to tighten it, serving through `serve_with_incoming_shutdown` with an incoming stream that drops the `TcpIncoming` when the signal fires (and `std::future::pending()` as tonic's own signal, so the drain stays graceful) refuses new connections straight away while in-flight streams finish. I tried that and it behaved as expected. Otherwise it might be enough to soften "stops accepting connections" in the README and the `http.rs` docs. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
