Add support for connecting tokio-console to binaries (#1915)

# Description

If compiled with `RUSTFLAGS="--cfg tokio_unstable"` we can now attact
[tokio-console](https://github.com/tokio-rs/console) to our binaries
which can help us to debug issues with debug issues with our async code

# Changes
- Adds https://github.com/tokio-rs/console/tree/main/console-subscriber
as a dependency to our observe crate
- Conditionally registers the console subscriber as an extra layer in
our tracing subscriber registry (this required some rewriting of the
existing layering)

Docs:
https://docs.rs/console-subscriber/latest/console_subscriber/#adding-the-console-subscriber

## How to test
```
RUSTFLAGS="--cfg tokio_unstable" cargo run --bin autopilot
```

In a separate console

```
cargo install --locked tokio-console
tokio-console
```

and observe tokio runtime information

<img width="1401" alt="image"
src="https://github.com/cowprotocol/services/assets/1200333/df47ff54-24df-49bf-ae29-f7df26045339">

README.md

@@ -162,6 +162,17 @@ ANVIL_IP_ADDR=0.0.0.0 anvil \
   --timestamp 1577836800
 ```
 
+### Profiling
+
+In order to attach [tokio-console](https://github.com/tokio-rs/console) to the running process, compile the binary using `RUSTFLAGS="--cfg tokio_unstable"`.
+
+Then in another shell, simply run
+
+```
+cargo install --locked tokio-console
+tokio-console
+```
+
 ## Running the Services Locally
 
 ### Prerequisites

crates/observe/Cargo.toml

@@ -7,6 +7,7 @@ license = "MIT OR Apache-2.0"
 
 [dependencies]
 atty = "0.2"
+console-subscriber = "0.2.0"
 once_cell = { workspace = true }
 prometheus = { workspace = true }
 prometheus-metric-storage = { workspace = true }

crates/observe/src/tracing.rs

@@ -2,7 +2,13 @@ use {
     std::{panic::PanicInfo, sync::Once},
     time::macros::format_description,
     tracing::level_filters::LevelFilter,
-    tracing_subscriber::fmt::{time::UtcTime, writer::MakeWriterExt as _},
+    tracing_subscriber::{
+        fmt::{time::UtcTime, writer::MakeWriterExt as _},
+        prelude::*,
+        util::SubscriberInitExt,
+        EnvFilter,
+        Layer,
+    },
 };
 
 /// Initializes tracing setup that is shared between the binaries.
@@ -25,22 +31,29 @@ pub fn initialize_reentrant(env_filter: &str) {
 }
 
 fn set_tracing_subscriber(env_filter: &str, stderr_threshold: LevelFilter) {
-    // This is what kibana uses to separate multi line log messages.
-    let subscriber_builder = tracing_subscriber::fmt::fmt()
+    let fmt_layer = tracing_subscriber::fmt::layer()
+        .with_writer(
+            std::io::stdout
+                .with_min_level(
+                    stderr_threshold
+                        .into_level()
+                        .unwrap_or(tracing::Level::ERROR),
+                )
+                .or_else(std::io::stderr),
+        )
         .with_timer(UtcTime::new(format_description!(
+            // This is what kibana uses to separate multi line log messages.
             "[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z"
         )))
-        .with_env_filter(env_filter)
-        .with_ansi(atty::is(atty::Stream::Stdout));
-    match stderr_threshold.into_level() {
-        Some(threshold) => subscriber_builder
-            .with_writer(
-                std::io::stderr
-                    .with_max_level(threshold)
-                    .or_else(std::io::stdout),
-            )
-            .init(),
-        None => subscriber_builder.init(),
+        .with_ansi(atty::is(atty::Stream::Stdout))
+        .with_filter::<EnvFilter>(env_filter.into());
+
+    let registry = tracing_subscriber::registry().with(fmt_layer);
+    if cfg!(tokio_unstable) {
+        // Start with tokio_console subscriber
+        registry.with(console_subscriber::spawn()).init();
+    } else {
+        registry.init()
     }
 }