milenkovicm commented on code in PR #1351:
URL:
https://github.com/apache/datafusion-ballista/pull/1351#discussion_r2683625333
##########
ballista/core/proto/ballista.proto:
##########
@@ -642,6 +642,7 @@ message JobStatus {
message GetJobStatusResult {
JobStatus status = 1;
+ optional string flight_proxy = 2;
Review Comment:
can we make this one of to represent proxy statuses
- no proxy
- in process (no need to specify ip/port client should use scheduler ip port)
- external process (ip / port specified) client should use given ip/port
that would remove check if for empty string on client side
##########
ballista/scheduler/src/flight_proxy_service.rs:
##########
@@ -0,0 +1,209 @@
+// 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.
+
+use crate::state::SchedulerState;
+use arrow_flight::flight_service_client::FlightServiceClient;
+use arrow_flight::flight_service_server::FlightService;
+use arrow_flight::{
+ Action, ActionType, Criteria, Empty, FlightData, FlightDescriptor,
FlightInfo,
+ HandshakeRequest, HandshakeResponse, PollInfo, PutResult, SchemaResult,
Ticket,
+};
+use ballista_core::error::BallistaError;
+use ballista_core::serde::decode_protobuf;
+use ballista_core::serde::scheduler::Action as BallistaAction;
+use ballista_core::utils::{GrpcClientConfig, create_grpc_client_connection};
+use datafusion_proto::logical_plan::AsLogicalPlan;
+use datafusion_proto::physical_plan::AsExecutionPlan;
+use futures::{Stream, TryFutureExt};
+use log::debug;
+use std::collections::HashSet;
+use std::pin::Pin;
+use std::sync::Arc;
+use tonic::{Request, Response, Status, Streaming};
+
+/// Service implementing a proxy from scheduler to executor Apache Arrow
Flight Protocol
Review Comment:
it would be great if we could add more comments here. describe how it can be
configured
##########
ballista/scheduler/src/scheduler_process.rs:
##########
@@ -131,7 +191,24 @@ pub async fn start_server(
) -> ballista_core::error::Result<()> {
info!("Ballista v{BALLISTA_VERSION} Scheduler listening on {address:?}");
let scheduler =
- create_scheduler::<LogicalPlanNode, PhysicalPlanNode>(cluster,
config).await?;
+ create_scheduler::<LogicalPlanNode, PhysicalPlanNode>(cluster,
config.clone())
+ .await?;
- start_grpc_service(address, scheduler).await
+ info!(
+ "advertise_flight_sql_endpoint: {:?}",
+ config.advertise_flight_sql_endpoint
+ );
+ match config.advertise_flight_sql_endpoint.clone() {
+ Some(s) if s != "" => {
Review Comment:
Sorry I might be unclear. If we specify different port (or ip port) that
would mean there is external process running proxy, not the scheduler process.
So we have three configuration options
- no proxy
- in process (no need to specify ip/port client should use scheduler ip port)
- external process (ip / port specified) client should use given ip/port
##########
ballista/scheduler/src/scheduler_server/grpc.rs:
##########
@@ -450,7 +450,22 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> SchedulerGrpc
let job_id = request.into_inner().job_id;
trace!("Received get_job_status request for job {}", job_id);
match self.state.task_manager.get_job_status(&job_id).await {
- Ok(status) => Ok(Response::new(GetJobStatusResult { status })),
+ Ok(status) => Ok(Response::new(GetJobStatusResult {
+ status,
+ flight_proxy: self
+ .state
+ .config
+ .advertise_flight_sql_endpoint
+ .clone()
+ .map(|s| match s {
+ s if s.is_empty() => format!(
Review Comment:
i guess same thing here, if configuration is empty client should dial back
on scheduler address / port
##########
ballista/scheduler/src/config.rs:
##########
@@ -41,7 +41,7 @@ pub struct Config {
/// Route for proxying flight results via scheduler (IP:PORT format).
#[arg(
long,
- help = "Route for proxying flight results via scheduler. Should be of
the form 'IP:PORT"
+ help = "Route for proxying flight results via scheduler. Should be of
the form 'IP:PORT'"
Review Comment:
same comment regarding empty address. if address/port not specified client
need to dial back on scheduler address
--
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]
