Tracing based communication tracking

[robinhundt]

With the switch to QUIC based channels (#28) we could also introduce tracing based communication tracking. The below code shows some of the likely required APIs. The idea is to instrument functions with a phase field and emit events with the number of bytes read/written in the AsyncRead/-Write implementations for the sub-streams. We then provide a tracing_subscriber::Layer implementation which records the byte counts for the events for the respective phases.

Open questions:

• How to deal with nested phases?
  • E.g. If we annotate the Executor::execute method with phase = online what happens if we use this method inside the function dependent setup impl of another protocol? (See aby impl)
  • Or what about interleaved setup and online phases?
• Allow more granular tracking of communication and use span target and name?
  • This could be used to track the communication of specific gates
• Performance impact of this. Especially if the Layer implementation uses a hashmap, potentially with a lock, what is the perf impact?
  • Benefit of tracing: If we don't include the layer, we don't pay these costs
• How exactly is determined which spans/events are even emitted? Especially in combination with a fmt layer
  • Ideally, we want these spans and events to only optionally be included in the fmt layer output for better performance. Can we have a level=INFO filter for the fmt layer, emit the events with a trace level and still process them in our tracking layer? I find tracing-subscriber a little confusing in this regard
• This tracking would operate on the level of QUIC streams, so it would not include QUIC overhead. We might be able to get the complete communication by using s2n-quic's events API.

use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt};

use std::fmt::Debug;

use tracing::field::{Field, Visit};
use tracing::span::{Attributes, Id};
use tracing_subscriber::layer::{Context, Layer};
use tracing::{event, info, instrument, Level};

#[instrument(fields(phase = "Outer"))]
pub fn foo(p: u8) {
    bar(p);
}

#[instrument(fields(phase = "Inner"))]
fn bar(p: u8) {
    dbg!(p);
    event!(Level::INFO, bytes = p);
}



struct TestVisitor(Option<String>);

impl Visit for TestVisitor {
    fn record_debug(&mut self, _field: &Field, _value: &dyn Debug) {
        // do nothing
    }

    fn record_str(&mut self, field: &Field, value: &str) {
        if field.name() == "phase" {
            self.0 = Some(value.to_owned());
        }
    }
}

struct Test(String);

pub struct MyTestLayer;

impl<S> Layer<S> for MyTestLayer
where
    S: tracing::Subscriber,
    S: for<'lookup> tracing_subscriber::registry::LookupSpan<'lookup>,
{
    fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) {
        let span = ctx.span(id).unwrap();
        let mut visitor = TestVisitor(None);
        attrs.record(&mut visitor);
        if let Some(test) = visitor.0 {
            dbg!(&test);
            span.extensions_mut().insert(Test(test));
        } else {
            // span has no test value
        }
    }

    fn on_event(&self, event: &tracing::Event<'_>, ctx: Context<'_, S>) {
        let span = ctx.event_span(event).unwrap();
        dbg!(span.metadata());
        for s in ctx.event_scope(event).unwrap() {
            dbg!(&s.extensions().get::<Test>().unwrap().0);
        }
    }
}


fn main() {
    tracing_subscriber::registry()
    .with(fmt::layer())
    .with(MyTestLayer)
    .init();

    foo(4);

}