Add stale keyspaces detector

Previous commits changed all the integration tests to drop keyspaces
after finishing. We want to enforce in CI that no keyspaces are left
after tests finish. I don't see better way of doing that than create a
separate small program that looks through existing keyspaces, filters
out non-test ones, and prints schemas of the stale ones. This commit
implements such program.
Now we will be able to add CI steps that call it after finishing tests.
This of course does not help with ccm tests. In the future we can solve
this in custom test framework, or by looking through cluster metadata
after each ccm test (before destroying the cluster).

Author: Lorak-mmk

Cargo.lock.msrv

@@ -5,5 +5,6 @@ members = [
     "scylla-macros",
     "scylla-cql",
     "scylla-proxy",
+    "utils",
 ]
 resolver = "2"

Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+edition = "2021"
+name = "utils"
+publish = false
+version = "0.0.0"
+
+[dependencies]
+scylla = { path = "../scylla" }
+tokio = { version = "1.34", features = ["rt", "rt-multi-thread", "macros"] }
+futures = "0.3.6"
+
+[[bin]]
+name = "find_stale_test_keyspaces"

utils/Cargo.toml

@@ -0,0 +1,54 @@
+use futures::{stream, StreamExt};
+use scylla::client::session::Session;
+use scylla::client::session_builder::SessionBuilder;
+
+#[tokio::main]
+async fn main() {
+    let uri = std::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
+        .expect("Can't connect to cluster");
+    session
+        .refresh_metadata()
+        .await
+        .expect("Can't refresh metadata");
+    let state = session.get_cluster_state();
+    let keyspaces_stream = stream::iter(state.keyspaces_iter());
+    let nr_stale = keyspaces_stream
+        .filter_map(|(name, _ks)| {
+            let session = &session;
+            async move {
+                if !name.starts_with("test_rust_") {
+                    return None;
+                }
+                let describe_result = session
+                    .query_unpaged(format!("DESCRIBE {name}"), &[])
+                    .await
+                    .expect("Can't perform DESCRIBE request")
+                    .into_rows_result()
+                    .expect("Wrong response type for DESCRIBE request");
+                println!("Found stale keyspace: {name}");
+                println!("Schema:");
+                for (_, _, _, describe_text) in describe_result
+                    .rows::<(&str, &str, &str, &str)>()
+                    .expect("Wrong response metadata for DESCRIBE request")
+                    .map(|r| r.expect("Failed to deserialize a row of RESPONSE request"))
+                {
+                    println!("{describe_text}");
+                }
+
+                Some(())
+            }
+        })
+        .count()
+        .await;
+
+    if nr_stale > 0 {
+        panic!("Found stale keyspaces");
+    }
+}