feat(subscriber): resource instrumentation

Cargo.toml

@@ -4,4 +4,7 @@ members = [
     "console-subscriber",
     "console-api"
 ]
-resolver = "2"
+resolver = "2"
+
+[patch.crates-io]
+tokio = { git = 'https://github.com/zaharidichev/tokio', branch = 'zd/instrument-sleep' }

console-api/build.rs

@@ -5,6 +5,10 @@ fn main() -> Result<(), Box<dyn Error>> {
         "../proto/trace.proto",
         "../proto/common.proto",
         "../proto/tasks.proto",
+        "../proto/instrument.proto",
+        "../proto/resources.proto",
+        "../proto/resource_ops.proto",
+        "../proto/async_ops.proto",
     ];
     let dirs = &["../proto"];
 

console-api/src/async_ops.rs

@@ -0,0 +1 @@
+tonic::include_proto!("rs.tokio.console.async_ops");

console-api/src/common.rs

@@ -32,19 +32,11 @@ impl<'a> From<&'a tracing_core::Metadata<'a>> for Metadata {
             metadata::Kind::Event
         };
 
-        let location = Location {
-            file: meta.file().map(String::from),
-            module_path: meta.module_path().map(String::from),
-            line: meta.line(),
-            column: None, // tracing doesn't support columns yet
-        };
-
         let field_names = meta.fields().iter().map(|f| f.name().to_string()).collect();
-
         Metadata {
             name: meta.name().to_string(),
             target: meta.target().to_string(),
-            location: Some(location),
+            location: Some(meta.into()),
             kind: kind as i32,
             level: metadata::Level::from(*meta.level()) as i32,
             field_names,
@@ -53,6 +45,17 @@ impl<'a> From<&'a tracing_core::Metadata<'a>> for Metadata {
     }
 }
 
+impl<'a> From<&'a tracing_core::Metadata<'a>> for Location {
+    fn from(meta: &'a tracing_core::Metadata<'a>) -> Self {
+        Location {
+            file: meta.file().map(String::from),
+            module_path: meta.module_path().map(String::from),
+            line: meta.line(),
+            column: None, // tracing doesn't support columns yet
+        }
+    }
+}
+
 impl<'a> From<&'a std::panic::Location<'a>> for Location {
     fn from(loc: &'a std::panic::Location<'a>) -> Self {
         Location {

console-api/src/instrument.rs

@@ -0,0 +1 @@
+tonic::include_proto!("rs.tokio.console.instrument");

console-api/src/lib.rs

@@ -1,4 +1,8 @@
+pub mod async_ops;
 mod common;
+pub mod instrument;
+pub mod resource_ops;
+pub mod resources;
 pub mod tasks;
 pub mod trace;
 pub use common::*;

console-api/src/resource_ops.rs

@@ -0,0 +1 @@
+tonic::include_proto!("rs.tokio.console.resource_ops");

console-api/src/resources.rs

@@ -0,0 +1 @@
+tonic::include_proto!("rs.tokio.console.resources");

console-subscriber/Cargo.toml

@@ -11,6 +11,7 @@ edition = "2018"
 tokio = { version = "^1.7", features = ["sync", "time", "macros", "tracing"]}
 tokio-stream = "0.1"
 tonic = { version = "0.4", features = ["transport"] }
+thread_local = "1.1.3"
 console-api = { path = "../console-api", features = ["transport"]}
 tracing-core = "0.1.18"
 tracing = "0.1.26"

console-subscriber/examples/dump.rs

@@ -1,4 +1,4 @@
-use console_api::tasks::{tasks_client::TasksClient, TasksRequest};
+use console_api::instrument::{instrument_client::InstrumentClient, InstrumentRequest};
 use futures::stream::StreamExt;
 
 #[tokio::main]
@@ -11,16 +11,16 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     });
 
     eprintln!("CONNECTING: {}", target);
-    let mut client = TasksClient::connect(target).await?;
+    let mut client = InstrumentClient::connect(target).await?;
 
-    let request = tonic::Request::new(TasksRequest {});
-    let mut stream = client.watch_tasks(request).await?.into_inner();
+    let request = tonic::Request::new(InstrumentRequest {});
+    let mut stream = client.watch_updates(request).await?.into_inner();
 
     let mut i: usize = 0;
     while let Some(update) = stream.next().await {
         match update {
             Ok(update) => {
-                eprintln!("UPDATE {}: {:#?}\n", i, update);
+                println!("UPDATE {}: {:#?}\n", i, update);
                 i += 1;
             }
             Err(e) => {