cluster.py

# 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.

"""
This module houses the main classes you will interact with,
:class:`.Cluster` and :class:`.Session`.
"""
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
from functools import partial, wraps
from itertools import groupby, count, chain
import json
import logging
from warnings import warn
from random import random
import six
from six.moves import filter, range, queue as Queue
import socket
import sys
import time
from threading import Lock, RLock, Thread, Event
import uuid

import weakref
from weakref import WeakValueDictionary

from cassandra import (ConsistencyLevel, AuthenticationFailed,
                       OperationTimedOut, UnsupportedOperation,
                       SchemaTargetType, DriverException, ProtocolVersion,
                       UnresolvableContactPoints)
from cassandra.auth import _proxy_execute_key, PlainTextAuthProvider
from cassandra.connection import (ConnectionException, ConnectionShutdown,
                                  ConnectionHeartbeat, ProtocolVersionUnsupported,
                                  EndPoint, DefaultEndPoint, DefaultEndPointFactory,
                                  ContinuousPagingState, SniEndPointFactory, ConnectionBusy)
from cassandra.cqltypes import UserType
from cassandra.encoder import Encoder
from cassandra.protocol import (QueryMessage, ResultMessage,
                                ErrorMessage, ReadTimeoutErrorMessage,
                                WriteTimeoutErrorMessage,
                                UnavailableErrorMessage,
                                OverloadedErrorMessage,
                                PrepareMessage, ExecuteMessage,
                                PreparedQueryNotFound,
                                IsBootstrappingErrorMessage,
                                TruncateError, ServerError,
                                BatchMessage, RESULT_KIND_PREPARED,
                                RESULT_KIND_SET_KEYSPACE, RESULT_KIND_ROWS,
                                RESULT_KIND_SCHEMA_CHANGE, ProtocolHandler,
                                RESULT_KIND_VOID)
from cassandra.metadata import Metadata, protect_name, murmur3, _NodeInfo
from cassandra.policies import (TokenAwarePolicy, DCAwareRoundRobinPolicy, SimpleConvictionPolicy,
                                ExponentialReconnectionPolicy, HostDistance,
                                RetryPolicy, IdentityTranslator, NoSpeculativeExecutionPlan,
                                NoSpeculativeExecutionPolicy, DefaultLoadBalancingPolicy,
                                NeverRetryPolicy)
from cassandra.pool import (Host, _ReconnectionHandler, _HostReconnectionHandler,
                            HostConnectionPool, HostConnection,
                            NoConnectionsAvailable)
from cassandra.query import (SimpleStatement, PreparedStatement, BoundStatement,
                             BatchStatement, bind_params, QueryTrace, TraceUnavailable,
                             named_tuple_factory, dict_factory, tuple_factory, FETCH_SIZE_UNSET,
                             HostTargetingStatement)
from cassandra.marshal import int64_pack
from cassandra.timestamps import MonotonicTimestampGenerator
from cassandra.compat import Mapping
from cassandra.util import _resolve_contact_points_to_string_map, Version

from cassandra.datastax.insights.reporter import MonitorReporter
from cassandra.datastax.insights.util import version_supports_insights

from cassandra.datastax.graph import (graph_object_row_factory, GraphOptions, GraphSON1Serializer,
                                      GraphProtocol, GraphSON2Serializer, GraphStatement, SimpleGraphStatement,
                                      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

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

def _is_eventlet_monkey_patched():
    if 'eventlet.patcher' not in sys.modules:
        return False
    import eventlet.patcher
    return eventlet.patcher.is_monkey_patched('socket')


def _is_gevent_monkey_patched():
    if 'gevent.monkey' not in sys.modules:
        return False
    import gevent.socket
    return socket.socket is gevent.socket.socket


# default to gevent when we are monkey patched with gevent, eventlet when
# monkey patched with eventlet, otherwise if libev is available, use that as
# the default because it's fastest. Otherwise, use asyncore.
if _is_gevent_monkey_patched():
    from cassandra.io.geventreactor import GeventConnection as DefaultConnection
elif _is_eventlet_monkey_patched():
    from cassandra.io.eventletreactor import EventletConnection as DefaultConnection
else:
    try:
        from cassandra.io.libevreactor import LibevConnection as DefaultConnection  # NOQA
    except ImportError:
        from cassandra.io.asyncorereactor import AsyncoreConnection as DefaultConnection  # NOQA

# Forces load of utf8 encoding module to avoid deadlock that occurs
# if code that is being imported tries to import the module in a seperate
# thread.
# See http://bugs.python.org/issue10923
"".encode('utf8')

log = logging.getLogger(__name__)


DEFAULT_MIN_REQUESTS = 5
DEFAULT_MAX_REQUESTS = 100

DEFAULT_MIN_CONNECTIONS_PER_LOCAL_HOST = 2
DEFAULT_MAX_CONNECTIONS_PER_LOCAL_HOST = 8

DEFAULT_MIN_CONNECTIONS_PER_REMOTE_HOST = 1
DEFAULT_MAX_CONNECTIONS_PER_REMOTE_HOST = 2

_GRAPH_PAGING_MIN_DSE_VERSION = Version('6.8.0')

_NOT_SET = object()


class NoHostAvailable(Exception):
    """
    Raised when an operation is attempted but all connections are
    busy, defunct, closed, or resulted in errors when used.
    """

    errors = None
    """
    A map of the form ``{ip: exception}`` which details the particular
    Exception that was caught for each host the operation was attempted
    against.
    """

    def __init__(self, message, errors):
        Exception.__init__(self, message, errors)
        self.errors = errors


def _future_completed(future):
    """ Helper for run_in_executor() """
    exc = future.exception()
    if exc:
        log.debug("Failed to run task on executor", exc_info=exc)


def run_in_executor(f):
    """
    A decorator to run the given method in the ThreadPoolExecutor.
    """

    @wraps(f)
    def new_f(self, *args, **kwargs):

        if self.is_shutdown:
            return
        try:
            future = self.executor.submit(f, self, *args, **kwargs)
            future.add_done_callback(_future_completed)
        except Exception:
            log.exception("Failed to submit task to executor")

    return new_f


_clusters_for_shutdown = set()


def _register_cluster_shutdown(cluster):
    _clusters_for_shutdown.add(cluster)


def _discard_cluster_shutdown(cluster):
    _clusters_for_shutdown.discard(cluster)


def _shutdown_clusters():
    clusters = _clusters_for_shutdown.copy()  # copy because shutdown modifies the global set "discard"
    for cluster in clusters:
        cluster.shutdown()


atexit.register(_shutdown_clusters)


def default_lbp_factory():
    if murmur3 is not None:
        return TokenAwarePolicy(DCAwareRoundRobinPolicy())
    return DCAwareRoundRobinPolicy()


class ContinuousPagingOptions(object):

    class PagingUnit(object):
        BYTES = 1
        ROWS = 2

    page_unit = None
    """
    Value of PagingUnit. Default is PagingUnit.ROWS.

    Units refer to the :attr:`~.Statement.fetch_size` or :attr:`~.Session.default_fetch_size`.
    """

    max_pages = None
    """
    Max number of pages to send
    """

    max_pages_per_second = None
    """
    Max rate at which to send pages
    """

    max_queue_size = None
    """
    The maximum queue size for caching pages, only honored for protocol version DSE_V2 and higher,
    by default it is 4 and it must be at least 2.
    """

    def __init__(self, page_unit=PagingUnit.ROWS, max_pages=0, max_pages_per_second=0, max_queue_size=4):
        self.page_unit = page_unit
        self.max_pages = max_pages
        self.max_pages_per_second = max_pages_per_second
        if max_queue_size < 2:
            raise ValueError('ContinuousPagingOptions.max_queue_size must be 2 or greater')
        self.max_queue_size = max_queue_size

    def page_unit_bytes(self):
        return self.page_unit == ContinuousPagingOptions.PagingUnit.BYTES


def _addrinfo_or_none(contact_point, port):
    """
    A helper function that wraps socket.getaddrinfo and returns None
    when it fails to, e.g. resolve one of the hostnames. Used to address
    PYTHON-895.
    """
    try:
        return socket.getaddrinfo(contact_point, port,
                                  socket.AF_UNSPEC, socket.SOCK_STREAM)
    except socket.gaierror:
        log.debug('Could not resolve hostname "{}" '
                  'with port {}'.format(contact_point, port))
        return None


def _execution_profile_to_string(name):
    default_profiles = {
        EXEC_PROFILE_DEFAULT: 'EXEC_PROFILE_DEFAULT',
        EXEC_PROFILE_GRAPH_DEFAULT: 'EXEC_PROFILE_GRAPH_DEFAULT',
        EXEC_PROFILE_GRAPH_SYSTEM_DEFAULT: 'EXEC_PROFILE_GRAPH_SYSTEM_DEFAULT',
        EXEC_PROFILE_GRAPH_ANALYTICS_DEFAULT: 'EXEC_PROFILE_GRAPH_ANALYTICS_DEFAULT',
    }

    if name in default_profiles:
        return default_profiles[name]

    return '"%s"' % (name,)


class ExecutionProfile(object):
    load_balancing_policy = None
    """
    An instance of :class:`.policies.LoadBalancingPolicy` or one of its subclasses.

    Used in determining host distance for establishing connections, and routing requests.

    Defaults to ``TokenAwarePolicy(DCAwareRoundRobinPolicy())`` if not specified
    """

    retry_policy = None
    """
    An instance of :class:`.policies.RetryPolicy` instance used when :class:`.Statement` objects do not have a
    :attr:`~.Statement.retry_policy` explicitly set.

    Defaults to :class:`.RetryPolicy` if not specified
    """

    consistency_level = ConsistencyLevel.LOCAL_ONE
    """
    :class:`.ConsistencyLevel` used when not specified on a :class:`.Statement`.
    """

    serial_consistency_level = None
    """
    Serial :class:`.ConsistencyLevel` used when not specified on a :class:`.Statement` (for LWT conditional statements).
    """

    request_timeout = 10.0
    """
    Request timeout used when not overridden in :meth:`.Session.execute`
    """

    row_factory = staticmethod(named_tuple_factory)
    """
    A callable to format results, accepting ``(colnames, rows)`` where ``colnames`` is a list of column names, and
    ``rows`` is a list of tuples, with each tuple representing a row of parsed values.

    Some example implementations:

    - :func:`cassandra.query.tuple_factory` - return a result row as a tuple
    - :func:`cassandra.query.named_tuple_factory` - return a result row as a named tuple
    - :func:`cassandra.query.dict_factory` - return a result row as a dict
    - :func:`cassandra.query.ordered_dict_factory` - return a result row as an OrderedDict
    """

    speculative_execution_policy = None
    """
    An instance of :class:`.policies.SpeculativeExecutionPolicy`

    Defaults to :class:`.NoSpeculativeExecutionPolicy` if not specified
    """

    continuous_paging_options = None
    """
    *Note:* This feature is implemented to facilitate server integration testing. It is not intended for general use in the Python driver.
    See :attr:`.Statement.fetch_size` or :attr:`Session.default_fetch_size` for configuring normal paging.

    When set, requests will use DSE's continuous paging, which streams multiple pages without
    intermediate requests.

    This has the potential to materialize all results in memory at once if the consumer cannot keep up. Use options
    to constrain page size and rate.

    This is only available for DSE clusters.
    """

    # 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=_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()

        if (serial_consistency_level is not None and
                not ConsistencyLevel.is_serial(serial_consistency_level)):
            raise ValueError("serial_consistency_level must be either "
                             "ConsistencyLevel.SERIAL "
                             "or ConsistencyLevel.LOCAL_SERIAL.")
        self.serial_consistency_level = serial_consistency_level

        self.request_timeout = request_timeout
        self.row_factory = row_factory
        self.speculative_execution_policy = speculative_execution_policy or NoSpeculativeExecutionPolicy()
        self.continuous_paging_options = continuous_paging_options


class GraphExecutionProfile(ExecutionProfile):
    graph_options = None
    """
    :class:`.GraphOptions` to use with this execution

    Default options for graph queries, initialized as follows by default::

        GraphOptions(graph_language=b'gremlin-groovy')

    See cassandra.graph.GraphOptions
    """

    def __init__(self, load_balancing_policy=_NOT_SET, retry_policy=None,
                 consistency_level=_NOT_SET, serial_consistency_level=None,
                 request_timeout=30.0, row_factory=None,
                 graph_options=None, continuous_paging_options=_NOT_SET):
        """
        Default execution profile for graph execution.

        See :class:`.ExecutionProfile` for base attributes. Note that if not explicitly set,
        the row_factory and graph_options.graph_protocol are resolved during the query execution.
        These options will resolve to graph_graphson3_row_factory and GraphProtocol.GRAPHSON_3_0
        for the core graph engine (DSE 6.8+), otherwise graph_object_row_factory and GraphProtocol.GRAPHSON_1_0

        In addition to default parameters shown in the signature, this profile also defaults ``retry_policy`` to
        :class:`cassandra.policies.NeverRetryPolicy`.
        """
        retry_policy = retry_policy or NeverRetryPolicy()
        super(GraphExecutionProfile, self).__init__(load_balancing_policy, retry_policy, consistency_level,
                                                    serial_consistency_level, request_timeout, row_factory,
                                                    continuous_paging_options=continuous_paging_options)
        self.graph_options = graph_options or GraphOptions(graph_source=b'g',
                                                           graph_language=b'gremlin-groovy')


class GraphAnalyticsExecutionProfile(GraphExecutionProfile):

    def __init__(self, load_balancing_policy=None, retry_policy=None,
                 consistency_level=_NOT_SET, serial_consistency_level=None,
                 request_timeout=3600. * 24. * 7., row_factory=None,
                 graph_options=None):
        """
        Execution profile with timeout and load balancing appropriate for graph analytics queries.

        See also :class:`~.GraphExecutionPolicy`.

        In addition to default parameters shown in the signature, this profile also defaults ``retry_policy`` to
        :class:`cassandra.policies.NeverRetryPolicy`, and ``load_balancing_policy`` to one that targets the current Spark
        master.

        Note: The graph_options.graph_source is set automatically to b'a' (analytics)
        when using GraphAnalyticsExecutionProfile. This is mandatory to target analytics nodes.
        """
        load_balancing_policy = load_balancing_policy or DefaultLoadBalancingPolicy(default_lbp_factory())
        graph_options = graph_options or GraphOptions(graph_language=b'gremlin-groovy')
        super(GraphAnalyticsExecutionProfile, self).__init__(load_balancing_policy, retry_policy, consistency_level,
                                                             serial_consistency_level, request_timeout, row_factory,
                                                             graph_options)
        # ensure the graph_source is analytics, since this is the purpose of the GraphAnalyticsExecutionProfile
        self.graph_options.set_source_analytics()


class ProfileManager(object):

    def __init__(self):
        self.profiles = dict()

    def _profiles_without_explicit_lbps(self):
        names = (profile_name for
                 profile_name, profile in self.profiles.items()
                 if not profile._load_balancing_policy_explicit)
        return tuple(
            'EXEC_PROFILE_DEFAULT' if n is EXEC_PROFILE_DEFAULT else n
            for n in names
        )

    def distance(self, host):
        distances = set(p.load_balancing_policy.distance(host) for p in self.profiles.values())
        return HostDistance.LOCAL if HostDistance.LOCAL in distances else \
            HostDistance.REMOTE if HostDistance.REMOTE in distances else \
            HostDistance.IGNORED

    def populate(self, cluster, hosts):
        for p in self.profiles.values():
            p.load_balancing_policy.populate(cluster, hosts)

    def check_supported(self):
        for p in self.profiles.values():
            p.load_balancing_policy.check_supported()

    def on_up(self, host):
        for p in self.profiles.values():
            p.load_balancing_policy.on_up(host)

    def on_down(self, host):
        for p in self.profiles.values():
            p.load_balancing_policy.on_down(host)

    def on_add(self, host):
        for p in self.profiles.values():
            p.load_balancing_policy.on_add(host)

    def on_remove(self, host):
        for p in self.profiles.values():
            p.load_balancing_policy.on_remove(host)

    @property
    def default(self):
        """
        internal-only; no checks are done because this entry is populated on cluster init
        """
        return self.profiles[EXEC_PROFILE_DEFAULT]


EXEC_PROFILE_DEFAULT = object()
"""
Key for the ``Cluster`` default execution profile, used when no other profile is selected in
``Session.execute(execution_profile)``.

Use this as the key in ``Cluster(execution_profiles)`` to override the default profile.
"""

EXEC_PROFILE_GRAPH_DEFAULT = object()
"""
Key for the default graph execution profile, used when no other profile is selected in
``Session.execute_graph(execution_profile)``.

Use this as the key in :doc:`Cluster(execution_profiles) </execution_profiles>`
to override the default graph profile.
"""

EXEC_PROFILE_GRAPH_SYSTEM_DEFAULT = object()
"""
Key for the default graph system execution profile. This can be used for graph statements using the DSE graph
system API.

Selected using ``Session.execute_graph(execution_profile=EXEC_PROFILE_GRAPH_SYSTEM_DEFAULT)``.
"""

EXEC_PROFILE_GRAPH_ANALYTICS_DEFAULT = object()
"""
Key for the default graph analytics execution profile. This can be used for graph statements intended to
use Spark/analytics as the traversal source.

Selected using ``Session.execute_graph(execution_profile=EXEC_PROFILE_GRAPH_ANALYTICS_DEFAULT)``.
"""


class _ConfigMode(object):
    UNCOMMITTED = 0
    LEGACY = 1
    PROFILES = 2


class Cluster(object):
    """
    The main class to use when interacting with a Cassandra cluster.
    Typically, one instance of this class will be created for each
    separate Cassandra cluster that your application interacts with.

    Example usage::

        >>> from cassandra.cluster import Cluster
        >>> cluster = Cluster(['192.168.1.1', '192.168.1.2'])
        >>> session = cluster.connect()
        >>> session.execute("CREATE KEYSPACE ...")
        >>> ...
        >>> cluster.shutdown()

    ``Cluster`` and ``Session`` also provide context management functions
    which implicitly handle shutdown when leaving scope.
    """

    contact_points = ['127.0.0.1']
    """
    The list of contact points to try connecting for cluster discovery. A
    contact point can be a string (ip or hostname), a tuple (ip/hostname, port) or a
    :class:`.connection.EndPoint` instance.

    Defaults to loopback interface.

    Note: When using :class:`.DCAwareLoadBalancingPolicy` with no explicit
    local_dc set (as is the default), the DC is chosen from an arbitrary
    host in contact_points. In this case, contact_points should contain
    only nodes from a single, local DC.

    Note: In the next major version, if you specify contact points, you will
    also be required to also explicitly specify a load-balancing policy. This
    change will help prevent cases where users had hard-to-debug issues
    surrounding unintuitive default load-balancing policy behavior.
    """
    # tracks if contact_points was set explicitly or with default values
    _contact_points_explicit = None

    port = 9042
    """
    The server-side port to open connections to. Defaults to 9042.
    """

    cql_version = None
    """
    If a specific version of CQL should be used, this may be set to that
    string version.  Otherwise, the highest CQL version supported by the
    server will be automatically used.
    """

    protocol_version = ProtocolVersion.DSE_V2
    """
    The maximum version of the native protocol to use.

    See :class:`.ProtocolVersion` for more information about versions.

    If not set in the constructor, the driver will automatically downgrade
    version based on a negotiation with the server, but it is most efficient
    to set this to the maximum supported by your version of Cassandra.
    Setting this will also prevent conflicting versions negotiated if your
    cluster is upgraded.

    """

    allow_beta_protocol_version = False

    no_compact = False

    """
    Setting true injects a flag in all messages that makes the server accept and use "beta" protocol version.
    Used for testing new protocol features incrementally before the new version is complete.
    """

    compression = True
    """
    Controls compression for communications between the driver and Cassandra.
    If left as the default of :const:`True`, either lz4 or snappy compression
    may be used, depending on what is supported by both the driver
    and Cassandra.  If both are fully supported, lz4 will be preferred.

    You may also set this to 'snappy' or 'lz4' to request that specific
    compression type.

    Setting this to :const:`False` disables compression.
    """

    _auth_provider = None
    _auth_provider_callable = None

    @property
    def auth_provider(self):
        """
        When :attr:`~.Cluster.protocol_version` is 2 or higher, this should
        be an instance of a subclass of :class:`~cassandra.auth.AuthProvider`,
        such as :class:`~.PlainTextAuthProvider`.

        When :attr:`~.Cluster.protocol_version` is 1, this should be
        a function that accepts one argument, the IP address of a node,
        and returns a dict of credentials for that node.

        When not using authentication, this should be left as :const:`None`.
        """
        return self._auth_provider

    @auth_provider.setter  # noqa
    def auth_provider(self, value):
        if not value:
            self._auth_provider = value
            return

        try:
            self._auth_provider_callable = value.new_authenticator
        except AttributeError:
            if self.protocol_version > 1:
                raise TypeError("auth_provider must implement the cassandra.auth.AuthProvider "
                                "interface when protocol_version >= 2")
            elif not callable(value):
                raise TypeError("auth_provider must be callable when protocol_version == 1")
            self._auth_provider_callable = value

        self._auth_provider = value

    _load_balancing_policy = None
    @property
    def load_balancing_policy(self):
        """
        An instance of :class:`.policies.LoadBalancingPolicy` or
        one of its subclasses.

        .. versionchanged:: 2.6.0

        Defaults to :class:`~.TokenAwarePolicy` (:class:`~.DCAwareRoundRobinPolicy`).
        when using CPython (where the murmur3 extension is available). :class:`~.DCAwareRoundRobinPolicy`
        otherwise. Default local DC will be chosen from contact points.

        **Please see** :class:`~.DCAwareRoundRobinPolicy` **for a discussion on default behavior with respect to
        DC locality and remote nodes.**
        """
        return self._load_balancing_policy

    @load_balancing_policy.setter
    def load_balancing_policy(self, lbp):
        if self._config_mode == _ConfigMode.PROFILES:
            raise ValueError("Cannot set Cluster.load_balancing_policy while using Configuration Profiles. Set this in a profile instead.")
        self._load_balancing_policy = lbp
        self._config_mode = _ConfigMode.LEGACY

    @property
    def _default_load_balancing_policy(self):
        return self.profile_manager.default.load_balancing_policy

    reconnection_policy = ExponentialReconnectionPolicy(1.0, 600.0)
    """
    An instance of :class:`.policies.ReconnectionPolicy`. Defaults to an instance
    of :class:`.ExponentialReconnectionPolicy` with a base delay of one second and
    a max delay of ten minutes.
    """

    _default_retry_policy = RetryPolicy()
    @property
    def default_retry_policy(self):
        """
        A default :class:`.policies.RetryPolicy` instance to use for all
        :class:`.Statement` objects which do not have a :attr:`~.Statement.retry_policy`
        explicitly set.
        """
        return self._default_retry_policy

    @default_retry_policy.setter
    def default_retry_policy(self, policy):
        if self._config_mode == _ConfigMode.PROFILES:
            raise ValueError("Cannot set Cluster.default_retry_policy while using Configuration Profiles. Set this in a profile instead.")
        self._default_retry_policy = policy
        self._config_mode = _ConfigMode.LEGACY

    conviction_policy_factory = SimpleConvictionPolicy
    """
    A factory function which creates instances of
    :class:`.policies.ConvictionPolicy`.  Defaults to
    :class:`.policies.SimpleConvictionPolicy`.
    """

    address_translator = IdentityTranslator()
    """
    :class:`.policies.AddressTranslator` instance to be used in translating server node addresses
    to driver connection addresses.
    """

    connect_to_remote_hosts = True
    """
    If left as :const:`True`, hosts that are considered :attr:`~.HostDistance.REMOTE`
    by the :attr:`~.Cluster.load_balancing_policy` will have a connection
    opened to them.  Otherwise, they will not have a connection opened to them.

    Note that the default load balancing policy ignores remote hosts by default.

    .. versionadded:: 2.1.0
    """

    metrics_enabled = False
    """
    Whether or not metric collection is enabled.  If enabled, :attr:`.metrics`
    will be an instance of :class:`~cassandra.metrics.Metrics`.
    """

    metrics = None
    """
    An instance of :class:`cassandra.metrics.Metrics` if :attr:`.metrics_enabled` is
    :const:`True`, else :const:`None`.
    """

    ssl_options = None
    """
    Using ssl_options without ssl_context is deprecated and will be removed in the
    next major release.

    An optional dict which will be used as kwargs for ``ssl.SSLContext.wrap_socket`` (or
    ``ssl.wrap_socket()`` if used without ssl_context) when new sockets are created.
    This should be used when client encryption is enabled in Cassandra.

    The following documentation only applies when ssl_options is used without ssl_context.

    By default, a ``ca_certs`` value should be supplied (the value should be
    a string pointing to the location of the CA certs file), and you probably
    want to specify ``ssl_version`` as ``ssl.PROTOCOL_TLSv1`` to match
    Cassandra's default protocol.

    .. versionchanged:: 3.3.0

    In addition to ``wrap_socket`` kwargs, clients may also specify ``'check_hostname': True`` to verify the cert hostname
    as outlined in RFC 2818 and RFC 6125. Note that this requires the certificate to be transferred, so
    should almost always require the option ``'cert_reqs': ssl.CERT_REQUIRED``. Note also that this functionality was not built into
    Python standard library until (2.7.9, 3.2). To enable this mechanism in earlier versions, patch ``ssl.match_hostname``
    with a custom or `back-ported function <https://pypi.org/project/backports.ssl_match_hostname/>`_.
    """

    ssl_context = None
    """
    An optional ``ssl.SSLContext`` instance which will be used when new sockets are created.
    This should be used when client encryption is enabled in Cassandra.

    ``wrap_socket`` options can be set using :attr:`~Cluster.ssl_options`. ssl_options will
    be used as kwargs for ``ssl.SSLContext.wrap_socket``.

    .. versionadded:: 3.17.0
    """

    sockopts = None
    """
    An optional list of tuples which will be used as arguments to
    ``socket.setsockopt()`` for all created sockets.

    Note: some drivers find setting TCPNODELAY beneficial in the context of
    their execution model. It was not found generally beneficial for this driver.
    To try with your own workload, set ``sockopts = [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]``
    """

    max_schema_agreement_wait = 10
    """
    The maximum duration (in seconds) that the driver will wait for schema
    agreement across the cluster. Defaults to ten seconds.
    If set <= 0, the driver will bypass schema agreement waits altogether.
    """

    metadata = None
    """
    An instance of :class:`cassandra.metadata.Metadata`.
    """

    connection_class = DefaultConnection
    """
    This determines what event loop system will be used for managing
    I/O with Cassandra.  These are the current options:

    * :class:`cassandra.io.asyncorereactor.AsyncoreConnection`
    * :class:`cassandra.io.libevreactor.LibevConnection`
    * :class:`cassandra.io.eventletreactor.EventletConnection` (requires monkey-patching - see doc for details)
    * :class:`cassandra.io.geventreactor.GeventConnection` (requires monkey-patching - see doc for details)
    * :class:`cassandra.io.twistedreactor.TwistedConnection`
    * EXPERIMENTAL: :class:`cassandra.io.asyncioreactor.AsyncioConnection`

    By default, ``AsyncoreConnection`` will be used, which uses
    the ``asyncore`` module in the Python standard library.

    If ``libev`` is installed, ``LibevConnection`` will be used instead.

    If ``gevent`` or ``eventlet`` monkey-patching is detected, the corresponding
    connection class will be used automatically.

    ``AsyncioConnection``, which uses the ``asyncio`` module in the Python
    standard library, is also available, but currently experimental. Note that
    it requires ``asyncio`` features that were only introduced in the 3.4 line
    in 3.4.6, and in the 3.5 line in 3.5.1.
    """

    control_connection_timeout = 2.0
    """
    A timeout, in seconds, for queries made by the control connection, such
    as querying the current schema and information about nodes in the cluster.
    If set to :const:`None`, there will be no timeout for these queries.
    """

    idle_heartbeat_interval = 30
    """
    Interval, in seconds, on which to heartbeat idle connections. This helps
    keep connections open through network devices that expire idle connections.
    It also helps discover bad connections early in low-traffic scenarios.
    Setting to zero disables heartbeats.
    """

    idle_heartbeat_timeout = 30
    """
    Timeout, in seconds, on which the heartbeat wait for idle connection responses.
    Lowering this value can help to discover bad connections earlier.
    """

    schema_event_refresh_window = 2
    """
    Window, in seconds, within which a schema component will be refreshed after
    receiving a schema_change event.

    The driver delays a random amount of time in the range [0.0, window)
    before executing the refresh. This serves two purposes:

    1.) Spread the refresh for deployments with large fanout from C* to client tier,
    preventing a 'thundering herd' problem with many clients refreshing simultaneously.

    2.) Remove redundant refreshes. Redundant events arriving within the delay period
    are discarded, and only one refresh is executed.

    Setting this to zero will execute refreshes immediately.

    Setting this negative will disable schema refreshes in response to push events
    (refreshes will still occur in response to schema change responses to DDL statements
    executed by Sessions of this Cluster).
    """

    topology_event_refresh_window = 10
    """
    Window, in seconds, within which the node and token list will be refreshed after
    receiving a topology_change event.

    Setting this to zero will execute refreshes immediately.

    Setting this negative will disable node refreshes in response to push events.

    See :attr:`.schema_event_refresh_window` for discussion of rationale
    """

    status_event_refresh_window = 2
    """
    Window, in seconds, within which the driver will start the reconnect after
    receiving a status_change event.

    Setting this to zero will connect immediately.

    This is primarily used to avoid 'thundering herd' in deployments with large fanout from cluster to clients.
    When nodes come up, clients attempt to reprepare prepared statements (depending on :attr:`.reprepare_on_up`), and
    establish connection pools. This can cause a rush of connections and queries if not mitigated with this factor.
    """

    prepare_on_all_hosts = True
    """
    Specifies whether statements should be prepared on all hosts, or just one.

    This can reasonably be disabled on long-running applications with numerous clients preparing statements on startup,
    where a randomized initial condition of the load balancing policy can be expected to distribute prepares from
    different clients across the cluster.
    """

    reprepare_on_up = True
    """
    Specifies whether all known prepared statements should be prepared on a node when it comes up.

    May be used to avoid overwhelming a node on return, or if it is supposed that the node was only marked down due to
    network. If statements are not reprepared, they are prepared on the first execution, causing
    an extra roundtrip for one or more client requests.
    """

    connect_timeout = 5
    """
    Timeout, in seconds, for creating new connections.

    This timeout covers the entire connection negotiation, including TCP
    establishment, options passing, and authentication.
    """

    timestamp_generator = None
    """
    An object, shared between all sessions created by this cluster instance,
    that generates timestamps when client-side timestamp generation is enabled.
    By default, each :class:`Cluster` uses a new
    :class:`~.MonotonicTimestampGenerator`.

    Applications can set this value for custom timestamp behavior. See the
    documentation for :meth:`Session.timestamp_generator`.
    """

    monitor_reporting_enabled = True
    """
    A boolean indicating if monitor reporting, which sends gathered data to
    Insights when running against DSE 6.8 and higher.
    """

    monitor_reporting_interval = 30
    """
    A boolean indicating if monitor reporting, which sends gathered data to
    Insights when running against DSE 6.8 and higher.
    """

    client_id = None
    """
    A UUID that uniquely identifies this Cluster object to Insights. This will
    be generated automatically unless the user provides one.
    """

    application_name = ''
    """
    A string identifying this application to Insights.
    """

    application_version = ''
    """
    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',

            # optional config options
            'use_default_tempdir': True  # use the system temp dir for the zip extraction
        }