spark/src/lib/logging/logging.cpp

#include "logging.h"

#include "LoggerHub.h"
#include "lib/config/config.h"
#include "lib/os/os.h"
#include "lib/utils/bts_assert.h"
#include "lib/utils/bts_chrono.h"
#include "lib/config/protos/logging.pb.h"
#include "lib/utils/file_utils.h"
#include "lib/utils/Dispatcher.h"

#include <google/protobuf/text_format.h>
#include <google/protobuf/io/zero_copy_stream_impl.h>
#include <google/protobuf/io/gzip_stream.h>
#include <boost/filesystem/path.hpp>
#include <boost/filesystem.hpp>
#include <boost/filesystem/operations.hpp>
#include <fstream>

namespace bts {
namespace log {

static LoggerHub _default_logger_hub("logger");
thread_local LoggerHandle _thread_logger(_default_logger_hub.make_logger());


template <class M>
typename M::const_iterator find_key(M const&m, typename M::mapped_type const &v)
{
   for (auto it = m.begin(); it != m.end();it++) 
      if (it->second == v)
         return it;
   return m.end();
}
   
boost::filesystem::path log_path ()
{
   auto store_root = os::bts_data_path();
   std::string base_path = store_root; 
   boost::filesystem::path result = utils::normalize_data_path (base_path);
   result /= "logging";
   return result;
}

void start_default_logging_thread()
{
   _default_logger_hub.start();
}

Filter get_filter (LoggingConfig &cfg,
                   config::Logging::SinkConfig const&scfg)
{
   Filter result;

   auto it = find_key(cfg.level_names,scfg.default_level());
   bts_assert(it != cfg.level_names.end());
   auto default_level = it->first;
   for (auto &l : result)
      l = default_level;
      
   for (auto const &sub_system : scfg.subsystem()) {

      auto ss_it = find_key(cfg.subsystem_names,sub_system.name());
      bts_assert(ss_it != cfg.subsystem_names.end());
      auto ss = ss_it->first;

      auto level_it = find_key(cfg.level_names,sub_system.level());
      bts_assert(level_it != cfg.level_names.end());
      auto level = level_it->first;

      result[ss.ordinal] = level;
   }
   return result;
}

void configure_logging(LoggingConfig const&cfg) {_default_logger_hub.config(cfg);}

void configure_logging(bts::config::Logging const&cfg)
{
   LoggingConfig real_cfg;
   real_cfg.logger_initial_queue_size = cfg.initial_queue_size();
   real_cfg.logger_max_queue_growth = cfg.max_queue_growth();

   real_cfg.level_names = { {Diag,"Diag"},{Info,"Info"},
                            {Warn,"Warn"},{Error,"Error"},
                            {Fatal,"Fatal"},{Sys,"Sys"} };

   real_cfg.subsystem_names = { {log::Gen,"Gen"},
                                {log::Cfg,"Cfg"},
                                {log::QBus,"QBus"},
                                {log::Log,"Log"},
                                {log::Jetfix,"Jetfix"},
                                {log::MarketData,"MarketData"},
                                {log::Api,"Api"},
                                {log::Persistence,"Persistence"},
                                {log::User,"User"},
                                {log::OrderMan,"OrderMan"},
                                {log::Latency,"Latency"},
                                {log::Assert,"Assert"},
                                {log::Engine,"Engine"},
                                {log::Ui,"UI"},
                                {log::Persist,"Persist"},
   };

   real_cfg.sinks["console"] =
      std::make_pair (console_sink(),get_filter(real_cfg,cfg.console()));
   real_cfg.sinks["syslog"] =
      std::make_pair (syslog_sink(),get_filter(real_cfg,cfg.syslog()));
   real_cfg.sinks["file"] =
      std::make_pair (file_sink(),get_filter(real_cfg,cfg.file()));
   configure_logging(real_cfg);
}

boost::filesystem::path log_cache_file_path ()
{
   auto store_root = os::bts_data_path();
   boost::filesystem::path result = utils::normalize_data_path (store_root);
   result /= "logging";
   result /= "log_control";
   return result;
}

LoggingConfigPoller::LoggingConfigPoller (bts::config::Logging const&cfg)
   : _poller(utils::Dispatcher::background(), 1000, [=]() { poll_once(); }), 
   _last_ts(std::time(nullptr)),
   _log_cache_file(log_cache_file_path()),
   _done(false)
{
   init(cfg);
}

LoggingConfigPoller::LoggingConfigPoller () : LoggingConfigPoller(config::get<config::Logging>())
{ }

void LoggingConfigPoller::init (bts::config::Logging const&cfg)
{
   auto parent = _log_cache_file.parent_path();
   if (!boost::filesystem::is_directory(parent)) 
      bts_assert(boost::filesystem::create_directories(parent));

   std::ofstream fs (_log_cache_file.string());
   google::protobuf::io::OstreamOutputStream gs (&fs);
   google::protobuf::TextFormat::Print (cfg,&gs);


   // initialize logging
   configure_logging(cfg);
   _poller.start();
}

void LoggingConfigPoller::poll_once() {
   if (_done) return;
   auto current_ts = 
      boost::filesystem::last_write_time(_log_cache_file) ;

   if (current_ts != _last_ts) {
      _last_ts = current_ts;
      bts::config::Logging cfg;
      std::ifstream fs (_log_cache_file.string());
      google::protobuf::io::IstreamInputStream gs (&fs);
      if (google::protobuf::TextFormat::Parse (&gs,&cfg))  {
         LOG(Log,Warn,"Detected logging configuration change. Reconfiguring logging.");
         configure_logging(cfg);
      }
      else
         LOG(Log,Error,"Detected invalid logging configuration. Ignoring.");
   }
}

LoggingConfigPoller::~LoggingConfigPoller ()
{
   _done = true;
   _poller.stop();
}

namespace interop {

extern "C" BTS_EXPORT void Logger_log (int level, int subsystem, const char *message, const char *file, int line)
{
   bts::log::_thread_logger.logger->enqueue(log::Subsystem{(uint8_t)subsystem}, log::Level{(uint8_t)level}, file, line, "{}", std::string(message));
}
}


}}