From 9bbce61e706ef15e033404ce7f59754ea5ec4d44 Mon Sep 17 00:00:00 2001 From: Adil Hafeez Date: Mon, 9 Feb 2026 11:32:12 -0800 Subject: [PATCH] fix header injector code --- crates/brightstaff/src/handlers/llm.rs | 18 +++------------- .../src/handlers/pipeline_processor.rs | 21 ++++--------------- .../src/router/plano_orchestrator.rs | 21 +++---------------- 3 files changed, 10 insertions(+), 50 deletions(-) diff --git a/crates/brightstaff/src/handlers/llm.rs b/crates/brightstaff/src/handlers/llm.rs index 7278c9fd..b9fe0ba3 100644 --- a/crates/brightstaff/src/handlers/llm.rs +++ b/crates/brightstaff/src/handlers/llm.rs @@ -11,8 +11,9 @@ use http_body_util::combinators::BoxBody; use http_body_util::{BodyExt, Full}; use hyper::header::{self}; use hyper::{Request, Response, StatusCode}; +use opentelemetry::global; use opentelemetry::trace::get_active_span; -use opentelemetry::{global, propagation::Injector}; +use opentelemetry_http::HeaderInjector; use std::collections::HashMap; use std::sync::Arc; use tokio::sync::RwLock; @@ -35,19 +36,6 @@ fn full>(chunk: T) -> BoxBody { .boxed() } -/// Adapter to inject OpenTelemetry trace context into Hyper HeaderMap -struct HeaderMapInjector<'a>(&'a mut header::HeaderMap); - -impl<'a> Injector for HeaderMapInjector<'a> { - fn set(&mut self, key: &str, value: String) { - if let Ok(name) = header::HeaderName::from_bytes(key.as_bytes()) { - if let Ok(val) = header::HeaderValue::from_str(&value) { - self.0.insert(name, val); - } - } - } -} - pub async fn llm_chat( request: Request, router_service: Arc, @@ -368,7 +356,7 @@ async fn llm_chat_inner( // Inject current LLM span's trace context so upstream spans are children of plano(llm) global::get_text_map_propagator(|propagator| { let cx = tracing_opentelemetry::OpenTelemetrySpanExt::context(&tracing::Span::current()); - propagator.inject_context(&cx, &mut HeaderMapInjector(&mut request_headers)); + propagator.inject_context(&cx, &mut HeaderInjector(&mut request_headers)); }); // Capture start time right before sending request to upstream diff --git a/crates/brightstaff/src/handlers/pipeline_processor.rs b/crates/brightstaff/src/handlers/pipeline_processor.rs index a68776da..d9e87095 100644 --- a/crates/brightstaff/src/handlers/pipeline_processor.rs +++ b/crates/brightstaff/src/handlers/pipeline_processor.rs @@ -8,7 +8,7 @@ use hermesllm::apis::openai::Message; use hermesllm::{ProviderRequest, ProviderRequestType}; use hyper::header::HeaderMap; use opentelemetry::global; -use opentelemetry::propagation::Injector; +use opentelemetry_http::HeaderInjector; use tracing::{debug, info, instrument, warn}; use crate::handlers::jsonrpc::{ @@ -51,19 +51,6 @@ pub enum PipelineError { }, } -/// Adapter to inject OpenTelemetry trace context into Hyper HeaderMap -struct HeaderMapInjector<'a>(&'a mut HeaderMap); - -impl<'a> Injector for HeaderMapInjector<'a> { - fn set(&mut self, key: &str, value: String) { - if let Ok(header_name) = hyper::header::HeaderName::from_bytes(key.as_bytes()) { - if let Ok(header_value) = hyper::header::HeaderValue::from_str(&value) { - self.0.insert(header_name, header_value); - } - } - } -} - /// Service for processing agent pipelines pub struct PipelineProcessor { client: reqwest::Client, @@ -169,7 +156,7 @@ impl PipelineProcessor { global::get_text_map_propagator(|propagator| { let cx = tracing_opentelemetry::OpenTelemetrySpanExt::context(&tracing::Span::current()); - propagator.inject_context(&cx, &mut HeaderMapInjector(&mut headers)); + propagator.inject_context(&cx, &mut HeaderInjector(&mut headers)); }); headers.insert( @@ -546,7 +533,7 @@ impl PipelineProcessor { global::get_text_map_propagator(|propagator| { let cx = tracing_opentelemetry::OpenTelemetrySpanExt::context(&tracing::Span::current()); - propagator.inject_context(&cx, &mut HeaderMapInjector(&mut agent_headers)); + propagator.inject_context(&cx, &mut HeaderInjector(&mut agent_headers)); }); agent_headers.insert( @@ -645,7 +632,7 @@ impl PipelineProcessor { global::get_text_map_propagator(|propagator| { let cx = tracing_opentelemetry::OpenTelemetrySpanExt::context(&tracing::Span::current()); - propagator.inject_context(&cx, &mut HeaderMapInjector(&mut agent_headers)); + propagator.inject_context(&cx, &mut HeaderInjector(&mut agent_headers)); }); agent_headers.insert( diff --git a/crates/brightstaff/src/router/plano_orchestrator.rs b/crates/brightstaff/src/router/plano_orchestrator.rs index bca6a5a0..cf2688b9 100644 --- a/crates/brightstaff/src/router/plano_orchestrator.rs +++ b/crates/brightstaff/src/router/plano_orchestrator.rs @@ -6,7 +6,8 @@ use common::{ }; use hermesllm::apis::openai::{ChatCompletionsResponse, Message}; use hyper::header; -use opentelemetry::{global, propagation::Injector}; +use opentelemetry::global; +use opentelemetry_http::HeaderInjector; use thiserror::Error; use tracing::{debug, info, warn}; @@ -14,19 +15,6 @@ use crate::router::orchestrator_model_v1::{self}; use super::orchestrator_model::OrchestratorModel; -/// Adapter to inject OpenTelemetry trace context into Hyper HeaderMap -struct HeaderMapInjector<'a>(&'a mut header::HeaderMap); - -impl<'a> Injector for HeaderMapInjector<'a> { - fn set(&mut self, key: &str, value: String) { - if let Ok(header_name) = header::HeaderName::from_bytes(key.as_bytes()) { - if let Ok(header_value) = header::HeaderValue::from_str(&value) { - self.0.insert(header_name, header_value); - } - } - } -} - pub struct OrchestratorService { orchestrator_url: String, client: reqwest::Client, @@ -110,10 +98,7 @@ impl OrchestratorService { global::get_text_map_propagator(|propagator| { let cx = tracing_opentelemetry::OpenTelemetrySpanExt::context(&tracing::Span::current()); - propagator.inject_context( - &cx, - &mut HeaderMapInjector(&mut orchestration_request_headers), - ); + propagator.inject_context(&cx, &mut HeaderInjector(&mut orchestration_request_headers)); }); if let Some(request_id) = request_id {