connection: introduce SelfIdentity and new options

SelfIdentity is intended to group STARTUP options that can be used
to uniquely identify driver, application and particular client instance.

DRIVER_NAME and DRIVER_VERSION were already present, but they were
hardcoded to "scylla-rust-driver" and current crate version, retrieved
from Cargo. Now it is possible to set custom values for them, which
can be useful for custom driver builds, as well as for drivers running
on top of Rust driver (think cpp-rust-driver).

The remaining options are inspired by cpp-driver, and added for
cpp-rust-driver purposes as well:
- APPLICATION_NAME and APPLICATION_VERSION identify a single build of
an application using the driver.
- CLIENT_ID uniquely identifies a single instance of an application.

All the above information are visible in Cassandra in
`system_views.clients` in `client_options` column.
DRIVER_NAME and DRIVER_VERSION are visible in ScyllaDB in
`system.clients`.

FAQ
Q: Where do APPLICATION_NAME, APPLICATION_VERSION and CLIENT_ID come from?
- there are no columns in system.clients dedicated to those attributes,
- APPLICATION_NAME / APPLICATION_VERSION are not present in Scylla's
  source code at all,
- only 2 results in Cassandra source is some example in docs;
  APPLICATION_NAME and APPLICATION_VERSION appears in client_options
  which is an arbitrary dict where client can send any keys.
- driver variables are mentioned in protocol v5, application variables
  are not.

A:
The following options are not exposed anywhere in Scylla tables.
They come directly from CPP driver, and they are supported in Cassandra.
As we want to support as big subset of its API as possible in
cpp-rust-driver, I decided to expose API for setting those particular
key-value pairs, similarly to what cpp-driver does, and not an API
to set arbitrary key-value pairs. Allowing users to set arbitrary
options could break the driver by overwriting options that bear special
meaning, e.g. the shard-aware port. Therefore, I'm against such liberal
API. OTOH, we need to expose APPLICATION_NAME, APPLICATION_VERSION
and CLIENT_ID for cpp-rust-driver.

scylla-cql/src/frame/request/options.rs

@@ -20,6 +20,9 @@ pub const COMPRESSION: &str = "COMPRESSION";
 pub const CQL_VERSION: &str = "CQL_VERSION";
 pub const DRIVER_NAME: &str = "DRIVER_NAME";
 pub const DRIVER_VERSION: &str = "DRIVER_VERSION";
+pub const APPLICATION_NAME: &str = "APPLICATION_NAME";
+pub const APPLICATION_VERSION: &str = "APPLICATION_VERSION";
+pub const CLIENT_ID: &str = "CLIENT_ID";
 
 /* Value names for options in SUPPORTED/STARTUP */
 pub const DEFAULT_CQL_PROTOCOL_VERSION: &str = "4.0.0";

scylla/src/transport/connection.rs

@@ -350,6 +350,188 @@ mod ssl_config {
     }
 }
 
+/// Driver and application self-identifying information,
+/// to be sent in STARTUP message.
+#[derive(Debug, Clone, Default)]
+pub struct SelfIdentity<'id> {
+    // Custom driver identity can be set if a custom driver build is running,
+    // or an entirely different driver is operating on top of Rust driver
+    // (e.g. cpp-rust-driver).
+    custom_driver_name: Option<Cow<'id, str>>,
+    custom_driver_version: Option<Cow<'id, str>>,
+
+    // ### Q: Where do APPLICATION_NAME, APPLICATION_VERSION and CLIENT_ID come from?
+    // - there are no columns in system.clients dedicated to those attributes,
+    // - APPLICATION_NAME / APPLICATION_VERSION are not present in Scylla's source code at all,
+    // - only 2 results in Cassandra source is some example in docs:
+    //   https://github.com/apache/cassandra/blob/d3cbf9c1f72057d2a5da9df8ed567d20cd272931/doc/modules/cassandra/pages/managing/operating/virtualtables.adoc?plain=1#L218.
+    //   APPLICATION_NAME and APPLICATION_VERSION appears in client_options which
+    //   is an arbitrary dict where client can send any keys.
+    // - driver variables are mentioned in protocol v5
+    //   (https://github.com/apache/cassandra/blob/d3cbf9c1f72057d2a5da9df8ed567d20cd272931/doc/native_protocol_v5.spec#L480),
+    //   application variables are not.
+    //
+    // ### A:
+    // The following options are not exposed anywhere in Scylla tables.
+    // They come directly from CPP driver, and they are supported in Cassandra
+    //
+    // See https://github.com/scylladb/cpp-driver/blob/fa0f27069a625057984d1fa58f434ea99b86c83f/include/cassandra.h#L2916.
+    // As we want to support as big subset of its API as possible in cpp-rust-driver, I decided to expose API for setting
+    // those particular key-value pairs, similarly to what cpp-driver does, and not an API to set arbitrary key-value pairs.
+    //
+    // Allowing users to set arbitrary options could break the driver by overwriting options that bear special meaning,
+    // e.g. the shard-aware port. Therefore, I'm against such liberal API. OTOH, we need to expose APPLICATION_NAME,
+    // APPLICATION_VERSION and CLIENT_ID for cpp-rust-driver.
+
+    // Application identity can be set to distinguish different applications
+    // connected to the same cluster.
+    application_name: Option<Cow<'id, str>>,
+    application_version: Option<Cow<'id, str>>,
+
+    // A (unique) client ID can be set to distinguish different instances
+    // of the same application connected to the same cluster.
+    client_id: Option<Cow<'id, str>>,
+}
+
+impl<'id> SelfIdentity<'id> {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Advertises a custom driver name, which can be used if a custom driver build is running,
+    /// or an entirely different driver is operating on top of Rust driver
+    /// (e.g. cpp-rust-driver).
+    pub fn set_custom_driver_name(&mut self, custom_driver_name: impl Into<Cow<'id, str>>) {
+        self.custom_driver_name = Some(custom_driver_name.into());
+    }
+
+    /// Advertises a custom driver name. See [Self::set_custom_driver_name] for use cases.
+    pub fn with_custom_driver_name(mut self, custom_driver_name: impl Into<Cow<'id, str>>) -> Self {
+        self.custom_driver_name = Some(custom_driver_name.into());
+        self
+    }
+
+    /// Custom driver name to be advertised. See [Self::set_custom_driver_name] for use cases.
+    pub fn get_custom_driver_name(&self) -> Option<&str> {
+        self.custom_driver_name.as_deref()
+    }
+
+    /// Advertises a custom driver version. See [Self::set_custom_driver_name] for use cases.
+    pub fn set_custom_driver_version(&mut self, custom_driver_version: impl Into<Cow<'id, str>>) {
+        self.custom_driver_version = Some(custom_driver_version.into());
+    }
+
+    /// Advertises a custom driver version. See [Self::set_custom_driver_name] for use cases.
+    pub fn with_custom_driver_version(
+        mut self,
+        custom_driver_version: impl Into<Cow<'id, str>>,
+    ) -> Self {
+        self.custom_driver_version = Some(custom_driver_version.into());
+        self
+    }
+
+    /// Custom driver version to be advertised. See [Self::set_custom_driver_version] for use cases.
+    pub fn get_custom_driver_version(&self) -> Option<&str> {
+        self.custom_driver_version.as_deref()
+    }
+
+    /// Advertises an application name, which can be used to distinguish different applications
+    /// connected to the same cluster.
+    pub fn set_application_name(&mut self, application_name: impl Into<Cow<'id, str>>) {
+        self.application_name = Some(application_name.into());
+    }
+
+    /// Advertises an application name. See [Self::set_application_name] for use cases.
+    pub fn with_application_name(mut self, application_name: impl Into<Cow<'id, str>>) -> Self {
+        self.application_name = Some(application_name.into());
+        self
+    }
+
+    /// Application name to be advertised. See [Self::set_application_name] for use cases.
+    pub fn get_application_name(&self) -> Option<&str> {
+        self.application_name.as_deref()
+    }
+
+    /// Advertises an application version. See [Self::set_application_name] for use cases.
+    pub fn set_application_version(&mut self, application_version: impl Into<Cow<'id, str>>) {
+        self.application_version = Some(application_version.into());
+    }
+
+    /// Advertises an application version. See [Self::set_application_name] for use cases.
+    pub fn with_application_version(
+        mut self,
+        application_version: impl Into<Cow<'id, str>>,
+    ) -> Self {
+        self.application_version = Some(application_version.into());
+        self
+    }
+
+    /// Application version to be advertised. See [Self::set_application_version] for use cases.
+    pub fn get_application_version(&self) -> Option<&str> {
+        self.application_version.as_deref()
+    }
+
+    /// Advertises a client ID, which can be set to distinguish different instances
+    /// of the same application connected to the same cluster.
+    pub fn set_client_id(&mut self, client_id: impl Into<Cow<'id, str>>) {
+        self.client_id = Some(client_id.into());
+    }
+
+    /// Advertises a client ID. See [Self::set_client_id] for use cases.
+    pub fn with_client_id(mut self, client_id: impl Into<Cow<'id, str>>) -> Self {
+        self.client_id = Some(client_id.into());
+        self
+    }
+
+    /// Client ID to be advertised. See [Self::set_client_id] for use cases.
+    pub fn get_client_id(&self) -> Option<&str> {
+        self.client_id.as_deref()
+    }
+}
+
+impl<'id: 'map, 'map> SelfIdentity<'id> {
+    fn add_startup_options(&'id self, options: &'map mut HashMap<Cow<'id, str>, Cow<'id, str>>) {
+        /* Driver identity. */
+        let driver_name = self
+            .custom_driver_name
+            .as_deref()
+            .unwrap_or(options::DEFAULT_DRIVER_NAME);
+        options.insert(
+            Cow::Borrowed(options::DRIVER_NAME),
+            Cow::Borrowed(driver_name),
+        );
+
+        let driver_version = self
+            .custom_driver_version
+            .as_deref()
+            .unwrap_or(options::DEFAULT_DRIVER_VERSION);
+        options.insert(
+            Cow::Borrowed(options::DRIVER_VERSION),
+            Cow::Borrowed(driver_version),
+        );
+
+        /* Application identity. */
+        if let Some(application_name) = self.application_name.as_deref() {
+            options.insert(
+                Cow::Borrowed(options::APPLICATION_NAME),
+                Cow::Borrowed(application_name),
+            );
+        }
+
+        if let Some(application_version) = self.application_version.as_deref() {
+            options.insert(
+                Cow::Borrowed(options::APPLICATION_VERSION),
+                Cow::Borrowed(application_version),
+            );
+        }
+
+        /* Client identity. */
+        if let Some(client_id) = self.client_id.as_deref() {
+            options.insert(Cow::Borrowed(options::CLIENT_ID), Cow::Borrowed(client_id));
+        }
+    }
+}
+
 #[derive(Clone)]
 pub(crate) struct ConnectionConfig {
     pub(crate) compression: Option<Compression>,
@@ -370,6 +552,8 @@ pub(crate) struct ConnectionConfig {
     pub(crate) keepalive_interval: Option<Duration>,
     pub(crate) keepalive_timeout: Option<Duration>,
     pub(crate) tablet_sender: Option<mpsc::Sender<(TableSpec<'static>, RawTablet)>>,
+
+    pub(crate) identity: SelfIdentity<'static>,
 }
 
 impl Default for ConnectionConfig {
@@ -394,6 +578,8 @@ impl Default for ConnectionConfig {
             keepalive_timeout: None,
 
             tablet_sender: None,
+
+            identity: SelfIdentity::default(),
         }
     }
 }
@@ -1526,25 +1712,9 @@ pub(crate) async fn open_connection(
     source_port: Option<u16>,
     config: &ConnectionConfig,
 ) -> Result<(Connection, ErrorReceiver), QueryError> {
+    /* Translate the address, if applicable. */
     let addr = maybe_translated_addr(endpoint, config.address_translator.as_deref()).await?;
-    open_named_connection(
-        addr,
-        source_port,
-        config,
-        Some(options::DEFAULT_DRIVER_NAME),
-        Some(options::DEFAULT_DRIVER_VERSION),
-    )
-    .await
-}
 
-/// The same as `open_connection`, but with customizable driver name and version.
-pub(crate) async fn open_named_connection(
-    addr: SocketAddr,
-    source_port: Option<u16>,
-    config: &ConnectionConfig,
-    driver_name: Option<&str>,
-    driver_version: Option<&str>,
-) -> Result<(Connection, ErrorReceiver), QueryError> {
     /* Setup connection on TCP level and prepare for sending/receiving CQL frames. */
     let (mut connection, error_receiver) =
         Connection::new(addr, source_port, config.clone()).await?;
@@ -1604,19 +1774,8 @@ pub(crate) async fn open_named_connection(
         Cow::Borrowed(options::DEFAULT_CQL_PROTOCOL_VERSION),
     );
 
-    // Driver identity.
-    if let Some(driver_name) = driver_name {
-        options.insert(
-            Cow::Borrowed(options::DRIVER_NAME),
-            Cow::Borrowed(driver_name),
-        );
-    }
-    if let Some(driver_version) = driver_version {
-        options.insert(
-            Cow::Borrowed(options::DRIVER_VERSION),
-            Cow::Borrowed(driver_version),
-        );
-    }
+    // Application & driver's identity.
+    config.identity.add_startup_options(&mut options);
 
     // Optional compression.
     if let Some(compression) = &config.compression {

scylla/src/transport/mod.rs

@@ -19,6 +19,7 @@ pub mod speculative_execution;
 pub mod topology;
 
 pub use crate::frame::{Authenticator, Compression};
+pub use connection::SelfIdentity;
 pub use execution_profile::ExecutionProfile;
 pub use scylla_cql::errors;
 

scylla/src/transport/session.rs

@@ -46,7 +46,7 @@ use super::node::KnownNode;
 use super::partitioner::PartitionerName;
 use super::query_result::MaybeFirstRowTypedError;
 use super::topology::UntranslatedPeer;
-use super::NodeRef;
+use super::{NodeRef, SelfIdentity};
 use crate::cql_to_rust::FromRow;
 use crate::frame::response::cql_to_rust::FromRowError;
 use crate::frame::response::result;
@@ -515,6 +515,8 @@ impl Session {
             keepalive_interval: config.keepalive_interval,
             keepalive_timeout: config.keepalive_timeout,
             tablet_sender: Some(tablet_sender),
+            // A temporary stub, removed in the next commit.
+            identity: SelfIdentity::default(),
         };
 
         let pool_config = PoolConfig {