Connect to a Cassandra cluster.

    :param contact_points: The Cassandra cluster addresses, can either be a string or a list of IPs.
    :type  contact_points: str or list of str
    :param cql_user:       The Cassandra user if authentication is turned on.
    :type  cql_user:       str
    :param cql_pass:       The Cassandra user password if authentication is turned on.
    :type  cql_pass:       str
    :param port:           The Cassandra cluster port, defaults to None.
    :type  port:           int
    :param protocol_version:  Cassandra protocol version to use.
    :type  port:           int
    :return:               The session and cluster objects.
    :rtype:                cluster object, session object

def _connect(contact_points=None, port=None, cql_user=None, cql_pass=None,
             protocol_version=None):
    '''
    Connect to a Cassandra cluster.

    :param contact_points: The Cassandra cluster addresses, can either be a string or a list of IPs.
    :type  contact_points: str or list of str
    :param cql_user:       The Cassandra user if authentication is turned on.
    :type  cql_user:       str
    :param cql_pass:       The Cassandra user password if authentication is turned on.
    :type  cql_pass:       str
    :param port:           The Cassandra cluster port, defaults to None.
    :type  port:           int
    :param protocol_version:  Cassandra protocol version to use.
    :type  port:           int
    :return:               The session and cluster objects.
    :rtype:                cluster object, session object
    '''
    # Lazy load the Cassandra cluster and session for this module by creating a
    # cluster and session when cql_query is called the first time. Get the
    # Cassandra cluster and session from this module's __context__ after it is
    # loaded the first time cql_query is called.
    #
    # TODO: Call cluster.shutdown() when the module is unloaded on
    # master/minion shutdown. Currently, Master.shutdown() and Minion.shutdown()
    # do nothing to allow loaded modules to gracefully handle resources stored
    # in __context__ (i.e. connection pools). This means that the connection
    # pool is orphaned and Salt relies on Cassandra to reclaim connections.
    # Perhaps if Master/Minion daemons could be enhanced to call an "__unload__"
    # function, or something similar for each loaded module, connection pools
    # and the like can be gracefully reclaimed/shutdown.
    if (__context__
        and 'cassandra_cql_returner_cluster' in __context__
        and 'cassandra_cql_returner_session' in __context__):
        return __context__['cassandra_cql_returner_cluster'], __context__['cassandra_cql_returner_session']
    else:

        contact_points = _load_properties(property_name=contact_points, config_option='cluster')
        contact_points = contact_points if isinstance(contact_points, list) else contact_points.split(',')
        port = _load_properties(property_name=port, config_option='port', set_default=True, default=9042)
        cql_user = _load_properties(property_name=cql_user, config_option='username', set_default=True, default="cassandra")
        cql_pass = _load_properties(property_name=cql_pass, config_option='password', set_default=True, default="cassandra")
        protocol_version = _load_properties(property_name=protocol_version,
                                            config_option='protocol_version',
                                            set_default=True, default=4)

        try:
            auth_provider = PlainTextAuthProvider(username=cql_user, password=cql_pass)
            ssl_opts = _get_ssl_opts()
            if ssl_opts:
                cluster = Cluster(contact_points,
                                  port=port,
                                  auth_provider=auth_provider,
                                  ssl_options=ssl_opts,
                                  protocol_version=protocol_version,
                                  compression=True)
            else:
                cluster = Cluster(contact_points, port=port,
                                  auth_provider=auth_provider,
                                  protocol_version=protocol_version,
                                  compression=True)
            for recontimes in range(1, 4):
                try:
                    session = cluster.connect()
                    break
                except OperationTimedOut:
                    log.warning('Cassandra cluster.connect timed out, try %s', recontimes)
                    if recontimes >= 3:
                        raise

            # TODO: Call cluster.shutdown() when the module is unloaded on shutdown.
            __context__['cassandra_cql_returner_cluster'] = cluster
            __context__['cassandra_cql_returner_session'] = session
            __context__['cassandra_cql_prepared'] = {}

            log.debug('Successfully connected to Cassandra cluster at %s', contact_points)
            return cluster, session
        except TypeError:
            pass
        except (ConnectionException, ConnectionShutdown, NoHostAvailable):
            log.error('Could not connect to Cassandra cluster at %s', contact_points)
            raise CommandExecutionError('ERROR: Could not connect to Cassandra cluster.')