Add Elasticsearch accessor unittests

Signed-off-by: Thibault Chataigner <t.chataigner@criteo.com>

.travis.yml

@@ -24,6 +24,8 @@ env:
     - CASSANDRA_VERSION=3.11.2
     - CASSANDRA_STRATIO_LUCENE_VERSION=3.11.1.0 #We should use ${CASSANDRA_VERSION}.0, but not released for 3.11.2 yet.
     - CASSANDRA_HOME="${TRAVIS_BUILD_DIR}/.deps/apache-cassandra-${CASSANDRA_VERSION}/"
+    - ES_VERSION=6.3.1
+    - ES_HOME="${TRAVIS_BUILD_DIR}/.deps/elasticsearch-${ES_VERSION}/"
 
 install:
   ### Java: We download now some dependencies to limit the number of messages later

biggraphite/test_utils.py

@@ -33,6 +33,7 @@
 from biggraphite import utils as bg_utils
 from biggraphite import accessor as bg_accessor
 from biggraphite.test_utils_cassandra import CassandraHelper
+from biggraphite.test_utils_elasticsearch import ElasticsearchHelper
 from biggraphite.drivers import memory as bg_memory
 from biggraphite import metadata_cache as bg_metadata_cache
 
@@ -177,16 +178,23 @@ class TestCaseWithAccessor(TestCaseWithTempDir):
     CACHE_CLASS = bg_metadata_cache.MemoryCache
 
     cassandra_helper = None
+    elasticsearch_helper = None
 
     @classmethod
     def setUpClass(cls):
         """Create the test Accessor."""
+        # TODO (t.chataigner) Handle hybrid accessor here.
         driver_name = cls.ACCESSOR_SETTINGS.get('driver', bg_utils.DEFAULT_DRIVER)
         if "cassandra" in driver_name:
             cls.cassandra_helper = CassandraHelper()
             cls.cassandra_helper.setUpClass()
             cls.ACCESSOR_SETTINGS.update(
                 cls.cassandra_helper.get_accessor_settings())
+        if "elasticsearch" in driver_name:
+            cls.elasticsearch_helper = ElasticsearchHelper()
+            cls.elasticsearch_helper.setUpClass()
+            cls.ACCESSOR_SETTINGS.update(
+                cls.elasticsearch_helper.get_accessor_settings())
 
         cls.accessor = bg_utils.accessor_from_settings(cls.ACCESSOR_SETTINGS)
         cls.accessor.syncdb()
@@ -199,6 +207,8 @@ def tearDownClass(cls):
         cls.accessor.shutdown()
         if cls.cassandra_helper:
             cls.cassandra_helper.tearDownClass()
+        if cls.elasticsearch_helper:
+            cls.elasticsearch_helper.tearDownClass()
 
     def setUp(self):
         """Create a new Accessor in self.acessor."""

biggraphite/test_utils_cassandra.py

@@ -12,7 +12,7 @@
 # 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.
-"""Utilities factoring code across tests.
+"""Cassandra utilities factoring code across tests.
 
 The dependencies for this module are not included in requirements.txt or in the package
 dependencies, instead one needs the elements of tests-requirements.txt .
@@ -128,8 +128,6 @@ def drop_keyspace(session, keyspace):
 
     def flush(self, accessor):
         """Flush all kind of buffers related to Cassandra."""
-        accessor.flush()
-
         # When using Lucene, we need to force a refresh on the index as the default
         # refresh period is 60s.
         if accessor.use_lucene:

biggraphite/test_utils_elasticsearch.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python
+# Copyright 2018 Criteo
+#
+# 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.
+"""Elasticsearch utilities factoring code across tests.
+
+The dependencies for this module are not included in requirements.txt or in the package
+dependencies, instead one needs the elements of tests-requirements.txt .
+"""
+from __future__ import absolute_import
+from __future__ import print_function
+
+import os
+import sys
+import logging
+
+
+HAS_ES_HOME = bool(os.getenv("ES_HOME"))
+ES_HOSTPORT = os.getenv("ES_HOSTPORT")
+HAS_ELASTICSEARCH = HAS_ES_HOME or ES_HOSTPORT
+
+# Only try to import elasticsearch if we are going to use it. This is better
+# than using try/except because the failure case is easier to handle.
+if HAS_ELASTICSEARCH:
+    from testing import elasticsearch as testing_elasticsearch
+
+
+class ElasticsearchHelper():
+    """Helper for an ephemeral Elasticsearch cluster."""
+
+    INDEX = "testindex"
+
+    @classmethod
+    def get_accessor_settings(cls):
+        """Prepare accessor settings for Elasticsearch driver."""
+        return {
+            "elasticsearch_index": cls.INDEX,
+            "elasticsearch_hosts": cls.hosts,
+            "elasticsearch_port": cls.port,
+            "elasticsearch_timeout": 60
+        }
+
+    @classmethod
+    def setUpClass(cls):
+        """Create the test Elasticsearch Cluster as cls.elasticsearch."""
+        cls.elasticsearch = None
+        if ES_HOSTPORT:
+            # Use existing and running instance.
+            host, cls.port = ES_HOSTPORT.split(':')
+            cls.hosts = [host]
+        else:
+            # Setup a new instance, and dynamically get its host and port.
+            cls.setUpElasticsearch()
+
+    @classmethod
+    def setUpElasticsearch(cls):
+        """Start Elasticsearch."""
+        cls.elasticsearch = testing_elasticsearch.Elasticsearch(
+            auto_start=False,
+        )
+        try:
+            cls.elasticsearch.setup()
+            cls.elasticsearch.start()
+        except Exception as e:
+            logging.exception(e)
+            print("fail to starting elasticsearch, logging potentially useful debug info",
+                  file=sys.stderr)
+            for attr in ["elasticsearch_home", "elasticsearch_yaml", "elasticsearch_major_version",
+                         "base_dir", "settings"]:
+                print(attr, ":", getattr(cls.elasticsearch,
+                                         attr, "Unknown"), file=sys.stderr)
+            cls.elasticsearch.cleanup()
+            raise
+
+        # testing.elasticsearch is meant to be used with the Thrift API, so we need to
+        # extract the IPs and native port for use with the native driver.
+        cls.hosts = [s.split(":")[0]
+                     for s in cls.elasticsearch.dsn()['hosts']]
+        cls.port = cls.elasticsearch.elasticsearch_yaml["http.port"]
+
+    @classmethod
+    def tearDownClass(cls):
+        """Stop the test Elasticsearch Cluster."""
+        if cls.elasticsearch:
+            cls.elasticsearch.stop()
+            cls.elasticsearch = None

tests-requirements.txt

@@ -10,6 +10,9 @@ unittest2
 # Cassandra
 testing.cassandra3
 
+# Elasticsearch
+testing.elasticsearch6
+
 # For bg-replay-traffic
 dpkt