Merge pull request scylladb#992 from Lorak-mmk/remove_with_current_su…

…bscriber

Remove `.with_current_subscriber()` calls

(cherry picked from commit 88bc8c1)

docs/source/logging/logging.md

@@ -1,9 +1,15 @@
 # Logging
 
 The driver uses the [tracing](https://github.com/tokio-rs/tracing) crate for all logs.\
-To view the logs you have to create a `tracing` subscriber to which all logs will be written.
+There are two ways to view the logs:
+- Create a `tracing` subscriber to which all logs will be written (recommended).
+- Enable `log` feature on `tracing` crate and use some logger from `log` ecosystem. \
+Only do this if you can't use `tracing` subscriber for some reason.
+
+## Using tracing subscriber
+
+To print the logs you can use the default subscriber:
 
-To just print the logs you can use the default subscriber:
 ```rust
 # extern crate scylla;
 # extern crate tokio;
@@ -45,4 +51,44 @@ To start this example execute:
 RUST_LOG=info cargo run
 ```
 
-The full [example](https://github.com/scylladb/scylla-rust-driver/tree/main/examples/logging.rs) is available in the `examples` folder
+The full [example](https://github.com/scylladb/scylla-rust-driver/tree/main/examples/logging.rs) is available in the `examples` folder.
+You can run it from main folder of driver repository using `RUST_LOG=trace SCYLLA_URI=<scylla_ip>:9042 cargo run --example logging`.
+
+## Using log
+
+To collect tracing events using log collector you first need to enable `log` feature on `tracing` crate.
+You can use `cargo add tracing -F log` or edit `Cargo.toml`:
+```toml
+tracing = { version = "0.1.40" , features = ["log"] }
+```
+then you can setup `env_logger` os some other logger and it will output logs from the driver:
+
+```rust
+# extern crate scylla;
+# extern crate tokio;
+# extern crate tracing;
+# extern crate env_logger;
+# use std::error::Error;
+# use scylla::{Session, SessionBuilder};
+use tracing::info;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    // Setup `log` collector that uses RUST_LOG env variable to configure
+    // verbosity.
+    env_logger::init();
+
+    let uri = std::env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+    info!("Connecting to {}", uri);
+
+    let session: Session = SessionBuilder::new().known_node(uri).build().await?;
+    session.query("CREATE KEYSPACE IF NOT EXISTS examples_ks WITH REPLICATION = {'class' : 'NetworkTopologyStrategy', 'replication_factor' : 1}", &[]).await?;
+
+    session.query("USE examples_ks", &[]).await?;
+
+    Ok(())
+}
+```
+
+The full [example](https://github.com/scylladb/scylla-rust-driver/tree/main/examples/logging_log.rs) is available in the `examples` folder.
+You can run it from main folder of driver repository using `RUST_LOG=trace SCYLLA_URI=<scylla_ip>:9042 cargo run --example logging_log`.

examples/Cargo.toml

@@ -12,7 +12,7 @@ rustyline = "9"
 rustyline-derive = "0.6"
 scylla = {path = "../scylla", features = ["ssl", "cloud", "chrono", "time", "num-bigint-03", "num-bigint-04", "bigdecimal-04"]}
 tokio = {version = "1.1.0", features = ["full"]}
-tracing = "0.1.25"
+tracing = { version = "0.1.25" , features = ["log"] }
 tracing-subscriber = { version = "0.3.14", features = ["env-filter"] }
 chrono = { version = "0.4", default-features = false }
 time = { version = "0.3.22" }
@@ -21,6 +21,7 @@ tower = "0.4"
 stats_alloc = "0.1"
 clap = { version = "3.2.4", features = ["derive"] }
 rand = "0.8.5"
+env_logger = "0.10"
 
 [[example]]
 name = "auth"
@@ -34,6 +35,10 @@ path = "basic.rs"
 name = "logging"
 path = "logging.rs"
 
+[[example]]
+name = "logging_log"
+path = "logging_log.rs"
+
 [[example]]
 name = "tls"
 path = "tls.rs"

examples/logging_log.rs

@@ -0,0 +1,27 @@
+use anyhow::Result;
+use scylla::transport::session::Session;
+use scylla::SessionBuilder;
+use std::env;
+use tracing::info;
+
+// To run this example, and view logged messages, RUST_LOG env var needs to be set
+// This can be done using shell command presented below
+// RUST_LOG=info cargo run --example logging_log
+#[tokio::main]
+async fn main() -> Result<()> {
+    // Driver uses `tracing` for logging purposes, but it's possible to use `log`
+    // ecosystem to view the messages. This requires adding `tracing` crate to
+    // dependencies and enabling its "log" feature. Then you will be able to use
+    // loggers like `env_logger` to see driver's messages.
+    env_logger::init();
+
+    let uri = env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+    info!("Connecting to {}", uri);
+
+    let session: Session = SessionBuilder::new().known_node(uri).build().await?;
+    session.query("CREATE KEYSPACE IF NOT EXISTS examples_ks WITH REPLICATION = {'class' : 'NetworkTopologyStrategy', 'replication_factor' : 1}", &[]).await?;
+
+    session.query("USE examples_ks", &[]).await?;
+
+    Ok(())
+}

scylla-proxy/examples/cmdline.rs

@@ -13,7 +13,6 @@ use std::{
 };
 
 use scylla_proxy::{Node, Proxy, ShardAwareness};
-use tracing::instrument::WithSubscriber;
 
 fn init_logger() {
     tracing_subscriber::fmt::fmt()
@@ -53,7 +52,7 @@ async fn main() {
         None,
         None,
     )]);
-    let running_proxy = proxy.run().with_current_subscriber().await.unwrap();
+    let running_proxy = proxy.run().await.unwrap();
 
     pause().await;
     running_proxy.finish().await.unwrap();

scylla-proxy/examples/identity_proxy.rs

@@ -1,7 +1,6 @@
 use std::{net::SocketAddr, str::FromStr};
 
 use scylla_proxy::{Node, Proxy, ShardAwareness};
-use tracing::instrument::WithSubscriber;
 
 fn init_logger() {
     tracing_subscriber::fmt::fmt()
@@ -30,7 +29,7 @@ async fn main() {
                 .build(),
         )
         .build();
-    let running_proxy = proxy.run().with_current_subscriber().await.unwrap();
+    let running_proxy = proxy.run().await.unwrap();
 
     pause().await;
     running_proxy.finish().await.unwrap();

scylla-proxy/examples/identity_shard_aware_proxy.rs

@@ -1,7 +1,6 @@
 use std::{net::SocketAddr, str::FromStr};
 
 use scylla_proxy::{Node, Proxy, ShardAwareness};
-use tracing::instrument::WithSubscriber;
 
 fn init_logger() {
     tracing_subscriber::fmt::fmt()
@@ -27,7 +26,7 @@ async fn main() {
         None,
         None,
     )]);
-    let running_proxy = proxy.run().with_current_subscriber().await.unwrap();
+    let running_proxy = proxy.run().await.unwrap();
 
     pause().await;
     running_proxy.finish().await.unwrap();

scylla/src/transport/cluster.rs

@@ -24,7 +24,6 @@ use std::collections::{HashMap, HashSet};
 use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::Duration;
-use tracing::instrument::WithSubscriber;
 use tracing::{debug, warn};
 use uuid::Uuid;
 
@@ -206,7 +205,7 @@ impl Cluster {
         };
 
         let (fut, worker_handle) = worker.work().remote_handle();
-        tokio::spawn(fut.with_current_subscriber());
+        tokio::spawn(fut);
 
         let result = Cluster {
             data: cluster_data,
@@ -647,7 +646,7 @@ impl ClusterWorker {
 
                             let cluster_data = self.cluster_data.load_full();
                             let use_keyspace_future = Self::handle_use_keyspace_request(cluster_data, request);
-                            tokio::spawn(use_keyspace_future.with_current_subscriber());
+                            tokio::spawn(use_keyspace_future);
                         },
                         None => return, // If use_keyspace_channel was closed then cluster was dropped, we can stop working
                     }

scylla/src/transport/connection.rs

@@ -13,7 +13,6 @@ use tokio::io::{split, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader, BufWrite
 use tokio::net::{TcpSocket, TcpStream};
 use tokio::sync::{mpsc, oneshot};
 use tokio::time::Instant;
-use tracing::instrument::WithSubscriber;
 use tracing::{debug, error, trace, warn};
 use uuid::Uuid;
 
@@ -1090,7 +1089,7 @@ impl Connection {
                 node_address,
             )
             .remote_handle();
-            tokio::task::spawn(task.with_current_subscriber());
+            tokio::task::spawn(task);
             return Ok(handle);
         }
 
@@ -1104,7 +1103,7 @@ impl Connection {
             node_address,
         )
         .remote_handle();
-        tokio::task::spawn(task.with_current_subscriber());
+        tokio::task::spawn(task);
         Ok(handle)
     }
 

scylla/src/transport/connection_pool.rs

@@ -27,7 +27,6 @@ use std::sync::{Arc, RwLock, Weak};
 use std::time::Duration;
 
 use tokio::sync::{broadcast, mpsc, Notify};
-use tracing::instrument::WithSubscriber;
 use tracing::{debug, error, trace, warn};
 
 /// The target size of a per-node connection pool.
@@ -212,7 +211,7 @@ impl NodeConnectionPool {
 
         let conns = refiller.get_shared_connections();
         let (fut, refiller_handle) = refiller.run(use_keyspace_request_receiver).remote_handle();
-        tokio::spawn(fut.with_current_subscriber());
+        tokio::spawn(fut);
 
         Self {
             conns,
@@ -1138,17 +1137,14 @@ impl PoolRefiller {
             Err(QueryError::IoError(io_error.unwrap()))
         };
 
-        tokio::task::spawn(
-            async move {
-                let res = fut.await;
-                match &res {
-                    Ok(()) => debug!("[{}] Successfully changed current keyspace", address),
-                    Err(err) => warn!("[{}] Failed to change keyspace: {:?}", address, err),
-                }
-                let _ = response_sender.send(res);
+        tokio::task::spawn(async move {
+            let res = fut.await;
+            match &res {
+                Ok(()) => debug!("[{}] Successfully changed current keyspace", address),
+                Err(err) => warn!("[{}] Failed to change keyspace: {:?}", address, err),
             }
-            .with_current_subscriber(),
-        );
+            let _ = response_sender.send(res);
+        });
     }
 
     // Requires the keyspace to be set

scylla/src/transport/iterator.rs

@@ -16,7 +16,6 @@ use scylla_cql::types::serialize::row::SerializedValues;
 use std::result::Result;
 use thiserror::Error;
 use tokio::sync::mpsc;
-use tracing::instrument::WithSubscriber;
 
 use super::errors::QueryError;
 use super::execution_profile::ExecutionProfileInner;
@@ -387,7 +386,7 @@ impl RowIterator {
         worker_task: impl Future<Output = PageSendAttemptedProof> + Send + 'static,
         mut receiver: mpsc::Receiver<Result<ReceivedPage, QueryError>>,
     ) -> Result<RowIterator, QueryError> {
-        tokio::task::spawn(worker_task.with_current_subscriber());
+        tokio::task::spawn(worker_task);
 
         // This unwrap is safe because:
         // - The future returned by worker.work sends at least one item

scylla/src/transport/load_balancing/default.rs

@@ -2269,7 +2269,7 @@ mod latency_awareness {
     use itertools::Either;
     use scylla_cql::errors::{DbError, QueryError};
     use tokio::time::{Duration, Instant};
-    use tracing::{instrument::WithSubscriber, trace, warn};
+    use tracing::{trace, warn};
     use uuid::Uuid;
 
     use crate::{load_balancing::NodeRef, routing::Shard, transport::node::Node};
@@ -2454,7 +2454,7 @@ mod latency_awareness {
                 }
             }
             .remote_handle();
-            tokio::task::spawn(updater_fut.with_current_subscriber());
+            tokio::task::spawn(updater_fut);
 
             Self {
                 _updater_handle: Some(updater_handle),

scylla/tests/integration/utils.rs

@@ -3,7 +3,6 @@ use std::collections::HashMap;
 use std::env;
 use std::net::SocketAddr;
 use std::str::FromStr;
-use tracing::instrument::WithSubscriber;
 
 use scylla_proxy::{Node, Proxy, ProxyError, RunningProxy, ShardAwareness};
 
@@ -53,7 +52,7 @@ where
     );
 
     let translation_map = proxy.translation_map();
-    let running_proxy = proxy.run().with_current_subscriber().await.unwrap();
+    let running_proxy = proxy.run().await.unwrap();
 
     let running_proxy = test(
         [proxy1_uri, proxy2_uri, proxy3_uri],