Skip to content

Commit

Permalink
Implement tracing::Event handling & parent_id for spans and events (p…
Browse files Browse the repository at this point in the history 
…aritytech#6672)

* implement events handling, implement parent_id for spans & events

* add events to sp_io::storage

* update test

* add tests

* adjust limit

* let tracing crate handle parent_ids

* re-enable current-id tracking

* add test for threads with CurrentSpan

* fix log level

* remove redundant check for non wasm traces

* remove duplicate definition in test

* Adding conditional events API

* prefer explicit parent_id over current,

enhance test

* limit changes to client::tracing event implementation

* remove From impl due to fallback required on parent_id

* implement SPAN_LIMIT

change event log output

* change version of tracing-core

* update dependancies

* revert limit

* remove duplicate dependency

* Apply suggestions from code review

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

Co-authored-by: Matt Rutherford <mattrutherford@users.noreply.github.com>
Co-authored-by: Benjamin Kampmann <ben@parity.io>
Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>
  • Loading branch information
@mattrutherford @gnunicorn @bkchr
4 people authored and Andrew Plaza committed Nov 9, 2020
1 parent 8efeb48 commit 66d828b
Show file tree
Hide file tree
Showing 3 changed files with 190 additions and 14 deletions.
5 changes: 4 additions & 1 deletion client/executor/src/integration_tests/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -662,14 +662,17 @@ fn parallel_execution(wasm_method: WasmExecutionMethod) {
fn wasm_tracing_should_work(wasm_method: WasmExecutionMethod) {

use std::sync::{Arc, Mutex};
use sc_tracing::SpanDatum;

use sc_tracing::{SpanDatum, TraceEvent};

struct TestTraceHandler(Arc<Mutex<Vec<SpanDatum>>>);

impl sc_tracing::TraceHandler for TestTraceHandler {
fn handle_span(&self, sd: SpanDatum) {
self.0.lock().unwrap().push(sd);
}

fn handle_event(&self, _event: TraceEvent) {}
}

let traces = Arc::new(Mutex::new(Vec::new()));
Expand Down
198 changes: 185 additions & 13 deletions client/tracing/src/lib.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ use std::time::{Duration, Instant};

use parking_lot::Mutex;
use serde::ser::{Serialize, Serializer, SerializeMap};
use tracing_core::{
use tracing::{
event::Event,
field::{Visit, Field},
Level,
Expand All @@ -51,6 +51,15 @@ pub struct ProfilingLayer {
current_span: CurrentSpan,
}

/// Responsible for assigning ids to new spans, which are not re-used.
pub struct ProfilingSubscriber {
next_id: AtomicU64,
targets: Vec<(String, Level)>,
trace_handler: Box<dyn TraceHandler>,
span_data: Mutex<FxHashMap<Id, SpanDatum>>,
current_span: CurrentSpan,
}

/// Used to configure how to receive the metrics
#[derive(Debug, Clone)]
pub enum TracingReceiver {
Expand All @@ -70,6 +79,18 @@ impl Default for TracingReceiver {
pub trait TraceHandler: Send + Sync {
/// Process a `SpanDatum`
fn handle_span(&self, span: SpanDatum);
/// Process a `TraceEvent`
fn handle_event(&self, event: TraceEvent);
}

/// Represents a tracing event, complete with values
#[derive(Debug)]
pub struct TraceEvent {
pub name: &'static str,
pub target: String,
pub level: Level,
pub values: Values,
pub parent_id: Option<Id>,
}


Expand All @@ -87,8 +108,10 @@ pub struct TraceEvent {
#[derive(Debug)]
pub struct SpanDatum {
/// id for this span
pub id: u64,
/// Name of the span
pub id: Id,
/// id of the parent span, if any
pub parent_id: Option<Id>,
/// Name of this span
pub name: String,
/// Target, typically module
pub target: String,
Expand Down Expand Up @@ -236,6 +259,7 @@ impl ProfilingLayer {
targets,
trace_handler,
span_data: Mutex::new(FxHashMap::default()),
current_span: Default::default()
}
}

Expand Down Expand Up @@ -271,7 +295,8 @@ impl<S: Subscriber> Layer<S> for ProfilingLayer {
let mut values = Values::default();
attrs.record(&mut values);
let span_datum = SpanDatum {
id,
id: id.clone(),
parent_id: attrs.parent().cloned().or_else(|| self.current_span.id()),
name: attrs.metadata().name().to_owned(),
target: attrs.metadata().target().to_owned(),
level: attrs.metadata().level().clone(),
Expand All @@ -285,7 +310,7 @@ impl<S: Subscriber> Layer<S> for ProfilingLayer {

fn on_record(&self, span: &Id, values: &Record<'_>, _ctx: Context<S>) {
let mut span_data = self.span_data.lock();
if let Some(s) = span_data.get_mut(&span.into_u64()) {
if let Some(s) = span_data.get_mut(span) {
values.record(&mut s.values);
}
}
Expand All @@ -307,7 +332,7 @@ impl<S: Subscriber> Layer<S> for ProfilingLayer {
self.current_span.enter(span.clone());
let mut span_data = self.span_data.lock();
let start_time = Instant::now();
if let Some(mut s) = span_data.get_mut(&span.into_u64()) {
if let Some(mut s) = span_data.get_mut(&span) {
s.start_time = start_time;
}
}
Expand All @@ -317,7 +342,7 @@ impl<S: Subscriber> Layer<S> for ProfilingLayer {
let end_time = Instant::now();
let span_datum = {
let mut span_data = self.span_data.lock();
span_data.remove(&span.into_u64())
span_data.remove(&span)
};

if let Some(mut span_datum) = span_datum {
Expand All @@ -330,8 +355,10 @@ impl<S: Subscriber> Layer<S> for ProfilingLayer {
if let Some(t) = span_datum.values.string_values.remove(WASM_TARGET_KEY) {
span_datum.target = t;
}
}
if self.check_target(&span_datum.target, &span_datum.level) {
if self.check_target(&span_datum.target, &span_datum.level) {
self.trace_handler.handle_span(span_datum);
}
} else {
self.trace_handler.handle_span(span_datum);
}
};
Expand Down Expand Up @@ -364,18 +391,32 @@ impl TraceHandler for LogTraceHandler {
span_datum.target,
span_datum.name,
span_datum.overall_time.as_nanos(),
span_datum.id.into_u64(),
span_datum.parent_id.map(|s| s.into_u64()),
);
} else {
log::log!(
log_level(span_datum.level),
"{}: {}, time: {}, {}",
"{}: {}, time: {}, id: {}, parent_id: {:?}, values: {}",
span_datum.target,
span_datum.name,
span_datum.overall_time.as_nanos(),
span_datum.id.into_u64(),
span_datum.parent_id.map(|s| s.into_u64()),
span_datum.values,
);
}
}

fn handle_event(&self, event: TraceEvent) {
log::log!(
log_level(event.level),
"{}, parent_id: {:?}, {}",
event.target,
event.parent_id.map(|s| s.into_u64()),
event.values,
);
}
}

/// TraceHandler for sending span data to telemetry,
Expand All @@ -388,11 +429,21 @@ impl TraceHandler for TelemetryTraceHandler {
telemetry!(SUBSTRATE_INFO; "tracing.profiling";
"name" => span_datum.name,
"target" => span_datum.target,
"line" => span_datum.line,
"time" => span_datum.overall_time.as_nanos(),
"id" => span_datum.id.into_u64(),
"parent_id" => span_datum.parent_id.map(|i| i.into_u64()),
"values" => span_datum.values
);
}

fn handle_event(&self, event: TraceEvent) {
telemetry!(SUBSTRATE_INFO; "tracing.event";
"name" => event.name,
"target" => event.target,
"parent_id" => event.parent_id.map(|i| i.into_u64()),
"values" => event.values
);
}
}

#[cfg(test)]
Expand All @@ -403,12 +454,17 @@ mod tests {

struct TestTraceHandler {
spans: Arc<Mutex<Vec<SpanDatum>>>,
events: Arc<Mutex<Vec<TraceEvent>>>,
}

impl TraceHandler for TestTraceHandler {
fn handle_span(&self, sd: SpanDatum) {
self.spans.lock().push(sd);
}

fn handle_event(&self, event: TraceEvent) {
self.events.lock().push(event);
}
}

type TestSubscriber = tracing_subscriber::layer::Layered<
Expand All @@ -418,8 +474,10 @@ mod tests {

fn setup_subscriber() -> (TestSubscriber, Arc<Mutex<Vec<SpanDatum>>>, Arc<Mutex<Vec<TraceEvent>>>) {
let spans = Arc::new(Mutex::new(Vec::new()));
let events = Arc::new(Mutex::new(Vec::new()));
let handler = TestTraceHandler {
spans: spans.clone(),
events: events.clone(),
};
let layer = ProfilingLayer::new_with_handler(
Box::new(handler),
Expand All @@ -431,25 +489,45 @@ mod tests {

#[test]
fn test_span() {
let (sub, spans) = setup_subscriber();
let (sub, spans, events) = setup_subscriber();
let _sub_guard = tracing::subscriber::set_default(sub);
let span = tracing::info_span!(target: "test_target", "test_span1");
assert_eq!(spans.lock().len(), 0);
assert_eq!(events.lock().len(), 0);
let _guard = span.enter();
assert_eq!(spans.lock().len(), 0);
assert_eq!(events.lock().len(), 0);
drop(_guard);
drop(span);
assert_eq!(spans.lock().len(), 1);
assert_eq!(events.lock().len(), 0);
let sd = spans.lock().remove(0);
assert_eq!(sd.name, "test_span1");
assert_eq!(sd.target, "test_target");
let time: u128 = sd.overall_time.as_nanos();
assert!(time > 0);
}

#[test]
fn test_span_parent_id() {
let (sub, spans, _events) = setup_subscriber();
let _sub_guard = tracing::subscriber::set_default(sub);
let span1 = tracing::info_span!(target: "test_target", "test_span1");
let _guard1 = span1.enter();
let span2 = tracing::info_span!(target: "test_target", "test_span2");
let _guard2 = span2.enter();
drop(_guard2);
drop(span2);
let sd2 = spans.lock().remove(0);
drop(_guard1);
drop(span1);
let sd1 = spans.lock().remove(0);
assert_eq!(sd1.id, sd2.parent_id.unwrap())
}

#[test]
fn test_span_values() {
let (sub, spans) = setup_subscriber();
let (sub, spans, _events) = setup_subscriber();
let _sub_guard = tracing::subscriber::set_default(sub);
let test_bool = true;
let test_u64 = 1u64;
Expand All @@ -475,4 +553,98 @@ mod tests {
assert_eq!(values.i64_values.get("test_i64").unwrap(), &test_i64);
assert_eq!(values.string_values.get("test_str").unwrap(), &test_str.to_owned());
}

#[test]
fn test_event() {
let (sub, _spans, events) = setup_subscriber();
let _sub_guard = tracing::subscriber::set_default(sub);
tracing::event!(target: "test_target", tracing::Level::INFO, "test_event");
let mut te1 = events.lock().remove(0);
assert_eq!(te1.values.string_values.remove(&"message".to_owned()).unwrap(), "test_event".to_owned());
}

#[test]
fn test_event_parent_id() {
let (sub, spans, events) = setup_subscriber();
let _sub_guard = tracing::subscriber::set_default(sub);

// enter span
let span1 = tracing::info_span!(target: "test_target", "test_span1");
let _guard1 = span1.enter();

// emit event
tracing::event!(target: "test_target", tracing::Level::INFO, "test_event");

//exit span
drop(_guard1);
drop(span1);

let sd1 = spans.lock().remove(0);
let te1 = events.lock().remove(0);

assert_eq!(sd1.id, te1.parent_id.unwrap());
}

#[test]
fn test_parent_id_with_threads() {
use std::sync::mpsc;
use std::thread;

let (sub, spans, events) = setup_subscriber();
let _sub_guard = tracing::subscriber::set_global_default(sub);
let span1 = tracing::info_span!(target: "test_target", "test_span1");
let _guard1 = span1.enter();

let (tx, rx) = mpsc::channel();
let handle = thread::spawn(move || {
let span2 = tracing::info_span!(target: "test_target", "test_span2");
let _guard2 = span2.enter();
// emit event
tracing::event!(target: "test_target", tracing::Level::INFO, "test_event1");
for msg in rx.recv() {
if msg == false {
break;
}
}
// gard2 and span2 dropped / exited
});

// wait for Event to be dispatched and stored
while events.lock().is_empty() {
thread::sleep(Duration::from_millis(1));
}

// emit new event (will be second item in Vec) while span2 still active in other thread
tracing::event!(target: "test_target", tracing::Level::INFO, "test_event2");

// stop thread and drop span
let _ = tx.send(false);
let _ = handle.join();

// wait for Span to be dispatched and stored
while spans.lock().is_empty() {
thread::sleep(Duration::from_millis(1));
}
let span2 = spans.lock().remove(0);
let event1 = events.lock().remove(0);
drop(_guard1);
drop(span1);

// emit event with no parent
tracing::event!(target: "test_target", tracing::Level::INFO, "test_event3");

let span1 = spans.lock().remove(0);
let event2 = events.lock().remove(0);

assert_eq!(event1.values.string_values.get("message").unwrap(), "test_event1");
assert_eq!(event2.values.string_values.get("message").unwrap(), "test_event2");
assert!(span1.parent_id.is_none());
assert!(span2.parent_id.is_none());
assert_eq!(span2.id, event1.parent_id.unwrap());
assert_eq!(span1.id, event2.parent_id.unwrap());
assert_ne!(span2.id, span1.id);

let event3 = events.lock().remove(0);
assert!(event3.parent_id.is_none());
}
}
1 change: 1 addition & 0 deletions primitives/runtime-interface/test/Cargo.toml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,3 +22,4 @@ sp-core = { version = "2.0.0-rc5", path = "../../core" }
sp-io = { version = "2.0.0-rc5", path = "../../io" }
tracing = "0.1.19"
tracing-core = "0.1.15"

0 comments on commit 66d828b

Please sign in to comment.