netget/server/otlp/
mod.rs

1//! OpenTelemetry HTTP/gRPC receiver — the model decides which exports are accepted.
2//!
3//! OTLP exporters `POST` to `/v1/traces`, `/v1/metrics` or `/v1/logs` in protobuf or JSON,
4//! optionally gzip-compressed. NetGet reads the body (bounded before and after inflation),
5//! decodes it, and raises one `otlp_export` event carrying a summary — never the payload. The
6//! model answers with a verdict (`accept_otlp`, `accept_otlp_partially`, `reject_otlp`), and
7//! NetGet encodes the response in the request's own encoding (see [`codec`]).
8//! Generated unary gRPC Export services use the same semantic verdict path on HTTP/2.
9//!
10//! Everything that is not a well-formed export is answered by NetGet without asking the model:
11//! another path (404), method (405), content type or content encoding (415), a body past the
12//! cap (413), bad gzip or an undecodable payload (400).
13//!
14//! See `src/server/otlp/AGENTS.md`.
15
16pub mod actions;
17pub mod codec;
18mod grpc;
19
20use std::convert::Infallible;
21use std::net::SocketAddr;
22use std::sync::Arc;
23
24use bytes::Bytes;
25use http_body_util::{BodyExt, Full};
26use hyper::body::{Body, Incoming};
27use hyper::header::{HeaderValue, ALLOW, CONTENT_ENCODING, CONTENT_TYPE, RETRY_AFTER};
28use hyper::service::service_fn;
29use hyper::{Method, Request, Response, StatusCode};
30use hyper_util::rt::TokioIo;
31use hyper_util::server::conn::auto;
32use tokio::sync::mpsc;
33use tracing::{debug, info};
34
35use crate::llm::action_helper::call_llm;
36use crate::llm::ollama_client::OllamaClient;
37use crate::protocol::Event;
38use crate::server::connection::ConnectionId;
39use crate::state::app_state::AppState;
40use crate::state::ServerId;
41use crate::{console_error, console_info};
42
43use actions::Verdict;
44use codec::{Encoding, Signal};
45
46pub use codec::MAX_BODY_BYTES;
47
48/// How long to wait for a peer's first byte after it connects. HTTP is client-speaks-first;
49/// enforced with `TcpStream::peek` before hyper sees the socket, so no deadline runs inside a
50/// model round-trip.
51const FIRST_BYTE_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
52
53/// How long a connection may do nothing at all between requests. OTLP exporters keep a
54/// connection and send a batch every few seconds (the SDKs' default schedule delay is 5 s for
55/// spans and 60 s for metrics), so two minutes keeps it across a metrics interval. A request in
56/// flight — the model thinking, or a `manual` rule parked for a human — is not idle.
57const IDLE_BETWEEN_REQUESTS_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(120);
58
59/// Concurrent connections this server admits.
60const MAX_CONNECTIONS: usize = crate::server::accept_bounded::DEFAULT_MAX_CONNECTIONS;
61
62/// What a peer over [`MAX_CONNECTIONS`] is told before the socket closes: a status OTLP
63/// clients retry.
64const CONNECTION_CAP_REFUSAL: &[u8] = b"HTTP/1.1 503 Service Unavailable\r\n\
65    Content-Length: 0\r\nRetry-After: 5\r\nConnection: close\r\n\r\n";
66
67/// The fixed message for an export the model answered with no verdict.
68const NO_DECISION: &str = "netget: the receiver reached no decision on this export";
69
70/// OTLP/HTTP receiver.
71pub struct OtlpServer;
72
73impl OtlpServer {
74    /// Bind, then spawn the accept loop. A bind failure reaches the caller, and the accept-loop
75    /// handle is registered so `stop_server` can abort it and release the port.
76    pub async fn spawn_with_llm_actions(
77        listen_addr: SocketAddr,
78        llm_client: OllamaClient,
79        app_state: Arc<AppState>,
80        status_tx: mpsc::UnboundedSender<String>,
81        server_id: ServerId,
82    ) -> anyhow::Result<SocketAddr> {
83        let listener =
84            crate::server::socket_helpers::create_reusable_tcp_listener(listen_addr).await?;
85        let local_addr = listener.local_addr()?;
86
87        console_info!(
88            status_tx,
89            "OTLP/HTTP receiver listening on http://{}/v1/{{traces,metrics,logs}}",
90            local_addr
91        );
92
93        let protocol = Arc::new(actions::OtlpProtocol::new());
94        let task_registrar = app_state.clone();
95        let limiter = crate::server::accept_bounded::ConnectionLimiter::new(MAX_CONNECTIONS);
96        let exports = Arc::new(tokio::sync::Semaphore::new(grpc::MAX_EXPORTS));
97        let accept_handle = tokio::spawn(async move {
98            loop {
99                match crate::server::accept_bounded::accept_bounded(
100                    &listener,
101                    &limiter,
102                    CONNECTION_CAP_REFUSAL,
103                    "OTLP",
104                    Some(&status_tx),
105                )
106                .await
107                {
108                    Ok((stream, remote_addr, permit)) => {
109                        let connection_id =
110                            ConnectionId::new(app_state.get_next_unified_id().await);
111                        let local_addr_conn = stream.local_addr().unwrap_or(local_addr);
112                        info!("OTLP connection {} from {}", connection_id, remote_addr);
113
114                        use crate::state::server::{
115                            ConnectionState as ServerConnectionState, ConnectionStatus,
116                            ProtocolConnectionInfo,
117                        };
118                        let now = crate::utils::clock::Instant::now();
119                        app_state
120                            .add_connection_to_server(
121                                server_id,
122                                ServerConnectionState {
123                                    id: connection_id,
124                                    remote_addr,
125                                    local_addr: local_addr_conn,
126                                    bytes_sent: 0,
127                                    bytes_received: 0,
128                                    packets_sent: 0,
129                                    packets_received: 0,
130                                    last_activity: now,
131                                    status: ConnectionStatus::Active,
132                                    status_changed_at: now,
133                                    protocol_info: ProtocolConnectionInfo::empty(),
134                                },
135                            )
136                            .await;
137                        let _ = status_tx.send("__UPDATE_UI__".to_string());
138
139                        let ctx = RequestContext {
140                            llm_client: llm_client.clone(),
141                            app_state: app_state.clone(),
142                            status_tx: status_tx.clone(),
143                            protocol: protocol.clone(),
144                            server_id,
145                            connection_id,
146                            exports: exports.clone(),
147                        };
148                        let app_state_for_close = app_state.clone();
149                        let status_for_close = status_tx.clone();
150
151                        let task_owner = app_state.clone();
152                        task_owner
153                            .spawn_server_task(server_id, async move {
154                                // Held for the life of the connection.
155                                let _permit = permit;
156                                match tokio::time::timeout(
157                                    FIRST_BYTE_READ_TIMEOUT,
158                                    stream.peek(&mut [0u8; 1]),
159                                )
160                                .await
161                                {
162                                    Ok(Ok(n)) if n > 0 => {
163                                        serve_connection(TokioIo::new(stream), ctx).await;
164                                    }
165                                    Ok(_) => {}
166                                    Err(_) => {
167                                        debug!(
168                                            "OTLP peer {} sent nothing for {}s; closing",
169                                            remote_addr,
170                                            FIRST_BYTE_READ_TIMEOUT.as_secs()
171                                        );
172                                    }
173                                }
174
175                                app_state_for_close
176                                    .close_connection_on_server(server_id, connection_id)
177                                    .await;
178                                let _ = status_for_close
179                                    .send(format!("[INFO] OTLP connection {connection_id} closed"));
180                                let _ = status_for_close.send("__UPDATE_UI__".to_string());
181                            })
182                            .await;
183                    }
184                    Err(e) => {
185                        console_error!(status_tx, "Failed to accept OTLP connection: {}", e);
186                        break;
187                    }
188                }
189            }
190        });
191
192        task_registrar
193            .register_server_task(server_id, accept_handle)
194            .await;
195
196        Ok(local_addr)
197    }
198}
199
200/// Per-connection dependencies, cloned into each hyper service call.
201#[derive(Clone)]
202struct RequestContext {
203    llm_client: OllamaClient,
204    app_state: Arc<AppState>,
205    status_tx: mpsc::UnboundedSender<String>,
206    protocol: Arc<actions::OtlpProtocol>,
207    server_id: ServerId,
208    connection_id: ConnectionId,
209    exports: Arc<tokio::sync::Semaphore>,
210}
211
212async fn serve_connection<T>(io: TokioIo<T>, ctx: RequestContext)
213where
214    T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
215{
216    let activity = Arc::new(crate::server::accept_bounded::ConnectionActivity::new());
217    let activity_for_service = Arc::clone(&activity);
218
219    let children = grpc::OwnedExecutor::default();
220    let _children_guard = grpc::ConnectionTasks(children.clone());
221    let service = service_fn(move |req: Request<Incoming>| {
222        let ctx = ctx.clone();
223        let activity = Arc::clone(&activity_for_service);
224        async move {
225            let _busy = activity.busy();
226            let reply = if req
227                .headers()
228                .get(CONTENT_TYPE)
229                .and_then(|v| v.to_str().ok())
230                .is_some_and(|v| {
231                    v.split(';').next().is_some_and(|v| {
232                        v.trim().eq_ignore_ascii_case("application/grpc")
233                            || v.trim().eq_ignore_ascii_case("application/grpc+proto")
234                    })
235                }) {
236                grpc::dispatch(req, ctx).await
237            } else {
238                handle_request(req, ctx)
239                    .await
240                    .map(|body| body.map_err(|never| match never {}).boxed_unsync())
241            };
242            Ok::<_, Infallible>(reply)
243        }
244    });
245
246    let mut builder = auto::Builder::new(children.clone());
247    builder
248        .http2()
249        .max_concurrent_streams(16)
250        .initial_stream_window_size(65536)
251        .initial_connection_window_size(1048576)
252        .max_header_list_size(32768)
253        .max_frame_size(16384);
254    let conn = builder.serve_connection(io, service);
255    tokio::pin!(conn);
256    tokio::select! {
257        result = &mut conn => {
258            if let Err(err) = result {
259                debug!("OTLP connection ended: {:?}", err);
260            }
261        }
262        _ = crate::server::accept_bounded::watch_idle(
263            Arc::clone(&activity),
264            IDLE_BETWEEN_REQUESTS_TIMEOUT,
265        ) => {
266            debug!(
267                "OTLP connection idle for {}s; closing",
268                IDLE_BETWEEN_REQUESTS_TIMEOUT.as_secs()
269            );
270        }
271    }
272}
273
274fn response(
275    status: StatusCode,
276    content_type: &'static str,
277    body: Vec<u8>,
278) -> Response<Full<Bytes>> {
279    let mut response = Response::new(Full::new(Bytes::from(body)));
280    *response.status_mut() = status;
281    response
282        .headers_mut()
283        .insert(CONTENT_TYPE, HeaderValue::from_static(content_type));
284    response
285}
286
287/// A refusal before the encoding is known: plain text.
288fn text(status: StatusCode, message: &'static str) -> Response<Full<Bytes>> {
289    response(
290        status,
291        "text/plain; charset=utf-8",
292        format!("{message}\n").into_bytes(),
293    )
294}
295
296/// A failure in the request's encoding: a `google.rpc.Status`.
297fn failure(encoding: Encoding, status: StatusCode, message: &str) -> Response<Full<Bytes>> {
298    response(
299        status,
300        encoding.content_type(),
301        codec::status_body(encoding, status.as_u16(), message),
302    )
303}
304
305async fn handle_request(req: Request<Incoming>, ctx: RequestContext) -> Response<Full<Bytes>> {
306    let log = crate::logging::emit::Log::new(Some(&ctx.status_tx));
307    let (parts, body) = req.into_parts();
308    let path = parts.uri.path().to_string();
309    let header = |name| {
310        parts
311            .headers
312            .get(name)
313            .and_then(|v: &HeaderValue| v.to_str().ok())
314            .map(str::to_string)
315    };
316    let _ = ctx
317        .status_tx
318        .send(format!("[DEBUG] OTLP {} {}", parts.method, path));
319
320    let Some(signal) = Signal::from_path(&path) else {
321        return text(StatusCode::NOT_FOUND, "404 page not found");
322    };
323    if parts.method != Method::POST {
324        let mut r = text(
325            StatusCode::METHOD_NOT_ALLOWED,
326            "netget: OTLP exports are POSTed",
327        );
328        r.headers_mut()
329            .insert(ALLOW, HeaderValue::from_static("POST"));
330        return r;
331    }
332    let Some(encoding) = header(CONTENT_TYPE)
333        .as_deref()
334        .and_then(Encoding::from_content_type)
335    else {
336        log.warn(format!(
337            "OTLP {path}: decision=fail_closed_unsupported_media_type"
338        ));
339        return text(
340            StatusCode::UNSUPPORTED_MEDIA_TYPE,
341            "netget: OTLP/HTTP takes application/x-protobuf or application/json",
342        );
343    };
344    let compressed = match header(CONTENT_ENCODING)
345        .map(|e| e.trim().to_ascii_lowercase())
346        .as_deref()
347    {
348        None | Some("") | Some("identity") => false,
349        Some("gzip") => true,
350        Some(_) => {
351            log.warn(format!(
352                "OTLP {path}: decision=fail_closed_unsupported_encoding"
353            ));
354            return failure(
355                encoding,
356                StatusCode::UNSUPPORTED_MEDIA_TYPE,
357                "netget: the only Content-Encoding accepted is gzip",
358            );
359        }
360    };
361
362    // The cap applies to what arrives, and again to what it inflates to.
363    let raw = match http_body_util::Limited::new(body, MAX_BODY_BYTES)
364        .collect()
365        .await
366    {
367        Ok(collected) => collected.to_bytes(),
368        Err(e) => {
369            if e.is::<http_body_util::LengthLimitError>() {
370                log.warn(format!(
371                    "OTLP {path}: decision=fail_closed_too_large (limit {MAX_BODY_BYTES} bytes)"
372                ));
373                return failure(
374                    encoding,
375                    StatusCode::PAYLOAD_TOO_LARGE,
376                    "netget: the export is larger than this receiver accepts",
377                );
378            }
379            debug!("OTLP {path}: body read failed: {e}");
380            return failure(
381                encoding,
382                StatusCode::BAD_REQUEST,
383                "netget: the request body could not be read",
384            );
385        }
386    };
387    ctx.app_state
388        .update_connection_stats(
389            ctx.server_id,
390            ctx.connection_id,
391            Some(raw.len() as u64),
392            None,
393            Some(1),
394            None,
395        )
396        .await;
397    let body: Vec<u8> = if compressed {
398        match codec::gunzip_bounded(&raw, MAX_BODY_BYTES) {
399            Ok(inflated) => inflated,
400            Err(codec::BodyError::TooLarge) => {
401                log.warn(format!(
402                    "OTLP {path}: gzip body of {} bytes inflates past {MAX_BODY_BYTES} \
403                     decision=fail_closed_too_large",
404                    raw.len()
405                ));
406                return failure(
407                    encoding,
408                    StatusCode::PAYLOAD_TOO_LARGE,
409                    "netget: the export is larger than this receiver accepts",
410                );
411            }
412            Err(codec::BodyError::BadGzip) => {
413                log.warn(format!("OTLP {path}: decision=fail_closed_bad_gzip"));
414                return failure(
415                    encoding,
416                    StatusCode::BAD_REQUEST,
417                    "netget: the body is not valid gzip",
418                );
419            }
420        }
421    } else {
422        raw.to_vec()
423    };
424
425    let summary = match codec::summarize(signal, encoding, &body) {
426        Ok(s) => s,
427        Err(e) => {
428            log.warn(format!(
429                "OTLP {path}: {} payload did not decode decision=fail_closed_bad_payload",
430                encoding.as_str()
431            ));
432            debug!("OTLP {path}: decode error: {e}");
433            return failure(
434                encoding,
435                StatusCode::BAD_REQUEST,
436                match encoding {
437                    Encoding::Protobuf => "netget: the export is not a valid OTLP protobuf message",
438                    Encoding::Json => "netget: the export is not valid OTLP JSON",
439                },
440            );
441        }
442    };
443
444    let body_bytes = body.len();
445    drop(body);
446    respond_to_export(
447        ctx, signal, encoding, compressed, body_bytes, summary, "http",
448    )
449    .await
450}
451
452async fn respond_to_export(
453    ctx: RequestContext,
454    signal: Signal,
455    encoding: Encoding,
456    compressed: bool,
457    body_bytes: usize,
458    summary: codec::Summary,
459    transport: &str,
460) -> Response<Full<Bytes>> {
461    let log = crate::logging::emit::Log::new(Some(&ctx.status_tx));
462    let path = format!("/v1/{}", signal.as_str());
463    let mut data = summary.to_event(signal, encoding, compressed, body_bytes);
464    data["transport"] = serde_json::json!(transport);
465    data["answer_with"] = serde_json::json!(actions::answer_with(signal, summary.item_count));
466    let event = Event::new(&actions::OTLP_EXPORT_EVENT, data);
467    let who = summary.services.first().cloned().unwrap_or_default();
468    let result = call_llm(
469        &ctx.llm_client,
470        &ctx.app_state,
471        ctx.server_id,
472        Some(ctx.connection_id),
473        &event,
474        ctx.protocol.as_ref(),
475    )
476    .await;
477
478    let reply = match result {
479        Err(e) => {
480            let failure_kind = crate::utils::WireFailure::classify(&e);
481            log.error(format!(
482                "OTLP {path} from {who:?}: decision=fail_closed_llm_error category={:?} error={}",
483                failure_kind, e
484            ));
485            console_error!(ctx.status_tx, "OTLP LLM call failed: {}", e);
486            match failure_kind {
487                crate::utils::WireFailure::Overloaded => {
488                    let mut r = failure(
489                        encoding,
490                        StatusCode::SERVICE_UNAVAILABLE,
491                        failure_kind.prefixed_text(),
492                    );
493                    r.headers_mut()
494                        .insert(RETRY_AFTER, HeaderValue::from_static("5"));
495                    r
496                }
497                crate::utils::WireFailure::Unavailable => failure(
498                    encoding,
499                    StatusCode::INTERNAL_SERVER_ERROR,
500                    failure_kind.prefixed_text(),
501                ),
502            }
503        }
504        Ok(execution) => {
505            let failures = execution.failure_summary();
506            match execution
507                .protocol_results
508                .iter()
509                .find_map(Verdict::from_result)
510            {
511                Some(Verdict::Accept) => {
512                    log.info(format!(
513                        "OTLP {path} from {who:?}: decision=model_answer items={}",
514                        summary.item_count
515                    ));
516                    response(
517                        StatusCode::OK,
518                        encoding.content_type(),
519                        codec::export_response(signal, encoding, None),
520                    )
521                }
522                Some(Verdict::Partial {
523                    rejected,
524                    error_message,
525                }) => {
526                    let rejected = rejected.clamp(0, summary.item_count as i64);
527                    log.info(format!(
528                        "OTLP {path} from {who:?}: decision=model_answer partial rejected={} of {}",
529                        rejected, summary.item_count
530                    ));
531                    response(
532                        StatusCode::OK,
533                        encoding.content_type(),
534                        codec::export_response(signal, encoding, Some((rejected, &error_message))),
535                    )
536                }
537                Some(Verdict::Reject {
538                    status,
539                    message,
540                    retry_after_secs,
541                }) => {
542                    log.info(format!(
543                        "OTLP {path} from {who:?}: decision=model_reject status={status}"
544                    ));
545                    let status =
546                        StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
547                    let mut r = failure(encoding, status, &message);
548                    if let Some(secs) =
549                        retry_after_secs.filter(|_| codec::retryable(status.as_u16()))
550                    {
551                        if let Ok(value) = HeaderValue::from_str(&secs.to_string()) {
552                            r.headers_mut().insert(RETRY_AFTER, value);
553                        }
554                    }
555                    r
556                }
557                None => {
558                    log.warn(format!(
559                        "OTLP {path} from {who:?}: decision=model_silent ({}); refusing 500",
560                        failures.unwrap_or_else(|| "no verdict".to_string())
561                    ));
562                    failure(encoding, StatusCode::INTERNAL_SERVER_ERROR, NO_DECISION)
563                }
564            }
565        }
566    };
567    let sent = reply.body().size_hint().exact().unwrap_or(0);
568    ctx.app_state
569        .update_connection_stats(
570            ctx.server_id,
571            ctx.connection_id,
572            None,
573            Some(sent),
574            None,
575            Some(1),
576        )
577        .await;
578    reply
579}
netget/server/otlp/mod.rs

netget/server/otlp/
mod.rs
