use std::collections::HashSet;
use std::marker::PhantomData;
use tracing::Subscriber;
use tracing_core::span;
use tracing_subscriber::{layer::Context, registry::LookupSpan, Layer};
use crate::field_mapping::ExtractedFields;
use crate::layer_common;
use crate::span_info::{SpanInfo, SPAN_INFO_KEY};
use crate::tracing_field::get_tracing_fields;
struct AsyncScopeState {
depth: usize,
}
pub struct AsyncDcontextLayer<S> {
include_span_info: bool,
_subscriber: PhantomData<fn(S)>,
}
impl<S> AsyncDcontextLayer<S> {
pub fn new() -> Self {
Self {
include_span_info: false,
_subscriber: PhantomData,
}
}
pub fn builder() -> AsyncDcontextLayerBuilder<S> {
AsyncDcontextLayerBuilder {
include_span_info: false,
_subscriber: PhantomData,
}
}
}
impl<S> Default for AsyncDcontextLayer<S> {
fn default() -> Self {
Self::new()
}
}
pub struct AsyncDcontextLayerBuilder<S> {
include_span_info: bool,
_subscriber: PhantomData<fn(S)>,
}
impl<S> AsyncDcontextLayerBuilder<S> {
pub fn include_span_info(mut self) -> Self {
self.include_span_info = true;
self
}
pub fn build(self) -> AsyncDcontextLayer<S> {
AsyncDcontextLayer {
include_span_info: self.include_span_info,
_subscriber: PhantomData,
}
}
}
use tracing_subscriber::registry::SpanRef;
fn apply_field_extraction<S>(span: &SpanRef<'_, S>)
where
S: for<'a> LookupSpan<'a>,
{
let metadata_fields = get_tracing_fields();
if !metadata_fields.iter().any(|e| e.extract.is_some()) {
return;
}
let extensions = span.extensions();
let fields = match extensions.get::<ExtractedFields>() {
Some(f) => f,
None => return,
};
for entry in metadata_fields {
if let Some(ref extract) = entry.extract {
let value = if let Some(v) = fields.string_values.get(entry.span_field) {
extract.from_str.as_ref().and_then(|f| f(v))
} else if let Some(&v) = fields.u64_values.get(entry.span_field) {
extract.from_u64.as_ref().and_then(|f| f(v))
} else if let Some(&v) = fields.i64_values.get(entry.span_field) {
extract.from_i64.as_ref().and_then(|f| f(v))
} else if let Some(&v) = fields.bool_values.get(entry.span_field) {
extract.from_bool.as_ref().and_then(|f| f(v))
} else {
None
};
if let Some(val) = value {
dcontext::async_ctx::set_raw_value(entry.context_key, val);
}
}
}
}
fn set_span_info<S>(span: &SpanRef<'_, S>)
where
S: for<'a> LookupSpan<'a>,
{
let metadata = span.metadata();
let info = SpanInfo {
name: metadata.name().to_string(),
target: metadata.target().to_string(),
level: metadata.level().to_string(),
};
dcontext::async_ctx::set_context(SPAN_INFO_KEY, info);
}
fn record_context_to_span<S>(span: &SpanRef<'_, S>)
where
S: for<'a> LookupSpan<'a>,
{
let metadata_fields = get_tracing_fields();
if !metadata_fields.iter().any(|e| e.span_fmt_fn.is_some()) {
return;
}
let user_set: HashSet<&str> = {
let extensions = span.extensions();
extensions
.get::<ExtractedFields>()
.map(|ef| ef.user_set_fields.iter().copied().collect())
.unwrap_or_default()
};
let to_record: Vec<(&'static str, String)> = metadata_fields
.iter()
.filter_map(|entry| {
let fmt_fn = entry.span_fmt_fn.as_ref()?;
if user_set.contains(entry.record_field) {
return None;
}
let formatted = dcontext::async_ctx::with_context_value(entry.context_key, |any_val| {
fmt_fn(any_val)
})
.flatten()?;
Some((entry.record_field, formatted))
})
.collect();
if !to_record.is_empty() {
let current = tracing::Span::current();
layer_common::SELF_RECORDING.with(|f| f.set(true));
for (field_name, value) in &to_record {
current.record(*field_name, value.as_str());
}
layer_common::SELF_RECORDING.with(|f| f.set(false));
}
}
impl<S> Layer<S> for AsyncDcontextLayer<S>
where
S: Subscriber + for<'a> LookupSpan<'a>,
{
fn on_new_span(&self, attrs: &span::Attributes<'_>, id: &span::Id, ctx: Context<'_, S>) {
if let Some(span) = ctx.span(id) {
layer_common::extract_span_fields(attrs, &span);
}
}
fn on_record(&self, id: &span::Id, values: &span::Record<'_>, ctx: Context<'_, S>) {
if let Some(span) = ctx.span(id) {
layer_common::merge_recorded_fields(values, &span);
}
}
fn on_enter(&self, id: &span::Id, ctx: Context<'_, S>) {
let span = match ctx.span(id) {
Some(s) => s,
None => return,
};
{
let extensions = span.extensions();
if extensions.get::<AsyncScopeState>().is_some() {
return;
}
}
let name = span.metadata().name();
let guard = match dcontext::async_ctx::try_push_scope(name) {
Some(guard) => guard,
None => return,
};
let depth = guard.expected_depth();
std::mem::forget(guard);
apply_field_extraction(&span);
if self.include_span_info {
set_span_info(&span);
}
record_context_to_span(&span);
let mut extensions = span.extensions_mut();
extensions.insert(AsyncScopeState { depth });
}
fn on_exit(&self, _id: &span::Id, _ctx: Context<'_, S>) {
}
fn on_close(&self, id: span::Id, ctx: Context<'_, S>) {
let span = match ctx.span(&id) {
Some(s) => s,
None => return,
};
let state = {
let mut extensions = span.extensions_mut();
extensions.remove::<AsyncScopeState>()
};
if let Some(state) = state {
let current = dcontext::async_ctx::current_depth();
if current == Some(state.depth) {
dcontext::async_ctx::pop_scope(state.depth);
}
}
{
let mut extensions = span.extensions_mut();
extensions.remove::<ExtractedFields>();
}
}
}