Merge pull request #1047 from oxarbitrage/elasticsearch_tests

ElasticSearch test cases framework

Author: oxarbitrage

libraries/plugins/elasticsearch/elasticsearch_plugin.cpp

@@ -44,6 +44,8 @@
 #include <boost/algorithm/string.hpp>
 #include <regex>
 
+#include <graphene/utilities/elasticsearch.hpp>
+
 namespace graphene { namespace elasticsearch {
 
 namespace detail
@@ -74,10 +76,9 @@ class elasticsearch_plugin_impl
       bool _elasticsearch_visitor = false;
       CURL *curl; // curl handler
       vector <string> bulk; //  vector of op lines
-   private:
+      vector<std::string> prepare;
+private:
       void add_elasticsearch( const account_id_type account_id, const optional<operation_history_object>& oho, const signed_block& b );
-      void createBulkLine(account_transaction_history_object ath, operation_history_struct os, int op_type, block_struct bs, visitor_struct vs);
-      void sendBulk(std::string _elasticsearch_node_url, bool _elasticsearch_logs);
 
 };
 
@@ -211,10 +212,28 @@ void elasticsearch_plugin_impl::add_elasticsearch( const account_id_type account
    else
       limit_documents = _elasticsearch_bulk_replay;
 
-   createBulkLine(ath, os, op_type, bs, vs); // we have everything, creating bulk line
+   // we have everything, creating bulk line
+   bulk_struct bulks;
+   bulks.account_history = ath;
+   bulks.operation_history = os;
+   bulks.operation_type = op_type;
+   bulks.block_data = bs;
+   bulks.additional_data = vs;
+
+   std::string data = fc::json::to_string(bulks);
+
+   auto block_date = bulks.block_data.block_time.to_iso_string();
+   std::vector<std::string> parts;
+   boost::split(parts, block_date, boost::is_any_of("-"));
+   std::string index_name = "graphene-" + parts[0] + "-" + parts[1]; // index name
+   std::string _id = fc::json::to_string(ath.id);
+
+   prepare = graphene::utilities::createBulk(index_name, data, _id, 0);
+   bulk.insert(bulk.end(), prepare.begin(), prepare.end());
 
    if (curl && bulk.size() >= limit_documents) { // we are in bulk time, ready to add data to elasticsearech
-      sendBulk(_elasticsearch_node_url, _elasticsearch_logs);
+      prepare.clear();
+      graphene::utilities::SendBulk(curl, bulk, _elasticsearch_node_url, _elasticsearch_logs, "logs-account-history");
    }
 
    // remove everything except current object from ath
@@ -243,97 +262,6 @@ void elasticsearch_plugin_impl::add_elasticsearch( const account_id_type account
    }
 }
 
-void elasticsearch_plugin_impl::createBulkLine(account_transaction_history_object ath, operation_history_struct os, int op_type, block_struct bs, visitor_struct vs)
-{
-   bulk_struct bulks;
-   bulks.account_history = ath;
-   bulks.operation_history = os;
-   bulks.operation_type = op_type;
-   bulks.block_data = bs;
-   bulks.additional_data = vs;
-
-   std::string alltogether = fc::json::to_string(bulks);
-
-   auto block_date = bulks.block_data.block_time.to_iso_string();
-   std::vector<std::string> parts;
-   boost::split(parts, block_date, boost::is_any_of("-"));
-   std::string index_name = "graphene-" + parts[0] + "-" + parts[1];
-
-   // bulk header before each line, op_type = create to avoid dups, index id will be ath id(2.9.X).
-   std::string _id = fc::json::to_string(ath.id);
-   bulk.push_back("{ \"index\" : { \"_index\" : \""+index_name+"\", \"_type\" : \"data\", \"op_type\" : \"create\", \"_id\" : "+_id+" } }"); // header
-   bulk.push_back(alltogether);
-}
-
-void elasticsearch_plugin_impl::sendBulk(std::string _elasticsearch_node_url, bool _elasticsearch_logs)
-{
-
-   // curl buffers to read
-   std::string readBuffer;
-   std::string readBuffer_logs;
-
-   std::string bulking = "";
-
-   bulking = boost::algorithm::join(bulk, "\n");
-   bulking = bulking + "\n";
-   bulk.clear();
-
-   //wlog((bulking));
-
-   struct curl_slist *headers = NULL;
-   headers = curl_slist_append(headers, "Content-Type: application/json");
-   std::string url = _elasticsearch_node_url + "_bulk";
-   curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
-   curl_easy_setopt(curl, CURLOPT_POST, true);
-   curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
-   curl_easy_setopt(curl, CURLOPT_POSTFIELDS, bulking.c_str());
-   curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, WriteCallback);
-   curl_easy_setopt(curl, CURLOPT_WRITEDATA, (void *)&readBuffer);
-   curl_easy_setopt(curl, CURLOPT_USERAGENT, "libcrp/0.1");
-   //curl_easy_setopt(curl, CURLOPT_VERBOSE, true);
-   curl_easy_perform(curl);
-
-   long http_code = 0;
-   curl_easy_getinfo (curl, CURLINFO_RESPONSE_CODE, &http_code);
-   if(http_code == 200) {
-      // all good, do nothing
-   }
-   else if(http_code == 429) {
-      // repeat request?
-   }
-   else {
-      // exit everything ?
-   }
-
-   if(_elasticsearch_logs) {
-      auto logs = readBuffer;
-      // do logs
-      std::string url_logs = _elasticsearch_node_url + "logs/data/";
-      curl_easy_setopt(curl, CURLOPT_URL, url_logs.c_str());
-      curl_easy_setopt(curl, CURLOPT_POST, true);
-      curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
-      curl_easy_setopt(curl, CURLOPT_POSTFIELDS, logs.c_str());
-      curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, WriteCallback);
-      curl_easy_setopt(curl, CURLOPT_WRITEDATA, (void *) &readBuffer_logs);
-      curl_easy_setopt(curl, CURLOPT_USERAGENT, "libcrp/0.1");
-      //curl_easy_setopt(curl, CURLOPT_VERBOSE, true);
-      //ilog("log here curl: ${output}", ("output", readBuffer_logs));
-      curl_easy_perform(curl);
-
-      http_code = 0;
-      curl_easy_getinfo (curl, CURLINFO_RESPONSE_CODE, &http_code);
-      if(http_code == 200) {
-         // all good, do nothing
-      }
-      else if(http_code == 429) {
-         // repeat request?
-      }
-      else {
-         // exit everything ?
-      }
-   }
-}
-
 } // end namespace detail
 
 elasticsearch_plugin::elasticsearch_plugin() :
@@ -394,6 +322,9 @@ void elasticsearch_plugin::plugin_initialize(const boost::program_options::varia
 
 void elasticsearch_plugin::plugin_startup()
 {
+   if(!graphene::utilities::checkES(my->curl, my->_elasticsearch_node_url))
+      FC_THROW_EXCEPTION(fc::exception, "ES database is not up in url ${url}", ("url", my->_elasticsearch_node_url));
+   ilog("elasticsearch ACCOUNT HISTORY: plugin_startup() begin");
 }
 
 } }

libraries/plugins/es_objects/es_objects.cpp

@@ -392,7 +392,9 @@ void es_objects_plugin::plugin_initialize(const boost::program_options::variable
 
 void es_objects_plugin::plugin_startup()
 {
-   ilog("elasticsearch objects: plugin_startup() begin");
+   if(!graphene::utilities::checkES(my->curl, my->_es_objects_elasticsearch_url))
+      FC_THROW_EXCEPTION(fc::exception, "ES database is not up in url ${url}", ("url", my->_es_objects_elasticsearch_url));
+   ilog("elasticsearch OBJECTS: plugin_startup() begin");
 }
 
 } }

libraries/utilities/elasticsearch.cpp

@@ -25,9 +25,46 @@
 
 #include <boost/algorithm/string/join.hpp>
 #include <fc/log/logger.hpp>
+#include <fc/io/json.hpp>
 
 namespace graphene { namespace utilities {
 
+
+bool checkES(CURL *curl, std::string elasticsearch_url)
+{
+   std::string readBuffer;
+   std::string url = elasticsearch_url + "_nodes";
+   curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
+   curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, WriteCallback);
+   curl_easy_setopt(curl, CURLOPT_WRITEDATA, (void *)&readBuffer);
+   curl_easy_setopt(curl, CURLOPT_USERAGENT, "libcrp/0.1");
+   curl_easy_perform(curl);
+   if(readBuffer.empty())
+      return false;
+   else
+      return true;
+}
+std::string simpleQuery(CURL *curl, std::string elasticsearch_url, std::string endpoint, std::string query)
+{
+   std::string readBuffer;
+   struct curl_slist *headers = NULL;
+   headers = curl_slist_append(headers, "Content-Type: application/json");
+   std::string url = elasticsearch_url + endpoint;
+   curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
+   curl_easy_setopt(curl, CURLOPT_POST, true);
+   curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
+   curl_easy_setopt(curl, CURLOPT_POSTFIELDS, query.c_str());
+   curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, WriteCallback);
+   curl_easy_setopt(curl, CURLOPT_WRITEDATA, (void *)&readBuffer);
+   curl_easy_setopt(curl, CURLOPT_USERAGENT, "libcrp/0.1");
+   curl_easy_perform(curl);
+
+   if(!readBuffer.empty())
+      return readBuffer;
+   return "";
+
+}
+
 bool SendBulk(CURL *curl, std::vector<std::string>& bulk, std::string elasticsearch_url, bool do_logs, std::string logs_index)
 {
   // curl buffers to read

libraries/utilities/include/graphene/utilities/elasticsearch.hpp

@@ -38,5 +38,7 @@ namespace graphene { namespace utilities {
 
    bool SendBulk(CURL *curl, std::vector <std::string>& bulk, std::string elasticsearch_url, bool do_logs, std::string logs_index);
    std::vector<std::string> createBulk(std::string type, std::string data, std::string id, bool onlycreate);
+   bool checkES(CURL *curl, std::string elasticsearch_url);
+   std::string simpleQuery(CURL *curl, std::string elasticsearch_url, std::string endpoint, std::string query);
 
 } } // end namespace graphene::utilities

tests/CMakeLists.txt

@@ -7,19 +7,19 @@ if( GPERFTOOLS_FOUND )
 endif()
 
 file(GLOB UNIT_TESTS "tests/*.cpp")
-add_executable( chain_test ${COMMON_SOURCES} ${UNIT_TESTS} )
-target_link_libraries( chain_test graphene_chain graphene_app graphene_account_history graphene_egenesis_none fc graphene_wallet ${PLATFORM_SPECIFIC_LIBS} )
+add_executable( chain_test ${UNIT_TESTS} ${COMMON_SOURCES} )
+target_link_libraries( chain_test graphene_chain graphene_app graphene_account_history graphene_elasticsearch graphene_egenesis_none fc graphene_wallet ${PLATFORM_SPECIFIC_LIBS} )
 if(MSVC)
   set_source_files_properties( tests/serialization_tests.cpp PROPERTIES COMPILE_FLAGS "/bigobj" )
 endif(MSVC)
 
 file(GLOB PERFORMANCE_TESTS "performance/*.cpp")
-add_executable( performance_test ${COMMON_SOURCES} ${PERFORMANCE_TESTS} )
-target_link_libraries( performance_test graphene_chain graphene_app graphene_account_history graphene_egenesis_none fc ${PLATFORM_SPECIFIC_LIBS} )
+add_executable( performance_test ${PERFORMANCE_TESTS} ${COMMON_SOURCES} )
+target_link_libraries( performance_test graphene_chain graphene_app graphene_account_history graphene_elasticsearch graphene_egenesis_none fc ${PLATFORM_SPECIFIC_LIBS} )
 
 file(GLOB BENCH_MARKS "benchmarks/*.cpp")
-add_executable( chain_bench ${COMMON_SOURCES} ${BENCH_MARKS} )
-target_link_libraries( chain_bench graphene_chain graphene_app graphene_account_history graphene_egenesis_none fc ${PLATFORM_SPECIFIC_LIBS} )
+add_executable( chain_bench ${BENCH_MARKS} ${COMMON_SOURCES} )
+target_link_libraries( chain_bench graphene_chain graphene_app graphene_account_history graphene_elasticsearch graphene_egenesis_none fc ${PLATFORM_SPECIFIC_LIBS} )
 
 file(GLOB APP_SOURCES "app/*.cpp")
 add_executable( app_test ${APP_SOURCES} )
@@ -29,4 +29,8 @@ file(GLOB CLI_SOURCES "cli/*.cpp")
 add_executable( cli_test ${CLI_SOURCES} )
 target_link_libraries( cli_test graphene_app graphene_wallet graphene_witness graphene_account_history graphene_net graphene_chain graphene_egenesis_none fc ${PLATFORM_SPECIFIC_LIBS} )
 
+file(GLOB ES_SOURCES "elasticsearch/*.cpp")
+add_executable( es_test ${ES_SOURCES} ${COMMON_SOURCES} )
+target_link_libraries( es_test graphene_chain graphene_app graphene_account_history graphene_elasticsearch graphene_egenesis_none fc ${PLATFORM_SPECIFIC_LIBS} )
+
 add_subdirectory( generate_empty_blocks )

tests/common/database_fixture.cpp

@@ -27,6 +27,7 @@
 #include <graphene/account_history/account_history_plugin.hpp>
 #include <graphene/market_history/market_history_plugin.hpp>
 #include <graphene/grouped_orders/grouped_orders_plugin.hpp>
+#include <graphene/elasticsearch/elasticsearch_plugin.hpp>
 
 #include <graphene/db/simple_index.hpp>
 
@@ -73,7 +74,6 @@ database_fixture::database_fixture()
       if( arg == "--show-test-names" )
          std::cout << "running test " << boost::unit_test::framework::current_test_case().p_name << std::endl;
    }
-   auto ahplugin = app.register_plugin<graphene::account_history::account_history_plugin>();
    auto mhplugin = app.register_plugin<graphene::market_history::market_history_plugin>();
    auto goplugin = app.register_plugin<graphene::grouped_orders::grouped_orders_plugin>();
    init_account_pub_key = init_account_priv_key.get_public_key();
@@ -126,8 +126,23 @@ database_fixture::database_fixture()
       options.insert(std::make_pair("track-account", boost::program_options::variable_value(track_account, false)));
    }
 
-   ahplugin->plugin_set_app(&app);
-   ahplugin->plugin_initialize(options);
+   if(boost::unit_test::framework::current_test_case().p_name.value == "es1") {
+      auto esplugin = app.register_plugin<graphene::elasticsearch::elasticsearch_plugin>();
+      esplugin->plugin_set_app(&app);
+
+      options.insert(std::make_pair("elasticsearch-node-url", boost::program_options::variable_value(string("http://localhost:9200/"), false)));
+      options.insert(std::make_pair("elasticsearch-bulk-replay", boost::program_options::variable_value(uint32_t(2), false)));
+      options.insert(std::make_pair("elasticsearch-bulk-sync", boost::program_options::variable_value(uint32_t(2), false)));
+
+      esplugin->plugin_initialize(options);
+      esplugin->plugin_startup();
+   }
+   else {
+      auto ahplugin = app.register_plugin<graphene::account_history::account_history_plugin>();
+      ahplugin->plugin_set_app(&app);
+      ahplugin->plugin_initialize(options);
+      ahplugin->plugin_startup();
+   }
 
    options.insert(std::make_pair("bucket-size", boost::program_options::variable_value(string("[15]"),false)));
    mhplugin->plugin_set_app(&app);
@@ -136,7 +151,6 @@ database_fixture::database_fixture()
    goplugin->plugin_set_app(&app);
    goplugin->plugin_initialize(options);
 
-   ahplugin->plugin_startup();
    mhplugin->plugin_startup();
    goplugin->plugin_startup();
 

tests/elasticsearch/main.cpp

@@ -0,0 +1,83 @@
+/*
+ * Copyright (c) 2018 oxarbitrage and contributors.
+ *
+ * The MIT License
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
+#include <graphene/app/api.hpp>
+#include <graphene/utilities/tempdir.hpp>
+#include <fc/crypto/digest.hpp>
+#include <fc/io/json.hpp>
+
+#include <graphene/utilities/elasticsearch.hpp>
+
+#include "../common/database_fixture.hpp"
+
+#define BOOST_TEST_MODULE Elastic Search Database Tests
+#include <boost/test/included/unit_test.hpp>
+
+using namespace graphene::chain;
+using namespace graphene::chain::test;
+using namespace graphene::app;
+
+BOOST_FIXTURE_TEST_SUITE( elasticsearch_tests, database_fixture )
+
+BOOST_AUTO_TEST_CASE(es1) {
+      try {
+         //account_id_type() do 3 ops
+         create_bitasset("USD", account_id_type());
+         create_account("dan");
+         create_account("bob");
+
+         // generate a few blocks
+         generate_block();
+         generate_block();
+
+         fc::usleep(fc::milliseconds(2000));
+
+         // for later use
+         //int asset_create_op_id = operation::tag<asset_create_operation>::value;
+         //int account_create_op_id = operation::tag<account_create_operation>::value;
+
+         string query = "{ \"query\" : { \"bool\" : { \"must\" : [{\"match_all\": {}}] } } }";
+
+         CURL *curl; // curl handler
+         curl = curl_easy_init();
+         auto res = graphene::utilities::simpleQuery(curl, "http://localhost:9200/", "graphene-*/data/_count", query);
+         variant j = fc::json::from_string(res);
+
+         auto total = j["_shards"]["total"].as_string();
+         BOOST_CHECK_EQUAL(total, "5");
+
+         res = graphene::utilities::simpleQuery(curl, "http://localhost:9200/", "graphene-*/data/_search", query);
+         j = fc::json::from_string(res);
+
+         auto first_id = j["hits"]["hits"][size_t(0)]["_id"].as_string() ;
+         BOOST_CHECK_EQUAL(first_id, "2.9.1"); // this should be 0? are they inserted in the right order?
+
+      }
+      catch (fc::exception &e) {
+         edump((e.to_detail_string()));
+         throw;
+      }
+}
+
+BOOST_AUTO_TEST_SUITE_END()