Initial implementation of the status interface reporter object.

dbsim/db_params.cpp

@@ -57,6 +57,9 @@ db::params db::parse_args(int argc, char** argv)
         ("wsrep-provider-options",
          po::value<std::string>(&params.wsrep_provider_options),
          "wsrep provider options")
+        ("status-file",
+         po::value<std::string>(&params.status_file),
+         "status output file")
         ("servers", po::value<size_t>(&params.n_servers)->required(),
          "number of servers to start")
         ("topology", po::value<std::string>(&params.topology),

dbsim/db_params.hpp

@@ -38,6 +38,7 @@ namespace db
         std::string topology;
         std::string wsrep_provider;
         std::string wsrep_provider_options;
+        std::string status_file;
         int debug_log_level;
         int fast_exit;
         int thread_instrumentation;
@@ -55,6 +56,7 @@ namespace db
             , topology()
             , wsrep_provider()
             , wsrep_provider_options()
+            , status_file("status.json")
             , debug_log_level(0)
             , fast_exit(0)
             , thread_instrumentation()

dbsim/db_server.cpp

@@ -24,10 +24,52 @@
 #include "db_simulator.hpp"
 
 #include "wsrep/logger.hpp"
+#include "wsrep/reporter.hpp"
+
+#include <ostream>
+#include <cstdio>
+
+static wsrep::default_mutex logger_mtx;
+static wsrep::reporter* reporter = nullptr;
+
+static void
+logger_fn(wsrep::log::level l, const char* pfx, const char* msg)
+{
+    wsrep::unique_lock<wsrep::mutex> lock(logger_mtx);
+
+    struct timespec time;
+    clock_gettime(CLOCK_REALTIME, &time);
+
+    time_t const tt(time.tv_sec);
+    struct tm    date;
+    localtime_r(&tt, &date);
+
+    char date_str[85] = { '\0', };
+    snprintf(date_str, sizeof(date_str) - 1,
+             "%04d-%02d-%02d %02d:%02d:%02d.%03d",
+             date.tm_year + 1900, date.tm_mon + 1, date.tm_mday,
+             date.tm_hour, date.tm_min, date.tm_sec, (int)time.tv_nsec/1000000);
+
+#define LOG_STR date_str << pfx << wsrep::log::to_c_string(l) << msg
+    if (l >= wsrep::log::error && reporter)
+    {
+        std::ostringstream os;
+        os << LOG_STR;
+        std::cerr << os.str() << std::endl;
+        auto const tstamp(double(time.tv_sec) + double(time.tv_nsec)*1.0e-9);
+        reporter->report_log_msg(wsrep::reporter::error, os.str(), tstamp);
+    }
+    else
+    {
+        std::cerr << LOG_STR << std::endl;
+    }
+#undef LOG_STR
+}
 
 db::server::server(simulator& simulator,
                    const std::string& name,
-                   const std::string& address)
+                   const std::string& address,
+                   const std::string& status_file)
     : simulator_(simulator)
     , storage_engine_(simulator_.params())
     , mutex_()
@@ -40,7 +82,15 @@ db::server::server(simulator& simulator,
     , appliers_()
     , clients_()
     , client_threads_()
-{ }
+{
+    wsrep::log::logger_fn(logger_fn);
+    reporter = new wsrep::reporter(mutex_, status_file, 3);
+}
+
+db::server::~server()
+{
+    delete reporter;
+}
 
 void db::server::applier_thread()
 {
@@ -129,3 +179,9 @@ wsrep::high_priority_service* db::server::streaming_applier_service()
     throw wsrep::not_implemented_error();
 }
 
+void db::server::log_state_change(enum wsrep::server_state::state from,
+                                  enum wsrep::server_state::state to)
+{
+    wsrep::log_info() << "State changed " << from << " -> " << to;
+    if (reporter) reporter->report_state(to, 0);
+}

dbsim/db_server.hpp

@@ -41,7 +41,9 @@ namespace db
     public:
         server(simulator& simulator,
                const std::string& name,
-               const std::string& address);
+               const std::string& address,
+               const std::string& status_file);
+        ~server();
         void applier_thread();
         void start_applier();
         void stop_applier();
@@ -58,6 +60,8 @@ namespace db
         wsrep::client_state* local_client_state();
         void release_client_state(wsrep::client_state*);
         wsrep::high_priority_service* streaming_applier_service();
+        void log_state_change(enum wsrep::server_state::state,
+                              enum wsrep::server_state::state);
     private:
         void start_client(size_t id);
 

dbsim/db_server_service.cpp

@@ -154,10 +154,9 @@ void db::server_service::log_state_change(
     enum wsrep::server_state::state prev_state,
     enum wsrep::server_state::state current_state)
 {
-
-    wsrep::log_info() << "State changed "
-                      << prev_state << " -> " << current_state;
+    server_.log_state_change(prev_state, current_state);
 }
+
 int db::server_service::wait_committing_transactions(int)
 {
     throw wsrep::not_implemented_error();

dbsim/db_simulator.cpp

@@ -137,7 +137,8 @@ void db::simulator::start()
                         std::make_unique<db::server>(
                             *this,
                             name_os.str(),
-                            address_os.str()))));
+                            address_os.str(),
+                            name_os.str() + "_" + params_.status_file))));
         if (it.second == false)
         {
             throw wsrep::runtime_error("Failed to add server");