Merge ticket branch t082
This commit is contained in:
+24
-15
@@ -106,25 +106,34 @@ pub async fn run_file(engine: EngineHandle, path: PathBuf) {
|
||||
}
|
||||
}
|
||||
|
||||
async fn fetch_metrics_body(client: &reqwest::Client, url: &str) -> Result<String, String> {
|
||||
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;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user