Fluent Logger for glog

fluent_logging_glog.cpp

#include <iostream>
#include <sstream>
#include <string>
#include <ctime>

#include <glog/logging.h>
#include <msgpack.hpp>

#include <pficommon/text/json.h>
#include <pficommon/network/socket.h>

struct FluentMessage {
  struct GlogMessage {
    GlogMessage(google::LogSeverity severity,
                const char* full_filename, const char* base_filename,
                int line,
                const struct::tm* tm_time,
                const char* message, size_t message_len)
      : severity_(severity), file_(full_filename), line_(line),
        time_(mktime(const_cast<struct::tm*>(tm_time))), message_(message, message_len) {
    }

    google::LogSeverity severity_;
    std::string file_;
    int line_;
    time_t time_;
    std::string message_;

    virtual bool skipToParseMessage(const std::string& message) const {
      size_t pos = 0;
      while (pos < message.size() && isspace(message[pos])) ++pos;
      return message[pos] != '{' && message[pos] != '[';
    }

    template<typename Packer>
    void msgpack_pack(Packer& packer) const {
      packer.pack_map(5);

      packer.pack(std::string("severity"));
      packer.pack(severity_);

      packer.pack(std::string("file"));
      packer.pack(file_);

      packer.pack(std::string("line"));
      packer.pack(line_);

      packer.pack(std::string("time"));
      packer.pack(time_);

      packer.pack(std::string("message"));

      if (skipToParseMessage(message_)) {
        packer.pack(message_);
        return ;
      }

      try {
        std::istringstream in(message_);
        pfi::text::json::json_parser parser(in);
        pfi::text::json::json js = parser.parse();
        msgpack_pack_impl<Packer>(packer, js);
      } catch (...) {
        packer.pack(message_);
      }
    }

    template<typename Packer>
    void msgpack_pack_impl(Packer& packer, const pfi::text::json::json& js) const {
      if (pfi::text::json::is<pfi::text::json::json_object>(js)) {
        const pfi::text::json::json_object* value = dynamic_cast<pfi::text::json::json_object*>(js.get());
        packer.pack_map(distance(value->begin(), value->end()));
        for (pfi::text::json::json_object::const_iterator it = value->begin(), end = value->end();
             it != end; ++it) {
          packer.pack(it->first);
          msgpack_pack_impl<Packer>(packer, it->second);
        }
      } else if (pfi::text::json::is<pfi::text::json::json_array>(js)) {
        const pfi::text::json::json_array* value = dynamic_cast<pfi::text::json::json_array*>(js.get());
        packer.pack_array(value->size());
        for (size_t i = 0; i < value->size(); ++i) {
          msgpack_pack_impl<Packer>(packer, value->operator[](i));
        }
      } else if (pfi::text::json::is<pfi::text::json::json_integer>(js)) {
        packer.pack(pfi::text::json::json_cast<int>(js));
      } else if (pfi::text::json::is<pfi::text::json::json_float>(js)) {
        packer.pack(pfi::text::json::json_cast<double>(js));
      } else if (pfi::text::json::is<pfi::text::json::json_string>(js)) {
        packer.pack(pfi::text::json::json_cast<std::string>(js));
      } else if (pfi::text::json::is<pfi::text::json::json_bool>(js)) {
        packer.pack(pfi::text::json::json_cast<bool>(js));
      } else if (pfi::text::json::is<pfi::text::json::json_null>(js)) {
        packer.pack_nil();
      } else {
        // TODO: throw exception ???                                                                                                                                                                                                       
      }
    }

    void msgpack_unpack(msgpack::object o) {
      // TODO:                                                                                                                                                                                                                             
    }
  };

  FluentMessage(const std::string& tag, time_t time, const GlogMessage& message)
    : tag_(tag), time_(time), message_(message) {}

  std::string tag_;
  time_t time_;
  GlogMessage message_;

  MSGPACK_DEFINE(tag_, time_, message_);
};

class FluentLogSink : public google::LogSink {
public:
  virtual ~FluentLogSink() {}
  
  virtual void send(google::LogSeverity severity, const char* full_filename,
                    const char* base_filename, int line,
                    const struct ::tm* tm_time,
                    const char* message, size_t message_len) = 0;
  
  virtual void WaitTillSent() {}
};

class FluentTCPLogSink : public FluentLogSink {
public:
  FluentTCPLogSink(const std::string& host, int port, const std::string& tag = "glog")
    : host_(host), port_(port), tag_(tag) {}
  
  virtual ~FluentTCPLogSink() {}
  
  virtual void send(google::LogSeverity severity, const char* full_filename,
                    const char* base_filename, int line,
                    const struct ::tm* tm_time,
                    const char* message, size_t message_len) {
    pfi::network::stream_socket ssocket;
    if (!ssocket.connect(host_, port_)) {
      std::cerr << "failed to connect fluentd(" << host_ << ":" << port_ << ")" << std::endl; // TODO:                                                                                                                                     
      return ;
    }

    {
      FluentMessage fluent_message(tag_,
                                   mktime(const_cast<tm*>(tm_time)),
                                   FluentMessage::GlogMessage(severity,
                                                              full_filename,
                                                              base_filename,
                                                              line,
                                                              tm_time,
                                                              message,
                                                              message_len));

      msgpack::sbuffer sbuf;
      msgpack::pack(sbuf, fluent_message);
      ssocket.write(sbuf.data(), sbuf.size());
    }
  }
  
  virtual void WaitTillSent() {}

protected:
  std::string host_;
  int port_;
  std::string tag_;
};

FluentTCPLogSink fluent_tcp_sink("localhost", 24224, "debug.glog");

int main(int argc, const char* argv[])
{
  google::InitGoogleLogging(argv[0]);
  google::AddLogSink(&fluent_tcp_sink);
  google::InstallFailureSignalHandler();

  LOG(INFO) << "start";
  LOG(INFO) << "{\"red\" : \"bull\", \"hoge\" : null, \"nested\" : { \"red bull\" : [1, 2.0, true] } }";

  return 0;
}