Merge branch 'oss-next' into ngdg_master_ft

Author: aboudreault

CHANGELOG.rst

@@ -98,11 +98,18 @@ Other
 
 3.20.0
 ======
-Unreleased
+October 28, 2019
+
+Features
+--------
+* DataStax Apollo Support (PYTHON-1074)
+* Use 4.0 schema parser in 4 alpha and snapshot builds (PYTHON-1158)
 
 Bug Fixes
 ---------
 * Connection setup methods prevent using ExecutionProfile in cqlengine (PYTHON-1009)
+* Driver deadlock if all connections dropped by heartbeat whilst request in flight and request times out (PYTHON-1044)
+* Exception when use pk__token__gt filter In python 3.7 (PYTHON-1121)
 
 3.19.0
 ======

README.rst

@@ -26,12 +26,11 @@ Features
 * Configurable `load balancing <http://datastax.github.io/python-driver/api/cassandra/policies.html#load-balancing>`_ and `retry policies <http://datastax.github.io/python-driver/api/cassandra/policies.html#retrying-failed-operations>`_
 * `Concurrent execution utilities <http://datastax.github.io/python-driver/api/cassandra/concurrent.html>`_
 * `Object mapper <http://datastax.github.io/python-driver/object_mapper.html>`_
+* `Connecting to DataStax Apollo database (cloud) <https://docs.datastax.com/en/developer/python-driver/latest/cloud/>`_
 * DSE Graph execution API
 * DSE Geometric type serialization
 * DSE PlainText and GSSAPI authentication
 
-A fluent API extension for DSE Graph is available in the ``dse-graph`` package. For more information, see `the documentation here <http://docs.datastax.com/en/developer/python-dse-graph/>`_.
-
 Installation
 ------------
 Installation through pip is recommended::

build.yaml

@@ -1,6 +1,7 @@
 schedules:
   nightly_master:
     schedule: nightly
+    disable_pull_requests: true
     branches:
       include: [master]
     env_vars: |
@@ -50,6 +51,7 @@ schedules:
 
   release_test:
     schedule: per_commit
+    disable_pull_requests: true
     branches:
       include: [/release-.+/]
     env_vars: |
@@ -69,6 +71,7 @@ ngdg:
 
   weekly_master:
     schedule: 0 10 * * 6
+    disable_pull_requests: true
     branches:
       include: [master]
     env_vars: |
@@ -80,6 +83,7 @@ ngdg:
 
   weekly_gevent:
     schedule: 0 14 * * 6
+    disable_pull_requests: true
     branches:
       include: [master]
     env_vars: |
@@ -91,6 +95,7 @@ ngdg:
 
   weekly_eventlet:
     schedule: 0 18 * * 6
+    disable_pull_requests: true
     branches:
       include: [master]
     env_vars: |
@@ -102,6 +107,7 @@ ngdg:
 
   weekly_asyncio:
     schedule: 0 22 * * 6
+    disable_pull_requests: true
     branches:
       include: [master]
     env_vars: |
@@ -113,6 +119,7 @@ ngdg:
 
   weekly_async:
     schedule: 0 10 * * 7
+    disable_pull_requests: true
     branches:
       include: [master]
     env_vars: |
@@ -124,6 +131,7 @@ ngdg:
 
   weekly_twister:
     schedule: 0 14 * * 7
+    disable_pull_requests: true
     branches:
       include: [master]
     env_vars: |
@@ -192,6 +200,9 @@ build:
       export PYTHONPATH=""
       export CCM_MAX_HEAP_SIZE=1024M
 
+      # Required for unix socket tests
+      sudo apt-get install socat
+
       # Install latest setuptools
       pip install --upgrade pip
       pip install -U setuptools
@@ -277,9 +288,13 @@ build:
         EVENT_LOOP_MANAGER=$EVENT_LOOP_MANAGER CASSANDRA_DIR=$CCM_INSTALL_DIR DSE_VERSION=$DSE_VERSION ADS_HOME=$HOME/ VERIFY_CYTHON=$FORCE_CYTHON nosetests -s -v --logging-format="[%(levelname)s] %(asctime)s %(thread)d: %(message)s" --with-ignore-docstrings --with-xunit --xunit-file=dse_results.xml tests/integration/advanced/ || true
       fi
 
+      echo "==========RUNNING ADVANCED AND CLOUD TESTS=========="
+      EVENT_LOOP_MANAGER=$EVENT_LOOP_MANAGER CLOUD_PROXY_PATH="$HOME/proxy/" CASSANDRA_VERSION=$CCM_CASSANDRA_VERSION MAPPED_CASSANDRA_VERSION=$MAPPED_CASSANDRA_VERSION VERIFY_CYTHON=$FORCE_CYTHON nosetests -s -v --logging-format="[%(levelname)s] %(asctime)s %(thread)d: %(message)s" --with-ignore-docstrings --with-xunit --xunit-file=advanced_results.xml tests/integration/advanced/ || true
+
       if [ -z "$EXCLUDE_LONG" ]; then
         echo "==========RUNNING LONG INTEGRATION TESTS=========="
         EVENT_LOOP_MANAGER=$EVENT_LOOP_MANAGER CCM_ARGS="$CCM_ARGS" DSE_VERSION=$DSE_VERSION CASSANDRA_VERSION=$CCM_CASSANDRA_VERSION MAPPED_CASSANDRA_VERSION=$MAPPED_CASSANDRA_VERSION VERIFY_CYTHON=$FORCE_CYTHON nosetests -s -v --logging-format="[%(levelname)s] %(asctime)s %(thread)d: %(message)s" --exclude-dir=tests/integration/long/upgrade --with-ignore-docstrings --with-xunit --xunit-file=long_results.xml tests/integration/long/ || true
       fi
+
   - xunit:
     - "*_results.xml"

cassandra/__init__.py

@@ -22,7 +22,7 @@ def emit(self, record):
 
 logging.getLogger('cassandra').addHandler(NullHandler())
 
-__version_info__ = (3, 19, 0, '20190910+labs')
+__version_info__ = (3, 20, 0, '20191030+labs')
 __version__ = '.'.join(map(str, __version_info__))
 
 

cassandra/cluster.py

@@ -51,7 +51,7 @@
 from cassandra.connection import (ConnectionException, ConnectionShutdown,
                                   ConnectionHeartbeat, ProtocolVersionUnsupported,
                                   EndPoint, DefaultEndPoint, DefaultEndPointFactory,
-                                  ContinuousPagingState)
+                                  ContinuousPagingState, SniEndPointFactory)
 from cassandra.cqltypes import UserType
 from cassandra.encoder import Encoder
 from cassandra.protocol import (QueryMessage, ResultMessage,
@@ -93,11 +93,11 @@
                                       graph_graphson2_row_factory, graph_graphson3_row_factory,
                                       GraphSON3Serializer)
 from cassandra.datastax.graph.query import _request_timeout_key, _GraphSONContextRowFactory
+from cassandra.datastax import cloud as dscloud
 
 if six.PY3:
     long = int
 
-
 def _is_eventlet_monkey_patched():
     if 'eventlet.patcher' not in sys.modules:
         return False
@@ -357,19 +357,28 @@ class ExecutionProfile(object):
 
     # indicates if lbp was set explicitly or uses default values
     _load_balancing_policy_explicit = False
+    _consistency_level_explicit = False
 
     def __init__(self, load_balancing_policy=_NOT_SET, retry_policy=None,
-                 consistency_level=ConsistencyLevel.LOCAL_ONE, serial_consistency_level=None,
+                 consistency_level=_NOT_SET, serial_consistency_level=None,
                  request_timeout=10.0, row_factory=named_tuple_factory, speculative_execution_policy=None,
                  continuous_paging_options=None):
+
         if load_balancing_policy is _NOT_SET:
             self._load_balancing_policy_explicit = False
             self.load_balancing_policy = default_lbp_factory()
         else:
             self._load_balancing_policy_explicit = True
             self.load_balancing_policy = load_balancing_policy
+
+        if consistency_level is _NOT_SET:
+            self._consistency_level_explicit = False
+            self.consistency_level = ConsistencyLevel.LOCAL_ONE
+        else:
+            self._consistency_level_explicit = True
+            self.consistency_level = consistency_level
+
         self.retry_policy = retry_policy or RetryPolicy()
-        self.consistency_level = consistency_level
 
         if (serial_consistency_level is not None and
                 not ConsistencyLevel.is_serial(serial_consistency_level)):
@@ -964,6 +973,19 @@ def default_retry_policy(self, policy):
     A string identifiying this application's version to Insights
     """
 
+    cloud = None
+    """
+    A dict of the cloud configuration. Example::
+        
+        {
+            # path to the secure connect bundle
+            'secure_connect_bundle': '/path/to/secure-connect-dbname.zip'
+        }
+
+    The zip file will be temporarily extracted in the same directory to
+    load the configuration and certificates.
+    """
+
     @property
     def schema_metadata_enabled(self):
         """
@@ -1064,13 +1086,34 @@ def __init__(self,
                  application_version=None,
                  monitor_reporting_enabled=True,
                  monitor_reporting_interval=30,
-                 client_id=None):
+                 client_id=None,
+                 cloud=None):
         """
         ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as
         extablishing connection pools or refreshing metadata.
 
         Any of the mutable Cluster attributes may be set as keyword arguments to the constructor.
         """
+
+        if cloud is not None:
+            if contact_points is not _NOT_SET or endpoint_factory or ssl_context or ssl_options:
+                raise ValueError("contact_points, endpoint_factory, ssl_context, and ssl_options "
+                                 "cannot be specified with a cloud configuration")
+
+            cloud_config = dscloud.get_cloud_config(cloud)
+
+            ssl_context = cloud_config.ssl_context
+            ssl_options = {'check_hostname': True}
+            if (auth_provider is None and cloud_config.username
+                    and cloud_config.password):
+                auth_provider = PlainTextAuthProvider(cloud_config.username, cloud_config.password)
+
+            endpoint_factory = SniEndPointFactory(cloud_config.sni_host, cloud_config.sni_port)
+            contact_points = [
+                endpoint_factory.create_from_sni(host_id)
+                for host_id in cloud_config.host_ids
+            ]
+
         if contact_points is not None:
             if contact_points is _NOT_SET:
                 self._contact_points_explicit = False
@@ -1160,12 +1203,12 @@ def __init__(self,
             self.timestamp_generator = MonotonicTimestampGenerator()
 
         self.profile_manager = ProfileManager()
-        self.profile_manager.profiles[EXEC_PROFILE_DEFAULT] = ExecutionProfile(self.load_balancing_policy,
-                                                                               self.default_retry_policy,
-                                                                               Session._default_consistency_level,
-                                                                               Session._default_serial_consistency_level,
-                                                                               Session._default_timeout,
-                                                                               Session._row_factory)
+        self.profile_manager.profiles[EXEC_PROFILE_DEFAULT] = ExecutionProfile(
+            self.load_balancing_policy,
+            self.default_retry_policy,
+            request_timeout=Session._default_timeout,
+            row_factory=Session._row_factory
+        )
 
         # legacy mode if either of these is not default
         if load_balancing_policy or default_retry_policy:
@@ -1430,6 +1473,7 @@ def add_execution_profile(self, name, profile, pool_wait_timeout=5):
             profile.load_balancing_policy.on_up(host)
         futures = set()
         for session in tuple(self.sessions):
+            self._set_default_dbaas_consistency(session)
             futures.update(session.update_created_pools())
         _, not_done = wait_futures(futures, pool_wait_timeout)
         if not_done:
@@ -1648,8 +1692,18 @@ def connect(self, keyspace=None, wait_for_all_pools=False):
         session = self._new_session(keyspace)
         if wait_for_all_pools:
             wait_futures(session._initial_connect_futures)
+
+        self._set_default_dbaas_consistency(session)
+
         return session
 
+    def _set_default_dbaas_consistency(self, session):
+        if session.cluster.metadata.dbaas:
+            for profile in self.profile_manager.profiles.values():
+                if not profile._consistency_level_explicit:
+                    profile.consistency_level = ConsistencyLevel.LOCAL_QUORUM
+            session._default_consistency_level = ConsistencyLevel.LOCAL_QUORUM
+
     def get_connection_holders(self):
         holders = []
         for s in tuple(self.sessions):
@@ -3341,7 +3395,7 @@ class ControlConnection(object):
     # Used only when token_metadata_enabled is set to False
     _SELECT_LOCAL_NO_TOKENS_RPC_ADDRESS = "SELECT rpc_address FROM system.local WHERE key='local'"
 
-    _SELECT_SCHEMA_PEERS_TEMPLATE = "SELECT peer, {nt_col_name}, schema_version FROM system.peers"
+    _SELECT_SCHEMA_PEERS_TEMPLATE = "SELECT peer, host_id, {nt_col_name}, schema_version FROM system.peers"
     _SELECT_SCHEMA_LOCAL = "SELECT schema_version FROM system.local WHERE key='local'"
 
     _MINIMUM_NATIVE_ADDRESS_VERSION = "4.0"
@@ -3393,6 +3447,8 @@ def connect(self):
         self._protocol_version = self._cluster.protocol_version
         self._set_new_connection(self._reconnect_internal())
 
+        self._cluster.metadata.dbaas = self._connection._product_type == dscloud.PRODUCT_APOLLO
+
     def _set_new_connection(self, conn):
         """
         Replace existing connection (if there is one) and close it.
@@ -4206,10 +4262,14 @@ def _on_timeout(self, _attempts=0):
         if self._connection is not None:
             try:
                 self._connection._requests.pop(self._req_id)
-            # This prevents the race condition of the
-            # event loop thread just receiving the waited message
-            # If it arrives after this, it will be ignored
+            # PYTHON-1044
+            # This request might have been removed from the connection after the latter was defunct by heartbeat.
+            # We should still raise OperationTimedOut to reject the future so that the main event thread will not
+            # wait for it endlessly
             except KeyError:
+                key = "Connection defunct by heartbeat"
+                errors = {key: "Client request timeout. See Session.execute[_async](timeout)"}
+                self._set_final_exception(OperationTimedOut(errors, self._current_host))
                 return
 
             pool = self.session._pools.get(self._current_host)