Apps/ImageApi
1
0 Fork 0
Code Issues 7 Pull Requests Releases Wiki Activity

Otel: dedupe LLM span error/status + traceparent helpers (#130 follow-up)

Browse Source 
Addresses the P2 refactoring follow-ups parked on issue #130:

- otel::with_traceparent(builder, &SpanContext) — one shared W3C
  traceparent injector. Replaces the 4-line inline header loop in
  llamacpp (3x) and openrouter (3x) and removes ollama's private
  inject_traceparent method (4 call sites now use the shared helper).
  Takes &SpanContext (not &dyn Span) because the Span trait is not dyn
  compatible in opentelemetry 0.31.

- otel::record_http_failure(span, status, message) and
  otel::set_http_status(span, status) — collapse the repeated
  'set http.status_code attribute + set_status(Status::error)'
  boilerplate across all three clients (~10 sites).

Net -45 lines. The remaining if-let-Some(span.as_mut()) guards (token
attributes, parse-failure status) stay inline rather than growing a
SpanHandle wrapper type for a pattern that's already readable.
This commit is contained in:
Cameron Cordes committed 2026-09-28 10:35:28 -04:00
1 parent 696d9fa8f3
commit 10ed84d7b4
4 files changed
+108 -153
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/ai/llamacpp.rs
+22 -44
View File
@@ -23,7 +23,7 @@
use anyhow::Context as _; use anyhow::Context as _;
use anyhow::{Result, anyhow, bail}; use anyhow::{Result, anyhow, bail};
use async_trait::async_trait; use async_trait::async_trait;
use opentelemetry::trace::{Span, Status, TraceContextExt}; use opentelemetry::trace::{Span, Status};
use opentelemetry::{Context, KeyValue}; use opentelemetry::{Context, KeyValue};
use reqwest::Client; use reqwest::Client;
use serde::Deserialize; use serde::Deserialize;
@@ -523,11 +523,7 @@ impl LlamaCppClient {
let _gpu = crate::ai::gpu::llm_lease().await; let _gpu = crate::ai::gpu::llm_lease().await;
let mut request = self.client.post(&url).json(&Value::Object(body)); let mut request = self.client.post(&url).json(&Value::Object(body));
if let Some(s) = span.as_mut() { if let Some(s) = span.as_mut() {
for (k, v) in crate::otel::traceparent_headers( request = crate::otel::with_traceparent(request, s.span_context());
&Context::new().with_remote_span_context(s.span_context().clone()),
) {
request = request.header(k, v);
}
} }
let resp = match request.send().await { let resp = match request.send().await {
Ok(r) => r, Ok(r) => r,
@@ -541,21 +537,15 @@ impl LlamaCppClient {
if !resp.status().is_success() { if !resp.status().is_success() {
let status = resp.status(); let status = resp.status();
if let Some(s) = span.as_mut() { crate::otel::record_http_failure(
s.set_attribute(KeyValue::new("http.status_code", status.as_u16() as i64)); span.as_deref_mut(),
s.set_status(Status::error(format!( status.as_u16(),
"llama-swap chat request failed: {status}" &format!("llama-swap chat request failed: {status}"),
))); );
}
let body = resp.text().await.unwrap_or_default(); let body = resp.text().await.unwrap_or_default();
bail!("llama-swap chat request failed: {} — {}", status, body); bail!("llama-swap chat request failed: {} — {}", status, body);
} }
if let Some(s) = span.as_mut() { crate::otel::set_http_status(span.as_deref_mut(), resp.status().as_u16());
s.set_attribute(KeyValue::new(
"http.status_code",
resp.status().as_u16() as i64,
));
}
let parsed: Value = resp.json().await.context("parsing chat response")?; let parsed: Value = resp.json().await.context("parsing chat response")?;
let choice = parsed let choice = parsed
@@ -700,11 +690,7 @@ impl LlmClient for LlamaCppClient {
// below so the lease spans the whole generation, not just the send. // below so the lease spans the whole generation, not just the send.
let gpu = crate::ai::gpu::llm_lease().await; let gpu = crate::ai::gpu::llm_lease().await;
let mut request = self.client.post(&url).json(&Value::Object(body)); let mut request = self.client.post(&url).json(&Value::Object(body));
for (k, v) in crate::otel::traceparent_headers( request = crate::otel::with_traceparent(request, span.span_context());
&Context::new().with_remote_span_context(span.span_context().clone()),
) {
request = request.header(k, v);
}
let resp = request let resp = request
.send() .send()
.await .await
@@ -712,17 +698,15 @@ impl LlmClient for LlamaCppClient {
if !resp.status().is_success() { if !resp.status().is_success() {
let status = resp.status(); let status = resp.status();
span.set_attribute(KeyValue::new("http.status_code", status.as_u16() as i64)); crate::otel::record_http_failure(
span.set_status(Status::error(format!( Some(&mut span),
"llama-swap stream request failed: {status}" status.as_u16(),
))); &format!("llama-swap stream request failed: {status}"),
);
let body = resp.text().await.unwrap_or_default(); let body = resp.text().await.unwrap_or_default();
bail!("llama-swap stream request failed: {} — {}", status, body); bail!("llama-swap stream request failed: {} — {}", status, body);
} }
span.set_attribute(KeyValue::new( crate::otel::set_http_status(Some(&mut span), resp.status().as_u16());
"http.status_code",
resp.status().as_u16() as i64,
));
let byte_stream = resp.bytes_stream(); let byte_stream = resp.bytes_stream();
let stream = async_stream::stream! { let stream = async_stream::stream! {
@@ -860,11 +844,7 @@ impl LlmClient for LlamaCppClient {
}); });
let mut request = self.client.post(&url).json(&body); let mut request = self.client.post(&url).json(&body);
for (k, v) in crate::otel::traceparent_headers( request = crate::otel::with_traceparent(request, span.span_context());
&Context::new().with_remote_span_context(span.span_context().clone()),
) {
request = request.header(k, v);
}
let resp = match request.send().await { let resp = match request.send().await {
Ok(r) => r, Ok(r) => r,
Err(e) => { Err(e) => {
@@ -875,17 +855,15 @@ impl LlmClient for LlamaCppClient {
if !resp.status().is_success() { if !resp.status().is_success() {
let status = resp.status(); let status = resp.status();
span.set_attribute(KeyValue::new("http.status_code", status.as_u16() as i64)); crate::otel::record_http_failure(
span.set_status(Status::error(format!( Some(&mut span),
"llama-swap embedding request failed: {status}" status.as_u16(),
))); &format!("llama-swap embedding request failed: {status}"),
);
let body = resp.text().await.unwrap_or_default(); let body = resp.text().await.unwrap_or_default();
bail!("llama-swap embedding request failed: {} — {}", status, body); bail!("llama-swap embedding request failed: {} — {}", status, body);
} }
span.set_attribute(KeyValue::new( crate::otel::set_http_status(Some(&mut span), resp.status().as_u16());
"http.status_code",
resp.status().as_u16() as i64,
));
#[derive(Deserialize)] #[derive(Deserialize)]
struct EmbedResponse { struct EmbedResponse {
src/ai/ollama.rs
+29 -64
View File
@@ -2,7 +2,7 @@ use anyhow::Context as _;
use anyhow::Result; use anyhow::Result;
use async_trait::async_trait; use async_trait::async_trait;
use chrono::NaiveDate; use chrono::NaiveDate;
use opentelemetry::trace::{Span, Status, TraceContextExt}; use opentelemetry::trace::{Span, Status};
use opentelemetry::{Context, KeyValue}; use opentelemetry::{Context, KeyValue};
use reqwest::Client; use reqwest::Client;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@@ -12,7 +12,6 @@ use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use crate::ai::llm_client::{LlmClient, LlmStreamEvent}; use crate::ai::llm_client::{LlmClient, LlmStreamEvent};
use crate::otel::traceparent_headers;
use futures::stream::{BoxStream, StreamExt}; use futures::stream::{BoxStream, StreamExt};
// Re-export shared types so existing `crate::ai::ollama::{...}` imports // Re-export shared types so existing `crate::ai::ollama::{...}` imports
@@ -392,7 +391,7 @@ impl OllamaClient {
.post(format!("{}/api/generate", url)) .post(format!("{}/api/generate", url))
.json(&request); .json(&request);
if let Some(s) = span.as_mut() { if let Some(s) = span.as_mut() {
request_builder = self.inject_traceparent(request_builder, s); request_builder = crate::otel::with_traceparent(request_builder, s.span_context());
} }
let response = match request_builder.send().await { let response = match request_builder.send().await {
Ok(r) => r, Ok(r) => r,
@@ -406,10 +405,11 @@ impl OllamaClient {
if !response.status().is_success() { if !response.status().is_success() {
let status = response.status(); let status = response.status();
if let Some(s) = span.as_mut() { crate::otel::record_http_failure(
s.set_attribute(KeyValue::new("http.status_code", status.as_u16() as i64)); span.as_deref_mut(),
s.set_status(Status::error(format!("Ollama request failed: {status}"))); status.as_u16(),
} &format!("Ollama request failed: {status}"),
);
let error_body = response.text().await.unwrap_or_default(); let error_body = response.text().await.unwrap_or_default();
return Err(anyhow::anyhow!( return Err(anyhow::anyhow!(
"Ollama request failed: {} - {}", "Ollama request failed: {} - {}",
@@ -417,12 +417,7 @@ impl OllamaClient {
error_body error_body
)); ));
} }
if let Some(s) = span.as_mut() { crate::otel::set_http_status(span.as_deref_mut(), response.status().as_u16());
s.set_attribute(KeyValue::new(
"http.status_code",
response.status().as_u16() as i64,
));
}
let result: OllamaResponse = match response.json().await { let result: OllamaResponse = match response.json().await {
Ok(r) => r, Ok(r) => r,
@@ -451,22 +446,6 @@ impl OllamaClient {
Ok(result.response) Ok(result.response)
} }
/// Attach the W3C traceparent header to an outgoing Ollama request so
/// a server that understands it can parent its own spans onto ours.
fn inject_traceparent(
&self,
builder: reqwest::RequestBuilder,
span: &opentelemetry::global::BoxedSpan,
) -> reqwest::RequestBuilder {
let mut builder = builder;
for (k, v) in traceparent_headers(
&Context::new().with_remote_span_context(span.span_context().clone()),
) {
builder = builder.header(k, v);
}
builder
}
/// Start the `llm.request` span for an outgoing Ollama call. /// Start the `llm.request` span for an outgoing Ollama call.
fn start_request_span( fn start_request_span(
&self, &self,
@@ -899,7 +878,7 @@ Analyze the image and use specific details from both the visual content and the
let response = { let response = {
let mut rb = self.client.post(&url).json(&request_body); let mut rb = self.client.post(&url).json(&request_body);
rb = self.inject_traceparent(rb, &span); rb = crate::otel::with_traceparent(rb, span.span_context());
match rb.send().await { match rb.send().await {
Ok(r) => r, Ok(r) => r,
Err(e) => { Err(e) => {
@@ -912,10 +891,11 @@ Analyze the image and use specific details from both the visual content and the
if !response.status().is_success() { if !response.status().is_success() {
let status = response.status(); let status = response.status();
span.set_attribute(KeyValue::new("http.status_code", status.as_u16() as i64)); crate::otel::record_http_failure(
span.set_status(Status::error(format!( Some(&mut span),
"Ollama stream request failed: {status}" status.as_u16(),
))); &format!("Ollama stream request failed: {status}"),
);
let body = response.text().await.unwrap_or_default(); let body = response.text().await.unwrap_or_default();
anyhow::bail!( anyhow::bail!(
"Ollama stream request failed with status {}: {}", "Ollama stream request failed with status {}: {}",
@@ -923,10 +903,7 @@ Analyze the image and use specific details from both the visual content and the
body body
); );
} }
span.set_attribute(KeyValue::new( crate::otel::set_http_status(Some(&mut span), response.status().as_u16());
"http.status_code",
response.status().as_u16() as i64,
));
// Ollama streams NDJSON: each line is a full `OllamaStreamChunk`. // Ollama streams NDJSON: each line is a full `OllamaStreamChunk`.
// We buffer partial lines across chunks from the byte stream. // We buffer partial lines across chunks from the byte stream.
@@ -1064,7 +1041,7 @@ Analyze the image and use specific details from both the visual content and the
let response = { let response = {
let mut rb = self.client.post(&url).json(&request_body); let mut rb = self.client.post(&url).json(&request_body);
if let Some(s) = span.as_mut() { if let Some(s) = span.as_mut() {
rb = self.inject_traceparent(rb, s); rb = crate::otel::with_traceparent(rb, s.span_context());
} }
match rb.send().await { match rb.send().await {
Ok(r) => r, Ok(r) => r,
@@ -1080,12 +1057,11 @@ Analyze the image and use specific details from both the visual content and the
if !response.status().is_success() { if !response.status().is_success() {
let status = response.status(); let status = response.status();
if let Some(s) = span.as_mut() { crate::otel::record_http_failure(
s.set_attribute(KeyValue::new("http.status_code", status.as_u16() as i64)); span.as_deref_mut(),
s.set_status(Status::error(format!( status.as_u16(),
"Ollama chat request failed: {status}" &format!("Ollama chat request failed: {status}"),
))); );
}
let body = response.text().await.unwrap_or_default(); let body = response.text().await.unwrap_or_default();
// warn, not error — the outer `chat_with_tools` may recover via // warn, not error — the outer `chat_with_tools` may recover via
// the fallback server. When both fail, the outer layer emits the // the fallback server. When both fail, the outer layer emits the
@@ -1102,12 +1078,7 @@ Analyze the image and use specific details from both the visual content and the
body body
); );
} }
if let Some(s) = span.as_mut() { crate::otel::set_http_status(span.as_deref_mut(), response.status().as_u16());
s.set_attribute(KeyValue::new(
"http.status_code",
response.status().as_u16() as i64,
));
}
let chat_response: OllamaChatResponse = match response.json().await { let chat_response: OllamaChatResponse = match response.json().await {
Ok(r) => r, Ok(r) => r,
@@ -1268,7 +1239,7 @@ Analyze the image and use specific details from both the visual content and the
.post(format!("{}/api/embed", url)) .post(format!("{}/api/embed", url))
.json(&request); .json(&request);
if let Some(s) = span.as_mut() { if let Some(s) = span.as_mut() {
rb = self.inject_traceparent(rb, s); rb = crate::otel::with_traceparent(rb, s.span_context());
} }
match rb.send().await { match rb.send().await {
Ok(r) => r, Ok(r) => r,
@@ -1283,12 +1254,11 @@ Analyze the image and use specific details from both the visual content and the
if !response.status().is_success() { if !response.status().is_success() {
let status = response.status(); let status = response.status();
if let Some(s) = span.as_mut() { crate::otel::record_http_failure(
s.set_attribute(KeyValue::new("http.status_code", status.as_u16() as i64)); span.as_deref_mut(),
s.set_status(Status::error(format!( status.as_u16(),
"Ollama batch embedding request failed: {status}" &format!("Ollama batch embedding request failed: {status}"),
))); );
}
let error_body = response.text().await.unwrap_or_default(); let error_body = response.text().await.unwrap_or_default();
return Err(anyhow::anyhow!( return Err(anyhow::anyhow!(
"Ollama batch embedding request failed: {} - {}", "Ollama batch embedding request failed: {} - {}",
@@ -1296,12 +1266,7 @@ Analyze the image and use specific details from both the visual content and the
error_body error_body
)); ));
} }
if let Some(s) = span.as_mut() { crate::otel::set_http_status(span.as_deref_mut(), response.status().as_u16());
s.set_attribute(KeyValue::new(
"http.status_code",
response.status().as_u16() as i64,
));
}
let result: OllamaEmbedResponse = match response.json().await { let result: OllamaEmbedResponse = match response.json().await {
Ok(r) => r, Ok(r) => r,
src/ai/openrouter.rs
+22 -44
View File
@@ -5,7 +5,7 @@
use anyhow::Context as _; use anyhow::Context as _;
use anyhow::{Result, anyhow, bail}; use anyhow::{Result, anyhow, bail};
use async_trait::async_trait; use async_trait::async_trait;
use opentelemetry::trace::{Span, Status, TraceContextExt}; use opentelemetry::trace::{Span, Status};
use opentelemetry::{Context, KeyValue}; use opentelemetry::{Context, KeyValue};
use reqwest::Client; use reqwest::Client;
use serde::Deserialize; use serde::Deserialize;
@@ -397,11 +397,7 @@ impl LlmClient for OpenRouterClient {
let mut request = self let mut request = self
.authed(self.client.post(&url)) .authed(self.client.post(&url))
.json(&Value::Object(body)); .json(&Value::Object(body));
for (k, v) in crate::otel::traceparent_headers( request = crate::otel::with_traceparent(request, span.span_context());
&Context::new().with_remote_span_context(span.span_context().clone()),
) {
request = request.header(k, v);
}
let resp = request let resp = request
.send() .send()
.await .await
@@ -409,17 +405,15 @@ impl LlmClient for OpenRouterClient {
if !resp.status().is_success() { if !resp.status().is_success() {
let status = resp.status(); let status = resp.status();
span.set_attribute(KeyValue::new("http.status_code", status.as_u16() as i64)); crate::otel::record_http_failure(
span.set_status(Status::error(format!( Some(&mut span),
"OpenRouter stream request failed: {status}" status.as_u16(),
))); &format!("OpenRouter stream request failed: {status}"),
);
let body = resp.text().await.unwrap_or_default(); let body = resp.text().await.unwrap_or_default();
bail!("OpenRouter stream request failed: {} — {}", status, body); bail!("OpenRouter stream request failed: {} — {}", status, body);
} }
span.set_attribute(KeyValue::new( crate::otel::set_http_status(Some(&mut span), resp.status().as_u16());
"http.status_code",
resp.status().as_u16() as i64,
));
// OpenAI-compat SSE stream. Each event is `data: <json>\n\n`, with // OpenAI-compat SSE stream. Each event is `data: <json>\n\n`, with
// `data: [DONE]` signalling completion. Tool calls arrive as // `data: [DONE]` signalling completion. Tool calls arrive as
@@ -612,11 +606,7 @@ impl LlmClient for OpenRouterClient {
}); });
let mut request = self.authed(self.client.post(&url)).json(&body); let mut request = self.authed(self.client.post(&url)).json(&body);
for (k, v) in crate::otel::traceparent_headers( request = crate::otel::with_traceparent(request, span.span_context());
&Context::new().with_remote_span_context(span.span_context().clone()),
) {
request = request.header(k, v);
}
let resp = request let resp = request
.send() .send()
.await .await
@@ -627,17 +617,15 @@ impl LlmClient for OpenRouterClient {
if !resp.status().is_success() { if !resp.status().is_success() {
let status = resp.status(); let status = resp.status();
span.set_attribute(KeyValue::new("http.status_code", status.as_u16() as i64)); crate::otel::record_http_failure(
span.set_status(Status::error(format!( Some(&mut span),
"OpenRouter embedding request failed: {status}" status.as_u16(),
))); &format!("OpenRouter embedding request failed: {status}"),
);
let body = resp.text().await.unwrap_or_default(); let body = resp.text().await.unwrap_or_default();
bail!("OpenRouter embedding request failed: {} — {}", status, body); bail!("OpenRouter embedding request failed: {} — {}", status, body);
} }
span.set_attribute(KeyValue::new( crate::otel::set_http_status(Some(&mut span), resp.status().as_u16());
"http.status_code",
resp.status().as_u16() as i64,
));
#[derive(Deserialize)] #[derive(Deserialize)]
struct EmbedResponse { struct EmbedResponse {
@@ -764,11 +752,7 @@ impl OpenRouterClient {
.authed(self.client.post(&url)) .authed(self.client.post(&url))
.json(&Value::Object(body)); .json(&Value::Object(body));
if let Some(s) = span.as_mut() { if let Some(s) = span.as_mut() {
for (k, v) in crate::otel::traceparent_headers( request = crate::otel::with_traceparent(request, s.span_context());
&Context::new().with_remote_span_context(s.span_context().clone()),
) {
request = request.header(k, v);
}
} }
let resp = match request.send().await { let resp = match request.send().await {
Ok(r) => r, Ok(r) => r,
@@ -782,21 +766,15 @@ impl OpenRouterClient {
if !resp.status().is_success() { if !resp.status().is_success() {
let status = resp.status(); let status = resp.status();
if let Some(s) = span.as_mut() { crate::otel::record_http_failure(
s.set_attribute(KeyValue::new("http.status_code", status.as_u16() as i64)); span.as_deref_mut(),
s.set_status(Status::error(format!( status.as_u16(),
"OpenRouter chat request failed: {status}" &format!("OpenRouter chat request failed: {status}"),
))); );
}
let body = resp.text().await.unwrap_or_default(); let body = resp.text().await.unwrap_or_default();
bail!("OpenRouter chat request failed: {} — {}", status, body); bail!("OpenRouter chat request failed: {} — {}", status, body);
} }
if let Some(s) = span.as_mut() { crate::otel::set_http_status(span.as_deref_mut(), resp.status().as_u16());
s.set_attribute(KeyValue::new(
"http.status_code",
resp.status().as_u16() as i64,
));
}
let parsed: Value = match resp.json().await { let parsed: Value = match resp.json().await {
Ok(v) => v, Ok(v) => v,
src/otel.rs
+35 -1
View File
@@ -2,7 +2,9 @@ use actix_web::HttpRequest;
use actix_web::http::header::HeaderMap; use actix_web::http::header::HeaderMap;
use opentelemetry::global::{BoxedSpan, BoxedTracer}; use opentelemetry::global::{BoxedSpan, BoxedTracer};
use opentelemetry::propagation::TextMapPropagator; use opentelemetry::propagation::TextMapPropagator;
use opentelemetry::trace::{Span, SpanBuilder, SpanKind, Status, TraceContextExt, Tracer}; use opentelemetry::trace::{
Span, SpanBuilder, SpanContext, SpanKind, Status, TraceContextExt, Tracer,
};
use opentelemetry::{Context, KeyValue, global}; use opentelemetry::{Context, KeyValue, global};
use opentelemetry_appender_log::OpenTelemetryLogBridge; use opentelemetry_appender_log::OpenTelemetryLogBridge;
use opentelemetry_otlp::WithExportConfig; use opentelemetry_otlp::WithExportConfig;
@@ -117,6 +119,38 @@ pub fn traceparent_headers(context: &Context) -> Vec<(String, String)> {
headers.into_iter().filter(|(_, v)| !v.is_empty()).collect() headers.into_iter().filter(|(_, v)| !v.is_empty()).collect()
} }
/// Attach the W3C traceparent header to an outgoing request so a downstream
/// service parents its own spans onto this trace. No-op when the span context
/// is not valid — e.g. debug builds, where the global tracer is a no-op and an
/// all-zero traceparent would produce garbage parent spans.
pub fn with_traceparent(
builder: reqwest::RequestBuilder,
span_context: &SpanContext,
) -> reqwest::RequestBuilder {
let mut b = builder;
for (k, v) in
traceparent_headers(&Context::new().with_remote_span_context(span_context.clone()))
{
b = b.header(k, v);
}
b
}
/// Record the HTTP status code on the span (success or failure path).
pub fn set_http_status(span: Option<&mut BoxedSpan>, status: u16) {
if let Some(s) = span {
s.set_attribute(KeyValue::new("http.status_code", status as i64));
}
}
/// Record an HTTP failure on the span: status code plus an error status.
pub fn record_http_failure(span: Option<&mut BoxedSpan>, status: u16, message: &str) {
if let Some(s) = span {
s.set_attribute(KeyValue::new("http.status_code", status as i64));
s.set_status(Status::error(message.to_string()));
}
}
pub fn trace_db_call<F, O>( pub fn trace_db_call<F, O>(
context: &Context, context: &Context,
query_type: &str, query_type: &str,