feat(scheduler): Redirect '/' to nighlies.a.o WebTUI

.github/workflows/web-tui.yml

@@ -90,7 +90,7 @@ jobs:
             --no-default-features --features web
 
       - name: Deploy at nightlies.a.o
-        if: ${{ github.event == 'push' && github.ref == 'refs/heads/main' }}
+        if: ${{ github.event_name == 'push' && github.ref_name == 'main' }}
         run: |
           TUI_DIR="datafusion/ballista/tui/${{ steps.cargo_metadata.outputs.ballista_version }}"
           REMOTE_TARGET_DIR="${{ secrets.NIGHTLIES_RSYNC_PATH }}/${TUI_DIR}/"
@@ -101,7 +101,7 @@ jobs:
 
           ssh -p ${{ secrets.NIGHTLIES_RSYNC_PORT }} -l ${{ secrets.NIGHTLIES_RSYNC_USER }} ${{ secrets.NIGHTLIES_RSYNC_HOST }} "mkdir -p ${REMOTE_TARGET_DIR}"
 
-          rsync --times --compress --delete --verbose -e "ssh -p ${{ secrets.NIGHTLIES_RSYNC_PORT }} -l ${{ secrets.NIGHTLIES_RSYNC_USER }}" ./target/web-tui/* ${{ secrets.NIGHTLIES_RSYNC_HOST }}:${REMOTE_TARGET_DIR}
+          rsync --times --compress --delete --recursive --verbose -e "ssh -p ${{ secrets.NIGHTLIES_RSYNC_PORT }} -l ${{ secrets.NIGHTLIES_RSYNC_USER }}" ./target/web-tui/* ${{ secrets.NIGHTLIES_RSYNC_HOST }}:${REMOTE_TARGET_DIR}
 
       - name: Upload WASM32 application
         uses: actions/upload-artifact@v7

Cargo.toml

@@ -68,6 +68,7 @@ async-trait = { version = "0.1" }
 serde = { version = "1.0" }
 tokio-stream = { version = "0.1" }
 url = { version = "2.5" }
+url-escape = { version = "0.1.2" }
 
 # cargo build --profile release-lto
 [profile.release-lto]

ballista-cli/Cargo.toml

@@ -46,7 +46,7 @@ config = { version = "0.15.23", default-features = false, features = ["yaml"], o
 dirs = "6.0"
 dotparser = { version = "0.3.0", optional = true }
 futures = { version = "0.3.31", optional = true }
-percent-encoding = { version = "2.3.2", optional = true }
+url-escape = { workspace = true, optional = true }
 prometheus-parse = { version = "0.2", optional = true }
 ratatui = { version = "0.30.1", default-features = false, optional = true }
 reqwest = { version = "0.13.4", features = ["json"], optional = true }
@@ -87,14 +87,14 @@ cli = [
 
 tui = [
     "dep:chrono", "dep:config", "dep:crossterm", "dep:dotparser", "dep:futures",
-    "dep:percent-encoding", "dep:prometheus-parse", "dep:ratatui", "dep:reqwest",
+    "dep:url-escape", "dep:prometheus-parse", "dep:ratatui", "dep:reqwest",
     "dep:serde", "dep:serde_json", "dep:tracing-appender",
     "dep:tui-shimmer", "ratatui/crossterm", "ratatui/all-widgets", "ratatui/macros",
     "ratatui/layout-cache", "ratatui/underline-color", "ratatui/serde"
 ]
 
 web = [
-    "dep:chrono", "dep:config", "dep:critical-section", "dep:dotparser", "dep:futures", "dep:percent-encoding",
+    "dep:chrono", "dep:config", "dep:critical-section", "dep:dotparser", "dep:futures", "dep:url-escape",
     "dep:prometheus-parse", "dep:ratatui", "ratatui/serde", "dep:reqwest",
     "dep:serde", "dep:serde_json", "dep:tui-shimmer",
     "dep:ratzilla", "dep:wasm-bindgen", "dep:wasm-bindgen-futures",

ballista-cli/src/tui/http_client.rs

@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use percent_encoding::{NON_ALPHANUMERIC, percent_encode};
 use reqwest::{Client, Response};
 use serde::de::DeserializeOwned;
 #[cfg(not(target_arch = "wasm32"))]
@@ -213,6 +212,6 @@ impl HttpClient {
     }
 
     fn url_encode(&self, job_id: &str) -> String {
-        percent_encode(job_id.as_bytes(), NON_ALPHANUMERIC).to_string()
+        url_escape::encode_path(job_id).to_string()
     }
 }

ballista-cli/src/tui/infrastructure/config.rs

@@ -114,9 +114,9 @@ mod web {
         pub(super) fn parse() -> File<FileSourceString, FileFormat> {
             let mut data_reload_interval_ms = 2000;
             let mut repaint_interval_ms = 50;
-            let mut scheduler_url = "http://localhost:50050";
+            let mut scheduler_url = String::from("http://localhost:50050");
             let mut http_timeout_ms = 2000;
-            let mut theme_name = "dark";
+            let mut theme_name = String::from("dark");
             let mut theme_app_background_bg = String::from("black");
 
             let query_string = Self::decode_request();
@@ -131,13 +131,18 @@ mod web {
                         "ballista_repaint_interval_ms" => {
                             repaint_interval_ms = value.parse::<u64>().unwrap_or(50)
                         }
-                        "ballista_scheduler_url" => scheduler_url = value,
+                        "ballista_scheduler_url" => {
+                            scheduler_url = url_escape::decode(value).to_string()
+                        }
                         "ballista_http_timeout" => {
                             http_timeout_ms = value.parse::<u64>().unwrap_or(2000)
                         }
-                        "ballista_theme_name" => theme_name = value,
+                        "ballista_theme_name" => {
+                            theme_name = url_escape::decode(value).to_string()
+                        }
                         "ballista_theme_overrides_app_background_bg" => {
-                            theme_app_background_bg = value.replace("%23", "#");
+                            theme_app_background_bg =
+                                url_escape::decode(value).to_string();
                         }
                         _ => {}
                     }

ballista/scheduler/Cargo.toml

@@ -77,6 +77,7 @@ tonic-prost = { workspace = true, optional = true }
 tracing = { workspace = true, optional = true }
 tracing-appender = { workspace = true, optional = true }
 tracing-subscriber = { workspace = true, optional = true }
+url-escape = { workspace = true }
 uuid = { workspace = true }
 
 [[test]]

ballista/scheduler/src/api/handlers.rs

@@ -17,6 +17,7 @@ use crate::state::execution_graph_dot::ExecutionGraphDot;
 use crate::state::execution_stage::TaskInfo;
 use crate::{api::SchedulerErrorResponse, scheduler_server::SchedulerServer};
 use axum::extract::Query;
+use axum::response::Redirect;
 use axum::{
     Json,
     extract::{Path, State},
@@ -46,8 +47,9 @@ use graphviz_rust::{
     exec,
     printer::PrinterContext,
 };
-use http::{StatusCode, header::CONTENT_TYPE};
+use http::{HeaderMap, StatusCode, header::CONTENT_TYPE};
 use serde::Serialize;
+use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -227,6 +229,59 @@ pub enum PlanFormat {
     Metrics,
 }
 
+/// A handler for GET requests to the root (`/`).
+/// It redirects to `https://nightlies.apache.org/datafusion/ballista/tui/<BALLISTA_VERSION>/`
+/// forwarding any query parameters
+pub async fn get_webtui<
+    T: AsLogicalPlan + Clone + Send + Sync + 'static,
+    U: AsExecutionPlan + Send + Sync + 'static,
+>(
+    header_map: HeaderMap,
+    Query(mut params): Query<HashMap<String, String>>,
+    State(data_server): State<Arc<SchedulerServer<T, U>>>,
+) -> Result<Redirect, (StatusCode, String)> {
+    const NIGHTLIES_URL: &str = "https://nightlies.apache.org/datafusion/ballista/tui";
+    let external_host = &data_server.state.config.external_host;
+    let bind_port = data_server.state.config.bind_port;
+
+    let ballista_scheduler_url =
+        params.remove("ballista_scheduler_url").unwrap_or_else(|| {
+            let default_scheduler_url = &format!("{external_host}:{bind_port}");
+            let proto = header_map
+                .get("x-forwarded-proto")
+                .and_then(|v| v.to_str().ok())
+                .unwrap_or("http");
+            let host = header_map
+                .get("x-forwarded-host")
+                .and_then(|hv| hv.to_str().ok())
+                .unwrap_or(external_host);
+            let port = header_map
+                .get("x-forwarded-port")
+                .and_then(|hv| hv.to_str().ok())
+                .and_then(|v| v.parse::<u16>().ok())
+                .unwrap_or(bind_port);
+            format!("{proto}://{host}:{port}")
+        });
+
+    let mut query_string = String::new();
+    query_string.push_str(&format!(
+        "ballista_scheduler_url={}",
+        url_escape::encode_query(&ballista_scheduler_url)
+    ));
+
+    for (k, v) in params.iter() {
+        query_string.push_str(&format!(
+            "&{}={}",
+            url_escape::encode_query(k),
+            url_escape::encode_query(v)
+        ));
+    }
+
+    let target = format!("{NIGHTLIES_URL}/{BALLISTA_VERSION}/?{query_string}");
+
+    Ok(Redirect::to(&target))
+}
+
 pub async fn get_scheduler_state<
     T: AsLogicalPlan + Clone + Send + Sync + 'static,
     U: AsExecutionPlan + Send + Sync + 'static,
@@ -546,7 +601,7 @@ pub async fn get_query_stages<
                         let metrics = running_stage.stage_metrics.as_deref().unwrap_or(&[]);
                         summary.stage_plan = Some(match plan_format {
                             PlanFormat::Default => displayable(running_stage.plan.as_ref()).indent(false).to_string(),
-                            PlanFormat::Tree    => displayable(running_stage.plan.as_ref()).tree_render().to_string(),
+                            PlanFormat::Tree => displayable(running_stage.plan.as_ref()).tree_render().to_string(),
                             PlanFormat::Metrics => format_stage_metrics(running_stage.plan.as_ref(), metrics),
                         });
                         summary.input_rows = running_stage
@@ -591,7 +646,7 @@ pub async fn get_query_stages<
                                         finish_time: info.finish_time as u64,
                                         input_rows,
                                         output_rows,
-                                        status: task_status
+                                        status: task_status,
                                     }
                                 })
                             })
@@ -600,7 +655,7 @@ pub async fn get_query_stages<
                     ExecutionStage::Successful(completed_stage) => {
                         summary.stage_plan = Some(match plan_format {
                             PlanFormat::Default => displayable(completed_stage.plan.as_ref()).indent(false).to_string(),
-                            PlanFormat::Tree    => displayable(completed_stage.plan.as_ref()).tree_render().to_string(),
+                            PlanFormat::Tree => displayable(completed_stage.plan.as_ref()).tree_render().to_string(),
                             PlanFormat::Metrics => format_stage_metrics(completed_stage.plan.as_ref(), &completed_stage.stage_metrics),
                         });
                         summary.input_rows = get_combined_count(
@@ -638,7 +693,7 @@ pub async fn get_query_stages<
                                     finish_time: task_info.finish_time as u64,
                                     input_rows,
                                     output_rows,
-                                    status: task_status
+                                    status: task_status,
                                 })
                             })
                             .collect();
@@ -908,7 +963,7 @@ pub async fn get_job_dot_graph<
         })?
     {
         ExecutionGraphDot::generate(graph.as_ref())
-            .map_err(|e|  {
+            .map_err(|e| {
                 tracing::error!("Error occurred while getting the dot graph for job '{job_id}' reason: {e:?}");
                 SchedulerErrorResponse::new(StatusCode::INTERNAL_SERVER_ERROR)
             })
@@ -953,10 +1008,10 @@ pub async fn get_job_svg_graph<
                 &mut PrinterContext::default(),
                 vec![CommandArg::Format(Format::Svg)],
             )
-            .map_err(|e| {
-                tracing::error!("Error occurred while getting job svg graph for job '{job_id}' reason: {e:?}");
-                SchedulerErrorResponse::new(StatusCode::INTERNAL_SERVER_ERROR)
-            })?;
+                .map_err(|e| {
+                    tracing::error!("Error occurred while getting job svg graph for job '{job_id}' reason: {e:?}");
+                    SchedulerErrorResponse::new(StatusCode::INTERNAL_SERVER_ERROR)
+                })?;
 
             let svg = String::from_utf8_lossy(&result).to_string();
             Ok(Response::builder()

ballista/scheduler/src/api/routes.rs

@@ -27,7 +27,10 @@ pub fn get_routes<
 >(
     scheduler_server: Arc<SchedulerServer<T, U>>,
 ) -> Router {
+    let web_tui_route = scheduler_server.state.config.web_tui_route.clone();
+
     let router = Router::new()
+        .route(&web_tui_route, get(handlers::get_webtui))
         .route("/api/state", get(handlers::get_scheduler_state::<T, U>))
         .route("/api/version", get(handlers::get_scheduler_version))
         .route("/api/executors", get(handlers::get_executors::<T, U>))

ballista/scheduler/src/config.rs

@@ -48,13 +48,26 @@ pub struct Config {
     )]
     pub advertise_flight_sql_endpoint: Option<String>,
     /// Namespace for the ballista cluster.
-    #[arg(short = 'n', long, default_value_t = String::from("ballista"), help = "Namespace for the ballista cluster that this executor will join.")]
+    #[arg(
+        short = 'n',
+        long,
+        default_value_t = String::from("ballista"),
+        help = "Namespace for the ballista cluster that this executor will join."
+    )]
     pub namespace: String,
     /// Local host name or IP address to bind to.
-    #[arg(long, default_value_t = String::from("0.0.0.0"), help = "Local host name or IP address to bind to.")]
+    #[arg(
+        long,
+        default_value_t = String::from("0.0.0.0"),
+        help = "Local host name or IP address to bind to."
+    )]
     pub bind_host: String,
     /// External host name for executors to connect to.
-    #[arg(long, default_value_t = String::from("localhost"), help = "Host name or IP address so that executors can connect to this scheduler.")]
+    #[arg(
+        long,
+        default_value_t = String::from("localhost"),
+        help = "Host name or IP address so that executors can connect to this scheduler."
+    )]
     pub external_host: String,
     /// Port to bind the scheduler gRPC service.
     #[arg(
@@ -211,6 +224,14 @@ pub struct Config {
         help = "Comma-separated list of allowed methods for CORS. By default, GET, PATCH, and OPTIONS are allowed."
     )]
     pub cors_allowed_methods: String,
+    #[cfg(feature = "rest-api")]
+    /// The HTTP path that will redirect to the WebTUI app at `https://nightlies.apache.org`
+    #[arg(
+        long,
+        default_value_t = String::from("/"),
+        help = "The HTTP path that will redirect to the WebTUI app at https://nightlies.apache.org."
+    )]
+    pub web_tui_route: String,
 }
 
 /// Configurations for the ballista scheduler of scheduling jobs and tasks
@@ -278,6 +299,9 @@ pub struct SchedulerConfig {
     #[cfg(feature = "rest-api")]
     /// Comma-separated list of allowed methods for CORS
     pub cors_allowed_methods: String,
+    #[cfg(feature = "rest-api")]
+    /// The HTTP path that will redirect to the WebTUI app at `https://nightlies.apache.org`
+    pub web_tui_route: String,
 }
 
 impl Default for SchedulerConfig {
@@ -314,6 +338,8 @@ impl Default for SchedulerConfig {
             cors_allowed_origins: String::default(),
             #[cfg(feature = "rest-api")]
             cors_allowed_methods: String::default(),
+            #[cfg(feature = "rest-api")]
+            web_tui_route: String::from("/"),
         }
     }
 }
@@ -546,6 +572,8 @@ impl TryFrom<Config> for SchedulerConfig {
             cors_allowed_origins: opt.cors_allowed_origins,
             #[cfg(feature = "rest-api")]
             cors_allowed_methods: opt.cors_allowed_methods,
+            #[cfg(feature = "rest-api")]
+            web_tui_route: opt.web_tui_route,
         };
 
         Ok(config)