IT: Adjust and enable "Requests" metrics test

Why does this test not work without adjustments? Well, this is because
rust-driver collects latencies with millisecond graunularity. In result,
most of the latencies during the tests in local setup are rounded to 0ms.

This is why, we somehow need to simulate higher latencies during the test.
There is one piece of code that user controls and is executed in rust-driver
in between start and end time measurements - namely `HistoryListener::log_attempt_start`.

This is where we can add a sleep to simulate higher latencies in local setup.
And so, I implemented `SleepingHistoryListener` that does just that. In addition,
I implemented the testing API to set such listener on the statement.

The "Requests" test is adjusted accordingly, and enabled.

Note: Since all latencies during the test in local setup are now expected to be around 1ms,
I loosened the stddev check to be `>= 0` instead of `> 0`.

Author: muzarski

Makefile

@@ -26,6 +26,7 @@ SCYLLA_TEST_FILTER := $(subst ${SPACE},${EMPTY},ClusterTests.*\
 :PreparedMetadataTests.*\
 :UseKeyspaceCaseSensitiveTests.*\
 :MetricsTests.Integration_Cassandra_ErrorsRequestTimeouts\
+:MetricsTests.Integration_Cassandra_Requests\
 :-PreparedTests.Integration_Cassandra_PreparedIDUnchangedDuringReprepare\
 :HeartbeatTests.Integration_Cassandra_HeartbeatFailed\
 :ControlConnectionTests.Integration_Cassandra_TopologyChange\
@@ -66,6 +67,7 @@ CASSANDRA_TEST_FILTER := $(subst ${SPACE},${EMPTY},ClusterTests.*\
 :PreparedMetadataTests.*\
 :UseKeyspaceCaseSensitiveTests.*\
 :MetricsTests.Integration_Cassandra_ErrorsRequestTimeouts\
+:MetricsTests.Integration_Cassandra_Requests\
 :-PreparedTests.Integration_Cassandra_PreparedIDUnchangedDuringReprepare\
 :PreparedTests.Integration_Cassandra_FailFastWhenPreparedIDChangesDuringReprepare\
 :HeartbeatTests.Integration_Cassandra_HeartbeatFailed\

scylla-rust-wrapper/src/integration_testing.rs

@@ -1,8 +1,16 @@
 use std::ffi::{CString, c_char};
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::time::Duration;
+
+use scylla::errors::{RequestAttemptError, RequestError};
+use scylla::observability::history::{AttemptId, HistoryListener, RequestId, SpeculativeId};
+use scylla::policies::retry::RetryDecision;
 
 use crate::argconv::{BoxFFI, CMut, CassBorrowedExclusivePtr};
 use crate::cluster::CassCluster;
-use crate::types::{cass_int32_t, cass_uint16_t, size_t};
+use crate::statement::{BoundStatement, CassStatement};
+use crate::types::{cass_int32_t, cass_uint16_t, cass_uint64_t, size_t};
 
 #[unsafe(no_mangle)]
 pub unsafe extern "C" fn testing_cluster_get_connect_timeout(
@@ -54,3 +62,57 @@ pub unsafe extern "C" fn testing_cluster_get_contact_points(
 pub unsafe extern "C" fn testing_free_contact_points(contact_points: *mut c_char) {
     let _ = unsafe { CString::from_raw(contact_points) };
 }
+
+#[derive(Debug)]
+struct SleepingHistoryListener(Duration);
+
+impl HistoryListener for SleepingHistoryListener {
+    fn log_request_start(&self) -> RequestId {
+        RequestId(0)
+    }
+
+    fn log_request_success(&self, _request_id: RequestId) {}
+
+    fn log_request_error(&self, _request_id: RequestId, _error: &RequestError) {}
+
+    fn log_new_speculative_fiber(&self, _request_id: RequestId) -> SpeculativeId {
+        SpeculativeId(0)
+    }
+
+    fn log_attempt_start(
+        &self,
+        _request_id: RequestId,
+        _speculative_id: Option<SpeculativeId>,
+        _node_addr: SocketAddr,
+    ) -> AttemptId {
+        // Sleep to simulate a delay in the request
+        std::thread::sleep(self.0);
+        AttemptId(0)
+    }
+
+    fn log_attempt_success(&self, _attempt_id: AttemptId) {}
+
+    fn log_attempt_error(
+        &self,
+        _attempt_id: AttemptId,
+        _error: &RequestAttemptError,
+        _retry_decision: &RetryDecision,
+    ) {
+    }
+}
+
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn testing_statement_set_sleeping_history_listener(
+    statement_raw: CassBorrowedExclusivePtr<CassStatement, CMut>,
+    sleep_time_ms: cass_uint64_t,
+) {
+    let sleep_time = Duration::from_millis(sleep_time_ms);
+    let history_listener = Arc::new(SleepingHistoryListener(sleep_time));
+
+    match &mut BoxFFI::as_mut_ref(statement_raw).unwrap().statement {
+        BoundStatement::Simple(inner) => inner.query.set_history_listener(history_listener),
+        BoundStatement::Prepared(inner) => Arc::make_mut(&mut inner.statement)
+            .statement
+            .set_history_listener(history_listener),
+    }
+}

src/testing.cpp

@@ -98,4 +98,8 @@ void set_record_attempted_hosts(CassStatement* statement, bool enable) {
   throw std::runtime_error("Unimplemented 'set_record_attempted_hosts'!");
 }
 
+void set_sleeping_history_listener_on_statement(CassStatement* statement, uint64_t sleep_time_ms) {
+  testing_statement_set_sleeping_history_listener(statement, sleep_time_ms);
+}
+
 }}} // namespace datastax::internal::testing

src/testing.hpp

@@ -50,6 +50,8 @@ CASS_EXPORT String get_server_name(CassFuture* future);
 
 CASS_EXPORT void set_record_attempted_hosts(CassStatement* statement, bool enable);
 
+CASS_EXPORT void set_sleeping_history_listener_on_statement(CassStatement* statement, uint64_t sleep_time_ms);
+
 }}} // namespace datastax::internal::testing
 
 #endif

src/testing_rust_impls.h

@@ -21,6 +21,11 @@ CASS_EXPORT void testing_cluster_get_contact_points(CassCluster* cluster, char**
                                                     size_t* contact_points_length);
 
 CASS_EXPORT void testing_free_contact_points(char* contact_points);
+
+// Sets a sleeping history listener on the statement.
+// This can be used to enforce a sleep time during statement execution, which increases the latency.
+CASS_EXPORT void testing_statement_set_sleeping_history_listener(CassStatement *statement,
+                                                                 cass_uint64_t sleep_time_ms);
 }
 
 #endif

tests/src/integration/objects/statement.hpp

@@ -165,6 +165,16 @@ class Statement : public Object<CassStatement, cass_statement_free> {
     ASSERT_EQ(CASS_OK, cass_statement_set_execution_profile(get(), name.c_str()));
   }
 
+  /**
+   * Set a sleeping history listener on the statement.
+   * This can be used to enforce a sleep time during statement execution, which increases the latency.
+   *
+   * @param sleep_time_ms Sleep time in milliseconds
+   */
+  void set_sleep_time(uint64_t sleep_time_ms) {
+    datastax::internal::testing::set_sleeping_history_listener_on_statement(get(), sleep_time_ms);
+  }
+
   /**
    * Enable/Disable whether the statement is idempotent. Idempotent statements
    * are able to be automatically retried after timeouts/errors and can be

tests/src/integration/tests/test_metrics.cpp

@@ -114,17 +114,20 @@ CASSANDRA_INTEGRATION_TEST_F(MetricsTests, ErrorsRequestTimeouts) {
 CASSANDRA_INTEGRATION_TEST_F(MetricsTests, Requests) {
   CHECK_FAILURE;
 
+  Statement statement(SELECT_ALL_SYSTEM_LOCAL_CQL);
+  statement.set_sleep_time(1); // Simulate >=1ms latency.
+
   CassMetrics metrics = session_.metrics();
   for (int i = 0; i < 600 && metrics.requests.fifteen_minute_rate == 0; ++i) {
-    session_.execute_async(SELECT_ALL_SYSTEM_LOCAL_CQL);
+    session_.execute_async(statement);
     metrics = session_.metrics();
     msleep(100);
   }
 
   EXPECT_LT(metrics.requests.min, CASS_UINT64_MAX);
   EXPECT_GT(metrics.requests.max, 0u);
   EXPECT_GT(metrics.requests.mean, 0u);
-  EXPECT_GT(metrics.requests.stddev, 0u);
+  EXPECT_GE(metrics.requests.stddev, 0u);
   EXPECT_GT(metrics.requests.median, 0u);
   EXPECT_GT(metrics.requests.percentile_75th, 0u);
   EXPECT_GT(metrics.requests.percentile_95th, 0u);