rapace_tracing/
lib.rs

1#![allow(clippy::type_complexity)]
2//! Tracing subscriber that forwards spans/events over rapace RPC.
3//!
4//! This crate enables plugins to use `tracing` normally while having all
5//! spans and events collected in the host process via rapace RPC.
6//!
7//! # Architecture
8//!
9//! ```text
10//! ┌─────────────────────────────────────────────────────────────────────────┐
11//! │                             PLUGIN PROCESS                              │
12//! │                                                                         │
13//! │   tracing::info!("hello") ──► RapaceTracingLayer ──► TracingSinkClient ─┤
14//! │                                      ▲                                  │
15//! │                                      │                                  │
16//! │                          TracingConfigServer ◄──────────────────────────┤
17//! │                          (applies host's filter)                        │
18//! └────────────────────────────────────────────────────────────────────────┬┘
19//!                                                                          │
20//!                              rapace transport (TCP/Unix/SHM)             │
21//!                                                                          │
22//! ┌────────────────────────────────────────────────────────────────────────┴┐
23//! │                              HOST PROCESS                               │
24//! │                                                                         │
25//! │   TracingSinkServer ──► HostTracingSink ──► tracing_subscriber / logs  │
26//! │                                                                         │
27//! │   TracingConfigClient ──► pushes filter changes to plugin              │
28//! │                                                                         │
29//! └─────────────────────────────────────────────────────────────────────────┘
30//! ```
31//!
32//! # Filter Flow
33//!
34//! The host is the single source of truth for log filtering:
35//! 1. Host decides what log levels/targets are enabled
36//! 2. Host pushes filter config to plugin via `TracingConfig::set_filter`
37//! 3. Plugin applies the filter locally (no spam over RPC)
38//! 4. When host changes filters dynamically, it pushes the update
39//!
40//! # Example
41//!
42//! ```ignore
43//! // Plugin side: install the layer
44//! let layer = RapaceTracingLayer::new(sink_client);
45//! tracing_subscriber::registry().with(layer).init();
46//!
47//! // Now all tracing calls are forwarded to the host
48//! tracing::info!("hello from plugin");
49//! ```
50
51use std::collections::HashMap;
52use std::pin::Pin;
53use std::sync::Arc;
54use std::sync::atomic::{AtomicU64, Ordering};
55
56use parking_lot::Mutex;
57use rapace::{Frame, RpcError, RpcSession, Transport};
58use tracing::span::{Attributes, Record};
59use tracing::{Event, Id, Subscriber};
60use tracing_subscriber::Layer;
61use tracing_subscriber::layer::Context;
62use tracing_subscriber::registry::LookupSpan;
63
64// ============================================================================
65// Facet Types (transport-agnostic)
66// ============================================================================
67
68/// A field value captured from tracing.
69#[derive(Debug, Clone, facet::Facet)]
70pub struct Field {
71    /// Field name
72    pub name: String,
73    /// Field value (stringified for v1)
74    pub value: String,
75}
76
77/// Metadata about a span.
78#[derive(Debug, Clone, facet::Facet)]
79pub struct SpanMeta {
80    /// Span name
81    pub name: String,
82    /// Target (module path)
83    pub target: String,
84    /// Level as string ("TRACE", "DEBUG", "INFO", "WARN", "ERROR")
85    pub level: String,
86    /// Source file, if available
87    pub file: Option<String>,
88    /// Line number, if available
89    pub line: Option<u32>,
90    /// Fields recorded at span creation
91    pub fields: Vec<Field>,
92}
93
94/// Metadata about an event.
95#[derive(Debug, Clone, facet::Facet)]
96pub struct EventMeta {
97    /// Event message (from the `message` field if present)
98    pub message: String,
99    /// Target (module path)
100    pub target: String,
101    /// Level as string
102    pub level: String,
103    /// Source file, if available
104    pub file: Option<String>,
105    /// Line number, if available
106    pub line: Option<u32>,
107    /// All fields including message
108    pub fields: Vec<Field>,
109    /// Parent span ID if inside a span
110    pub parent_span_id: Option<u64>,
111}
112
113// ============================================================================
114// TracingSink Service (plugin calls host)
115// ============================================================================
116
117/// Service for receiving tracing data from plugins.
118///
119/// The host implements this, the plugin calls it via RPC.
120#[allow(async_fn_in_trait)]
121#[rapace::service]
122pub trait TracingSink {
123    /// Called when a new span is created.
124    /// Returns a span ID that the plugin should use for subsequent calls.
125    async fn new_span(&self, span: crate::SpanMeta) -> u64;
126
127    /// Called when fields are recorded on an existing span.
128    async fn record(&self, span_id: u64, fields: Vec<crate::Field>);
129
130    /// Called when an event is emitted.
131    async fn event(&self, event: crate::EventMeta);
132
133    /// Called when a span is entered.
134    async fn enter(&self, span_id: u64);
135
136    /// Called when a span is exited.
137    async fn exit(&self, span_id: u64);
138
139    /// Called when a span is dropped/closed.
140    async fn drop_span(&self, span_id: u64);
141}
142
143// ============================================================================
144// TracingConfig Service (host calls plugin)
145// ============================================================================
146
147/// Service for configuring tracing in plugins.
148///
149/// The plugin implements this, the host calls it via RPC to push filter config.
150/// This allows the host to be the single source of truth for log filtering.
151#[allow(async_fn_in_trait)]
152#[rapace::service]
153pub trait TracingConfig {
154    /// Set the tracing filter.
155    ///
156    /// The filter string uses the same format as RUST_LOG (e.g., "info,mymodule=debug").
157    /// The plugin should apply this filter to all subsequent tracing calls.
158    async fn set_filter(&self, filter: String);
159}
160
161// ============================================================================
162// Plugin Side: Shared Filter State
163// ============================================================================
164
165use tracing_subscriber::EnvFilter;
166
167/// Shared filter state that can be updated by the host.
168///
169/// This is wrapped in Arc and shared between `RapaceTracingLayer` and `TracingConfigImpl`.
170#[derive(Clone)]
171pub struct SharedFilter {
172    inner: Arc<parking_lot::RwLock<EnvFilter>>,
173}
174
175impl SharedFilter {
176    /// Create a new shared filter with default (allow all) settings.
177    pub fn new() -> Self {
178        // Default to allowing everything - host will push the real filter
179        // Note: EnvFilter::builder().parse("") returns a filter that blocks everything,
180        // so we explicitly use "trace" to allow all levels.
181        let filter = EnvFilter::new("trace");
182        Self {
183            inner: Arc::new(parking_lot::RwLock::new(filter)),
184        }
185    }
186
187    /// Update the filter from a filter string (RUST_LOG format).
188    pub fn set_filter(&self, filter_str: &str) {
189        match EnvFilter::builder().parse(filter_str) {
190            Ok(filter) => {
191                *self.inner.write() = filter;
192            }
193            Err(e) => {
194                // Log the error but don't crash - keep existing filter
195                eprintln!("rapace-tracing: invalid filter '{}': {}", filter_str, e);
196            }
197        }
198    }
199
200    /// Check if a given level is enabled (max level check).
201    ///
202    /// This is a quick check based on the filter's max level hint.
203    /// Target-specific filtering happens through the Layer's `enabled` method.
204    pub fn max_level_enabled(&self, level: tracing::level_filters::LevelFilter) -> bool {
205        let filter = self.inner.read();
206        if let Some(max) = filter.max_level_hint() {
207            level <= max
208        } else {
209            true
210        }
211    }
212
213    /// Get the current max level hint.
214    pub fn max_level_hint(&self) -> Option<tracing::level_filters::LevelFilter> {
215        self.inner.read().max_level_hint()
216    }
217}
218
219impl Default for SharedFilter {
220    fn default() -> Self {
221        Self::new()
222    }
223}
224
225// ============================================================================
226// Plugin Side: TracingConfig Implementation
227// ============================================================================
228
229/// Plugin-side implementation of TracingConfig.
230///
231/// Host calls this to push filter updates to the plugin.
232#[derive(Clone)]
233pub struct TracingConfigImpl {
234    filter: SharedFilter,
235}
236
237impl TracingConfigImpl {
238    /// Create a new TracingConfig implementation with the given shared filter.
239    pub fn new(filter: SharedFilter) -> Self {
240        Self { filter }
241    }
242}
243
244impl TracingConfig for TracingConfigImpl {
245    async fn set_filter(&self, filter: String) {
246        self.filter.set_filter(&filter);
247    }
248}
249
250/// Create a dispatcher for TracingConfig service (plugin side).
251pub fn create_tracing_config_dispatcher(
252    config: TracingConfigImpl,
253) -> impl Fn(
254    u32,
255    u32,
256    Vec<u8>,
257) -> Pin<Box<dyn std::future::Future<Output = Result<Frame, RpcError>> + Send>>
258+ Send
259+ Sync
260+ 'static {
261    move |_channel_id, method_id, payload| {
262        let config = config.clone();
263        Box::pin(async move {
264            let server = TracingConfigServer::new(config);
265            server.dispatch(method_id, &payload).await
266        })
267    }
268}
269
270// ============================================================================
271// Plugin Side: RapaceTracingLayer
272// ============================================================================
273
274/// A tracing Layer that forwards spans/events to a TracingSink via RPC.
275///
276/// Install this layer in the plugin's tracing_subscriber registry to have
277/// all tracing data forwarded to the host process.
278///
279/// The layer uses a `SharedFilter` to apply host-controlled filtering locally,
280/// avoiding unnecessary RPC calls for filtered events.
281pub struct RapaceTracingLayer<T: Transport + Send + Sync + 'static> {
282    session: Arc<RpcSession<T>>,
283    /// Maps local tracing span IDs to our u64 IDs used in RPC
284    span_ids: Mutex<HashMap<u64, u64>>,
285    /// Counter for generating local span IDs
286    next_span_id: AtomicU64,
287    /// Runtime handle for spawning async tasks
288    rt: tokio::runtime::Handle,
289    /// Shared filter state (updated by host via TracingConfig)
290    filter: SharedFilter,
291}
292
293impl<T: Transport + Send + Sync + 'static> RapaceTracingLayer<T> {
294    /// Create a new layer that forwards to the given RPC session.
295    ///
296    /// The session should be connected to a host that implements TracingSink.
297    /// Use the returned `SharedFilter` to create a `TracingConfigImpl` for the
298    /// host to push filter updates.
299    pub fn new(session: Arc<RpcSession<T>>, rt: tokio::runtime::Handle) -> (Self, SharedFilter) {
300        let filter = SharedFilter::new();
301        let layer = Self {
302            session,
303            span_ids: Mutex::new(HashMap::new()),
304            next_span_id: AtomicU64::new(1),
305            rt,
306            filter: filter.clone(),
307        };
308        (layer, filter)
309    }
310
311    /// Create a new layer with an existing shared filter.
312    pub fn with_filter(
313        session: Arc<RpcSession<T>>,
314        rt: tokio::runtime::Handle,
315        filter: SharedFilter,
316    ) -> Self {
317        Self {
318            session,
319            span_ids: Mutex::new(HashMap::new()),
320            next_span_id: AtomicU64::new(1),
321            rt,
322            filter,
323        }
324    }
325
326    /// Call TracingSink.new_span via RPC (fire-and-forget from sync context).
327    fn call_new_span(&self, meta: SpanMeta) -> u64 {
328        let client = TracingSinkClient::new(self.session.clone());
329        let local_id = self.next_span_id.fetch_add(1, Ordering::Relaxed);
330
331        self.rt.spawn(async move {
332            let _ = client.new_span(meta).await;
333        });
334
335        local_id
336    }
337
338    /// Call TracingSink.record via RPC.
339    fn call_record(&self, span_id: u64, fields: Vec<Field>) {
340        let client = TracingSinkClient::new(self.session.clone());
341        self.rt.spawn(async move {
342            let _ = client.record(span_id, fields).await;
343        });
344    }
345
346    /// Call TracingSink.event via RPC.
347    fn call_event(&self, event: EventMeta) {
348        let client = TracingSinkClient::new(self.session.clone());
349        self.rt.spawn(async move {
350            let _ = client.event(event).await;
351        });
352    }
353
354    /// Call TracingSink.enter via RPC.
355    fn call_enter(&self, span_id: u64) {
356        let client = TracingSinkClient::new(self.session.clone());
357        self.rt.spawn(async move {
358            let _ = client.enter(span_id).await;
359        });
360    }
361
362    /// Call TracingSink.exit via RPC.
363    fn call_exit(&self, span_id: u64) {
364        let client = TracingSinkClient::new(self.session.clone());
365        self.rt.spawn(async move {
366            let _ = client.exit(span_id).await;
367        });
368    }
369
370    /// Call TracingSink.drop_span via RPC.
371    fn call_drop_span(&self, span_id: u64) {
372        let client = TracingSinkClient::new(self.session.clone());
373        self.rt.spawn(async move {
374            let _ = client.drop_span(span_id).await;
375        });
376    }
377}
378
379impl<S, T> Layer<S> for RapaceTracingLayer<T>
380where
381    S: Subscriber + for<'a> LookupSpan<'a>,
382    T: Transport + Send + Sync + 'static,
383{
384    fn enabled(&self, metadata: &tracing::Metadata<'_>, _ctx: Context<'_, S>) -> bool {
385        // Avoid infinite recursion: don't forward events about rapace_tracing itself
386        // or rapace_core (which handles the RPC session internals)
387        let target = metadata.target();
388        if target.starts_with("rapace_tracing")
389            || target.starts_with("rapace_core")
390            || target.starts_with("rapace_transport_shm")
391        {
392            return false;
393        }
394
395        // Check against the host-controlled filter
396        let level = match *metadata.level() {
397            tracing::Level::ERROR => tracing::level_filters::LevelFilter::ERROR,
398            tracing::Level::WARN => tracing::level_filters::LevelFilter::WARN,
399            tracing::Level::INFO => tracing::level_filters::LevelFilter::INFO,
400            tracing::Level::DEBUG => tracing::level_filters::LevelFilter::DEBUG,
401            tracing::Level::TRACE => tracing::level_filters::LevelFilter::TRACE,
402        };
403        self.filter.max_level_enabled(level)
404    }
405
406    fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, _ctx: Context<'_, S>) {
407        let meta = attrs.metadata();
408
409        // Collect fields
410        let mut visitor = FieldVisitor::new();
411        attrs.record(&mut visitor);
412
413        let span_meta = SpanMeta {
414            name: meta.name().to_string(),
415            target: meta.target().to_string(),
416            level: meta.level().to_string(),
417            file: meta.file().map(|s| s.to_string()),
418            line: meta.line(),
419            fields: visitor.fields,
420        };
421
422        let local_id = self.call_new_span(span_meta);
423
424        // Store mapping from tracing's Id to our local ID
425        self.span_ids.lock().insert(id.into_u64(), local_id);
426    }
427
428    fn on_record(&self, id: &Id, values: &Record<'_>, _ctx: Context<'_, S>) {
429        let span_id = match self.span_ids.lock().get(&id.into_u64()) {
430            Some(&id) => id,
431            None => return,
432        };
433
434        let mut visitor = FieldVisitor::new();
435        values.record(&mut visitor);
436
437        if !visitor.fields.is_empty() {
438            self.call_record(span_id, visitor.fields);
439        }
440    }
441
442    fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) {
443        let meta = event.metadata();
444
445        // Collect fields
446        let mut visitor = FieldVisitor::new();
447        event.record(&mut visitor);
448
449        // Extract message from fields
450        let message = visitor
451            .fields
452            .iter()
453            .find(|f| f.name == "message")
454            .map(|f| f.value.clone())
455            .unwrap_or_default();
456
457        // Get parent span ID if any
458        let parent_span_id = ctx
459            .current_span()
460            .id()
461            .and_then(|id| self.span_ids.lock().get(&id.into_u64()).copied());
462
463        let event_meta = EventMeta {
464            message,
465            target: meta.target().to_string(),
466            level: meta.level().to_string(),
467            file: meta.file().map(|s| s.to_string()),
468            line: meta.line(),
469            fields: visitor.fields,
470            parent_span_id,
471        };
472
473        self.call_event(event_meta);
474    }
475
476    fn on_enter(&self, id: &Id, _ctx: Context<'_, S>) {
477        if let Some(&span_id) = self.span_ids.lock().get(&id.into_u64()) {
478            self.call_enter(span_id);
479        }
480    }
481
482    fn on_exit(&self, id: &Id, _ctx: Context<'_, S>) {
483        if let Some(&span_id) = self.span_ids.lock().get(&id.into_u64()) {
484            self.call_exit(span_id);
485        }
486    }
487
488    fn on_close(&self, id: Id, _ctx: Context<'_, S>) {
489        if let Some(span_id) = self.span_ids.lock().remove(&id.into_u64()) {
490            self.call_drop_span(span_id);
491        }
492    }
493}
494
495/// Visitor for collecting tracing fields into our Field type.
496struct FieldVisitor {
497    fields: Vec<Field>,
498}
499
500impl FieldVisitor {
501    fn new() -> Self {
502        Self { fields: Vec::new() }
503    }
504}
505
506impl tracing::field::Visit for FieldVisitor {
507    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
508        self.fields.push(Field {
509            name: field.name().to_string(),
510            value: format!("{:?}", value),
511        });
512    }
513
514    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
515        self.fields.push(Field {
516            name: field.name().to_string(),
517            value: value.to_string(),
518        });
519    }
520
521    fn record_i64(&mut self, field: &tracing::field::Field, value: i64) {
522        self.fields.push(Field {
523            name: field.name().to_string(),
524            value: value.to_string(),
525        });
526    }
527
528    fn record_u64(&mut self, field: &tracing::field::Field, value: u64) {
529        self.fields.push(Field {
530            name: field.name().to_string(),
531            value: value.to_string(),
532        });
533    }
534
535    fn record_bool(&mut self, field: &tracing::field::Field, value: bool) {
536        self.fields.push(Field {
537            name: field.name().to_string(),
538            value: value.to_string(),
539        });
540    }
541}
542
543// ============================================================================
544// Host Side: TracingSink Implementation
545// ============================================================================
546
547/// Collected trace data for inspection/testing.
548#[derive(Debug, Clone)]
549pub enum TraceRecord {
550    NewSpan { id: u64, meta: SpanMeta },
551    Record { span_id: u64, fields: Vec<Field> },
552    Event(EventMeta),
553    Enter { span_id: u64 },
554    Exit { span_id: u64 },
555    DropSpan { span_id: u64 },
556}
557
558/// Host-side implementation of TracingSink.
559///
560/// Collects all trace data into a buffer for inspection/testing.
561/// In a real application, you might forward to a real tracing subscriber.
562#[derive(Clone)]
563pub struct HostTracingSink {
564    records: Arc<Mutex<Vec<TraceRecord>>>,
565    next_span_id: Arc<AtomicU64>,
566}
567
568impl HostTracingSink {
569    /// Create a new sink that collects trace data.
570    pub fn new() -> Self {
571        Self {
572            records: Arc::new(Mutex::new(Vec::new())),
573            next_span_id: Arc::new(AtomicU64::new(1)),
574        }
575    }
576
577    /// Get all collected trace records.
578    pub fn records(&self) -> Vec<TraceRecord> {
579        self.records.lock().clone()
580    }
581
582    /// Clear all collected records.
583    pub fn clear(&self) {
584        self.records.lock().clear();
585    }
586}
587
588impl Default for HostTracingSink {
589    fn default() -> Self {
590        Self::new()
591    }
592}
593
594impl TracingSink for HostTracingSink {
595    async fn new_span(&self, span: SpanMeta) -> u64 {
596        let id = self.next_span_id.fetch_add(1, Ordering::Relaxed);
597        self.records
598            .lock()
599            .push(TraceRecord::NewSpan { id, meta: span });
600        id
601    }
602
603    async fn record(&self, span_id: u64, fields: Vec<Field>) {
604        self.records
605            .lock()
606            .push(TraceRecord::Record { span_id, fields });
607    }
608
609    async fn event(&self, event: EventMeta) {
610        self.records.lock().push(TraceRecord::Event(event));
611    }
612
613    async fn enter(&self, span_id: u64) {
614        self.records.lock().push(TraceRecord::Enter { span_id });
615    }
616
617    async fn exit(&self, span_id: u64) {
618        self.records.lock().push(TraceRecord::Exit { span_id });
619    }
620
621    async fn drop_span(&self, span_id: u64) {
622        self.records.lock().push(TraceRecord::DropSpan { span_id });
623    }
624}
625
626// ============================================================================
627// Dispatcher Helper
628// ============================================================================
629
630/// Create a dispatcher for TracingSink service.
631pub fn create_tracing_sink_dispatcher(
632    sink: HostTracingSink,
633) -> impl Fn(
634    u32,
635    u32,
636    Vec<u8>,
637) -> Pin<Box<dyn std::future::Future<Output = Result<Frame, RpcError>> + Send>>
638+ Send
639+ Sync
640+ 'static {
641    move |_channel_id, method_id, payload| {
642        let sink = sink.clone();
643        Box::pin(async move {
644            let server = TracingSinkServer::new(sink);
645            server.dispatch(method_id, &payload).await
646        })
647    }
648}
649
650// TracingSinkClient is generated by the rapace::service attribute.
651// Use TracingSinkClient::new(session) to create a client.
rapace_tracing/lib.rs

rapace_tracing/
lib.rs
