use reqwest::{Client, Response};
use serde::de::DeserializeOwned;
#[cfg(not(target_arch = "wasm32"))]
use std::time::Duration;
use crate::tui::domain::jobs::stages::StagePlanTab;
use crate::tui::{
TuiResult,
domain::{
SchedulerState,
executors::Executor,
jobs::{
CancelJobResponse, Job, JobConfigResponse, JobDetails,
stages::JobStagesResponse,
},
metrics::{Metric, MetricsResponse},
},
error::TuiError,
infrastructure::Settings,
};
pub struct HttpClient {
config: Settings,
client: reqwest::Client,
}
impl HttpClient {
pub fn new(config: Settings) -> TuiResult<Self> {
let builder = Client::builder();
#[cfg(not(target_arch = "wasm32"))]
let builder = builder.timeout(Duration::from_millis(config.http.timeout));
Ok(Self {
client: builder.build()?,
config,
})
}
pub fn scheduler_url(&self) -> &str {
&self.config.scheduler.url
}
pub async fn get_scheduler_state(&self) -> TuiResult<SchedulerState> {
let url = self.url("state");
self.json::<SchedulerState>(&url).await
}
pub async fn get_executors(&self) -> TuiResult<Vec<Executor>> {
let url = self.url("executors");
self.json::<Vec<Executor>>(&url).await
}
pub async fn get_executor(&self, executor_id: &str) -> TuiResult<Executor> {
let url = self.url(&format!("executor/{}", self.url_encode(executor_id)));
tracing::trace!("Going to GET details for '{}'", &url);
self.json::<Executor>(&url).await
}
pub async fn get_jobs(&self) -> TuiResult<Vec<Job>> {
let url = self.url("jobs");
self.json::<Vec<Job>>(&url).await.map(|mut jobs| {
jobs.sort_by_key(|b| std::cmp::Reverse(b.start_time)); jobs
})
}
pub async fn cancel_job(&self, job_id: &str) -> TuiResult<CancelJobResponse> {
let url = self.url(&format!("job/{}", self.url_encode(job_id)));
tracing::trace!("Going to PATCH {}", &url);
let response = self
.client
.patch(&url)
.send()
.await
.inspect_err(|err| tracing::error!("The HTTP PATCH request failed: {err:?}"))
.map_err(TuiError::from)?;
let response = response
.error_for_status()
.map_err(TuiError::from)
.inspect_err(|err| tracing::error!("HTTP error status: {err:?}"))?;
response
.json::<CancelJobResponse>()
.await
.map_err(TuiError::from)
.inspect(|data| tracing::trace!("Cancel response: {data:?}"))
.inspect_err(|err| {
tracing::error!("Failed to parse cancel response: {err:?}")
})
}
pub async fn get_job_details(
&self,
job_id: &str,
plan_format: &StagePlanTab,
) -> TuiResult<JobDetails> {
#[derive(serde::Deserialize, Debug)]
struct JobDetailResponse {
logical_plan: Option<String>,
physical_plan: Option<String>,
stage_plan: Option<String>,
}
let url = self.url(&format!(
"job/{}?{}",
self.url_encode(job_id),
plan_format.as_query_param(),
));
let resp = self.json::<JobDetailResponse>(&url).await?;
Ok(JobDetails {
job_id: job_id.to_string(),
logical_plan: resp.logical_plan,
physical_plan: resp.physical_plan,
physical_plan_tree: None,
stage_plan: resp.stage_plan,
})
}
pub async fn get_job_dot(&self, job_id: &str) -> TuiResult<String> {
let url = self.url(&format!("job/{}/dot", self.url_encode(job_id)));
self.text(&url).await
}
pub async fn get_job_stages(
&self,
job_id: &str,
plan_format: &StagePlanTab,
) -> TuiResult<JobStagesResponse> {
let url = self.url(&format!(
"job/{}/stages?{}",
self.url_encode(job_id),
plan_format.as_query_param(),
));
self.json::<JobStagesResponse>(&url).await
}
pub async fn get_job_config(&self, job_id: &str) -> TuiResult<JobConfigResponse> {
let url = self.url(&format!("job/{}/config", self.url_encode(job_id)));
self.json::<JobConfigResponse>(&url).await
}
pub async fn get_metrics(&self) -> TuiResult<Vec<Metric>> {
let url = self.url("metrics");
let body: String = self.text(&url).await?;
let response = body.parse::<MetricsResponse>().map_err(TuiError::from)?;
Ok(response.metrics)
}
async fn text(&self, url: &str) -> TuiResult<String> {
let response = self.get(url).await?;
let response = response
.error_for_status()
.map_err(TuiError::from)
.inspect_err(|err| tracing::error!("HTTP error status: {err:?}"))?;
response
.text()
.await
.map_err(TuiError::from)
.inspect(|data| tracing::trace!("Loaded: {data:?}"))
.inspect_err(|err| tracing::error!("The HTTP request failed: {err:?}"))
}
async fn json<R>(&self, url: &str) -> TuiResult<R>
where
R: std::fmt::Debug + DeserializeOwned,
{
let response = self.get(url).await?;
let response = response
.error_for_status()
.map_err(TuiError::from)
.inspect_err(|err| tracing::error!("HTTP error status: {err:?}"))?;
response
.json::<R>()
.await
.map_err(TuiError::from)
.inspect(|data| tracing::trace!("Loaded: {data:?}"))
.inspect_err(|err| tracing::error!("The HTTP request failed: {err:?}"))
}
async fn get(&self, url: &str) -> TuiResult<Response> {
tracing::trace!("Going to make a request to {}", &url);
self.client
.get(url)
.send()
.await
.inspect(|data| tracing::trace!("Got: {data:?}"))
.inspect_err(|err| tracing::error!("The HTTP GET request failed: {err:?}"))
.map_err(TuiError::from)
}
fn url(&self, path: &str) -> String {
format!("{}/api/{}", self.config.scheduler.url, path)
}
fn url_encode(&self, job_id: &str) -> String {
url_escape::encode_path(job_id).to_string()
}
}