diff --git a/src/engine/ingest.rs b/src/engine/ingest.rs index 1a94383..c01edd7 100644 --- a/src/engine/ingest.rs +++ b/src/engine/ingest.rs @@ -106,25 +106,34 @@ pub async fn run_file(engine: EngineHandle, path: PathBuf) { } } +async fn fetch_metrics_body(client: &reqwest::Client, url: &str) -> Result { + let response = client + .get(url) + .send() + .await + .map_err(|err| format!("warning: failed to poll metrics endpoint {url}: {err}"))?; + let response = response + .error_for_status() + .map_err(|err| format!("warning: metrics endpoint {url} returned an error: {err}"))?; + response + .text() + .await + .map_err(|err| format!("warning: failed to read metrics response from {url}: {err}")) +} + pub async fn run_metrics_poll(engine: EngineHandle, url: String) { let client = reqwest::Client::new(); loop { - match client.get(&url).send().await { - Ok(response) => match response.error_for_status() { - Ok(response) => match response.text().await { - Ok(body) => { - let now = Timestamp::now(); - for sample in parse_openmetrics(&body, now) { - if engine.tx.send(EngineMessage::IngestMetricSample(sample)).await.is_err() { - return; - } - } + match fetch_metrics_body(&client, &url).await { + Ok(body) => { + let now = Timestamp::now(); + for sample in parse_openmetrics(&body, now) { + if engine.tx.send(EngineMessage::IngestMetricSample(sample)).await.is_err() { + return; } - Err(err) => eprintln!("warning: failed to read metrics response from {url}: {err}"), - }, - Err(err) => eprintln!("warning: metrics endpoint {url} returned an error: {err}"), - }, - Err(err) => eprintln!("warning: failed to poll metrics endpoint {url}: {err}"), + } + } + Err(warning) => eprintln!("{warning}"), } tokio::time::sleep(METRICS_POLL_INTERVAL).await; }