lbp: construct CassHostFilter

In cpp-driver, the following rule is upheld: if a host is rejected
by **all** policies, the connection to that host is not opened at all.

We can achieve this by:
- taking the union of all whitelists (per hosts and per dcs)
- taking the intersection of all blacklists (per hosts and per dcs)

Now, if a host is not in the union of whitelists, it is rejected.
If a host is in the intersection of blacklists, it is rejected.

Note: if the execution profile does not have a base LBP defined
(one of: round-robin, dc-aware, rack-aware), all of the extensions
are ignored - including the filtering rules. This is achieved by filtering
out such profiles in CassCluster::build_host_filter.

Author: muzarski

scylla-rust-wrapper/src/cluster.rs

@@ -3,7 +3,9 @@ use crate::cass_error::CassError;
 use crate::cass_types::CassConsistency;
 use crate::exec_profile::{CassExecProfile, ExecProfileName, exec_profile_builder_modify};
 use crate::future::CassFuture;
-use crate::load_balancing::{LoadBalancingConfig, LoadBalancingKind, set_filtering};
+use crate::load_balancing::{
+    CassHostFilter, LoadBalancingConfig, LoadBalancingKind, set_filtering,
+};
 use crate::retry_policy::CassRetryPolicy;
 use crate::retry_policy::RetryPolicy::*;
 use crate::ssl::CassSsl;
@@ -15,6 +17,7 @@ use scylla::client::execution_profile::ExecutionProfileBuilder;
 use scylla::client::session_builder::SessionBuilder;
 use scylla::client::{PoolSize, SelfIdentity, WriteCoalescingDelay};
 use scylla::frame::Compression;
+use scylla::policies::host_filter::HostFilter;
 use scylla::policies::load_balancing::LatencyAwarenessBuilder;
 use scylla::policies::retry::RetryPolicy;
 use scylla::policies::speculative_execution::SimpleSpeculativeExecutionPolicy;
@@ -95,6 +98,25 @@ impl CassCluster {
     pub(crate) fn get_client_id(&self) -> Option<uuid::Uuid> {
         self.client_id
     }
+
+    pub(crate) fn build_host_filter(&self) -> Arc<dyn HostFilter> {
+        CassHostFilter::new_from_lbp_configs(
+            std::iter::once(&self.load_balancing_config).chain(
+                self.execution_profile_map
+                    .values()
+                    // Filter out the profiles that do not have specified base LBP.
+                    // If base LBP is not specified, the extensions such as filtering
+                    // are simply ignored - default (cluster) LBP is used.
+                    .filter_map(|exec_profile| {
+                        exec_profile
+                            .load_balancing_config
+                            .load_balancing_kind
+                            .as_ref()
+                            .map(|_| &exec_profile.load_balancing_config)
+                    }),
+            ),
+        )
+    }
 }
 
 // Utilities for integration testing

scylla-rust-wrapper/src/exec_profile.rs

@@ -37,7 +37,7 @@ use crate::types::{
 #[derive(Clone, Debug)]
 pub struct CassExecProfile {
     inner: ExecutionProfileBuilder,
-    load_balancing_config: LoadBalancingConfig,
+    pub(crate) load_balancing_config: LoadBalancingConfig,
 }
 
 impl FFI for CassExecProfile {

scylla-rust-wrapper/src/load_balancing.rs

@@ -331,6 +331,57 @@ where
     intersection.filter(|v| !v.is_empty())
 }
 
+impl CassHostFilter {
+    /// Computes the filtering rules from the load balancing policies, and
+    /// returns a corresponding `HostFilter`.
+    ///
+    /// In cpp-driver, the following rule is upheld: if a host is rejected
+    /// by **all** policies, the connection to that host is not opened at all.
+    ///
+    /// We can achieve this by:
+    /// - taking the union of all whitelists (per hosts and per dcs)
+    /// - taking the intersection of all blacklists (per hosts and per dcs)
+    ///
+    /// Now, if a host is not in the union of whitelists, it is rejected.
+    /// If a host is in the intersection of blacklists, it is rejected.
+    pub(crate) fn new_from_lbp_configs<'a>(
+        configs: impl Iterator<Item = &'a LoadBalancingConfig> + Clone,
+    ) -> Arc<dyn HostFilter> {
+        let whitelist_hosts = nonempty_union(
+            configs
+                .clone()
+                .map(|lbp_config| &lbp_config.filtering.whitelist_hosts),
+        );
+
+        let blacklist_hosts = nonempty_intersection(
+            configs
+                .clone()
+                .map(|lbp_config| &lbp_config.filtering.blacklist_hosts),
+        );
+
+        let whitelist_dc = nonempty_union(
+            configs
+                .clone()
+                .map(|lbp_config| &lbp_config.filtering.whitelist_dc),
+        );
+
+        let blacklist_dc = nonempty_intersection(
+            configs
+                .clone()
+                .map(|lbp_config| &lbp_config.filtering.blacklist_dc),
+        );
+
+        Arc::new(Self {
+            filtering: FilteringInfo {
+                whitelist_hosts,
+                blacklist_hosts,
+                whitelist_dc,
+                blacklist_dc,
+            },
+        })
+    }
+}
+
 #[cfg(test)]
 mod tests {
     #[test]

scylla-rust-wrapper/src/session.rs

@@ -21,6 +21,7 @@ use scylla::cluster::metadata::ColumnType;
 use scylla::errors::ExecutionError;
 use scylla::frame::types::Consistency;
 use scylla::observability::metrics::MetricsError;
+use scylla::policies::host_filter::HostFilter;
 use scylla::response::PagingStateResponse;
 use scylla::response::query_result::QueryResult;
 use scylla::statement::unprepared::Statement;
@@ -78,11 +79,13 @@ impl CassSessionInner {
     ) -> CassOwnedSharedPtr<CassFuture, CMut> {
         let session_builder = build_session_builder(cluster);
         let exec_profile_map = cluster.execution_profile_map().clone();
+        let host_filter = cluster.build_host_filter();
 
         CassFuture::make_raw(Self::connect_fut(
             session_opt,
             session_builder,
             exec_profile_map,
+            host_filter,
             cluster
                 .get_client_id()
                 // If user did not set a client id, generate a random uuid v4.
@@ -95,6 +98,7 @@ impl CassSessionInner {
         session_opt: Arc<RwLock<Option<CassSessionInner>>>,
         session_builder_fut: impl Future<Output = SessionBuilder>,
         exec_profile_builder_map: HashMap<ExecProfileName, CassExecProfile>,
+        host_filter: Arc<dyn HostFilter>,
         client_id: uuid::Uuid,
         keyspace: Option<String>,
     ) -> CassFutureResult {
@@ -124,6 +128,7 @@ impl CassSessionInner {
         }
 
         let session = session_builder
+            .host_filter(host_filter)
             .build()
             .await
             .map_err(|err| (err.to_cass_error(), err.msg()))?;