add timely-console feature

.cargo/config

@@ -1,3 +1,6 @@
+[build]
+rustflags = ["--cfg", "tokio_unstable"]
+
 [target."x86_64-unknown-linux-gnu"]
 # Compressing debug information can yield hundreds of megabytes of savings.
 # The Rust toolchain does not currently perform dead code elimination on

Cargo.toml

@@ -93,3 +93,9 @@ prost = { git = "https://github.com/MaterializeInc/prost.git" }
 prost-types = { git = "https://github.com/MaterializeInc/prost.git" }
 prost-build = { git = "https://github.com/MaterializeInc/prost.git" }
 prost-derive = { git = "https://github.com/MaterializeInc/prost.git" }
+
+[patch."https://github.com/TimelyDataflow/timely-dataflow"]
+timely = { path = "../timely-dataflow/timely" }
+[patch."https://github.com/TimelyDataflow/differential-dataflow"]
+differential-dataflow = { path = "../differential-dataflow" }
+dogsdogsdogs = { path = "../differential-dataflow/dogsdogsdogs" }

src/materialized/Cargo.toml

@@ -144,6 +144,7 @@ walkdir = "2.3.2"
 # access to the file system.
 dev-web = []
 tokio-console = ["console-subscriber", "tokio/tracing"]
+timely-console = ["tokio-console", "timely/tracing"]
 
 [package.metadata.cargo-udeps.ignore]
 # krb5 used to build a binary

src/materialized/src/bin/materialized/main.rs

@@ -377,10 +377,20 @@ struct Args {
     #[clap(long, env = "MZ_TELEMETRY_INTERVAL", parse(try_from_str = mz_repr::util::parse_duration), hide = true)]
     telemetry_interval: Option<Duration>,
 
+    #[clap(flatten)]
+    console_args: ConsoleArgs,
+}
+
+#[derive(Parser)]
+struct ConsoleArgs {
     #[cfg(feature = "tokio-console")]
     /// Turn on the console-subscriber to use materialize with `tokio-console`
     #[clap(long, hide = true)]
     tokio_console: bool,
+
+    #[cfg(feature = "timely-console")]
+    #[clap(long, hide = true)]
+    disable_timely_console: bool,
 }
 
 /// This type is a hack to allow a dynamic default for the `--workers` argument,
@@ -423,6 +433,59 @@ fn main() {
     }
 }
 
+fn finish_and_init_tracing<
+    L: tracing_subscriber::layer::Layer<S> + Send + Sync + 'static,
+    S: ::tracing::Subscriber + Send + Sync + 'static,
+>(
+    stack: tracing_subscriber::layer::Layered<L, S>,
+    #[allow(unused)] args: ConsoleArgs,
+) where
+    tracing_subscriber::layer::Layered<L, S>: tracing_subscriber::util::SubscriberInitExt,
+    for<'ls> S: tracing_subscriber::registry::LookupSpan<'ls>,
+{
+    use tracing_subscriber::util::SubscriberInitExt;
+
+    #[cfg(not(feature = "tokio-console"))]
+    let tokio_console = false;
+    #[cfg(not(feature = "timely-console"))]
+    let disable_timely_console = false;
+
+    #[cfg(feature = "tokio-console")]
+    let tokio_console = args.tokio_console;
+    #[cfg(feature = "timely-console")]
+    let disable_timely_console = args.disable_timely_console;
+
+    #[allow(clippy::if_same_then_else)]
+    if disable_timely_console && tokio_console {
+        #[cfg(feature = "tokio-console")]
+        {
+            use tracing_subscriber::filter::FilterExt;
+            use tracing_subscriber::filter::Targets;
+            use tracing_subscriber::layer::SubscriberExt;
+            use tracing_subscriber::Layer;
+            stack
+                .with(tokio_console.then(|| {
+                    console_subscriber::spawn().with_filter(
+                        Targets::from_str("timely.console.span=trace")
+                            .unwrap()
+                            .not(),
+                    )
+                }))
+                .init();
+        }
+    } else if tokio_console {
+        #[cfg(feature = "tokio-console")]
+        {
+            use tracing_subscriber::layer::SubscriberExt;
+            stack
+                .with(tokio_console.then(|| console_subscriber::spawn()))
+                .init();
+        }
+    } else {
+        stack.init()
+    }
+}
+
 fn run(args: Args) -> Result<(), anyhow::Error> {
     panic::set_hook(Box::new(handle_panic));
 
@@ -551,7 +614,6 @@ fn run(args: Args) -> Result<(), anyhow::Error> {
         use tracing_subscriber::filter::{LevelFilter, Targets};
         use tracing_subscriber::fmt;
         use tracing_subscriber::layer::{Layer, SubscriberExt};
-        use tracing_subscriber::util::SubscriberInitExt;
 
         let filter = Targets::from_str(&args.log_filter)
             .context("parsing --log-filter option")?
@@ -581,10 +643,7 @@ fn run(args: Args) -> Result<(), anyhow::Error> {
                             .with_filter(filter),
                     );
 
-                #[cfg(feature = "tokio-console")]
-                let stack = stack.with(args.tokio_console.then(|| console_subscriber::spawn()));
-
-                stack.init()
+                finish_and_init_tracing(stack, args.console_args);
             }
             log_file => {
                 // Logging to a file. If the user did not explicitly specify
@@ -627,10 +686,7 @@ fn run(args: Args) -> Result<(), anyhow::Error> {
                             .with_filter(filter),
                     );
 
-                #[cfg(feature = "tokio-console")]
-                let stack = stack.with(args.tokio_console.then(|| console_subscriber::spawn()));
-
-                stack.init()
+                finish_and_init_tracing(stack, args.console_args);
             }
         }
     }