1pub mod actions;
2pub mod exposition;
3use crate::client::{command_support, llm_budget::call_llm_for_client};
4use crate::logging::emit::Log;
5use crate::protocol::{ConnectContext, Event};
6use crate::state::{
7 client_handles::{ClientCommand, ClientSendOutcome},
8 AccessLogOwner, ClientStatus,
9};
10pub use actions::PrometheusClientProtocol;
11use anyhow::{ensure, Context, Result};
12use serde_json::{json, Value};
13use std::{net::SocketAddr, time::Duration};
14use tokio::sync::mpsc;
15pub const DEFAULT_PATH: &str = "/metrics";
16pub const DEFAULT_TIMEOUT_SECS: u64 = 15;
17pub const MAX_TIMEOUT_SECS: u64 = 30;
18pub const MAX_FOLLOWUPS: u8 = 4;
19pub const QUEUE_CAPACITY: usize = 8;
20pub const MAX_PATH: usize = 4096;
21#[derive(Debug, Clone)]
22pub struct Request {
23 pub path: String,
24 pub format: String,
25}
26pub fn request(action: &Value, default_path: &str) -> Result<Request> {
27 ensure!(
28 action["type"] == "scrape_metrics",
29 "expected scrape_metrics action"
30 );
31 let path = match action.get("path") {
32 None => default_path,
33 Some(v) => v.as_str().context("path must be string")?,
34 };
35 ensure!(
36 !path.is_empty()
37 && path.len() <= MAX_PATH
38 && path.starts_with('/')
39 && !path.starts_with("//")
40 && !path.contains('#'),
41 "scrape path must be a bounded origin path/query"
42 );
43 let uri: hyper::Uri = path.parse().context("invalid scrape path")?;
44 ensure!(
45 uri.scheme().is_none() && uri.authority().is_none(),
46 "scrape path must stay on exporter origin"
47 );
48 let format = match action.get("format") {
49 None => "auto",
50 Some(v) => v.as_str().context("format must be string")?,
51 };
52 ensure!(
53 ["auto", "text", "openmetrics"].contains(&format),
54 "format must be auto/text/openmetrics"
55 );
56 Ok(Request {
57 path: path.into(),
58 format: format.into(),
59 })
60}
61fn accept(format: &str) -> &'static str {
62 match format {
63 "text"=>"text/plain;version=0.0.4;escaping=underscores",
64 "openmetrics"=>"application/openmetrics-text;version=1.0.0;escaping=underscores",
65 _=>"application/openmetrics-text;version=1.0.0;escaping=underscores;q=1.0,text/plain;version=0.0.4;escaping=underscores;q=0.8",
66 }
67}
68fn parse_response(
69 status: u16,
70 headers: &hyper::HeaderMap,
71 body: &[u8],
72 requested: &str,
73) -> Result<Value> {
74 ensure!(status == 200, "exporter returned HTTP {status}");
75 ensure!(
76 headers.get_all("content-type").iter().count() == 1,
77 "exporter must send exactly one Content-Type"
78 );
79 let content_type = headers
80 .get("content-type")
81 .context("missing Content-Type")?
82 .to_str()?;
83 let format = exposition::content_type(content_type)?;
84 ensure!(
85 requested == "auto" || requested == format.name(),
86 "exporter returned an unrequested metrics format"
87 );
88 for encoding in headers.get_all("content-encoding") {
89 ensure!(
90 encoding.to_str()?.eq_ignore_ascii_case("identity"),
91 "unsupported metrics Content-Encoding; requested identity"
92 );
93 }
94 let mut data = exposition::parse(body, format)?;
95 data["content_type"] = json!(content_type);
96 Ok(data)
97}
98#[cfg(not(target_arch = "wasm32"))]
99async fn fetch(
100 client: &reqwest::Client,
101 origin: &str,
102 request: &Request,
103 timeout: Duration,
104) -> Result<Value> {
105 let mut response = client
106 .get(format!("{origin}{}", request.path))
107 .header("Accept", accept(&request.format))
108 .header("Accept-Encoding", "identity")
109 .header(
110 "X-Prometheus-Scrape-Timeout-Seconds",
111 timeout.as_secs().to_string(),
112 )
113 .header("Connection", "close")
114 .send()
115 .await
116 .context("send exporter scrape")?;
117 let status = response.status().as_u16();
118 ensure!(status == 200, "exporter returned HTTP {status}");
119 if let Some(length) = response.content_length() {
120 ensure!(length <= exposition::MAX_BODY as u64, "metrics body limit");
121 }
122 let headers = response.headers().clone();
123 let mut body = Vec::new();
124 while let Some(chunk) = response.chunk().await.context("read exporter body")? {
125 ensure!(
126 body.len().saturating_add(chunk.len()) <= exposition::MAX_BODY,
127 "metrics body limit"
128 );
129 body.extend_from_slice(&chunk);
130 }
131 parse_response(status, &headers, &body, &request.format)
132}
133#[cfg(target_arch = "wasm32")]
134async fn fetch(_client: &(), origin: &str, request: &Request, timeout: Duration) -> Result<Value> {
135 use http_body_util::{BodyExt, Full, Limited};
136 use hyper_util::rt::TokioIo;
137 let target =
138 crate::client::http_fetch::transport::parse_http_url(&format!("{origin}{}", request.path))?;
139 let stream = tokio::net::TcpStream::connect((target.host.as_str(), target.port)).await?;
140 let (mut sender, connection) =
141 hyper::client::conn::http1::handshake(TokioIo::new(stream)).await?;
142 let exchange = async {
144 let response = sender
145 .send_request(
146 hyper::Request::builder()
147 .method("GET")
148 .uri(target.path_and_query.as_str())
149 .header("Host", target.authority.as_str())
150 .header("Accept", accept(&request.format))
151 .header("Accept-Encoding", "identity")
152 .header(
153 "X-Prometheus-Scrape-Timeout-Seconds",
154 timeout.as_secs().to_string(),
155 )
156 .header("Connection", "close")
157 .body(Full::new(bytes::Bytes::new()))?,
158 )
159 .await?;
160 let (parts, body) = response.into_parts();
161 let body = Limited::new(body, exposition::MAX_BODY)
162 .collect()
163 .await?
164 .to_bytes();
165 parse_response(
166 parts.status.as_u16(),
167 &parts.headers,
168 &body,
169 &request.format,
170 )
171 };
172 tokio::pin!(exchange, connection);
173 tokio::select! {result=&mut exchange=>result,result=&mut connection=>{result?;anyhow::bail!("exporter connection closed before complete scrape")}}
174}
175type HandlerEvent = (Event, u8);
176type HandlerAction = (Value, u8);
177pub async fn connect(ctx: ConnectContext) -> Result<SocketAddr> {
178 let address = if ctx.remote_addr.contains("://") {
179 ctx.remote_addr.clone()
180 } else {
181 format!("http://{}", ctx.remote_addr)
182 };
183 let url = url::Url::parse(&address).context("invalid exporter URL")?;
184 ensure!(
185 ["http", "https"].contains(&url.scheme()) && url.host_str().is_some(),
186 "exporter URL must be HTTP(S)"
187 );
188 ensure!(url.username().is_empty() && url.password().is_none() && url.fragment().is_none() && url.query().is_none() && url.path()=="/","use an exporter origin without credentials/path/query; metrics_path selects the scrape endpoint");
189 let origin = url.origin().ascii_serialization();
190 let mut default_path = DEFAULT_PATH.to_owned();
191 let mut timeout_secs = DEFAULT_TIMEOUT_SECS;
192 if let Some(p) = &ctx.startup_params {
193 default_path = p
194 .get_optional_string("metrics_path")?
195 .unwrap_or_else(|| DEFAULT_PATH.into());
196 timeout_secs = p
197 .get_optional_u64("scrape_timeout_secs")?
198 .unwrap_or(DEFAULT_TIMEOUT_SECS);
199 }
200 request(
201 &json!({"type":"scrape_metrics","path":default_path}),
202 DEFAULT_PATH,
203 )?;
204 ensure!(
205 (1..=MAX_TIMEOUT_SECS).contains(&timeout_secs),
206 "scrape_timeout_secs must be 1..30"
207 );
208 let timeout = Duration::from_secs(timeout_secs);
209 #[cfg(target_arch = "wasm32")]
210 crate::client::http_fetch::transport::parse_http_url(&origin)?;
211 let host = url
212 .host_str()
213 .unwrap()
214 .trim_start_matches('[')
215 .trim_end_matches(']');
216 let remote = tokio::time::timeout(
217 timeout,
218 tokio::net::lookup_host((host, url.port_or_known_default().unwrap())),
219 )
220 .await
221 .context("exporter resolution deadline")??
222 .next()
223 .context("exporter resolved no addresses")?;
224 #[cfg(not(target_arch = "wasm32"))]
225 let client = reqwest::Client::builder()
226 .no_proxy()
227 .no_gzip()
228 .no_brotli()
229 .no_deflate()
230 .no_zstd()
231 .redirect(reqwest::redirect::Policy::none())
232 .timeout(timeout)
233 .pool_max_idle_per_host(0)
234 .http1_only()
235 .user_agent("NetGet-Prometheus/1.0")
236 .build()?;
237 #[cfg(target_arch = "wasm32")]
238 let client = ();
239 let external = command_support::register_command_channel(&ctx.state, ctx.client_id).await;
240 let (actions_tx, actions_rx) = mpsc::channel::<HandlerAction>(QUEUE_CAPACITY);
241 let (events_tx, mut events_rx) = mpsc::channel::<HandlerEvent>(QUEUE_CAPACITY);
242 let handler_ctx = ctx.clone();
243 let handler = tokio::spawn(async move {
244 while let Some((event, depth)) = events_rx.recv().await {
245 let instruction = handler_ctx
246 .state
247 .get_instruction_for_client(handler_ctx.client_id)
248 .await
249 .unwrap_or_default();
250 let memory = handler_ctx
251 .state
252 .get_memory_for_client(handler_ctx.client_id)
253 .await
254 .unwrap_or_default();
255 match call_llm_for_client(
256 &handler_ctx.llm_client,
257 &handler_ctx.state,
258 handler_ctx.client_id.to_string(),
259 &instruction,
260 &memory,
261 Some(&event),
262 &PrometheusClientProtocol,
263 &handler_ctx.status_tx,
264 )
265 .await
266 {
267 Ok(result) => {
268 if let Some(memory) = result.memory_updates {
269 handler_ctx
270 .state
271 .set_memory_for_client(handler_ctx.client_id, memory)
272 .await;
273 }
274 for action in result.actions {
275 if depth >= MAX_FOLLOWUPS && action["type"] != "disconnect" {
276 Log::new(Some(&handler_ctx.status_tx))
277 .warn("Prometheus handler followup limit reached");
278 continue;
279 }
280 if actions_tx.send((action, depth + 1)).await.is_err() {
281 return;
282 }
283 }
284 }
285 Err(e) => {
286 Log::new(Some(&handler_ctx.status_tx)).warn(format!("Prometheus handler: {e}"))
287 }
288 }
289 }
290 });
291 let handler_abort = handler.abort_handle();
292 ctx.state.register_client_task(ctx.client_id, handler).await;
293 let session_ctx = ctx.clone();
294 let task = tokio::spawn(async move {
295 let result = session(
296 &session_ctx,
297 &client,
298 &origin,
299 &default_path,
300 timeout,
301 external,
302 actions_rx,
303 events_tx,
304 )
305 .await;
306 handler_abort.abort();
307 let status = match result {
308 Ok(()) => ClientStatus::Disconnected,
309 Err(e) => {
310 Log::new(Some(&session_ctx.status_tx)).warn(format!("Prometheus ended: {e}"));
311 ClientStatus::Error(e.to_string())
312 }
313 };
314 session_ctx
315 .state
316 .update_client_status(session_ctx.client_id, status)
317 .await;
318 session_ctx
319 .state
320 .remove_client_handle(session_ctx.client_id)
321 .await;
322 let _ = session_ctx.status_tx.send("__UPDATE_UI__".into());
323 });
324 ctx.state.register_client_task(ctx.client_id, task).await;
325 Ok(remote)
326}
327#[cfg(not(target_arch = "wasm32"))]
328type HttpClient = reqwest::Client;
329#[cfg(target_arch = "wasm32")]
330type HttpClient = ();
331#[allow(clippy::too_many_arguments)]
332async fn session(
333 ctx: &ConnectContext,
334 client: &HttpClient,
335 origin: &str,
336 default_path: &str,
337 timeout: Duration,
338 mut external: mpsc::Receiver<ClientCommand>,
339 mut internal: mpsc::Receiver<HandlerAction>,
340 events: mpsc::Sender<HandlerEvent>,
341) -> Result<()> {
342 events
343 .try_send((
344 Event::new(
345 &actions::CONNECTED_EVENT,
346 json!({"origin":origin,"metrics_path":default_path}),
347 ),
348 0,
349 ))
350 .context("Prometheus event queue full")?;
351 loop {
352 let (action, depth, command) = tokio::select! {
353 command=external.recv()=>match command {Some(c)=>(c.action.clone(),0,Some(c)),None=>return Ok(())},
354 action=internal.recv()=>match action {Some((a,d))=>(a,d,None),None=>return Ok(())},
355 };
356 if action["type"] == "disconnect" {
357 if let Some(c) = command {
358 command_support::reply(c, Ok(ClientSendOutcome::Disconnected));
359 }
360 return Ok(());
361 }
362 let request = match request(&action, default_path) {
363 Ok(r) => r,
364 Err(e) => {
365 if let Some(c) = command {
366 command_support::reply(
367 c,
368 Ok(ClientSendOutcome::Rejected {
369 error: e.to_string(),
370 }),
371 );
372 } else {
373 Log::new(Some(&ctx.status_tx)).warn(format!("Prometheus action rejected: {e}"));
374 }
375 continue;
376 }
377 };
378 if let Some(c) = command {
379 ctx.state
380 .record_access_log(
381 AccessLogOwner::Client(ctx.client_id.as_u32()),
382 "Prometheus",
383 None,
384 "injected_action",
385 action.clone(),
386 vec![json!({"accepted":true})],
387 )
388 .await;
389 command_support::reply(
390 c,
391 Ok(ClientSendOutcome::Executed {
392 detail: "scrape accepted; result follows as an event".into(),
393 }),
394 );
395 }
396 let result = {
397 let scrape = tokio::time::timeout(timeout, fetch(client, origin, &request, timeout));
398 tokio::pin!(scrape);
399 loop {
400 tokio::select! {biased;
401 command=external.recv()=>{
402 let Some(c)=command else{return Ok(())};
403 if c.action["type"]=="disconnect" {command_support::reply(c,Ok(ClientSendOutcome::Disconnected));return Ok(());}
404 command_support::reply(c,Ok(ClientSendOutcome::Rejected{error:"Prometheus scrape pending; retry after its event".into()}));
405 },
406 result=&mut scrape=>break result.context("whole exporter scrape deadline").and_then(|r|r),
407 }
408 }
409 };
410 let event = match result {
411 Ok(mut data) => {
412 data["request"] = action;
413 data["path"] = json!(request.path);
414 Event::new(&actions::METRICS_EVENT, data)
415 }
416 Err(e) => Event::new(
417 &actions::ERROR_EVENT,
418 json!({"request":action,"path":request.path,"error":format!("{e:#}")}),
419 ),
420 };
421 events
422 .try_send((event, depth))
423 .context("Prometheus event queue full; disconnecting bounded client")?;
424 }
425}