ipx-directory traces, and ipx's trace follows its calls there (#213)
The exporter setup and the traceparent parser move from ipx into ipx_directory::otel, which ipx already depends on, so both binaries share them. The directory makes a span a request, a child of the caller's traceparent, and a trace a refresher round with a span a feed; ipx sends the current span's traceparent on every call to the directory and wraps each share in a span. The dashboard's Traces row and the prod-check skill cover all three services. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -17,8 +17,6 @@ futures-util.workspace = true
|
||||
jsonwebtoken.workspace = true
|
||||
librqbit.workspace = true
|
||||
opentelemetry.workspace = true
|
||||
opentelemetry-otlp.workspace = true
|
||||
opentelemetry_sdk.workspace = true
|
||||
opml.workspace = true
|
||||
percent-encoding.workspace = true
|
||||
quick-xml.workspace = true
|
||||
|
||||
@@ -58,6 +58,10 @@ pub fn configured(cfg: &config::Config) -> bool {
|
||||
async fn call<T: DeserializeOwned>(ctx: &Ctx, method: reqwest::Method, path: &str, body: Option<&[SnapshotEntry]>) -> Result<T, DirError> {
|
||||
let (base, token) = target(&ctx.cfg()).ok_or(DirError::NotConfigured)?;
|
||||
let mut req = ctx.client.request(method, format!("{base}{path}")).bearer_auth(token).timeout(std::time::Duration::from_secs(20));
|
||||
// The directory's spans join this request's trace (#213).
|
||||
if let Some(tp) = ipx_directory::otel::traceparent() {
|
||||
req = req.header("traceparent", tp);
|
||||
}
|
||||
if let Some(b) = body {
|
||||
req = req.json(b);
|
||||
}
|
||||
@@ -147,7 +151,8 @@ pub async fn run(ctx: std::sync::Arc<Ctx>, mut stop: tokio::sync::watch::Receive
|
||||
Ok(entries) => {
|
||||
let key = serde_json::to_vec(&(target(&cfg), &entries)).unwrap_or_default();
|
||||
if resend(last.as_ref(), &key) {
|
||||
let got = share(&ctx, &entries).await;
|
||||
let span = tracing::info_span!(target: "ipx::directory", "directory_share", feeds = entries.len());
|
||||
let got = tracing::Instrument::instrument(share(&ctx, &entries), span).await;
|
||||
if got.is_ok() {
|
||||
last = Some((key, std::time::Instant::now()));
|
||||
}
|
||||
|
||||
@@ -254,7 +254,7 @@ async fn main() -> Result<()> {
|
||||
// The daemon only: `ipx status` runs every half minute as the healthcheck, and a trace
|
||||
// for each would bury the ones worth looking at.
|
||||
let otel = match cli.command {
|
||||
Command::Daemon { .. } => otel_provider()?,
|
||||
Command::Daemon { .. } => ipx_directory::otel::provider("ipx")?,
|
||||
_ => None,
|
||||
};
|
||||
// Everything goes to stderr as before, and is mirrored into a ring the UI can read.
|
||||
@@ -424,23 +424,6 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
/// Traces over OTLP, to Tempo for one, when OTEL_EXPORTER_OTLP_ENDPOINT names a collector
|
||||
/// (`http://host:4318`: the exporter speaks OTLP over HTTP and adds `/v1/traces`). The exporter
|
||||
/// reads that and the other `OTEL_` variables itself.
|
||||
fn otel_provider() -> Result<Option<opentelemetry_sdk::trace::SdkTracerProvider>> {
|
||||
if std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").unwrap_or_default().is_empty() {
|
||||
return Ok(None);
|
||||
}
|
||||
let exporter = opentelemetry_otlp::SpanExporter::builder().with_http().build()?;
|
||||
let resource = opentelemetry_sdk::Resource::builder().with_service_name("ipx").build();
|
||||
Ok(Some(
|
||||
opentelemetry_sdk::trace::SdkTracerProvider::builder()
|
||||
.with_batch_exporter(exporter)
|
||||
.with_resource(resource)
|
||||
.build(),
|
||||
))
|
||||
}
|
||||
|
||||
/// Accounts. Passwords come in on stdin so they never reach a shell history or a `ps`
|
||||
/// listing.
|
||||
async fn user_cmd(ctx: &Arc<Ctx>, cmd: UserCmd) -> Result<()> {
|
||||
|
||||
@@ -1039,28 +1039,6 @@ mod tests {
|
||||
}
|
||||
|
||||
|
||||
#[test]
|
||||
fn a_requests_trace_is_the_pages_from_its_traceparent() {
|
||||
use opentelemetry::trace::TraceContextExt;
|
||||
let with = |v: &str| {
|
||||
let mut h = HeaderMap::new();
|
||||
h.insert("traceparent", v.parse().unwrap());
|
||||
remote_parent(&h, None)
|
||||
};
|
||||
let cx = with("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01").expect("parsed");
|
||||
assert_eq!(cx.span().span_context().trace_id().to_string(), "4bf92f3577b34da6a3ce929d0e0e4736");
|
||||
assert!(cx.span().span_context().is_remote());
|
||||
// Anything else from the browser starts a trace of its own.
|
||||
for bad in ["", "00-xyz-00f067aa0ba902b7-01", "01-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01",
|
||||
"00-00000000000000000000000000000000-00f067aa0ba902b7-01", "00-4bf92f3577b34da6a3ce929d0e0e4736"] {
|
||||
assert!(with(bad).is_none(), "{bad}");
|
||||
}
|
||||
assert!(remote_parent(&HeaderMap::new(), None).is_none());
|
||||
// From the query, where a beacon or Safari's player cannot send a header (#192).
|
||||
let q = remote_parent(&HeaderMap::new(), Some("x=1&traceparent=00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"));
|
||||
assert_eq!(q.unwrap().span().span_context().trace_id().to_string(), "4bf92f3577b34da6a3ce929d0e0e4736");
|
||||
assert!(remote_parent(&HeaderMap::new(), Some("traceparent=nonsense")).is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_page_reports_only_what_it_is_meant_to() {
|
||||
@@ -2231,30 +2209,6 @@ async fn client_event(user: crate::db::User, Json(e): Json<ClientEvent>) -> Stat
|
||||
StatusCode::NO_CONTENT
|
||||
}
|
||||
|
||||
/// The trace a request from the page belongs to, from its `traceparent` (#190), so the requests
|
||||
/// one action sets off are one trace. None for a missing or malformed one. In the query where no
|
||||
/// header can be set: a position saved by sendBeacon, and the playlist Safari's player fetches
|
||||
/// itself (#192).
|
||||
fn remote_parent(headers: &HeaderMap, query: Option<&str>) -> Option<opentelemetry::Context> {
|
||||
use opentelemetry::trace::{SpanContext, SpanId, TraceContextExt, TraceFlags, TraceId, TraceState};
|
||||
let v = match headers.get("traceparent") {
|
||||
Some(h) => h.to_str().ok()?,
|
||||
None => query?.split('&').find_map(|kv| kv.strip_prefix("traceparent="))?,
|
||||
};
|
||||
let mut parts = v.split('-');
|
||||
let (version, trace, span) = (parts.next()?, parts.next()?, parts.next()?);
|
||||
if version != "00" || trace.len() != 32 || span.len() != 16 {
|
||||
return None;
|
||||
}
|
||||
let sc = SpanContext::new(
|
||||
TraceId::from_hex(trace).ok()?,
|
||||
SpanId::from_hex(span).ok()?,
|
||||
TraceFlags::SAMPLED,
|
||||
true,
|
||||
TraceState::default(),
|
||||
);
|
||||
sc.is_valid().then(|| opentelemetry::Context::new().with_remote_span_context(sc))
|
||||
}
|
||||
|
||||
/// Names a request's trace by its route, `POST /api/entries/{feed_id}/{guid}/flags`, once routing
|
||||
/// has found it. Named by the path `access_log` sees, every item's GUID was a trace name of its
|
||||
@@ -2295,7 +2249,7 @@ async fn access_log(req: Request, next: Next) -> Response {
|
||||
http.response.status_code = tracing::field::Empty,
|
||||
)
|
||||
};
|
||||
if let Some(parent) = remote_parent(req.headers(), req.uri().query()) {
|
||||
if let Some(parent) = ipx_directory::otel::remote_parent(req.headers(), req.uri().query()) {
|
||||
use tracing_opentelemetry::OpenTelemetrySpanExt;
|
||||
let _ = span.set_parent(parent);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user