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/cluster.py

@@ -511,6 +511,19 @@ def __init__(self, load_balancing_policy=None, retry_policy=None,
         self.graph_options.set_source_analytics()
 
 
+class ApplicationInfo(object):
+    def __init__(self, application_name=None, application_version=None, client_id=None):
+        if not isinstance(application_name, str):
+            raise TypeError('application_name must be a string')
+        if not isinstance(application_version, str):
+            raise TypeError('application_version must be a string')
+        if not isinstance(client_id, str):
+            raise TypeError('client_id must be a string')
+        self.application_name = application_name
+        self.application_version = application_version
+        self.client_id = client_id
+
+
 class ProfileManager(object):
 
     def __init__(self):
@@ -706,6 +719,15 @@ class Cluster(object):
     Setting this to :const:`False` disables compression.
     """
 
+    application_info = None
+    """
+    An instance of :class:`.cluster.ApplicationInfo` or one of its subclasses.
+
+    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 +1226,7 @@ def __init__(self,
                  shard_aware_options=None,
                  metadata_request_timeout=None,
                  column_encryption_policy=None,
+                 application_info=None
                  ):
         """
         ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as
@@ -1329,6 +1352,14 @@ 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 isinstance(application_info, type):
+                raise TypeError("application_info should not be a class, it should be an instance of ApplicationInfo class")
+            if not isinstance(application_info, ApplicationInfo):
+                raise TypeError(
+                    "application_info should be an instance of ApplicationInfo 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 +1810,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

@@ -774,7 +774,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=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 +797,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 {})
@@ -1439,6 +1440,13 @@ def _send_startup_message(self, compression=None, no_compact=False, extra_option
         opts = {'DRIVER_NAME': DRIVER_NAME,
                 'DRIVER_VERSION': DRIVER_VERSION,
                 **extra_options}
+        if self.application_info:
+            if self.application_info.application_name:
+                opts['APPLICATION_NAME'] = self.application_info.application_name
+            if self.application_info.application_version:
+                opts['APPLICATION_VERSION'] = self.application_info.application_version
+            if self.application_info.client_id:
+                opts['CLIENT_ID'] = self.application_info.client_id
         if compression:
             opts['COMPRESSION'] = compression
         if no_compact: