metrale_bench/http/
get.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! 2026-09-26: Plain GET requests: `/v1/models` (`list_models`, `probe`),
4//! `get_json`, `get_text` and `fetch_hardware`.
5//!
6//! Owner: bench (HTTP client).
7//! Invariants:
8//! - Every request sends `Connection: close` and is bounded by the caller's
9//!   `timeout`.
10//! - Every read loop stops at EOF or once past a byte bound (512 B for
11//!   `probe`, 64 KiB for `/hardware`, `TEXT_BODY_LIMIT` for `get_text`, 1 MiB
12//!   otherwise). `get_text` fails past its bound rather than truncate.
13
14use super::*;
15
16pub async fn list_models(target: &TargetEndpoint, timeout: Duration) -> Result<Vec<String>> {
17    let body = get_models(target, timeout).await?;
18    let start = body
19        .find('{')
20        .context("no JSON in the /v1/models response")?;
21    // 2026-09-26: Parse the first value and ignore whatever follows. A
22    // chunked reply carries hex length prefixes and a terminating
23    // `0\r\n\r\n`, on which a plain `from_str` fails.
24    let doc: serde_json::Value = serde_json::Deserializer::from_str(&body[start..])
25        .into_iter()
26        .next()
27        .context("/v1/models returned an empty body")?
28        .context("/v1/models did not return JSON")?;
29    Ok(doc
30        .get("data")
31        .and_then(|d| d.as_array())
32        .map(|rows| {
33            rows.iter()
34                .filter_map(|r| r.get("id").and_then(|i| i.as_str()))
35                .map(str::to_string)
36                .collect()
37        })
38        .unwrap_or_default())
39}
40
41/// 2026-09-26: Reachability check: `GET /v1/models` must answer 200 within
42/// `timeout`. Every benchmark driver except vision and video calls it.
43pub async fn probe(target: &TargetEndpoint, timeout: Duration) -> Result<()> {
44    let (host, port) = target.host_port()?;
45    let fut = async {
46        let mut sock = TcpStream::connect((host.as_str(), port)).await?;
47        let req =
48            format!("GET /v1/models HTTP/1.1\r\nHost: {host}:{port}\r\nConnection: close\r\n\r\n");
49        sock.write_all(req.as_bytes()).await?;
50        let mut head = Vec::new();
51        let mut buf = [0u8; 1024];
52        while head.len() < 512 {
53            let n = sock.read(&mut buf).await?;
54            if n == 0 {
55                break;
56            }
57            head.extend_from_slice(&buf[..n]);
58        }
59        anyhow::Ok(String::from_utf8_lossy(&head).into_owned())
60    };
61    let head = tokio::time::timeout(timeout, fut)
62        .await
63        .map_err(|_| anyhow!("{} did not answer within {:?}", target.base_url, timeout))?
64        .with_context(|| format!("probing {}", target.base_url))?;
65    let status = head.lines().next().unwrap_or_default();
66    if !status_is_success(status) {
67        bail!("{} /v1/models returned {status:?}", target.base_url);
68    }
69    Ok(())
70}
71
72/// 2026-09-26: The whole `/v1/models` response, headers included.
73async fn get_models(target: &TargetEndpoint, timeout: Duration) -> Result<String> {
74    let (host, port) = target.host_port()?;
75    let fut = async {
76        let mut sock = TcpStream::connect((host.as_str(), port)).await?;
77        let req =
78            format!("GET /v1/models HTTP/1.1\r\nHost: {host}:{port}\r\nConnection: close\r\n\r\n");
79        sock.write_all(req.as_bytes()).await?;
80        let mut body = Vec::new();
81        let mut buf = [0u8; 4096];
82        // 2026-09-26: Read to EOF, bounded at 1 MiB: a short fixed read would
83        // cut a long model list into invalid JSON.
84        loop {
85            let n = sock.read(&mut buf).await?;
86            if n == 0 {
87                break;
88            }
89            body.extend_from_slice(&buf[..n]);
90            if body.len() > 1 << 20 {
91                break;
92            }
93        }
94        anyhow::Ok(String::from_utf8_lossy(&body).into_owned())
95    };
96    tokio::time::timeout(timeout, fut)
97        .await
98        .map_err(|_| anyhow!("{} did not answer within {:?}", target.base_url, timeout))?
99        .with_context(|| format!("reading models from {}", target.base_url))
100}
101
102/// 2026-09-26: `GET <path>` and parse the first JSON value of the reply (a
103/// chunked reply carries framing after the document). A status other than
104/// 200 is an error that quotes the document.
105pub async fn get_json(
106    target: &TargetEndpoint,
107    path: &str,
108    timeout: Duration,
109) -> Result<serde_json::Value> {
110    let (host, port) = target.host_port()?;
111    let fut = async {
112        let mut sock = TcpStream::connect((host.as_str(), port)).await?;
113        let req =
114            format!("GET {path} HTTP/1.1\r\nHost: {host}:{port}\r\nConnection: close\r\n\r\n");
115        sock.write_all(req.as_bytes()).await?;
116        let mut body = Vec::new();
117        let mut buf = [0u8; 4096];
118        loop {
119            let n = sock.read(&mut buf).await?;
120            if n == 0 {
121                break;
122            }
123            body.extend_from_slice(&buf[..n]);
124            if body.len() > 1 << 20 {
125                break;
126            }
127        }
128        anyhow::Ok(String::from_utf8_lossy(&body).into_owned())
129    };
130    let raw = tokio::time::timeout(timeout, fut)
131        .await
132        .map_err(|_| {
133            anyhow!(
134                "{} did not answer {path} within {:?}",
135                target.base_url,
136                timeout
137            )
138        })?
139        .with_context(|| format!("reading {path} from {}", target.base_url))?;
140    let status = raw
141        .split_whitespace()
142        .nth(1)
143        .and_then(|s| s.parse::<u16>().ok())
144        .unwrap_or(0);
145    let start = raw
146        .find('{')
147        .with_context(|| format!("{path} returned no JSON (status {status})"))?;
148    let doc: serde_json::Value = serde_json::Deserializer::from_str(&raw[start..])
149        .into_iter()
150        .next()
151        .with_context(|| format!("{path} returned an empty body"))?
152        .with_context(|| format!("{path} did not return JSON"))?;
153    if status != 200 {
154        anyhow::bail!("{path} answered {status}: {doc}");
155    }
156    Ok(doc)
157}
158
159/// 2026-10-04: The most bytes `get_text` reads. A telemetry-on `/metrics` page carries
160/// one histogram per scheduler phase, so it is larger than the JSON documents above.
161const TEXT_BODY_LIMIT: usize = 8 << 20;
162
163/// 2026-10-04: `GET <path>` and return the body as text, chunked framing decoded, lines
164/// joined with `\n` and each trimmed. A status other than 200 is an error that quotes the
165/// body (the streaming client's `Reader` does both). A body past `TEXT_BODY_LIMIT` is an
166/// error, so a cut-off page never reads as a page without the series that were cut.
167pub async fn get_text(target: &TargetEndpoint, path: &str, timeout: Duration) -> Result<String> {
168    let (host, port) = target.host_port()?;
169    let fut = async {
170        let mut sock = TcpStream::connect((host.as_str(), port)).await?;
171        let req =
172            format!("GET {path} HTTP/1.1\r\nHost: {host}:{port}\r\nConnection: close\r\n\r\n");
173        sock.write_all(req.as_bytes()).await?;
174        let mut reader = Reader::default();
175        let mut lines = Vec::new();
176        let mut read = 0usize;
177        let mut buf = [0u8; 16 * 1024];
178        loop {
179            let n = sock.read(&mut buf).await?;
180            if n == 0 {
181                reader.finish()?;
182                break;
183            }
184            read += n;
185            if read > TEXT_BODY_LIMIT {
186                bail!("{path} is larger than {TEXT_BODY_LIMIT} bytes");
187            }
188            lines.extend(reader.push(&buf[..n])?);
189        }
190        // 2026-10-04: A last line with no newline stays in the reader's buffer.
191        let tail = String::from_utf8_lossy(&reader.body).trim().to_string();
192        if !tail.is_empty() {
193            lines.push(tail);
194        }
195        anyhow::Ok(lines.join("\n"))
196    };
197    tokio::time::timeout(timeout, fut)
198        .await
199        .map_err(|_| {
200            anyhow!(
201                "{} did not answer {path} within {:?}",
202                target.base_url,
203                timeout
204            )
205        })?
206        .with_context(|| format!("reading {path} from {}", target.base_url))
207}
208
209/// 2026-09-26: `GET /hardware`: the serving box's hardware fingerprint.
210///
211/// Fetched from the endpoint rather than probed locally because a benchmark
212/// number belongs to the box that did the inference, which is not
213/// necessarily the box running the benchmark CLI. Any failure (bad target,
214/// connect, timeout, no JSON, wrong shape) returns
215/// [`crate::hardware::Hardware::unknown`], so a server without the endpoint
216/// does not make a run unrecordable.
217pub async fn fetch_hardware(
218    target: &TargetEndpoint,
219    timeout: Duration,
220) -> crate::hardware::Hardware {
221    let (host, port) = match target.host_port() {
222        Ok(hp) => hp,
223        Err(_) => return crate::hardware::Hardware::unknown(),
224    };
225    let fut = async {
226        let mut sock = TcpStream::connect((host.as_str(), port)).await?;
227        let req =
228            format!("GET /hardware HTTP/1.1\r\nHost: {host}:{port}\r\nConnection: close\r\n\r\n");
229        sock.write_all(req.as_bytes()).await?;
230        let mut body = Vec::new();
231        let mut buf = [0u8; 4096];
232        loop {
233            let n = sock.read(&mut buf).await?;
234            if n == 0 {
235                break;
236            }
237            body.extend_from_slice(&buf[..n]);
238            if body.len() > 64 * 1024 {
239                break;
240            }
241        }
242        anyhow::Ok(String::from_utf8_lossy(&body).into_owned())
243    };
244    let Ok(raw) = tokio::time::timeout(timeout, fut).await else {
245        return crate::hardware::Hardware::unknown();
246    };
247    let Ok(raw) = raw else {
248        return crate::hardware::Hardware::unknown();
249    };
250    // 2026-09-26: Chunked framing: parse the first JSON value and ignore the
251    // rest, as `list_models` does.
252    let Some(start) = raw.find('{') else {
253        return crate::hardware::Hardware::unknown();
254    };
255    serde_json::Deserializer::from_str(&raw[start..])
256        .into_iter()
257        .next()
258        .and_then(|r| r.ok())
259        .and_then(|doc: serde_json::Value| serde_json::from_value(doc).ok())
260        .unwrap_or_else(crate::hardware::Hardware::unknown)
261}
metrale_bench/http/get.rs

metrale_bench/http/
get.rs
