def get_status():
    """Returns the status of the elastic cluster
    """
    es = get_elasticsearch()
    try:
        cluster_health = es.cluster.health()
    except ConnectionError as err:
        logging.getLogger(__name__).error("Failed to connect to ES: %s", err)
        cluster_health = {}
        es_reachable = False
    else:
        es_reachable = True

    es_cluster_health = cluster_health.get("status") in ES_RUNNING_STATUS

    # the 'ready' is used as the readyness probe.
    # for the moment idunn is ready if ES is reachable
    ready = es_cluster_health
    return {"es": {"reachable": es_reachable, "running": es_cluster_health}, "ready": ready} 

def search(self, project_id, user_id=None, doc_id=None, search=None,
               offset=0, limit=10):
        search = search or {}
        query_dsl = self.get_search_query(
            project_id=project_id,
            user_id=user_id,
            doc_id=doc_id,
            search=search
        )
        try:
            res = self.es.search(index=self.index, doc_type=self.doc_type,
                                 size=limit, from_=offset, body=query_dsl)
        except elasticsearch.ConnectionError:
            raise freezer_api_exc.StorageEngineError(
                message='unable to connect to db server')
        except Exception as e:
            raise freezer_api_exc.StorageEngineError(
                message='search operation failed: {0}'.format(e))
        hit_list = res['hits']['hits']
        return [x['_source'] for x in hit_list] 

def test_execute_single_with_connection_error_continues(self):
        import elasticsearch
        es = None
        params = None
        # ES client uses pseudo-status "N/A" in this case...
        runner = mock.Mock(side_effect=as_future(exception=elasticsearch.ConnectionError("N/A", "no route to host", None)))

        ops, unit, request_meta_data = await driver.execute_single(
            self.context_managed(runner), es, params, on_error="continue")

        self.assertEqual(0, ops)
        self.assertEqual("ops", unit)
        self.assertEqual({
            # Look ma: No http-status!
            "error-description": "no route to host",
            "error-type": "transport",
            "success": False
        }, request_meta_data) 

def test_search_raise_StorageEngineError_when_ConnectionError(self,
                                                                  mock_es):
        self.mock_es.search.side_effect = elasticsearch.ConnectionError(
            'regular test failure')
        self.assertRaises(exceptions.StorageEngineError,
                          self.type_manager.search, project_id='tecs',
                          user_id='my_user_id', doc_id='mydocid') 

def test_search_raise_StorageEngineError_when_ConnectionError(self,
                                                                  mock_es):
        self.mock_es.search.side_effect = elasticsearch.ConnectionError(
            'regular test failure')
        self.assertRaises(exceptions.StorageEngineError,
                          self.type_manager.search,
                          user_id='my_user_id', doc_id='mydocid') 

def search(self, user_id, doc_id=None, search=None, offset=0, limit=10):
        search = search or {}
        query_dsl = self.get_search_query(user_id, doc_id, search)
        try:
            res = self.es.search(index=self.index, doc_type=self.doc_type,
                                 size=limit, from_=offset, body=query_dsl)
        except elasticsearch.ConnectionError:
            raise freezer_api_exc.StorageEngineError(
                message=_i18n._('unable to connect to db server'))
        except Exception as e:
            raise freezer_api_exc.StorageEngineError(
                message=_i18n._('search operation failed: %s') % e)
        hit_list = res['hits']['hits']
        return [x['_source'] for x in hit_list] 

def wait_for_rest_layer(es, max_attempts=40):
    """
    Waits for ``max_attempts`` until Elasticsearch's REST API is available.

    :param es: Elasticsearch client to use for connecting.
    :param max_attempts: The maximum number of attempts to check whether the REST API is available.
    :return: True iff Elasticsearch's REST API is available.
    """
    # assume that at least the hosts that we expect to contact should be available. Note that this is not 100%
    # bullet-proof as a cluster could have e.g. dedicated masters which are not contained in our list of target hosts
    # but this is still better than just checking for any random node's REST API being reachable.
    expected_node_count = len(es.transport.hosts)
    logger = logging.getLogger(__name__)
    for attempt in range(max_attempts):
        logger.debug("REST API is available after %s attempts", attempt)
        import elasticsearch
        try:
            # see also WaitForHttpResource in Elasticsearch tests. Contrary to the ES tests we consider the API also
            # available when the cluster status is RED (as long as all required nodes are present)
            es.cluster.health(wait_for_nodes=">={}".format(expected_node_count))
            logger.info("REST API is available for >= [%s] nodes after [%s] attempts.", expected_node_count, attempt)
            return True
        except elasticsearch.ConnectionError as e:
            if "SSL: UNKNOWN_PROTOCOL" in str(e):
                raise exceptions.SystemSetupError("Could not connect to cluster via https. Is this an https endpoint?", e)
            else:
                logger.debug("Got connection error on attempt [%s]. Sleeping...", attempt)
                time.sleep(3)
        except elasticsearch.TransportError as e:
            # cluster block, x-pack not initialized yet, our wait condition is not reached
            if e.status_code in (503, 401, 408):
                logger.debug("Got status code [%s] on attempt [%s]. Sleeping...", e.status_code, attempt)
                time.sleep(3)
            else:
                logger.warning("Got unexpected status code [%s] on attempt [%s].", e.status_code, attempt)
                raise e
    return False 

def test_ssl_error(self, es):
        import elasticsearch
        import urllib3.exceptions

        es.cluster.health.side_effect = elasticsearch.ConnectionError("N/A",
                                                            "[SSL: UNKNOWN_PROTOCOL] unknown protocol (_ssl.c:719)",
                                                            urllib3.exceptions.SSLError(
                                                                "[SSL: UNKNOWN_PROTOCOL] unknown protocol (_ssl.c:719)"))
        with self.assertRaisesRegex(expected_exception=exceptions.SystemSetupError,
                                    expected_regex="Could not connect to cluster via https. Is this an https endpoint?"):
            client.wait_for_rest_layer(es, max_attempts=3) 

def test_execute_single_with_connection_error_aborts_as_fatal(self):
        import elasticsearch
        es = None
        params = None
        # ES client uses pseudo-status "N/A" in this case...
        runner = mock.Mock(side_effect=as_future(exception=elasticsearch.ConnectionError("N/A", "no route to host", None)))

        with self.assertRaises(exceptions.RallyAssertionError) as ctx:
            await driver.execute_single(self.context_managed(runner), es, params, on_error="continue-on-non-fatal")
        self.assertEqual(
            "Request returned an error. Error type: transport, Description: no route to host",
            ctx.exception.args[0]) 

def test_retries_on_timeout_if_wanted_and_raises_if_no_recovery(self):
        import elasticsearch

        delegate = mock.Mock(side_effect=[
            as_future(exception=elasticsearch.ConnectionError("N/A", "no route to host")),
            as_future(exception=elasticsearch.ConnectionError("N/A", "no route to host")),
            as_future(exception=elasticsearch.ConnectionError("N/A", "no route to host")),
            as_future(exception=elasticsearch.ConnectionError("N/A", "no route to host"))
        ])
        es = None
        params = {
            "retries": 3,
            "retry-wait-period": 0.01,
            "retry-on-timeout": True,
            "retry-on-error": True
        }
        retrier = runner.Retry(delegate)

        with self.assertRaises(elasticsearch.ConnectionError):
            await retrier(es, params)

        delegate.assert_has_calls([
            mock.call(es, params),
            mock.call(es, params),
            mock.call(es, params)
        ]) 

def test_retries_on_timeout_if_wanted_and_returns_first_call(self):
        import elasticsearch
        failed_return_value = {"weight": 1, "unit": "ops", "success": False}

        delegate = mock.Mock(side_effect=[
            as_future(exception=elasticsearch.ConnectionError("N/A", "no route to host")),
            as_future(failed_return_value)
        ])
        es = None
        params = {
            "retries": 3,
            "retry-wait-period": 0.01,
            "retry-on-timeout": True,
            "retry-on-error": False
        }
        retrier = runner.Retry(delegate)

        result = await retrier(es, params)
        self.assertEqual(failed_return_value, result)

        delegate.assert_has_calls([
            # has returned a connection error
            mock.call(es, params),
            # has returned normally
            mock.call(es, params)
        ]) 

def test_retries_mixed_timeout_and_application_errors(self):
        import elasticsearch
        connection_error = elasticsearch.ConnectionError("N/A", "no route to host")
        failed_return_value = {"weight": 1, "unit": "ops", "success": False}
        success_return_value = {"weight": 1, "unit": "ops", "success": False}

        delegate = mock.Mock(side_effect=[
            as_future(exception=connection_error),
            as_future(failed_return_value),
            as_future(exception=connection_error),
            as_future(exception=connection_error),
            as_future(failed_return_value),
            as_future(success_return_value)
        ])
        es = None
        params = {
            # we try exactly as often as there are errors to also test the semantics of "retry".
            "retries": 5,
            "retry-wait-period": 0.01,
            "retry-on-timeout": True,
            "retry-on-error": True
        }
        retrier = runner.Retry(delegate)

        result = await retrier(es, params)
        self.assertEqual(success_return_value, result)

        delegate.assert_has_calls([
            # connection error
            mock.call(es, params),
            # application error
            mock.call(es, params),
            # connection error
            mock.call(es, params),
            # connection error
            mock.call(es, params),
            # application error
            mock.call(es, params),
            # success
            mock.call(es, params)
        ]) 

def init_elasticsearch(uri):
    # init ElasticSearch
    es_conn = elasticsearch.Elasticsearch(uri)
    try:
        es_conn.info()
    except elasticsearch.ConnectionError:
        raise ElasticsearchConnectionError(uri)

    return es_conn 

def wait_until_up(self):
        for i in xrange(1024):
            try:
                ret = self.client.cluster.health(wait_for_status='green', request_timeout=15 * 60)
                if ret.get('status') == 'green':
                    break
                sleep(10)
            except ConnectionError:
                pass 

def main_loop(self, method, url, params, body, headers=None, ignore=(), timeout=None):
        for attempt in range(self.max_retries + 1):
            connection = self.get_connection()

            try:
                status, headers, data = yield from connection.perform_request(
                        method, url, params, body, headers=headers, ignore=ignore, timeout=timeout)
            except TransportError as e:
                if method == 'HEAD' and e.status_code == 404:
                    return False

                retry = False
                if isinstance(e, ConnectionTimeout):
                    retry = self.retry_on_timeout
                elif isinstance(e, ConnectionError):
                    retry = True
                elif e.status_code in self.retry_on_status:
                    retry = True

                if retry:
                    # only mark as dead if we are retrying
                    self.mark_dead(connection)
                    # raise exception on last retry
                    if attempt == self.max_retries:
                        raise
                else:
                    raise

            else:
                if method == 'HEAD':
                    return 200 <= status < 300

                # connection didn't fail, confirm it's live status
                self.connection_pool.mark_live(connection)
                if data:
                    data = self.deserializer.loads(data, headers.get('content-type'))
                return data 

def execute(self, size=10):
        """Executes the knowledge base search with provided criteria and returns matching documents.

        Args:
            size (int): The maximum number of records to fetch, default to 10.

        Returns:
            a list of matching documents.
        """
        try:
            # TODO: move the ES API call logic to ES helper
            es_query = self._build_es_query(size=size)

            response = self.client.search(index=self.index, body=es_query)
            results = [hit["_source"] for hit in response["hits"]["hits"]]
            return results
        except EsConnectionError as e:
            logger.error(
                "Unable to connect to Elasticsearch: %s details: %s", e.error, e.info
            )
            raise KnowledgeBaseConnectionError(es_host=self.client.transport.hosts)
        except TransportError as e:
            logger.error(
                "Unexpected error occurred when sending requests to Elasticsearch: %s "
                "Status code: %s details: %s",
                e.error,
                e.status_code,
                e.info,
            )
            raise KnowledgeBaseError
        except ElasticsearchException:
            raise KnowledgeBaseError 

def does_index_exist(
    app_namespace, index_name, es_host=None, es_client=None, connect_timeout=2
):
    """Return boolean flag to indicate whether the specified index exists."""

    es_client = es_client or create_es_client(es_host)
    scoped_index_name = get_scoped_index_name(app_namespace, index_name)

    try:
        # Confirm ES connection with a shorter timeout
        es_client.cluster.health(request_timeout=connect_timeout)
        return es_client.indices.exists(index=scoped_index_name)
    except EsConnectionError as e:
        logger.debug(
            "Unable to connect to Elasticsearch: %s details: %s", e.error, e.info
        )
        raise KnowledgeBaseConnectionError(es_host=es_client.transport.hosts)
    except TransportError as e:
        logger.error(
            "Unexpected error occurred when sending requests to Elasticsearch: %s "
            "Status code: %s details: %s",
            e.error,
            e.status_code,
            e.info,
        )
        raise KnowledgeBaseError
    except ElasticsearchException:
        raise KnowledgeBaseError 

def get_field_names(
    app_namespace, index_name, es_host=None, es_client=None, connect_timeout=2
):
    """Return a list of field names available in the specified index."""

    es_client = es_client or create_es_client(es_host)
    scoped_index_name = get_scoped_index_name(app_namespace, index_name)

    try:
        if not does_index_exist(
            app_namespace, index_name, es_host, es_client, connect_timeout
        ):
            raise ValueError(
                "Elasticsearch index '{}' does not exist.".format(index_name)
            )

        res = es_client.indices.get(index=scoped_index_name)

        if is_es_version_7(es_client):
            all_field_info = res[scoped_index_name]["mappings"]["properties"]
        else:
            all_field_info = res[scoped_index_name]["mappings"][DOC_TYPE]["properties"]
        return all_field_info.keys()
    except EsConnectionError as e:
        logger.debug(
            "Unable to connect to Elasticsearch: %s details: %s", e.error, e.info
        )
        raise KnowledgeBaseConnectionError(es_host=es_client.transport.hosts)
    except TransportError as e:
        logger.error(
            "Unexpected error occurred when sending requests to Elasticsearch: %s "
            "Status code: %s details: %s",
            e.error,
            e.status_code,
            e.info,
        )
        raise KnowledgeBaseError
    except ElasticsearchException:
        raise KnowledgeBaseError 

def get(self, request, *args, **kwargs):
        try:
            return super(ElasticsearchListView, self).get(request, *args, **kwargs)
        except (TransportError, ConnectionError):
            self.es_failed = True
            if self.db_fallback:
                return super(ElasticsearchListView, self).get(request, *args, **kwargs)
            else:
                raise 

def get(self, request, *args, **kwargs):
        try:
            return super(ElasticsearchDetailView, self).get(request, *args, **kwargs)
        except (TransportError, ConnectionError):
            self.es_failed = True
            if self.db_fallback:
                return super(ElasticsearchDetailView, self).get(request, *args, **kwargs)
            else:
                raise 

def dispatch(self, request, *args, **kwargs):
        try:
            r = super(IndexableModelMixin, self).dispatch(request, *args, **kwargs)
        except (ConnectionError, TransportError), e:
            # reset object list
            self.queryset = None
            self.es_failed = True
            # db fallback
            r = super(IndexableModelMixin, self).dispatch(request, *args, **kwargs)
            if settings.DEBUG and isinstance(r.data, dict):
                r.data["filter_fail_cause"] = str(e)

        # Add a failed message in case something went wrong with elasticsearch
        # for example if the cluster went down. 

def dispatch(self, request, *args, **kwargs):
        try:
            r = super(IndexableModelMixin, self).dispatch(request, *args, **kwargs)
        except (ConnectionError, TransportError), e:
            # reset object list
            self.queryset = None
            self.es_failed = True
            # db fallback
            r = super(IndexableModelMixin, self).dispatch(request, *args, **kwargs)
            if settings.DEBUG and isinstance(r.data, dict):
                r.data["filter_fail_cause"] = str(e)

        # Add a failed message in case something went wrong with elasticsearch
        # for example if the cluster went down. 

def execute_single(runner, es, params, on_error):
    """
    Invokes the given runner once and provides the runner's return value in a uniform structure.

    :return: a triple of: total number of operations, unit of operations, a dict of request meta data (may be None).
    """
    import elasticsearch
    fatal_error = False
    try:
        async with runner:
            return_value = await runner(es, params)
        if isinstance(return_value, tuple) and len(return_value) == 2:
            total_ops, total_ops_unit = return_value
            request_meta_data = {"success": True}
        elif isinstance(return_value, dict):
            total_ops = return_value.pop("weight", 1)
            total_ops_unit = return_value.pop("unit", "ops")
            request_meta_data = return_value
            if "success" not in request_meta_data:
                request_meta_data["success"] = True
        else:
            total_ops = 1
            total_ops_unit = "ops"
            request_meta_data = {"success": True}
    except elasticsearch.TransportError as e:
        # we *specifically* want to distinguish connection refused (a node died?) from connection timeouts
        # pylint: disable=unidiomatic-typecheck
        if type(e) is elasticsearch.ConnectionError:
            fatal_error = True

        total_ops = 0
        total_ops_unit = "ops"
        request_meta_data = {
            "success": False,
            "error-type": "transport"
        }
        # The ES client will sometimes return string like "N/A" or "TIMEOUT" for connection errors.
        if isinstance(e.status_code, int):
            request_meta_data["http-status"] = e.status_code
        # connection timeout errors don't provide a helpful description
        if isinstance(e, elasticsearch.ConnectionTimeout):
            request_meta_data["error-description"] = "network connection timed out"
        elif e.info:
            request_meta_data["error-description"] = "%s (%s)" % (e.error, e.info)
        else:
            request_meta_data["error-description"] = e.error
    except KeyError as e:
        logging.getLogger(__name__).exception("Cannot execute runner [%s]; most likely due to missing parameters.", str(runner))
        msg = "Cannot execute [%s]. Provided parameters are: %s. Error: [%s]." % (str(runner), list(params.keys()), str(e))
        raise exceptions.SystemSetupError(msg)

    if not request_meta_data["success"]:
        if on_error == "abort" or (on_error == "continue-on-non-fatal" and fatal_error):
            msg = "Request returned an error. Error type: %s" % request_meta_data.get("error-type", "Unknown")
            description = request_meta_data.get("error-description")
            if description:
                msg += ", Description: %s" % description
            raise exceptions.RallyAssertionError(msg)
    return total_ops, total_ops_unit, request_meta_data 

def _get_sniff_data(self, initial=False):
        previous_sniff = self.last_sniff

        # reset last_sniff timestamp
        self.last_sniff = time.time()

        # use small timeout for the sniffing request, should be a fast api call
        timeout = self.sniff_timeout if not initial else None

        tasks = [
            c.perform_request('GET', '/_nodes/_all/http', timeout=timeout)
            # go through all current connections as well as the
            # seed_connections for good measure
            for c in chain(self.connection_pool.connections, (c for c in self.seed_connections if c not in self.connection_pool.connections))
        ]

        done = ()
        try:
            while tasks:
                # execute sniff requests in parallel, wait for first to return
                done, tasks = yield from asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED, loop=self.loop)
                # go through all the finished tasks
                for t in done:
                    try:
                        _, headers, node_info = t.result()
                        node_info = self.deserializer.loads(node_info, headers.get('content-type'))
                    except (ConnectionError, SerializationError) as e:
                        logger.warn('Sniffing request failed with %r', e)
                        continue
                    node_info = list(node_info['nodes'].values())
                    return node_info
            else:
                # no task has finished completely
                raise TransportError("N/A", "Unable to sniff hosts.")
        except:
            # keep the previous value on error
            self.last_sniff = previous_sniff
            raise
        finally:
            # clean up pending futures
            for t in chain(done, tasks):
                t.cancel() 

def _load_field_info(self, index):
        """load knowledge base field metadata information for the specified index.

        Args:
            index (str): index name.
        """

        # load field info from local cache
        index_info = self._es_field_info.get(index, {})

        if not index_info:
            try:
                # TODO: move the ES API call logic to ES helper
                self._es_field_info[index] = {}
                res = self._es_client.indices.get(index=index)
                if is_es_version_7(self._es_client):
                    all_field_info = res[index]["mappings"]["properties"]
                else:
                    all_field_info = res[index]["mappings"][DOC_TYPE]["properties"]
                for field_name in all_field_info:
                    field_type = all_field_info[field_name].get("type")
                    self._es_field_info[index][field_name] = FieldInfo(
                        field_name, field_type
                    )
            except EsConnectionError as e:
                logger.error(
                    "Unable to connect to Elasticsearch: %s details: %s",
                    e.error,
                    e.info,
                )
                raise KnowledgeBaseConnectionError(
                    es_host=self._es_client.transport.hosts
                )
            except TransportError as e:
                logger.error(
                    "Unexpected error occurred when sending requests to Elasticsearch: %s "
                    "Status code: %s details: %s",
                    e.error,
                    e.status_code,
                    e.info,
                )
                raise KnowledgeBaseError
            except ElasticsearchException:
                raise KnowledgeBaseError 

def create_index(
    app_namespace, index_name, mapping, es_host=None, es_client=None, connect_timeout=2
):
    """Creates a new index.

    Args:
        app_namespace (str): The namespace of the app
        index_name (str): The name of the new index to be created
        mapping (str): The Elasticsearch index mapping to use
        es_host (str): The Elasticsearch host server
        es_client: The Elasticsearch client
        connect_timeout (int, optional): The amount of time for a connection to the
            Elasticsearch host
    """
    es_client = es_client or create_es_client(es_host)
    scoped_index_name = get_scoped_index_name(app_namespace, index_name)

    try:
        if not does_index_exist(
            app_namespace, index_name, es_host, es_client, connect_timeout
        ):
            template = resolve_es_config_for_version(
                DEFAULT_ES_INDEX_TEMPLATE, es_client
            )
            es_client.indices.put_template(
                name=DEFAULT_ES_INDEX_TEMPLATE_NAME, body=template
            )
            logger.info("Creating index %r", index_name)
            es_client.indices.create(scoped_index_name, body=mapping)
        else:
            logger.error("Index %r already exists.", index_name)
    except EsConnectionError as e:
        logger.debug(
            "Unable to connect to Elasticsearch: %s details: %s", e.error, e.info
        )
        raise KnowledgeBaseConnectionError(es_host=es_client.transport.hosts)
    except TransportError as e:
        logger.error(
            "Unexpected error occurred when sending requests to Elasticsearch: %s "
            "Status code: %s details: %s",
            e.error,
            e.status_code,
            e.info,
        )
        raise KnowledgeBaseError(
            "Unexpected error occurred when sending requests to "
            "Elasticsearch: {} Status code: {} details: "
            "{}".format(e.error, e.status_code, e.info)
        )
    except ElasticsearchException:
        raise KnowledgeBaseError