Merge pull request scylladb#33 from riptano/merge_ossnext_ngdg

Merge ossnext ngdg

Author: TheRealFalcon

CHANGELOG.rst

@@ -29,9 +29,22 @@ Unreleased
 
 Features
 --------
+* Allow passing ssl context for Twisted (PYTHON-1161)
+* ssl context and cloud support for Eventlet (PYTHON-1162)
+* Cloud Twisted support (PYTHON-1163)
+* Add additional_write_policy and read_repair to system schema parsing (PYTHON-1048)
+* Handle prepared id mismatch when repreparing on the fly (PYTHON-1124)
+* Remove *read_repair_chance table options (PYTHON-1140)
+* Flexible version parsing (PYTHON-1174)
+* Support NULL in collection deserializer (PYTHON-1123)
+* Avoid warnings about unspecified load balancing policy when connecting to a cloud cluster (PYTHON-1177)
+* [GRAPH] Ability to execute Fluent Graph queries asynchronously (PYTHON-1129)
 
 Bug Fixes
 ---------
+* re-raising the CQLEngineException will fail on Python 3 (PYTHON-1166)
+* asyncio message chunks can be processed discontinuously (PYTHON-1185)
+* Reconnect attempts persist after downed node removed from peers (PYTHON-1181)
 
 Others
 ------
@@ -96,6 +109,22 @@ Other
 * Write documentation examples for DSE 2.0 features (PYTHON-732)
 * DSE_V1 protocol should not include all of protocol v5 (PYTHON-694)
 
+3.20.2
+======
+November 19, 2019
+
+Bug Fixes
+---------
+* Fix import error for old python installation without SSLContext (PYTHON-1183)
+
+3.20.1
+======
+November 6, 2019
+
+Bug Fixes
+---------
+* ValueError: too many values to unpack (expected 2)" when there are two dashes in server version number (PYTHON-1172)
+
 3.20.0
 ======
 October 28, 2019

build.yaml

@@ -46,8 +46,8 @@ schedules:
       EXCLUDE_LONG=1
     matrix:
       exclude:
-        - python: [2.7, 3.4, 3.6, 3.7]
-        - cassandra: ['2.0', '2.1', '2.2', '3.0', 'test-dse']
+        - python: [2.7, 3.4, 3.7]
+        - cassandra: ['2.0', '2.1', '2.2', '3.0', 'test-dse', dse-4.8', 'dse-5.0']
 
   release_test:
     schedule: per_commit
@@ -57,17 +57,17 @@ schedules:
     env_vars: |
       EVENT_LOOP_MANAGER='libev'
 
-ngdg:
+  tmpngdg:
     schedule: adhoc
     branches:
-      include: [ngdg_master_ft]
+      include: [merge_ossnext_ngdg]
     env_vars: |
       EVENT_LOOP_MANAGER='libev'
       EXCLUDE_LONG=1
     matrix:
       exclude:
         - python: [2.7, 3.4, 3.6, 3.7]
-        - cassandra: ['dse-4.8', 'dse-5.0', dse-6.0', 'dse-6.7']
+        - cassandra: ['2.1', '2.2', '3.0', '3.11', 'dse-4.8', 'dse-5.0', 'dse-5.1', 'dse-6.0', 'dse-6.7']
 
   weekly_master:
     schedule: 0 10 * * 6
@@ -258,6 +258,7 @@ build:
           "tests/integration/standard/test_metrics.py"
           "tests/integration/standard/test_query.py"
           "tests/integration/simulacron/test_endpoint.py"
+          "tests/integration/long/test_ssl.py"
         )
         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" --with-ignore-docstrings --with-xunit --xunit-file=standard_results.xml ${EVENT_LOOP_TESTS[@]} || true
         exit 0

cassandra/__init__.py

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

cassandra/cluster.py

@@ -19,6 +19,7 @@
 from __future__ import absolute_import
 
 import atexit
+from binascii import hexlify
 from collections import defaultdict
 from concurrent.futures import ThreadPoolExecutor, FIRST_COMPLETED, wait as wait_futures
 from copy import copy
@@ -39,16 +40,6 @@
 import weakref
 from weakref import WeakValueDictionary
 
-try:
-    from cassandra.io.twistedreactor import TwistedConnection
-except ImportError:
-    TwistedConnection = None
-
-try:
-    from weakref import WeakSet
-except ImportError:
-    from cassandra.util import WeakSet  # NOQA
-
 from cassandra import (ConsistencyLevel, AuthenticationFailed,
                        OperationTimedOut, UnsupportedOperation,
                        SchemaTargetType, DriverException, ProtocolVersion,
@@ -101,6 +92,21 @@
 from cassandra.datastax.graph.query import _request_timeout_key, _GraphSONContextRowFactory
 from cassandra.datastax import cloud as dscloud
 
+try:
+    from cassandra.io.twistedreactor import TwistedConnection
+except ImportError:
+    TwistedConnection = None
+
+try:
+    from cassandra.io.eventletreactor import EventletConnection
+except ImportError:
+    EventletConnection = None
+
+try:
+    from weakref import WeakSet
+except ImportError:
+    from cassandra.util import WeakSet  # NOQA
+
 if six.PY3:
     long = int
 
@@ -1104,14 +1110,14 @@ def __init__(self,
             self.connection_class = connection_class
 
         if cloud is not None:
+            self.cloud = cloud
             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,
-                create_pyopenssl_context=self.connection_class is TwistedConnection
-            )
+            uses_twisted = TwistedConnection and issubclass(self.connection_class, TwistedConnection)
+            uses_eventlet = EventletConnection and issubclass(self.connection_class, EventletConnection)
+            cloud_config = dscloud.get_cloud_config(cloud, create_pyopenssl_context=uses_twisted or uses_eventlet)
 
             ssl_context = cloud_config.ssl_context
             ssl_options = {'check_hostname': True}
@@ -1241,7 +1247,7 @@ def __init__(self,
             profiles.setdefault(EXEC_PROFILE_GRAPH_ANALYTICS_DEFAULT,
                                 GraphAnalyticsExecutionProfile(load_balancing_policy=lbp))
 
-        if self._contact_points_explicit:
+        if self._contact_points_explicit and not self.cloud:  # avoid this warning for cloud users.
             if self._config_mode is _ConfigMode.PROFILES:
                 default_lbp_profiles = self.profile_manager._profiles_without_explicit_lbps()
                 if default_lbp_profiles:
@@ -2027,6 +2033,10 @@ def on_remove(self, host):
             listener.on_remove(host)
         self.control_connection.on_remove(host)
 
+        reconnection_handler = host.get_and_set_reconnection_handler(None)
+        if reconnection_handler:
+            reconnection_handler.cancel()
+
     def signal_connection_failure(self, host, connection_exc, is_host_addition, expect_host_to_be_down=False):
         is_down = host.signal_connection_failure(connection_exc)
         if is_down:
@@ -4648,6 +4658,15 @@ def _execute_after_prepare(self, host, connection, pool, response):
         if isinstance(response, ResultMessage):
             if response.kind == RESULT_KIND_PREPARED:
                 if self.prepared_statement:
+                    if self.prepared_statement.query_id != response.query_id:
+                        self._set_final_exception(DriverException(
+                            "ID mismatch while trying to reprepare (expected {expected}, got {got}). "
+                            "This prepared statement won't work anymore. "
+                            "This usually happens when you run a 'USE...' "
+                            "query after the statement was prepared.".format(
+                                expected=hexlify(self.prepared_statement.query_id), got=hexlify(response.query_id)
+                            )
+                        ))
                     self.prepared_statement.result_metadata = response.column_metadata
                     new_metadata_id = response.result_metadata_id
                     if new_metadata_id is not None:

cassandra/connection.py

@@ -26,11 +26,7 @@
 import sys
 from threading import Thread, Event, RLock, Condition
 import time
-
-try:
-    import ssl
-except ImportError:
-    ssl = None  # NOQA
+import ssl
 
 if 'gevent.monkey' in sys.modules:
     from gevent.queue import Queue, Empty
@@ -772,6 +768,15 @@ def factory(cls, endpoint, timeout, *args, **kwargs):
         else:
             return conn
 
+    def _wrap_socket_from_context(self):
+        self._socket = self.ssl_context.wrap_socket(self._socket, **(self.ssl_options or {}))
+
+    def _initiate_connection(self, sockaddr):
+        self._socket.connect(sockaddr)
+
+    def _match_hostname(self):
+        ssl.match_hostname(self._socket.getpeercert(), self.endpoint.address)
+
     def _get_socket_addresses(self):
         address, port = self.endpoint.resolve()
 
@@ -791,17 +796,16 @@ def _connect_socket(self):
             try:
                 self._socket = self._socket_impl.socket(af, socktype, proto)
                 if self.ssl_context:
-                    self._socket = self.ssl_context.wrap_socket(self._socket,
-                                                                **(self.ssl_options or {}))
+                    self._wrap_socket_from_context()
                 elif self.ssl_options:
                     if not self._ssl_impl:
                         raise RuntimeError("This version of Python was not compiled with SSL support")
                     self._socket = self._ssl_impl.wrap_socket(self._socket, **self.ssl_options)
                 self._socket.settimeout(self.connect_timeout)
-                self._socket.connect(sockaddr)
+                self._initiate_connection(sockaddr)
                 self._socket.settimeout(None)
                 if self._check_hostname:
-                    ssl.match_hostname(self._socket.getpeercert(), self.endpoint.address)
+                    self._match_hostname()
                 sockerr = None
                 break
             except socket.error as err:

cassandra/cqlengine/management.py

@@ -232,7 +232,7 @@ def _sync_table(model, connection=None):
         except CQLEngineException as ex:
             # 1.2 doesn't return cf names, so we have to examine the exception
             # and ignore if it says the column family already exists
-            if "Cannot add already existing column family" not in unicode(ex):
+            if "Cannot add already existing column family" not in six.text_type(ex):
                 raise
     else:
         log.debug(format_log_context("sync_table checking existing table %s", keyspace=ks_name, connection=connection), cf_name)

cassandra/cqltypes.py

@@ -813,9 +813,12 @@ def deserialize_safe(cls, byts, protocol_version):
         for _ in range(numelements):
             itemlen = unpack(byts[p:p + length])
             p += length
-            item = byts[p:p + itemlen]
-            p += itemlen
-            result.append(subtype.from_binary(item, inner_proto))
+            if itemlen < 0:
+                result.append(None)
+            else:
+                item = byts[p:p + itemlen]
+                p += itemlen
+                result.append(subtype.from_binary(item, inner_proto))
         return cls.adapter(result)
 
     @classmethod
@@ -867,14 +870,23 @@ def deserialize_safe(cls, byts, protocol_version):
         for _ in range(numelements):
             key_len = unpack(byts[p:p + length])
             p += length
-            keybytes = byts[p:p + key_len]
-            p += key_len
+            if key_len < 0:
+                keybytes = None
+                key = None
+            else:
+                keybytes = byts[p:p + key_len]
+                p += key_len
+                key = key_type.from_binary(keybytes, inner_proto)
+
             val_len = unpack(byts[p:p + length])
             p += length
-            valbytes = byts[p:p + val_len]
-            p += val_len
-            key = key_type.from_binary(keybytes, inner_proto)
-            val = value_type.from_binary(valbytes, inner_proto)
+            if val_len < 0:
+                val = None
+            else:
+                valbytes = byts[p:p + val_len]
+                p += val_len
+                val = value_type.from_binary(valbytes, inner_proto)
+
             themap._insert_unchecked(key, keybytes, val)
         return themap
 

cassandra/datastax/cloud/__init__.py

@@ -15,8 +15,10 @@
 import os
 import logging
 import json
+import sys
 import tempfile
 import shutil
+import six
 from six.moves.urllib.request import urlopen
 
 _HAS_SSL = True
@@ -135,7 +137,8 @@ def read_metadata_info(config, cloud_config):
         response = urlopen(url, context=config.ssl_context, timeout=timeout)
     except Exception as e:
         log.exception(e)
-        raise DriverException("Unable to connect to the metadata service at %s" % url)
+        raise DriverException("Unable to connect to the metadata service at %s. "
+                              "Check the cluster status in the Constellation cloud console. " % url)
 
     if response.code != 200:
         raise DriverException(("Error while fetching the metadata at: %s. "
@@ -177,8 +180,12 @@ def _ssl_context_from_cert(ca_cert_location, cert_location, key_location):
 def _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location):
     try:
         from OpenSSL import SSL
-    except ImportError:
-        return None
+    except ImportError as e:
+        six.reraise(
+            ImportError,
+            ImportError("PyOpenSSL must be installed to connect to Apollo with the Eventlet or Twisted event loops"),
+            sys.exc_info()[2]
+        )
     ssl_context = SSL.Context(SSL.TLSv1_METHOD)
     ssl_context.set_verify(SSL.VERIFY_PEER, callback=lambda _1, _2, _3, _4, ok: ok)
     ssl_context.use_certificate_file(cert_location)