Repository: scylladb/scylla-rust-driver
Pull request #199 (2021-03-18 03:20:06)

CQL Tracing
## Description
This PR adds support for CQL tracing.
Tracing can be used like this:
```rust
// Create a simple query and enable tracing for it
let mut query: Query = Query::new("SELECT peer from system.peers".to_string());
query.set_tracing(true);

// QueryResult will contain a tracing_id which can be used to query tracing information
let query_result: QueryResult = session.query(query.clone(), &[]).await?;
let query_tracing_id: Uuid = query_result
    .tracing_id
    .ok_or_else(|| anyhow!("Tracing id is None!"))?;

// Get tracing information for this query and print it
let tracing_info: TracingInfo = session.get_tracing_info(&query_tracing_id).await?;
println!("Query tracing info: {:#?}\n", tracing_info);
```
Queries that support tracing are:
* `Session::query()`
* `Session::prepare()`
* `Session::execute()`
* `Session::batch()`
* `Session::query_iter()`
* `Session::execute_iter()`

~~In the [specification](https://github.com/apache/cassandra/blob/28e8632c9793f8d77247fc40f2c5003810df16f9/doc/native_protocol_v4.spec#L116) it's written that only QUERY, PREPARE and EXECUTE queries support tracing so `Session::batch()` doesn't support it.~~

## Notes
`TracingInfo` contains columns from `system_traces` that are the same in Cassandra and Scylla. Scylla has some scylla specific entries that the user can query themselves if they really need to.

Retrying tracing info queries and its configuration is based on [DataStax Java Driver](https://docs.datastax.com/en/developer/java-driver/4.0/manual/core/tracing/)

## Fixed issues
Fixes: #189

## Pre-review checklist

<!--
    Make sure you took care of the issues on the list.
    Put 'x' into those boxes which apply.
    You can also create the PR now and click on all relevant checkboxes.
    See CONTRIBUTING.md for more details.
-->

- [x] I have split my patch into logically separate commits.
- [x] All commit messages clearly explain what they change and why.
- [x] I added relevant tests for new features and bug fixes.
- [x] All commits compile, pass static checks and pass test.
- [x] PR description sums up the changes and reasons why they should be introduced.
- [x] I added appropriate `Fixes:` annotations to PR description.


cvybhu: The tests pass on my machine but fail in CI for some reason 0_o. Gonna investigate.
psarna: Probably related to timing - tracing data ends up in the tracing tables asynchronously, so you might need to try fetching the appropriate tracing id in a loop.

By the way, contrary to the protocol spec Scylla does support tracing batches, it's an important debugging mechanism, so we'd want to eventually support it in the driver too.
cvybhu: > Probably related to timing - tracing data ends up in the tracing tables asynchronously, so you might need to try fetching the appropriate tracing id in a loop.

:+1: Added retrying in tests, also Cassandra sometimes had a `null` in the table which the code handled incorrectly, all fixed now.

> By the way, contrary to the protocol spec Scylla does support tracing batches, it's an important debugging mechanism, so we'd want to eventually support it in the driver too.

Added

cvybhu: @piodul @psarna Rebased and ready for review
psarna: @cvybhu Looks good, but please also rebase on top of /main - after merging TLS, a small manual translation of this layer to `QueryResult` is in order.

Patch:
diff --git a/README.md b/README.md
index 1b7819df..a65dd57f 100644
--- a/README.md
+++ b/README.md
@@ -14,7 +14,7 @@ let uri = "127.0.0.1:9042";
 
 let session: Session = SessionBuilder::new().known_node(uri).build().await?;
 
-if let Some(rows) = session.query("SELECT a, b, c FROM ks.t", &[]).await? {
+if let Some(rows) = session.query("SELECT a, b, c FROM ks.t", &[]).await?.rows {
     for row in rows.into_typed::<(i32, i32, String)>() {
         let (a, b, c) = row?;
         println!("a, b, c: {}, {}, {}", a, b, c);
diff --git a/examples/Cargo.toml b/examples/Cargo.toml
index 6eae04cf..accb9a53 100644
--- a/examples/Cargo.toml
+++ b/examples/Cargo.toml
@@ -14,6 +14,7 @@ tokio = {version = "1.1.0", features = ["full"]}
 tracing = "0.1.25"
 tracing-subscriber = "0.2.16"
 chrono = "0.4"
+uuid = "0.8.1"
 
 [[example]]
 name = "basic"
@@ -54,3 +55,8 @@ path = "user-defined-type.rs"
 [[example]]
 name = "cql-time-types"
 path = "cql-time-types.rs"
+
+[[example]]
+name = "tracing"
+path = "tracing.rs"
+
diff --git a/examples/basic.rs b/examples/basic.rs
index 732fc602..e31c3141 100644
--- a/examples/basic.rs
+++ b/examples/basic.rs
@@ -44,7 +44,7 @@ async fn main() -> Result<()> {
         .await?;
 
     // Rows can be parsed as tuples
-    if let Some(rows) = session.query("SELECT a, b, c FROM ks.t", &[]).await? {
+    if let Some(rows) = session.query("SELECT a, b, c FROM ks.t", &[]).await?.rows {
         for row in rows.into_typed::<(i32, i32, String)>() {
             let (a, b, c) = row?;
             println!("a, b, c: {}, {}, {}", a, b, c);
@@ -59,7 +59,7 @@ async fn main() -> Result<()> {
         c: String,
     }
 
-    if let Some(rows) = session.query("SELECT a, b, c FROM ks.t", &[]).await? {
+    if let Some(rows) = session.query("SELECT a, b, c FROM ks.t", &[]).await?.rows {
         for row_data in rows.into_typed::<RowData>() {
             let row_data = row_data?;
             println!("row_data: {:?}", row_data);
@@ -67,7 +67,7 @@ async fn main() -> Result<()> {
     }
 
     // Or simply as untyped rows
-    if let Some(rows) = session.query("SELECT a, b, c FROM ks.t", &[]).await? {
+    if let Some(rows) = session.query("SELECT a, b, c FROM ks.t", &[]).await?.rows {
         for row in rows {
             let a = row.columns[0].as_ref().unwrap().as_int().unwrap();
             let b = row.columns[1].as_ref().unwrap().as_int().unwrap();
diff --git a/examples/compare-tokens.rs b/examples/compare-tokens.rs
index 467b1f1e..d2f7ad04 100644
--- a/examples/compare-tokens.rs
+++ b/examples/compare-tokens.rs
@@ -34,6 +34,7 @@ async fn main() -> Result<()> {
         let qt = session
             .query(format!("SELECT token(pk) FROM ks.t where pk = {}", pk), &[])
             .await?
+            .rows
             .unwrap()
             .get(0)
             .expect("token query no rows!")
diff --git a/examples/cql-time-types.rs b/examples/cql-time-types.rs
index 20a542bd..46b7a2d6 100644
--- a/examples/cql-time-types.rs
+++ b/examples/cql-time-types.rs
@@ -36,7 +36,7 @@ async fn main() -> Result<()> {
         .query("INSERT INTO ks.dates (d) VALUES (?)", (example_date,))
         .await?;
 
-    if let Some(rows) = session.query("SELECT d from ks.dates", &[]).await? {
+    if let Some(rows) = session.query("SELECT d from ks.dates", &[]).await?.rows {
         for row in rows.into_typed::<(NaiveDate,)>() {
             let (read_date,): (NaiveDate,) = match row {
                 Ok(read_date) => read_date,
@@ -53,7 +53,7 @@ async fn main() -> Result<()> {
         .query("INSERT INTO ks.dates (d) VALUES (?)", (example_big_date,))
         .await?;
 
-    if let Some(rows) = session.query("SELECT d from ks.dates", &[]).await? {
+    if let Some(rows) = session.query("SELECT d from ks.dates", &[]).await?.rows {
         for row in rows {
             let read_days: u32 = match row.columns[0] {
                 Some(CqlValue::Date(days)) => days,
@@ -79,7 +79,7 @@ async fn main() -> Result<()> {
         .query("INSERT INTO ks.times (t) VALUES (?)", (Time(example_time),))
         .await?;
 
-    if let Some(rows) = session.query("SELECT t from ks.times", &[]).await? {
+    if let Some(rows) = session.query("SELECT t from ks.times", &[]).await?.rows {
         for row in rows.into_typed::<(Duration,)>() {
             let (read_time,): (Duration,) = row?;
 
@@ -105,7 +105,11 @@ async fn main() -> Result<()> {
         )
         .await?;
 
-    if let Some(rows) = session.query("SELECT t from ks.timestamps", &[]).await? {
+    if let Some(rows) = session
+        .query("SELECT t from ks.timestamps", &[])
+        .await?
+        .rows
+    {
         for row in rows.into_typed::<(Duration,)>() {
             let (read_time,): (Duration,) = row?;
 
diff --git a/examples/tls.rs b/examples/tls.rs
index 65907285..d581409a 100644
--- a/examples/tls.rs
+++ b/examples/tls.rs
@@ -80,7 +80,7 @@ async fn main() -> Result<()> {
         .await?;
 
     // Rows can be parsed as tuples
-    if let Some(rows) = session.query("SELECT a, b, c FROM ks.t", &[]).await? {
+    if let Some(rows) = session.query("SELECT a, b, c FROM ks.t", &[]).await?.rows {
         for row in rows.into_typed::<(i32, i32, String)>() {
             let (a, b, c) = row?;
             println!("a, b, c: {}, {}, {}", a, b, c);
diff --git a/examples/tracing.rs b/examples/tracing.rs
new file mode 100644
index 00000000..652e010e
--- /dev/null
+++ b/examples/tracing.rs
@@ -0,0 +1,105 @@
+// CQL Tracing allows to see each step during execution of a query
+// query() prepare() execute() batch() query_iter() and execute_iter() can be traced
+
+use anyhow::{anyhow, Result};
+use futures::StreamExt;
+use scylla::batch::Batch;
+use scylla::statement::{prepared_statement::PreparedStatement, query::Query, Consistency};
+use scylla::tracing::{GetTracingConfig, TracingInfo};
+use scylla::transport::iterator::RowIterator;
+use scylla::{BatchResult, QueryResult};
+use scylla::{Session, SessionBuilder};
+use std::env;
+use std::num::NonZeroU32;
+use std::time::Duration;
+use uuid::Uuid;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let uri = env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+
+    println!("Connecting to {} ...", uri);
+    let session: Session = SessionBuilder::new().known_node(uri).build().await?;
+
+    session.query("CREATE KEYSPACE IF NOT EXISTS ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[]).await?;
+
+    session
+        .query(
+            "CREATE TABLE IF NOT EXISTS ks.tracing_example (val text primary key)",
+            &[],
+        )
+        .await?;
+
+    // QUERY
+    // Create a simple query and enable tracing for it
+    let mut query: Query = Query::new("SELECT val from ks.tracing_example".to_string());
+    query.set_tracing(true);
+
+    // QueryResult will contain a tracing_id which can be used to query tracing information
+    let query_result: QueryResult = session.query(query.clone(), &[]).await?;
+    let query_tracing_id: Uuid = query_result
+        .tracing_id
+        .ok_or_else(|| anyhow!("Tracing id is None!"))?;
+
+    // Get tracing information for this query and print it
+    let tracing_info: TracingInfo = session.get_tracing_info(&query_tracing_id).await?;
+    println!("Query tracing info: {:#?}\n", tracing_info);
+
+    // PREPARE
+    // Now prepare a query - query to be prepared has tracing set so the prepare will be traced
+    let mut prepared: PreparedStatement = session.prepare(query.clone()).await?;
+
+    // prepared.prepare_tracing_id contains tracing ids of all prepare requests
+    let prepare_ids: &Vec<Uuid> = &prepared.prepare_tracing_ids;
+    println!("Prepare tracing ids: {:?}\n", prepare_ids);
+
+    // EXECUTE
+    // To trace execution of a prepared statement tracing must be enabled for it
+    prepared.set_tracing(true);
+
+    let execute_result: QueryResult = session.execute(&prepared, &[]).await?;
+    println!("Execute tracing id: {:?}", execute_result.tracing_id);
+
+    // PAGED QUERY_ITER EXECUTE_ITER
+    // It's also possible to trace paged queries like query_iter or execute_iter
+    // After iterating through all rows iterator.get_tracing_ids() will give tracing ids
+    // for all page queries
+    let mut row_iterator: RowIterator = session.query_iter(query, &[]).await?;
+
+    while let Some(_row) = row_iterator.next().await {
+        // Receive rows
+    }
+
+    // Now print tracing ids for all page queries:
+    println!(
+        "Paged row iterator tracing ids: {:?}\n",
+        row_iterator.get_tracing_ids()
+    );
+
+    // BATCH
+    // Create a simple batch and enable tracing
+    let mut batch: Batch = Batch::default();
+    batch.append_statement("INSERT INTO ks.tracing_example (val) VALUES('val')");
+    batch.set_tracing(true);
+
+    // Run the batch and print its tracing_id
+    let batch_result: BatchResult = session.batch(&batch, ((),)).await?;
+    println!("Batch tracing id: {:?}\n", batch_result.tracing_id);
+
+    // CUSTOM
+    // GetTracingConfig allows to specify a custom settings for querying tracing info
+    // Tracing info might not immediately be available on queried node
+    // so the driver performs a few attempts with sleeps in between.
+
+    let custom_config: GetTracingConfig = GetTracingConfig {
+        attempts: NonZeroU32::new(8).unwrap(),
+        interval: Duration::from_millis(100),
+        consistency: Consistency::One,
+    };
+
+    let _custom_info: TracingInfo = session
+        .get_tracing_info_custom(&query_tracing_id, &custom_config)
+        .await?;
+
+    Ok(())
+}
diff --git a/examples/user-defined-type.rs b/examples/user-defined-type.rs
index 6f04d938..d369fdd4 100644
--- a/examples/user-defined-type.rs
+++ b/examples/user-defined-type.rs
@@ -47,7 +47,7 @@ async fn main() -> Result<()> {
         .await?;
 
     // And read like any normal value
-    if let Some(rows) = session.query("SELECT my FROM ks.udt_tab", &[]).await? {
+    if let Some(rows) = session.query("SELECT my FROM ks.udt_tab", &[]).await?.rows {
         for row in rows.into_typed::<(MyType,)>() {
             let (my_val,) = row?;
             println!("{:?}", my_val)
diff --git a/scylla/src/frame/mod.rs b/scylla/src/frame/mod.rs
index 49337cbe..9ea6373c 100644
--- a/scylla/src/frame/mod.rs
+++ b/scylla/src/frame/mod.rs
@@ -124,6 +124,7 @@ pub struct RequestBodyWithExtensions {
 pub fn prepare_request_body_with_extensions(
     body_with_ext: RequestBodyWithExtensions,
     compression: Option<Compression>,
+    tracing: bool,
 ) -> Result<(u8, Bytes), FrameError> {
     let mut flags = 0;
 
@@ -133,6 +134,10 @@ pub fn prepare_request_body_with_extensions(
         body = compress(&body, compression)?.into();
     }
 
+    if tracing {
+        flags |= FLAG_TRACING;
+    }
+
     Ok((flags, body))
 }
 
diff --git a/scylla/src/lib.rs b/scylla/src/lib.rs
index d6fb6d33..38fd0059 100644
--- a/scylla/src/lib.rs
+++ b/scylla/src/lib.rs
@@ -6,6 +6,7 @@ pub mod macros;
 pub mod frame;
 pub mod routing;
 pub mod statement;
+pub mod tracing;
 pub mod transport;
 
 pub use macros::*;
@@ -15,5 +16,6 @@ pub use statement::query;
 
 pub use frame::response::cql_to_rust;
 
+pub use transport::connection::{BatchResult, QueryResult};
 pub use transport::session::{IntoTypedRows, Session, SessionConfig};
 pub use transport::session_builder::SessionBuilder;
diff --git a/scylla/src/statement/batch.rs b/scylla/src/statement/batch.rs
index 1b0c3780..9d678160 100644
--- a/scylla/src/statement/batch.rs
+++ b/scylla/src/statement/batch.rs
@@ -14,6 +14,7 @@ pub struct Batch {
     pub serial_consistency: Option<Consistency>,
     pub is_idempotent: bool,
     pub retry_policy: Option<Box<dyn RetryPolicy + Send + Sync>>,
+    pub tracing: bool,
 }
 
 impl Batch {
@@ -86,6 +87,18 @@ impl Batch {
     pub fn get_retry_policy(&self) -> &Option<Box<dyn RetryPolicy + Send + Sync>> {
         &self.retry_policy
     }
+
+    /// Enable or disable CQL Tracing for this batch  
+    /// If enabled session.batch() will return a BatchResult containing tracing_id
+    /// which can be used to query tracing information about the execution of this query
+    pub fn set_tracing(&mut self, should_trace: bool) {
+        self.tracing = should_trace;
+    }
+
+    /// Gets whether tracing is enabled for this batch
+    pub fn get_tracing(&self) -> bool {
+        self.tracing
+    }
 }
 
 impl Default for Batch {
@@ -97,6 +110,7 @@ impl Default for Batch {
             serial_consistency: None,
             is_idempotent: false,
             retry_policy: None,
+            tracing: false,
         }
     }
 }
@@ -138,6 +152,7 @@ impl Clone for Batch {
                 .retry_policy
                 .as_ref()
                 .map(|policy| policy.clone_boxed()),
+            tracing: self.tracing,
         }
     }
 }
diff --git a/scylla/src/statement/prepared_statement.rs b/scylla/src/statement/prepared_statement.rs
index 07004786..7765e7f3 100644
--- a/scylla/src/statement/prepared_statement.rs
+++ b/scylla/src/statement/prepared_statement.rs
@@ -5,17 +5,20 @@ use crate::transport::retry_policy::RetryPolicy;
 use bytes::{BufMut, Bytes, BytesMut};
 use std::convert::TryInto;
 use thiserror::Error;
+use uuid::Uuid;
 
 /// Represents a statement prepared on the server.
 pub struct PreparedStatement {
     id: Bytes,
     metadata: PreparedMetadata,
     statement: String,
+    pub prepare_tracing_ids: Vec<Uuid>,
     page_size: Option<i32>,
     pub consistency: Consistency,
     pub serial_consistency: Option<Consistency>,
     pub is_idempotent: bool,
     pub retry_policy: Option<Box<dyn RetryPolicy + Send + Sync>>,
+    pub tracing: bool,
 }
 
 impl PreparedStatement {
@@ -24,11 +27,13 @@ impl PreparedStatement {
             id,
             metadata,
             statement,
+            prepare_tracing_ids: Vec::new(),
             page_size: None,
             consistency: Default::default(),
             serial_consistency: None,
             is_idempotent: false,
             retry_policy: None,
+            tracing: false,
         }
     }
 
@@ -103,6 +108,23 @@ impl PreparedStatement {
         &self.retry_policy
     }
 
+    /// Enable or disable CQL Tracing for this prepared statement  
+    /// If enabled session.execute() will return QueryResult containing tracing_id
+    /// which can be used to query tracing information about the execution of this statement
+    pub fn set_tracing(&mut self, should_trace: bool) {
+        self.tracing = should_trace;
+    }
+
+    /// Gets whether tracing is enabled for this prepared statement
+    pub fn get_tracing(&self) -> bool {
+        self.tracing
+    }
+
+    /// Gets tracing ids of queries used to prepare this statement
+    pub fn get_prepare_tracing_ids(&self) -> &[Uuid] {
+        &self.prepare_tracing_ids
+    }
+
     /// Computes the partition key of the target table from given values
     /// Partition keys have a specific serialization rules.
     /// Ref: https://github.com/scylladb/scylla/blob/40adf38915b6d8f5314c621a94d694d172360833/compound_compat.hh#L33-L47
@@ -181,6 +203,7 @@ impl Clone for PreparedStatement {
             id: self.id.clone(),
             metadata: self.metadata.clone(),
             statement: self.statement.clone(),
+            prepare_tracing_ids: self.prepare_tracing_ids.clone(),
             page_size: self.page_size,
             consistency: self.consistency,
             serial_consistency: self.serial_consistency,
@@ -189,6 +212,7 @@ impl Clone for PreparedStatement {
                 .retry_policy
                 .as_ref()
                 .map(|policy| policy.clone_boxed()),
+            tracing: self.tracing,
         }
     }
 }
diff --git a/scylla/src/statement/query.rs b/scylla/src/statement/query.rs
index 97ba6581..2ca3ac04 100644
--- a/scylla/src/statement/query.rs
+++ b/scylla/src/statement/query.rs
@@ -11,6 +11,7 @@ pub struct Query {
     pub serial_consistency: Option<Consistency>,
     pub is_idempotent: bool,
     pub retry_policy: Option<Box<dyn RetryPolicy + Send + Sync>>,
+    pub tracing: bool,
 }
 
 impl Query {
@@ -23,6 +24,7 @@ impl Query {
             serial_consistency: None,
             is_idempotent: false,
             retry_policy: None,
+            tracing: false,
         }
     }
 
@@ -93,6 +95,18 @@ impl Query {
     pub fn get_retry_policy(&self) -> &Option<Box<dyn RetryPolicy + Send + Sync>> {
         &self.retry_policy
     }
+
+    /// Enable or disable CQL Tracing for this query  
+    /// If enabled session.query() will return QueryResult containing tracing_id
+    /// which can be used to query tracing information about the execution of this query
+    pub fn set_tracing(&mut self, should_trace: bool) {
+        self.tracing = should_trace;
+    }
+
+    /// Gets whether tracing is enabled for this query
+    pub fn get_tracing(&self) -> bool {
+        self.tracing
+    }
 }
 
 impl From<String> for Query {
@@ -119,6 +133,7 @@ impl Clone for Query {
                 .retry_policy
                 .as_ref()
                 .map(|policy| policy.clone_boxed()),
+            tracing: self.tracing,
         }
     }
 }
diff --git a/scylla/src/tracing.rs b/scylla/src/tracing.rs
new file mode 100644
index 00000000..63822b23
--- /dev/null
+++ b/scylla/src/tracing.rs
@@ -0,0 +1,117 @@
+use crate::statement::Consistency;
+use std::collections::HashMap;
+use std::net::IpAddr;
+use std::num::NonZeroU32;
+use std::time::Duration;
+use uuid::Uuid;
+
+use crate::cql_to_rust::{FromRow, FromRowError};
+use crate::frame::response::result::Row;
+
+/// Tracing info retrieved from `system_traces.sessions`
+/// with all events from `system_traces.events`
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct TracingInfo {
+    pub client: Option<IpAddr>,
+    pub command: Option<String>,
+    pub coordinator: Option<IpAddr>,
+    pub duration: Option<i32>,
+    pub parameters: Option<HashMap<String, String>>,
+    pub request: Option<String>,
+    /// started_at is a timestamp - time since unix epoch
+    pub started_at: Option<chrono::Duration>,
+
+    pub events: Vec<TracingEvent>,
+}
+
+/// A single event happening during a traced query
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct TracingEvent {
+    pub event_id: Uuid,
+    pub activity: Option<String>,
+    pub source: Option<IpAddr>,
+    pub source_elapsed: Option<i32>,
+    pub thread: Option<String>,
+}
+
+/// Used to configure a custom retry strategy when querying tracing info
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct GetTracingConfig {
+    /// Number of attempts to be made before giving up.
+    /// Default value: 5
+    pub attempts: NonZeroU32,
+    /// Interval to wait between each attempt.
+    /// Default value: 3 milliseconds
+    pub interval: Duration,
+    /// Consistency to use in queries that read TracingInfo.
+    /// Default value: One
+    pub consistency: Consistency,
+}
+
+impl Default for GetTracingConfig {
+    fn default() -> GetTracingConfig {
+        GetTracingConfig {
+            attempts: NonZeroU32::new(5).unwrap(),
+            interval: Duration::from_millis(3),
+            consistency: Consistency::One,
+        }
+    }
+}
+
+// A query used to query TracingInfo from system_traces.sessions
+pub(crate) const TRACES_SESSION_QUERY_STR: &str =
+    "SELECT client, command, coordinator, duration, parameters, request, started_at \
+    FROM system_traces.sessions WHERE session_id = ?";
+
+// A query used to query TracingEvent from system_traces.events
+pub(crate) const TRACES_EVENTS_QUERY_STR: &str =
+    "SELECT event_id, activity, source, source_elapsed, thread \
+    FROM system_traces.events WHERE session_id = ?";
+
+// Converts a row received by performing TRACES_SESSION_QUERY_STR to TracingInfo
+impl FromRow for TracingInfo {
+    fn from_row(row: Row) -> Result<TracingInfo, FromRowError> {
+        let (client, command, coordinator, duration, parameters, request, started_at) =
+            <(
+                Option<IpAddr>,
+                Option<String>,
+                Option<IpAddr>,
+                Option<i32>,
+                Option<HashMap<String, String>>,
+                Option<String>,
+                Option<chrono::Duration>,
+            )>::from_row(row)?;
+
+        Ok(TracingInfo {
+            client,
+            command,
+            coordinator,
+            duration,
+            parameters,
+            request,
+            started_at,
+            events: Vec::new(),
+        })
+    }
+}
+
+// Converts a row received by performing TRACES_SESSION_QUERY_STR to TracingInfo
+impl FromRow for TracingEvent {
+    fn from_row(row: Row) -> Result<TracingEvent, FromRowError> {
+        let (event_id, activity, source, source_elapsed, thread) = <(
+            Uuid,
+            Option<String>,
+            Option<IpAddr>,
+            Option<i32>,
+            Option<String>,
+        )>::from_row(row)?;
+
+        Ok(TracingEvent {
+            event_id,
+            activity,
+            source,
+            source_elapsed,
+            thread,
+        })
+    }
+}
diff --git a/scylla/src/transport/connection.rs b/scylla/src/transport/connection.rs
index d6489299..36c7227e 100644
--- a/scylla/src/transport/connection.rs
+++ b/scylla/src/transport/connection.rs
@@ -4,6 +4,7 @@ use tokio::io::{split, AsyncRead, AsyncWrite};
 use tokio::net::{TcpSocket, TcpStream};
 use tokio::sync::{mpsc, oneshot};
 use tracing::{error, warn};
+use uuid::Uuid;
 
 #[cfg(feature = "ssl")]
 use openssl::ssl::{Ssl, SslContext};
@@ -63,6 +64,53 @@ struct TaskResponse {
     body: Bytes,
 }
 
+pub struct QueryResponse {
+    pub response: Response,
+    pub tracing_id: Option<Uuid>,
+    pub warnings: Vec<String>,
+}
+
+/// Result of a single query  
+/// Contains all rows returned by the database and some more information
+#[derive(Default, Debug)]
+pub struct QueryResult {
+    /// Rows returned by the database
+    pub rows: Option<Vec<result::Row>>,
+    /// Warnings returned by the database
+    pub warnings: Vec<String>,
+    /// CQL Tracing uuid - can only be Some if tracing is enabled for this query
+    pub tracing_id: Option<Uuid>,
+}
+
+/// Result of Session::batch(). Contains no rows, only some useful information.
+pub struct BatchResult {
+    /// Warnings returned by the database
+    pub warnings: Vec<String>,
+    /// CQL Tracing uuid - can only be Some if tracing is enabled for this batch
+    pub tracing_id: Option<Uuid>,
+}
+
+impl QueryResponse {
+    pub fn into_query_result(self) -> Result<QueryResult, QueryError> {
+        let rows: Option<Vec<result::Row>> = match self.response {
+            Response::Error(err) => return Err(err.into()),
+            Response::Result(result::Result::Rows(rs)) => Some(rs.rows),
+            Response::Result(_) => None,
+            _ => {
+                return Err(QueryError::ProtocolError(
+                    "Unexpected server response, expected Result or Error",
+                ))
+            }
+        };
+
+        Ok(QueryResult {
+            rows,
+            warnings: self.warnings,
+            tracing_id: self.tracing_id,
+        })
+    }
+}
+
 #[derive(Clone)]
 pub struct ConnectionConfig {
     pub compression: Option<Compression>,
@@ -133,52 +181,64 @@ impl Connection {
     }
 
     pub async fn startup(&self, options: HashMap<String, String>) -> Result<Response, QueryError> {
-        self.send_request(&request::Startup { options }, false)
-            .await
+        Ok(self
+            .send_request(&request::Startup { options }, false, false)
+            .await?
+            .response)
     }
 
     pub async fn get_options(&self) -> Result<Response, QueryError> {
-        self.send_request(&request::Options {}, false).await
-    }
+        Ok(self
+            .send_request(&request::Options {}, false, false)
+            .await?
+            .response)
+    }
+
+    pub async fn prepare(&self, query: &Query) -> Result<PreparedStatement, QueryError> {
+        let query_response = self
+            .send_request(
+                &request::Prepare {
+                    query: &query.get_contents(),
+                },
+                true,
+                query.tracing,
+            )
+            .await?;
 
-    pub async fn prepare(&self, query: &str) -> Result<PreparedStatement, QueryError> {
-        let result = self.send_request(&request::Prepare { query }, true).await?;
-        match result {
-            Response::Error(err) => Err(err.into()),
-            Response::Result(result::Result::Prepared(p)) => Ok(PreparedStatement::new(
-                p.id,
-                p.prepared_metadata,
-                query.to_owned(),
-            )),
-            _ => Err(QueryError::ProtocolError(
-                "PREPARE: Unexpected server response",
-            )),
+        let mut prepared_statement = match query_response.response {
+            Response::Error(err) => return Err(err.into()),
+            Response::Result(result::Result::Prepared(p)) => {
+                PreparedStatement::new(p.id, p.prepared_metadata, query.get_contents().to_owned())
+            }
+            _ => {
+                return Err(QueryError::ProtocolError(
+                    "PREPARE: Unexpected server response",
+                ))
+            }
+        };
+
+        if let Some(tracing_id) = query_response.tracing_id {
+            prepared_statement.prepare_tracing_ids.push(tracing_id);
         }
+
+        Ok(prepared_statement)
     }
 
     pub async fn query_single_page(
         &self,
         query: impl Into<Query>,
         values: impl ValueList,
-    ) -> Result<Option<Vec<result::Row>>, QueryError> {
+    ) -> Result<QueryResult, QueryError> {
         let query: Query = query.into();
-        self.query_single_page_by_ref(&query, &values).await
+        self.query(&query, &values, None).await?.into_query_result()
     }
 
     pub async fn query_single_page_by_ref(
         &self,
         query: &Query,
         values: &impl ValueList,
-    ) -> Result<Option<Vec<result::Row>>, QueryError> {
-        let result = self.query(query, values, None).await?;
-        match result {
-            Response::Error(err) => Err(err.into()),
-            Response::Result(result::Result::Rows(rs)) => Ok(Some(rs.rows)),
-            Response::Result(_) => Ok(None),
-            _ => Err(QueryError::ProtocolError(
-                "QUERY: Unexpected server response",
-            )),
-        }
+    ) -> Result<QueryResult, QueryError> {
+        self.query(query, values, None).await?.into_query_result()
     }
 
     pub async fn query(
@@ -186,7 +246,7 @@ impl Connection {
         query: &Query,
         values: impl ValueList,
         paging_state: Option<Bytes>,
-    ) -> Result<Response, QueryError> {
+    ) -> Result<QueryResponse, QueryError> {
         let serialized_values = values.serialized()?;
 
         let query_frame = query::Query {
@@ -200,24 +260,17 @@ impl Connection {
             },
         };
 
-        self.send_request(&query_frame, true).await
+        self.send_request(&query_frame, true, query.tracing).await
     }
 
     pub async fn execute_single_page(
         &self,
         prepared_statement: &PreparedStatement,
         values: impl ValueList,
-    ) -> Result<Option<Vec<result::Row>>, QueryError> {
-        let response = self.execute(prepared_statement, values, None).await?;
-
-        match response {
-            Response::Error(err) => Err(err.into()),
-            Response::Result(result::Result::Rows(rs)) => Ok(Some(rs.rows)),
-            Response::Result(_) => Ok(None),
-            _ => Err(QueryError::ProtocolError(
-                "EXECUTE: Unexpected server response",
-            )),
-        }
+    ) -> Result<QueryResult, QueryError> {
+        self.execute(prepared_statement, values, None)
+            .await?
+            .into_query_result()
     }
 
     pub async fn execute(
@@ -225,7 +278,7 @@ impl Connection {
         prepared_statement: &PreparedStatement,
         values: impl ValueList,
         paging_state: Option<Bytes>,
-    ) -> Result<Response, QueryError> {
+    ) -> Result<QueryResponse, QueryError> {
         let serialized_values = values.serialized()?;
 
         let execute_frame = execute::Execute {
@@ -239,12 +292,15 @@ impl Connection {
             },
         };
 
-        let response = self.send_request(&execute_frame, true).await?;
+        let query_response = self
+            .send_request(&execute_frame, true, prepared_statement.tracing)
+            .await?;
 
-        if let Response::Error(err) = &response {
+        if let Response::Error(err) = &query_response.response {
             if err.error == DbError::Unprepared {
                 // Repreparation of a statement is needed
-                let reprepared = self.prepare(prepared_statement.get_statement()).await?;
+                let reprepare_query: Query = prepared_statement.get_statement().into();
+                let reprepared = self.prepare(&reprepare_query).await?;
                 // Reprepared statement should keep its id - it's the md5 sum
                 // of statement contents
                 if reprepared.get_id() != prepared_statement.get_id() {
@@ -253,14 +309,20 @@ impl Connection {
                     ));
                 }
 
-                return self.send_request(&execute_frame, true).await;
+                return self
+                    .send_request(&execute_frame, true, prepared_statement.tracing)
+                    .await;
             }
         }
 
-        Ok(response)
+        Ok(query_response)
     }
 
-    pub async fn batch(&self, batch: &Batch, values: impl BatchValues) -> Result<(), QueryError> {
+    pub async fn batch(
+        &self,
+        batch: &Batch,
+        values: impl BatchValues,
+    ) -> Result<BatchResult, QueryError> {
         let statements_count = batch.get_statements().len();
         if statements_count != values.len() {
             return Err(QueryError::BadQuery(BadQuery::ValueLenMismatch(
@@ -287,11 +349,14 @@ impl Connection {
             serial_consistency: batch.get_serial_consistency(),
         };
 
-        let response = self.send_request(&batch_frame, true).await?;
+        let query_response = self.send_request(&batch_frame, true, batch.tracing).await?;
 
-        match response {
+        match query_response.response {
             Response::Error(err) => Err(err.into()),
-            Response::Result(_) => Ok(()),
+            Response::Result(_) => Ok(BatchResult {
+                warnings: query_response.warnings,
+                tracing_id: query_response.tracing_id,
+            }),
             _ => Err(QueryError::ProtocolError(
                 "BATCH: Unexpected server response",
             )),
@@ -311,7 +376,7 @@ impl Connection {
 
         let query_response = self.query(&query, (), None).await?;
 
-        match query_response {
+        match query_response.response {
             Response::Result(result::Result::SetKeyspace(set_keyspace)) => {
                 if set_keyspace.keyspace_name.to_lowercase()
                     != keyspace_name.as_str().to_lowercase()
@@ -334,7 +399,8 @@ impl Connection {
         &self,
         request: &R,
         compress: bool,
-    ) -> Result<Response, QueryError> {
+        tracing: bool,
+    ) -> Result<QueryResponse, QueryError> {
         let body = request.to_bytes()?;
         let compression = if compress {
             self.config.compression
@@ -344,7 +410,7 @@ impl Connection {
         let body_with_ext = RequestBodyWithExtensions { body };
 
         let (flags, raw_request) =
-            frame::prepare_request_body_with_extensions(body_with_ext, compression)?;
+            frame::prepare_request_body_with_extensions(body_with_ext, compression, tracing)?;
 
         let (sender, receiver) = oneshot::channel();
 
@@ -375,13 +441,17 @@ impl Connection {
             task_response.body,
         )?;
 
-        for warn_description in body_with_ext.warnings {
+        for warn_description in &body_with_ext.warnings {
             warn!(warning = warn_description.as_str());
         }
 
         let response = Response::deserialize(task_response.opcode, &mut &*body_with_ext.body)?;
 
-        Ok(response)
+        Ok(QueryResponse {
+            response,
+            warnings: body_with_ext.warnings,
+            tracing_id: body_with_ext.trace_id,
+        })
     }
 
     #[cfg(feature = "ssl")]
diff --git a/scylla/src/transport/iterator.rs b/scylla/src/transport/iterator.rs
index 4d776c20..4de08052 100644
--- a/scylla/src/transport/iterator.rs
+++ b/scylla/src/transport/iterator.rs
@@ -25,16 +25,23 @@ use crate::routing::Token;
 use crate::statement::Consistency;
 use crate::statement::{prepared_statement::PreparedStatement, query::Query};
 use crate::transport::cluster::ClusterData;
-use crate::transport::connection::Connection;
+use crate::transport::connection::{Connection, QueryResponse};
 use crate::transport::load_balancing::{LoadBalancingPolicy, Statement};
 use crate::transport::metrics::Metrics;
 use crate::transport::node::Node;
 use crate::transport::retry_policy::{QueryInfo, RetryDecision, RetrySession};
+use uuid::Uuid;
 
 pub struct RowIterator {
     current_row_idx: usize,
     current_page: Rows,
-    page_receiver: mpsc::Receiver<Result<Rows, QueryError>>,
+    page_receiver: mpsc::Receiver<Result<ReceivedPage, QueryError>>,
+    tracing_ids: Vec<Uuid>,
+}
+
+struct ReceivedPage {
+    pub rows: Rows,
+    pub tracing_id: Option<Uuid>,
 }
 
 impl Stream for RowIterator {
@@ -45,9 +52,13 @@ impl Stream for RowIterator {
 
         if s.is_current_page_exhausted() {
             match Pin::new(&mut s.page_receiver).poll_recv(cx) {
-                Poll::Ready(Some(Ok(rows))) => {
-                    s.current_page = rows;
+                Poll::Ready(Some(Ok(received_page))) => {
+                    s.current_page = received_page.rows;
                     s.current_row_idx = 0;
+
+                    if let Some(tracing_id) = received_page.tracing_id {
+                        s.tracing_ids.push(tracing_id);
+                    }
                 }
                 Poll::Ready(Some(Err(err))) => return Poll::Ready(Some(Err(err))),
                 Poll::Ready(None) => return Poll::Ready(None),
@@ -119,6 +130,7 @@ impl RowIterator {
             current_row_idx: 0,
             current_page: Default::default(),
             page_receiver: receiver,
+            tracing_ids: Vec::new(),
         }
     }
 
@@ -173,9 +185,15 @@ impl RowIterator {
             current_row_idx: 0,
             current_page: Default::default(),
             page_receiver: receiver,
+            tracing_ids: Vec::new(),
         }
     }
 
+    /// Returns tracing ids of all finished queries that had tracing enabled
+    pub fn get_tracing_ids(&self) -> &[Uuid] {
+        &self.tracing_ids
+    }
+
     fn is_current_page_exhausted(&self) -> bool {
         self.current_row_idx >= self.current_page.rows.len()
     }
@@ -184,14 +202,14 @@ impl RowIterator {
 // RowIteratorWorker works in the background to fetch pages
 // RowIterator receives them through a channel
 struct RowIteratorWorker<'a, ConnFunc, QueryFunc> {
-    sender: mpsc::Sender<Result<Rows, QueryError>>,
+    sender: mpsc::Sender<Result<ReceivedPage, QueryError>>,
 
     // Closure used to choose a connection from a node
     // AsyncFn(Arc<Node>) -> Result<Arc<Connection>, QueryError>
     choose_connection: ConnFunc,
 
     // Closure used to perform a single page query
-    // AsyncFn(Arc<Connection>, Option<Bytes>) -> Result<Response, QueryError>
+    // AsyncFn(Arc<Connection>, Option<Bytes>) -> Result<QueryResponse, QueryError>
     page_query: QueryFunc,
 
     statement_info: Statement<'a>,
@@ -210,7 +228,7 @@ where
     ConnFunc: Fn(Arc<Node>) -> ConnFut,
     ConnFut: Future<Output = Result<Arc<Connection>, QueryError>>,
     QueryFunc: Fn(Arc<Connection>, Option<Bytes>) -> QueryFut,
-    QueryFut: Future<Output = Result<Response, QueryError>>,
+    QueryFut: Future<Output = Result<QueryResponse, QueryError>>,
 {
     async fn work(mut self, cluster_data: Arc<ClusterData>) {
         let query_plan = self.load_balancer.plan(&self.statement_info, &cluster_data);
@@ -270,10 +288,10 @@ where
             self.metrics.inc_total_paged_queries();
             let query_start = std::time::Instant::now();
 
-            let response: Response =
+            let query_response: QueryResponse =
                 (self.page_query)(connection.clone(), self.paging_state.clone()).await?;
 
-            match response {
+            match query_response.response {
                 Response::Result(result::Result::Rows(mut rows)) => {
                     let _ = self
                         .metrics
@@ -281,8 +299,13 @@ where
 
                     self.paging_state = rows.metadata.paging_state.take();
 
+                    let received_page = ReceivedPage {
+                        rows,
+                        tracing_id: query_response.tracing_id,
+                    };
+
                     // Send next page to RowIterator
-                    if self.sender.send(Ok(rows)).await.is_err() {
+                    if self.sender.send(Ok(received_page)).await.is_err() {
                         // channel was closed, RowIterator was dropped - should shutdown
                         return Ok(());
                     }
@@ -316,6 +339,13 @@ pub struct TypedRowIterator<RowT> {
     phantom_data: std::marker::PhantomData<RowT>,
 }
 
+impl<RowT> TypedRowIterator<RowT> {
+    /// Returns tracing ids of all finished queries that had tracing enabled
+    pub fn get_tracing_ids(&self) -> &[Uuid] {
+        self.row_iterator.get_tracing_ids()
+    }
+}
+
 /// Couldn't get next typed row from the iterator
 #[derive(Error, Debug, Clone)]
 pub enum NextRowError {
diff --git a/scylla/src/transport/session.rs b/scylla/src/transport/session.rs
index 5bf1b7d6..a30c7787 100644
--- a/scylla/src/transport/session.rs
+++ b/scylla/src/transport/session.rs
@@ -4,6 +4,7 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 use tokio::net::lookup_host;
 use tracing::warn;
+use uuid::Uuid;
 
 use super::errors::{BadQuery, NewSessionError, QueryError};
 use crate::batch::Batch;
@@ -15,9 +16,10 @@ use crate::prepared_statement::{PartitionKeyError, PreparedStatement};
 use crate::query::Query;
 use crate::routing::{murmur3_token, Token};
 use crate::statement::Consistency;
+use crate::tracing::{GetTracingConfig, TracingEvent, TracingInfo};
 use crate::transport::{
     cluster::Cluster,
-    connection::{Connection, ConnectionConfig, VerifiedKeyspaceName},
+    connection::{BatchResult, Connection, ConnectionConfig, QueryResult, VerifiedKeyspaceName},
     iterator::RowIterator,
     load_balancing::{LoadBalancingPolicy, RoundRobinPolicy, Statement, TokenAwarePolicy},
     metrics::{Metrics, MetricsView},
@@ -278,7 +280,7 @@ impl Session {
         &self,
         query: impl Into<Query>,
         values: impl ValueList,
-    ) -> Result<Option<Vec<result::Row>>, QueryError> {
+    ) -> Result<QueryResult, QueryError> {
         let query: Query = query.into();
         let query_text: &str = query.get_contents();
         let serialized_values = values.serialized();
@@ -294,7 +296,7 @@ impl Session {
             return self
                 .use_keyspace(keyspace_name, case_sensitive)
                 .await
-                .map(|_| None);
+                .map(|_| QueryResult::default());
         }
 
         let retry_session = match &query.retry_policy {
@@ -348,11 +350,13 @@ impl Session {
     /// which can later be used to perform more efficient queries
     /// # Arguments
     ///#
-    pub async fn prepare(&self, query: &str) -> Result<PreparedStatement, QueryError> {
+    pub async fn prepare(&self, query: impl Into<Query>) -> Result<PreparedStatement, QueryError> {
+        let query: Query = query.into();
+
         let connections = self.cluster.get_working_connections().await?;
 
         // Prepare statements on all connections concurrently
-        let handles = connections.iter().map(|c| c.prepare(query));
+        let handles = connections.iter().map(|c| c.prepare(&query));
         let mut results = join_all(handles).await;
 
         // If at least one prepare was succesfull prepare returns Ok
@@ -370,7 +374,7 @@ impl Session {
             }
         }
 
-        let prepared: PreparedStatement = first_ok.unwrap()?;
+        let mut prepared: PreparedStatement = first_ok.unwrap()?;
 
         // Validate prepared ids equality
         for res in results {
@@ -380,6 +384,11 @@ impl Session {
                         "Prepared statement Ids differ, all should be equal",
                     ));
                 }
+
+                // Collect all tracing ids from prepare() queries in the final result
+                prepared
+                    .prepare_tracing_ids
+                    .extend(statement.prepare_tracing_ids);
             }
         }
 
@@ -395,7 +404,7 @@ impl Session {
         &self,
         prepared: &PreparedStatement,
         values: impl ValueList,
-    ) -> Result<Option<Vec<result::Row>>, QueryError> {
+    ) -> Result<QueryResult, QueryError> {
         let serialized_values = values.serialized()?;
         let values_ref = &serialized_values;
 
@@ -455,7 +464,11 @@ impl Session {
     ///
     /// * `batch` - batch to be performed
     /// * `values` - values bound to the query
-    pub async fn batch(&self, batch: &Batch, values: impl BatchValues) -> Result<(), QueryError> {
+    pub async fn batch(
+        &self,
+        batch: &Batch,
+        values: impl BatchValues,
+    ) -> Result<BatchResult, QueryError> {
         let values_ref = &values;
 
         let retry_session = match &batch.retry_policy {
@@ -530,6 +543,108 @@ impl Session {
         MetricsView::new(self.metrics.clone())
     }
 
+    /// Get [`TracingInfo`] of a traced query performed earlier
+    pub async fn get_tracing_info(&self, tracing_id: &Uuid) -> Result<TracingInfo, QueryError> {
+        self.get_tracing_info_custom(tracing_id, &GetTracingConfig::default())
+            .await
+    }
+
+    /// Queries tracing info with custom retry settings.  
+    /// Tracing info might not be available immediately on queried node -
+    /// that's why the driver performs a few attempts with sleeps in between.
+    /// [`GetTracingConfig`] allows to specify a custom querying strategy.
+    pub async fn get_tracing_info_custom(
+        &self,
+        tracing_id: &Uuid,
+        config: &GetTracingConfig,
+    ) -> Result<TracingInfo, QueryError> {
+        // config.attempts is NonZeroU32 so at least one attempt will be made
+        for _ in 0..config.attempts.get() {
+            let current_try: Option<TracingInfo> = self
+                .try_getting_tracing_info(tracing_id, config.consistency)
+                .await?;
+
+            match current_try {
+                Some(tracing_info) => return Ok(tracing_info),
+                None => tokio::time::sleep(config.interval).await,
+            };
+        }
+
+        return Err(QueryError::ProtocolError(
+            "All tracing queries returned an empty result, \
+            maybe information didnt reach this node yet. \
+            Consider using get_tracing_info_custom with \
+            bigger interval in GetTracingConfig",
+        ));
+    }
+
+    // Tries getting the tracing info
+    // If the queries return 0 rows then returns None - the information didn't reach this node yet
+    // If there is some other error returns this error
+    async fn try_getting_tracing_info(
+        &self,
+        tracing_id: &Uuid,
+        consistency: Consistency,
+    ) -> Result<Option<TracingInfo>, QueryError> {
+        // Query system_traces.sessions for TracingInfo
+        let mut traces_session_query =
+            Query::new(crate::tracing::TRACES_SESSION_QUERY_STR.to_string());
+        traces_session_query.consistency = consistency;
+
+        // Query system_traces.events for TracingEvents
+        let mut traces_events_query =
+            Query::new(crate::tracing::TRACES_EVENTS_QUERY_STR.to_string());
+        traces_events_query.consistency = Consistency::One;
+        traces_events_query.consistency = consistency;
+
+        let (traces_session_res, traces_events_res) = tokio::try_join!(
+            self.query(traces_session_query, (tracing_id,)),
+            self.query(traces_events_query, (tracing_id,))
+        )?;
+
+        // Get tracing info
+        let tracing_info_row_res: Option<Result<TracingInfo, _>> = traces_session_res
+            .rows
+            .ok_or(QueryError::ProtocolError(
+                "Response to system_traces.sessions query was not Rows",
+            ))?
+            .into_typed::<TracingInfo>()
+            .next();
+
+        let mut tracing_info: TracingInfo = match tracing_info_row_res {
+            Some(tracing_info_row_res) => tracing_info_row_res.map_err(|_| {
+                QueryError::ProtocolError(
+                    "Columns from system_traces.session have an unexpected type",
+                )
+            })?,
+            None => return Ok(None),
+        };
+
+        // Get tracing events
+        let tracing_event_rows = traces_events_res
+            .rows
+            .ok_or(QueryError::ProtocolError(
+                "Response to system_traces.events query was not Rows",
+            ))?
+            .into_typed::<TracingEvent>();
+
+        for event in tracing_event_rows {
+            let tracing_event: TracingEvent = event.map_err(|_| {
+                QueryError::ProtocolError(
+                    "Columns from system_traces.events have an unexpected type",
+                )
+            })?;
+
+            tracing_info.events.push(tracing_event);
+        }
+
+        if tracing_info.events.is_empty() {
+            return Ok(None);
+        }
+
+        Ok(Some(tracing_info))
+    }
+
     // This method allows to easily run a query using load balancing, retry policy etc.
     // Requires some information about the query and two closures
     // First closure is used to choose a connection
diff --git a/scylla/src/transport/topology.rs b/scylla/src/transport/topology.rs
index af0cedcc..dbaaa88c 100644
--- a/scylla/src/transport/topology.rs
+++ b/scylla/src/transport/topology.rs
@@ -168,11 +168,11 @@ async fn query_peers(conn: &Connection, connect_port: u16) -> Result<Vec<Peer>,
 
     let (peers_res, local_res) = tokio::try_join!(peers_query, local_query)?;
 
-    let peers_rows = peers_res.ok_or(QueryError::ProtocolError(
+    let peers_rows = peers_res.rows.ok_or(QueryError::ProtocolError(
         "system.peers query response was not Rows",
     ))?;
 
-    let local_rows = local_res.ok_or(QueryError::ProtocolError(
+    let local_rows = local_res.rows.ok_or(QueryError::ProtocolError(
         "system.local query response was not Rows",
     ))?;
 
@@ -222,6 +222,7 @@ async fn query_keyspaces(conn: &Connection) -> Result<HashMap<String, Keyspace>,
             &[],
         )
         .await?
+        .rows
         .ok_or(QueryError::ProtocolError(
             "system_schema.keyspaces query response was not Rows",
         ))?;


Test patch:
diff --git a/scylla/src/frame/cql_collections_test.rs b/scylla/src/frame/cql_collections_test.rs
index 589f9150..0cb43a9a 100644
--- a/scylla/src/frame/cql_collections_test.rs
+++ b/scylla/src/frame/cql_collections_test.rs
@@ -50,6 +50,7 @@ async fn test_cql_collections() {
         .query("SELECT team FROM ks.teams", &[])
         .await
         .unwrap()
+        .rows
     {
         for row in rows.into_typed::<(HashMap<i32, String>,)>() {
             let teams: HashMap<i32, String> = row.unwrap().0;
@@ -83,6 +84,7 @@ async fn test_cql_collections() {
         .query("SELECT tags FROM ks.images", &[])
         .await
         .unwrap()
+        .rows
     {
         for row in rows.into_typed::<(Vec<String>,)>() {
             let tag: Vec<String> = row.unwrap().0;
@@ -101,6 +103,7 @@ async fn test_cql_collections() {
         .query("SELECT tags FROM ks.images", &[])
         .await
         .unwrap()
+        .rows
     {
         for row in rows.into_typed::<(Vec<String>,)>() {
             let tag: Vec<String> = row.unwrap().0;
@@ -130,6 +133,7 @@ async fn test_cql_collections() {
         .query("SELECT scores FROM ks.plays", &[])
         .await
         .unwrap()
+        .rows
     {
         for row in rows.into_typed::<(Vec<i32>,)>() {
             let scores: Vec<i32> = row.unwrap().0;
@@ -151,6 +155,7 @@ async fn test_cql_collections() {
         .query("SELECT scores FROM ks.plays", &[])
         .await
         .unwrap()
+        .rows
     {
         for row in rows.into_typed::<(Vec<i32>,)>() {
             let scores: Vec<i32> = row.unwrap().0;
@@ -180,6 +185,7 @@ async fn test_cql_collections() {
         .query("SELECT duration FROM ks.durations", &[])
         .await
         .unwrap()
+        .rows
     {
         for row in rows.into_typed::<((i32, String),)>() {
             let duration = row.unwrap().0;
diff --git a/scylla/src/frame/cql_types_test.rs b/scylla/src/frame/cql_types_test.rs
index 33b25e85..5f3539e2 100644
--- a/scylla/src/frame/cql_types_test.rs
+++ b/scylla/src/frame/cql_types_test.rs
@@ -89,6 +89,7 @@ where
             .query(select_values, &[])
             .await
             .unwrap()
+            .rows
             .unwrap()
             .into_typed::<(T,)>()
             .map(Result::unwrap)
@@ -178,6 +179,7 @@ async fn test_counter() {
             .query(select_values, (i as i32,))
             .await
             .unwrap()
+            .rows
             .unwrap()
             .into_typed::<(Counter,)>()
             .map(Result::unwrap)
@@ -233,6 +235,7 @@ async fn test_naive_date() {
             .query("SELECT val from ks.naive_date", &[])
             .await
             .unwrap()
+            .rows
             .unwrap()
             .into_typed::<(NaiveDate,)>()
             .next()
@@ -256,6 +259,7 @@ async fn test_naive_date() {
                 .query("SELECT val from ks.naive_date", &[])
                 .await
                 .unwrap()
+                .rows
                 .unwrap()
                 .into_typed::<(NaiveDate,)>()
                 .next()
@@ -313,6 +317,7 @@ async fn test_date() {
             .query("SELECT val from ks.date_tests", &[])
             .await
             .unwrap()
+            .rows
             .unwrap()[0]
             .columns[0]
             .as_ref()
@@ -361,6 +366,7 @@ async fn test_time() {
             .query("SELECT val from ks.time_tests", &[])
             .await
             .unwrap()
+            .rows
             .unwrap()
             .into_typed::<(Duration,)>()
             .next()
@@ -382,6 +388,7 @@ async fn test_time() {
             .query("SELECT val from ks.time_tests", &[])
             .await
             .unwrap()
+            .rows
             .unwrap()
             .into_typed::<(Duration,)>()
             .next()
@@ -467,6 +474,7 @@ async fn test_timestamp() {
             .query("SELECT val from ks.timestamp_tests", &[])
             .await
             .unwrap()
+            .rows
             .unwrap()
             .into_typed::<(Duration,)>()
             .next()
@@ -488,6 +496,7 @@ async fn test_timestamp() {
             .query("SELECT val from ks.timestamp_tests", &[])
             .await
             .unwrap()
+            .rows
             .unwrap()
             .into_typed::<(Duration,)>()
             .next()
@@ -541,6 +550,7 @@ async fn test_timeuuid() {
             .query("SELECT val from ks.timeuuid_tests", &[])
             .await
             .unwrap()
+            .rows
             .unwrap()
             .into_typed::<(Uuid,)>()
             .next()
@@ -563,6 +573,7 @@ async fn test_timeuuid() {
             .query("SELECT val from ks.timeuuid_tests", &[])
             .await
             .unwrap()
+            .rows
             .unwrap()
             .into_typed::<(Uuid,)>()
             .next()
@@ -631,6 +642,7 @@ async fn test_inet() {
             .query("SELECT val from ks.inet_tests WHERE id = 0", &[])
             .await
             .unwrap()
+            .rows
             .unwrap()
             .into_typed::<(IpAddr,)>()
             .next()
@@ -649,6 +661,7 @@ async fn test_inet() {
             .query("SELECT val from ks.inet_tests WHERE id = 0", &[])
             .await
             .unwrap()
+            .rows
             .unwrap()
             .into_typed::<(IpAddr,)>()
             .next()
@@ -705,6 +718,7 @@ async fn test_blob() {
             .query("SELECT val from ks.blob_tests WHERE id = 0", &[])
             .await
             .unwrap()
+            .rows
             .unwrap()
             .into_typed::<(Vec<u8>,)>()
             .next()
@@ -723,6 +737,7 @@ async fn test_blob() {
             .query("SELECT val from ks.blob_tests WHERE id = 0", &[])
             .await
             .unwrap()
+            .rows
             .unwrap()
             .into_typed::<(Vec<u8>,)>()
             .next()
diff --git a/scylla/src/transport/session_test.rs b/scylla/src/transport/session_test.rs
index 1c4c8741..5077eecd 100644
--- a/scylla/src/transport/session_test.rs
+++ b/scylla/src/transport/session_test.rs
@@ -1,7 +1,14 @@
+use crate::batch::Batch;
 use crate::frame::value::ValueList;
+use crate::query::Query;
 use crate::routing::hash3_x64_128;
+use crate::statement::Consistency;
+use crate::tracing::TracingInfo;
+use crate::transport::connection::{BatchResult, QueryResult};
 use crate::transport::errors::{BadKeyspaceName, BadQuery, DbError, QueryError};
 use crate::{IntoTypedRows, Session, SessionBuilder};
+use futures::StreamExt;
+use uuid::Uuid;
 
 #[tokio::test]
 async fn test_unprepared_statement() {
@@ -39,6 +46,7 @@ async fn test_unprepared_statement() {
         .query("SELECT a, b, c FROM ks.t", &[])
         .await
         .unwrap()
+        .rows
         .unwrap();
 
     let mut results: Vec<(i32, i32, &String)> = rs
@@ -112,6 +120,7 @@ async fn test_prepared_statement() {
             .query("SELECT token(a) FROM ks.t2", &[])
             .await
             .unwrap()
+            .rows
             .unwrap();
         let token: i64 = rs.first().unwrap().columns[0]
             .as_ref()
@@ -131,6 +140,7 @@ async fn test_prepared_statement() {
             .query("SELECT token(a,b,c) FROM ks.complex_pk", &[])
             .await
             .unwrap()
+            .rows
             .unwrap();
         let token: i64 = rs.first().unwrap().columns[0]
             .as_ref()
@@ -152,6 +162,7 @@ async fn test_prepared_statement() {
             .query("SELECT a,b,c FROM ks.t2", &[])
             .await
             .unwrap()
+            .rows
             .unwrap();
         let r = rs.first().unwrap();
         let a = r.columns[0].as_ref().unwrap().as_int().unwrap();
@@ -164,6 +175,7 @@ async fn test_prepared_statement() {
             .query("SELECT a,b,c,d,e FROM ks.complex_pk", &[])
             .await
             .unwrap()
+            .rows
             .unwrap();
         let r = rs.first().unwrap();
         let a = r.columns[0].as_ref().unwrap().as_int().unwrap();
@@ -218,6 +230,7 @@ async fn test_batch() {
         .query("SELECT a, b, c FROM ks.t_batch", &[])
         .await
         .unwrap()
+        .rows
         .unwrap();
 
     let mut results: Vec<(i32, i32, &String)> = rs
@@ -276,6 +289,7 @@ async fn test_token_calculation() {
             .query("SELECT token(a) FROM ks.t3 WHERE a = ?", &values)
             .await
             .unwrap()
+            .rows
             .unwrap();
         let token: i64 = rs.first().unwrap().columns[0]
             .as_ref()
@@ -331,6 +345,7 @@ async fn test_use_keyspace() {
         .query("SELECT * FROM tab", &[])
         .await
         .unwrap()
+        .rows
         .unwrap()
         .into_typed::<(String,)>()
         .map(|res| res.unwrap().0)
@@ -381,6 +396,7 @@ async fn test_use_keyspace() {
         .query("SELECT * FROM tab", &[])
         .await
         .unwrap()
+        .rows
         .unwrap()
         .into_typed::<(String,)>()
         .map(|res| res.unwrap().0)
@@ -447,6 +463,7 @@ async fn test_use_keyspace_case_sensitivity() {
         .query("SELECT * from tab", &[])
         .await
         .unwrap()
+        .rows
         .unwrap()
         .into_typed::<(String,)>()
         .map(|row| row.unwrap().0)
@@ -462,6 +479,7 @@ async fn test_use_keyspace_case_sensitivity() {
         .query("SELECT * from tab", &[])
         .await
         .unwrap()
+        .rows
         .unwrap()
         .into_typed::<(String,)>()
         .map(|row| row.unwrap().0)
@@ -511,6 +529,7 @@ async fn test_raw_use_keyspace() {
         .query("SELECT * FROM tab", &[])
         .await
         .unwrap()
+        .rows
         .unwrap()
         .into_typed::<(String,)>()
         .map(|res| res.unwrap().0)
@@ -597,3 +616,232 @@ async fn test_db_errors() {
         }
     );
 }
+
+#[tokio::test]
+async fn test_tracing() {
+    let uri = std::env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+    let session = SessionBuilder::new().known_node(uri).build().await.unwrap();
+
+    session.query("CREATE KEYSPACE IF NOT EXISTS test_tracing_ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[]).await.unwrap();
+
+    session
+        .query(
+            "CREATE TABLE IF NOT EXISTS test_tracing_ks.tab (a text primary key)",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    test_tracing_query(&session).await;
+    test_tracing_execute(&session).await;
+    test_tracing_prepare(&session).await;
+    test_get_tracing_info(&session).await;
+    test_tracing_query_iter(&session).await;
+    test_tracing_execute_iter(&session).await;
+    test_tracing_batch(&session).await;
+}
+
+async fn test_tracing_query(session: &Session) {
+    // A query without tracing enabled has no tracing uuid in result
+    let untraced_query: Query = Query::new("SELECT * FROM test_tracing_ks.tab".to_string());
+    let untraced_query_result: QueryResult = session.query(untraced_query, &[]).await.unwrap();
+
+    assert!(untraced_query_result.tracing_id.is_none());
+
+    // A query with tracing enabled has a tracing uuid in result
+    let mut traced_query: Query = Query::new("SELECT * FROM test_tracing_ks.tab".to_string());
+    traced_query.tracing = true;
+
+    let traced_query_result: QueryResult = session.query(traced_query, &[]).await.unwrap();
+    assert!(traced_query_result.tracing_id.is_some());
+
+    // Querying this uuid from tracing table gives some results
+    assert_in_tracing_table(session, traced_query_result.tracing_id.unwrap()).await;
+}
+
+async fn test_tracing_execute(session: &Session) {
+    // Executing a prepared statement without tracing enabled has no tracing uuid in result
+    let untraced_prepared = session
+        .prepare("SELECT * FROM test_tracing_ks.tab")
+        .await
+        .unwrap();
+
+    let untraced_prepared_result: QueryResult =
+        session.execute(&untraced_prepared, &[]).await.unwrap();
+
+    assert!(untraced_prepared_result.tracing_id.is_none());
+
+    // Executing a prepared statement with tracing enabled has a tracing uuid in result
+    let mut traced_prepared = session
+        .prepare("SELECT * FROM test_tracing_ks.tab")
+        .await
+        .unwrap();
+
+    traced_prepared.tracing = true;
+
+    let traced_prepared_result: QueryResult = session.execute(&traced_prepared, &[]).await.unwrap();
+    assert!(traced_prepared_result.tracing_id.is_some());
+
+    // Querying this uuid from tracing table gives some results
+    assert_in_tracing_table(session, traced_prepared_result.tracing_id.unwrap()).await;
+}
+
+async fn test_tracing_prepare(session: &Session) {
+    // Preparing a statement without tracing enabled has no tracing uuids in result
+    let untraced_prepared = session
+        .prepare("SELECT * FROM test_tracing_ks.tab")
+        .await
+        .unwrap();
+
+    assert!(untraced_prepared.prepare_tracing_ids.is_empty());
+
+    // Preparing a statement with tracing enabled has tracing uuids in result
+    let mut to_prepare_traced = Query::new("SELECT * FROM test_tracing_ks.tab".to_string());
+    to_prepare_traced.tracing = true;
+
+    let traced_prepared = session.prepare(to_prepare_traced).await.unwrap();
+    assert!(!traced_prepared.prepare_tracing_ids.is_empty());
+
+    // Querying this uuid from tracing table gives some results
+    for tracing_id in traced_prepared.prepare_tracing_ids {
+        assert_in_tracing_table(session, tracing_id).await;
+    }
+}
+
+async fn test_get_tracing_info(session: &Session) {
+    // A query with tracing enabled has a tracing uuid in result
+    let mut traced_query: Query = Query::new("SELECT * FROM test_tracing_ks.tab".to_string());
+    traced_query.tracing = true;
+
+    let traced_query_result: QueryResult = session.query(traced_query, &[]).await.unwrap();
+    let tracing_id: Uuid = traced_query_result.tracing_id.unwrap();
+
+    // Getting tracing info from session using this uuid works
+    let tracing_info: TracingInfo = session.get_tracing_info(&tracing_id).await.unwrap();
+    assert!(!tracing_info.events.is_empty());
+}
+
+async fn test_tracing_query_iter(session: &Session) {
+    // A query without tracing enabled has no tracing ids
+    let untraced_query: Query = Query::new("SELECT * FROM test_tracing_ks.tab".to_string());
+
+    let mut untraced_row_iter = session.query_iter(untraced_query, &[]).await.unwrap();
+    while let Some(_row) = untraced_row_iter.next().await {
+        // Receive rows
+    }
+
+    assert!(untraced_row_iter.get_tracing_ids().is_empty());
+
+    // The same is true for TypedRowIter
+    let untraced_typed_row_iter = untraced_row_iter.into_typed::<(i32,)>();
+    assert!(untraced_typed_row_iter.get_tracing_ids().is_empty());
+
+    // A query with tracing enabled has a tracing ids in result
+    let mut traced_query: Query = Query::new("SELECT * FROM test_tracing_ks.tab".to_string());
+    traced_query.tracing = true;
+
+    let mut traced_row_iter = session.query_iter(traced_query, &[]).await.unwrap();
+    while let Some(_row) = traced_row_iter.next().await {
+        // Receive rows
+    }
+
+    assert!(!traced_row_iter.get_tracing_ids().is_empty());
+
+    // The same is true for TypedRowIter
+    let traced_typed_row_iter = traced_row_iter.into_typed::<(i32,)>();
+    assert!(!traced_typed_row_iter.get_tracing_ids().is_empty());
+
+    for tracing_id in traced_typed_row_iter.get_tracing_ids() {
+        assert_in_tracing_table(session, *tracing_id).await;
+    }
+}
+
+async fn test_tracing_execute_iter(session: &Session) {
+    // A prepared statement without tracing enabled has no tracing ids
+    let untraced_prepared = session
+        .prepare("SELECT * FROM test_tracing_ks.tab")
+        .await
+        .unwrap();
+
+    let mut untraced_row_iter = session.execute_iter(untraced_prepared, &[]).await.unwrap();
+    while let Some(_row) = untraced_row_iter.next().await {
+        // Receive rows
+    }
+
+    assert!(untraced_row_iter.get_tracing_ids().is_empty());
+
+    // The same is true for TypedRowIter
+    let untraced_typed_row_iter = untraced_row_iter.into_typed::<(i32,)>();
+    assert!(untraced_typed_row_iter.get_tracing_ids().is_empty());
+
+    // A prepared statement with tracing enabled has a tracing ids in result
+    let mut traced_prepared = session
+        .prepare("SELECT * FROM test_tracing_ks.tab")
+        .await
+        .unwrap();
+    traced_prepared.tracing = true;
+
+    let mut traced_row_iter = session.execute_iter(traced_prepared, &[]).await.unwrap();
+    while let Some(_row) = traced_row_iter.next().await {
+        // Receive rows
+    }
+
+    assert!(!traced_row_iter.get_tracing_ids().is_empty());
+
+    // The same is true for TypedRowIter
+    let traced_typed_row_iter = traced_row_iter.into_typed::<(i32,)>();
+    assert!(!traced_typed_row_iter.get_tracing_ids().is_empty());
+
+    for tracing_id in traced_typed_row_iter.get_tracing_ids() {
+        assert_in_tracing_table(session, *tracing_id).await;
+    }
+}
+
+async fn test_tracing_batch(session: &Session) {
+    // A batch without tracing enabled has no tracing id
+    let mut untraced_batch: Batch = Default::default();
+    untraced_batch.append_statement("INSERT INTO test_tracing_ks.tab (a) VALUES('a')");
+
+    let untraced_batch_result: BatchResult = session.batch(&untraced_batch, ((),)).await.unwrap();
+    assert!(untraced_batch_result.tracing_id.is_none());
+
+    // Batch with tracing enabled has a tracing uuid in result
+    let mut traced_batch: Batch = Default::default();
+    traced_batch.append_statement("INSERT INTO test_tracing_ks.tab (a) VALUES('a')");
+    traced_batch.tracing = true;
+
+    let traced_batch_result: BatchResult = session.batch(&traced_batch, ((),)).await.unwrap();
+    assert!(traced_batch_result.tracing_id.is_some());
+
+    assert_in_tracing_table(session, traced_batch_result.tracing_id.unwrap()).await;
+}
+
+async fn assert_in_tracing_table(session: &Session, tracing_uuid: Uuid) {
+    let mut traces_query =
+        Query::new("SELECT * FROM system_traces.sessions WHERE session_id = ?".to_string());
+    traces_query.consistency = Consistency::One;
+
+    // Tracing info might not be immediately available
+    // If rows are empty perform 8 retries with a 32ms wait in between
+
+    for _ in 0..8 {
+        let row_opt = session
+            .query(traces_query.clone(), (tracing_uuid,))
+            .await
+            .unwrap()
+            .rows
+            .into_iter()
+            .next();
+
+        if row_opt.is_some() {
+            // Ok there was some row for this tracing_uuid
+            return;
+        }
+
+        // Otherwise retry
+        tokio::time::sleep(std::time::Duration::from_millis(32)).await;
+    }
+
+    // If all retries failed panic with an error
+    panic!("No rows for tracing with this session id!");
+}