andygrove commented on code in PR #2577: URL: https://github.com/apache/datafusion-ballista/pull/2577#discussion_r4251363872
########## dev/docker/result-service-entrypoint.sh: ########## @@ -0,0 +1,22 @@ +#!/bin/bash +# +# 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. + +set -e + +/root/ballista-result-service "$@" Review Comment: I think this needs an `exec` for the graceful shutdown to work in the image. Without it, bash stays as PID 1 and runs the service as a child. Bash as PID 1 doesn't forward SIGTERM, so the service never sees it. I checked with an `ubuntu:26.04` image running this same script around a stand-in binary that traps SIGTERM. `docker stop` never reached it, and it was SIGKILLed when the timeout ran out (exit 137). With `exec /root/ballista-result-service "$@"` the service runs as PID 1, gets the SIGTERM and exits straight away. In the Kubernetes example this means the drain never runs. Each pod keeps serving until the kubelet SIGKILLs it at `terminationGracePeriodSeconds`, so every rollout step takes the full 60s and any stream still running at that point gets cut off. The chaos manifests set `command` to the binary directly, which is probably why the k8s scenarios didn't catch it. The scheduler and executor entrypoints follow the same pattern, and the executor handles SIGTERM too. That's not from this PR, so a separate follow-up seems fine for those. ########## ballista/result-service/src/metrics.rs: ########## @@ -0,0 +1,234 @@ +// 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( Review Comment: With `prometheus-metrics` enabled, a second `serve` call in the same process fails with `Error registering metric: AlreadyReg`. Each call registers the metrics in the default registry again. I hit it with a quick test that calls `serve` twice in a row. Now that `serve` is a library API, embedders could run into this from integration tests or an in-process restart. The scheduler avoids it by caching its collector in a `OnceCell` (see `PrometheusMetricsCollector::current()`). Would it make sense to do the same here? ########## ballista/result-service/src/lib.rs: ########## @@ -0,0 +1,410 @@ +// 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 Review Comment: Nit: now that this is a library crate, would it be worth adding `#![warn(missing_docs)]` like the other library crates in the workspace? Everything public here is already documented, so it would just keep it that way. -- 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]
