cluster: add application_info

Implement clustr.application_info to make driver send following startup options to server:
1. `APPLICATION_NAME` - ID what application is using driver, example: repo of the application
2. `APPLICATION_VERSION` - Version of the application, example: release version or commit id of the application
3. `CLIENT_ID` - unique id of the client instance, example: pod name

All strings.

Author: dkropachev

cassandra/application_info.py

@@ -0,0 +1,45 @@
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+class ApplicationInfoBase(object):
+    """
+    A class that holds application information and adds it to startup message options
+    """
+    def add_startup_options(self, options: dict[str, str]):
+        raise NotImplementedError()
+
+
+class ApplicationInfo(ApplicationInfoBase):
+    def __init__(self, application_name: str = None, application_version: str = None, client_id: str = None):
+        if application_name:
+            if not isinstance(application_name, str):
+                raise TypeError('application_name must be a string')
+            self.application_name = application_name
+        if application_version:
+            if not isinstance(application_version, str):
+                raise TypeError('application_version must be a string')
+            self.application_version = application_version
+        if client_id:
+            if not isinstance(client_id, str):
+                raise TypeError('client_id must be a string')
+            self.client_id = client_id
+
+    def add_startup_options(self, options: dict[str, str]):
+        if self.application_name:
+            options['APPLICATION_NAME'] = self.application_name
+        if self.application_version:
+            options['APPLICATION_VERSION'] = self.application_version
+        if self.client_id:
+            options['CLIENT_ID'] = self.client_id

cassandra/cluster.py

@@ -95,6 +95,7 @@
 from cassandra.datastax.graph.query import _request_timeout_key, _GraphSONContextRowFactory
 from cassandra.datastax import cloud as dscloud
 from cassandra.scylla.cloud import CloudConfiguration
+from cassandra.application_info import ApplicationInfoBase
 
 try:
     from cassandra.io.twistedreactor import TwistedConnection
@@ -706,6 +707,15 @@ class Cluster(object):
     Setting this to :const:`False` disables compression.
     """
 
+    application_info: ApplicationInfoBase = None
+    """
+    An instance of any subclass of :class:`.cluster.ApplicationInfoBase`.
+
+    Defaults to None
+
+    When not None makes driver sends information about application that uses driver in startup frame
+    """
+
     _auth_provider = None
     _auth_provider_callable = None
 
@@ -1204,6 +1214,7 @@ def __init__(self,
                  shard_aware_options=None,
                  metadata_request_timeout=None,
                  column_encryption_policy=None,
+                 application_info:ApplicationInfoBase=None
                  ):
         """
         ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as
@@ -1329,6 +1340,12 @@ def __init__(self,
                 raise TypeError("address_translator should not be a class, it should be an instance of that class")
             self.address_translator = address_translator
 
+        if application_info is not None:
+            if not isinstance(application_info, ApplicationInfoBase):
+                raise TypeError(
+                    "application_info should be an instance of any ApplicationInfoBase class")
+            self.application_info = application_info
+
         if timestamp_generator is not None:
             if not callable(timestamp_generator):
                 raise ValueError("timestamp_generator must be callable")
@@ -1779,6 +1796,7 @@ def _make_connection_kwargs(self, endpoint, kwargs_dict):
         kwargs_dict.setdefault('user_type_map', self._user_types)
         kwargs_dict.setdefault('allow_beta_protocol_version', self.allow_beta_protocol_version)
         kwargs_dict.setdefault('no_compact', self.no_compact)
+        kwargs_dict.setdefault('application_info', self.application_info)
 
         return kwargs_dict
 

cassandra/connection.py

@@ -29,6 +29,7 @@
 import random
 import itertools
 
+from cassandra.application_info import ApplicationInfoBase
 from cassandra.protocol_features import ProtocolFeatures
 
 if 'gevent.monkey' in sys.modules:
@@ -774,7 +775,7 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None,
                  cql_version=None, protocol_version=ProtocolVersion.MAX_SUPPORTED, is_control_connection=False,
                  user_type_map=None, connect_timeout=None, allow_beta_protocol_version=False, no_compact=False,
                  ssl_context=None, owning_pool=None, shard_id=None, total_shards=None,
-                 on_orphaned_stream_released=None):
+                 on_orphaned_stream_released=None, application_info: ApplicationInfoBase = None):
         # TODO next major rename host to endpoint and remove port kwarg.
         self.endpoint = host if isinstance(host, EndPoint) else DefaultEndPoint(host, port)
 
@@ -797,6 +798,7 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None,
         self._socket_writable = True
         self.orphaned_request_ids = set()
         self._on_orphaned_stream_released = on_orphaned_stream_released
+        self._application_info = application_info
 
         if ssl_options:
             self.ssl_options.update(self.endpoint.ssl_options or {})
@@ -1380,6 +1382,8 @@ def _handle_options_response(self, options_response):
 
         options = {}
         self.features.add_startup_options(options)
+        if self._application_info:
+            self._application_info.add_startup_options(options)
 
         if self.cql_version:
             if self.cql_version not in supported_cql_versions:

tests/integration/__init__.py

@@ -405,7 +405,7 @@ def _id_and_mark(f):
 notdse = unittest.skipIf(DSE_VERSION, "DSE not supported")
 requiredse = unittest.skipUnless(DSE_VERSION, "DSE required")
 requirescloudproxy = unittest.skipIf(CLOUD_PROXY_PATH is None, "Cloud Proxy path hasn't been specified")
-
+notscylla = unittest.skipIf(SCYLLA_VERSION, "Does not support scylla")
 libevtest = unittest.skipUnless(EVENT_LOOP_MANAGER=="libev", "Test timing designed for libev loop")
 
 def wait_for_node_socket(node, timeout):

tests/integration/standard/test_application_info.py

@@ -0,0 +1,34 @@
+import unittest
+
+from cassandra.application_info import ApplicationInfo
+from tests.integration import TestCluster, notscylla
+
+
+@notscylla
+class ApplicationInfoTest(unittest.TestCase):
+    def setUp(self):
+        self.cluster = TestCluster(application_info=ApplicationInfo(
+            application_name='TestApplicationInfo',
+            application_version='1.0.0-test',
+            client_id='some-client-id',
+        ))
+        self.session = self.cluster.connect()
+
+    def tearDown(self):
+        self.cluster.shutdown()
+
+    def test_application_info_endup_in_system_views_clients(self):
+        """
+        Test to ensure that ApplicationInfo user provides endup in `client_options` of `system_views.clients` table
+        """
+        found = False
+
+        for row in self.session.execute("select client_options from system_views.clients"):
+            if not row['client_options']:
+                continue
+            for marker in ("'APPLICATION_NAME': 'app_name'", "'APPLICATION_VERSION': 'app_version'", "'CLIENT_ID': 'client_id'"):
+                if marker not in row['client_options']:
+                    break
+            else:
+                found = True
+        assert found