Adding upstream version 18.2.2.upstream/18.2.2

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 364 insertions, 0 deletions

diff --git a/src/crimson/common/logclient.cc b/src/crimson/common/logclient.cc
new file mode 100644
index 000000000..d402ecd19
--- /dev/null
+++ b/src/crimson/common/logclient.cc
@@ -0,0 +1,364 @@
+#include "crimson/common/logclient.h"
+#include <fmt/ranges.h>
+#include "include/str_map.h"
+#include "messages/MLog.h"
+#include "messages/MLogAck.h"
+#include "messages/MMonGetVersion.h"
+#include "crimson/net/Messenger.h"
+#include "crimson/mon/MonClient.h"
+#include "mon/MonMap.h"
+#include "common/Graylog.h"
+
+using std::map;
+using std::ostream;
+using std::ostringstream;
+using std::string;
+using crimson::common::local_conf;
+
+namespace {
+  seastar::logger& logger()
+  {
+    return crimson::get_logger(ceph_subsys_monc);
+  }
+}
+
+//TODO: in order to avoid unnecessary maps declarations and moving around,
+//	create a named structure containing the maps and return optional
+//	fit to it.
+int parse_log_client_options(CephContext *cct,
+			     map<string,string> &log_to_monitors,
+			     map<string,string> &log_to_syslog,
+			     map<string,string> &log_channels,
+			     map<string,string> &log_prios,
+			     map<string,string> &log_to_graylog,
+			     map<string,string> &log_to_graylog_host,
+			     map<string,string> &log_to_graylog_port,
+			     uuid_d &fsid,
+			     string &host)
+{
+  ostringstream oss;
+
+  int r = get_conf_str_map_helper(
+    cct->_conf.get_val<string>("clog_to_monitors"), oss,
+    &log_to_monitors, CLOG_CONFIG_DEFAULT_KEY);
+  if (r < 0) {
+    logger().error("{} error parsing 'clog_to_monitors'", __func__);
+    return r;
+  }
+
+  r = get_conf_str_map_helper(
+    cct->_conf.get_val<string>("clog_to_syslog"), oss,
+                              &log_to_syslog, CLOG_CONFIG_DEFAULT_KEY);
+  if (r < 0) {
+    logger().error("{} error parsing 'clog_to_syslog'", __func__);
+    return r;
+  }
+
+  r = get_conf_str_map_helper(
+    cct->_conf.get_val<string>("clog_to_syslog_facility"), oss,
+    &log_channels, CLOG_CONFIG_DEFAULT_KEY);
+  if (r < 0) {
+    logger().error("{} error parsing 'clog_to_syslog_facility'", __func__);
+    return r;
+  }
+
+  r = get_conf_str_map_helper(
+    cct->_conf.get_val<string>("clog_to_syslog_level"), oss,
+    &log_prios, CLOG_CONFIG_DEFAULT_KEY);
+  if (r < 0) {
+    logger().error("{} error parsing 'clog_to_syslog_level'", __func__);
+    return r;
+  }
+
+  r = get_conf_str_map_helper(
+    cct->_conf.get_val<string>("clog_to_graylog"), oss,
+    &log_to_graylog, CLOG_CONFIG_DEFAULT_KEY);
+  if (r < 0) {
+    logger().error("{} error parsing 'clog_to_graylog'", __func__);
+    return r;
+  }
+
+  r = get_conf_str_map_helper(
+    cct->_conf.get_val<string>("clog_to_graylog_host"), oss,
+    &log_to_graylog_host, CLOG_CONFIG_DEFAULT_KEY);
+  if (r < 0) {
+    logger().error("{} error parsing 'clog_to_graylog_host'", __func__);
+    return r;
+  }
+
+  r = get_conf_str_map_helper(
+    cct->_conf.get_val<string>("clog_to_graylog_port"), oss,
+    &log_to_graylog_port, CLOG_CONFIG_DEFAULT_KEY);
+  if (r < 0) {
+    logger().error("{} error parsing 'clog_to_graylog_port'", __func__);
+    return r;
+  }
+
+  fsid = cct->_conf.get_val<uuid_d>("fsid");
+  host = cct->_conf->host;
+  return 0;
+}
+
+LogChannel::LogChannel(LogClient *lc, const string &channel)
+  : parent(lc), log_channel(channel), log_to_syslog(false),
+    log_to_monitors(false)
+{
+}
+
+LogChannel::LogChannel(LogClient *lc, const string &channel,
+                       const string &facility, const string &prio)
+  : parent(lc), log_channel(channel), log_prio(prio),
+    syslog_facility(facility), log_to_syslog(false),
+    log_to_monitors(false)
+{
+}
+
+LogClient::LogClient(crimson::net::Messenger *m,
+		     logclient_flag_t flags)
+  : messenger(m), is_mon(flags & FLAG_MON),
+    last_log_sent(0), last_log(0)
+{
+}
+
+void LogChannel::set_log_to_monitors(bool v)
+{
+  if (log_to_monitors != v) {
+    parent->reset();
+    log_to_monitors = v;
+  }
+}
+
+void LogChannel::update_config(map<string,string> &log_to_monitors,
+			       map<string,string> &log_to_syslog,
+			       map<string,string> &log_channels,
+			       map<string,string> &log_prios,
+			       map<string,string> &log_to_graylog,
+			       map<string,string> &log_to_graylog_host,
+			       map<string,string> &log_to_graylog_port,
+			       uuid_d &fsid,
+			       string &host)
+{
+  logger().debug(
+    "{} log_to_monitors {} log_to_syslog {} log_channels {} log_prios {}",
+    __func__, log_to_monitors, log_to_syslog, log_channels, log_prios);
+  bool to_monitors = (get_str_map_key(log_to_monitors, log_channel,
+                                      &CLOG_CONFIG_DEFAULT_KEY) == "true");
+  bool to_syslog = (get_str_map_key(log_to_syslog, log_channel,
+                                    &CLOG_CONFIG_DEFAULT_KEY) == "true");
+  string syslog_facility = get_str_map_key(log_channels, log_channel,
+					   &CLOG_CONFIG_DEFAULT_KEY);
+  string prio = get_str_map_key(log_prios, log_channel,
+				&CLOG_CONFIG_DEFAULT_KEY);
+  bool to_graylog = (get_str_map_key(log_to_graylog, log_channel,
+				     &CLOG_CONFIG_DEFAULT_KEY) == "true");
+  string graylog_host = get_str_map_key(log_to_graylog_host, log_channel,
+				       &CLOG_CONFIG_DEFAULT_KEY);
+  string graylog_port_str = get_str_map_key(log_to_graylog_port, log_channel,
+					    &CLOG_CONFIG_DEFAULT_KEY);
+  int graylog_port = atoi(graylog_port_str.c_str());
+
+  set_log_to_monitors(to_monitors);
+  set_log_to_syslog(to_syslog);
+  set_syslog_facility(syslog_facility);
+  set_log_prio(prio);
+
+  if (to_graylog && !graylog) { /* should but isn't */
+    graylog = seastar::make_shared<ceph::logging::Graylog>("clog");
+  } else if (!to_graylog && graylog) { /* shouldn't but is */
+    graylog = nullptr;
+  }
+
+  if (to_graylog && graylog) {
+    graylog->set_fsid(fsid);
+    graylog->set_hostname(host);
+  }
+
+  if (graylog && (!graylog_host.empty()) && (graylog_port != 0)) {
+    graylog->set_destination(graylog_host, graylog_port);
+  }
+
+  logger().debug("{} to_monitors: {} to_syslog: {}"
+	  "syslog_facility: {} prio: {} to_graylog: {} graylog_host: {}"
+	  "graylog_port: {}", __func__, (to_monitors ? "true" : "false"),
+	  (to_syslog ? "true" : "false"), syslog_facility, prio,
+	  (to_graylog ? "true" : "false"), graylog_host, graylog_port);
+}
+
+void LogChannel::do_log(clog_type prio, std::stringstream& ss)
+{
+  while (!ss.eof()) {
+    string s;
+    getline(ss, s);
+    if (!s.empty()) {
+      do_log(prio, s);
+    }
+  }
+}
+
+void LogChannel::do_log(clog_type prio, const std::string& s)
+{
+  if (CLOG_ERROR == prio) {
+    logger().error("log {} : {}", prio, s);
+  } else {
+    logger().warn("log {} : {}", prio, s);
+  }
+  LogEntry e;
+  e.stamp = ceph_clock_now();
+  e.addrs = parent->get_myaddrs();
+  e.name = parent->get_myname();
+  e.rank = parent->get_myrank();
+  e.prio = prio;
+  e.msg = s;
+  e.channel = get_log_channel();
+
+  // seq and who should be set for syslog/graylog/log_to_mon
+  // log to monitor?
+  if (log_to_monitors) {
+    e.seq = parent->queue(e);
+  } else {
+    e.seq = parent->get_next_seq();
+  }
+
+  // log to syslog?
+  if (do_log_to_syslog()) {
+    logger().warn("{} log to syslog", __func__);
+    e.log_to_syslog(get_log_prio(), get_syslog_facility());
+  }
+
+  // log to graylog?
+  if (do_log_to_graylog()) {
+    logger().warn("{} log to graylog", __func__);
+    graylog->log_log_entry(&e);
+  }
+}
+
+MessageURef LogClient::get_mon_log_message(log_flushing_t flush_flag)
+{
+  if (flush_flag == log_flushing_t::FLUSH) {
+    if (log_queue.empty()) {
+      return {};
+    }
+    // reset session
+    last_log_sent = log_queue.front().seq;
+  }
+  return _get_mon_log_message();
+}
+
+bool LogClient::are_pending() const
+{
+  return last_log > last_log_sent;
+}
+
+MessageURef LogClient::_get_mon_log_message()
+{
+  if (log_queue.empty()) {
+    return {};
+  }
+
+  // only send entries that haven't been sent yet during this mon
+  // session!  monclient needs to call reset_session() on mon session
+  // reset for this to work right.
+
+  if (last_log_sent == last_log) {
+    return {};
+  }
+
+  // limit entries per message
+  const int64_t num_unsent = last_log - last_log_sent;
+  int64_t num_to_send;
+  if (local_conf()->mon_client_max_log_entries_per_message > 0) {
+    num_to_send = std::min(num_unsent,
+		 local_conf()->mon_client_max_log_entries_per_message);
+  } else {
+    num_to_send = num_unsent;
+  }
+
+  logger().debug("log_queue is {} last_log {} sent {} num {} unsent {}"
+		" sending {}", log_queue.size(), last_log,
+		last_log_sent, log_queue.size(), num_unsent, num_to_send);
+  ceph_assert((unsigned)num_unsent <= log_queue.size());
+  auto log_iter = log_queue.begin();
+  std::deque<LogEntry> out_log_queue; /* will send the logs contained here */
+  while (log_iter->seq <= last_log_sent) {
+    ++log_iter;
+    ceph_assert(log_iter != log_queue.end());
+  }
+  while (num_to_send--) {
+    ceph_assert(log_iter != log_queue.end());
+    out_log_queue.push_back(*log_iter);
+    last_log_sent = log_iter->seq;
+    logger().debug(" will send {}", *log_iter);
+    ++log_iter;
+  }
+  
+  return crimson::make_message<MLog>(m_fsid,
+				  std::move(out_log_queue));
+}
+
+version_t LogClient::queue(LogEntry &entry)
+{
+  entry.seq = ++last_log;
+  log_queue.push_back(entry);
+
+  return entry.seq;
+}
+
+void LogClient::reset()
+{
+  if (log_queue.size()) {
+    log_queue.clear();
+  }
+  last_log_sent = last_log;
+}
+
+uint64_t LogClient::get_next_seq()
+{
+  return ++last_log;
+}
+
+entity_addrvec_t LogClient::get_myaddrs() const
+{
+  return messenger->get_myaddrs();
+}
+
+entity_name_t LogClient::get_myrank()
+{
+  return messenger->get_myname();
+}
+
+const EntityName& LogClient::get_myname() const
+{
+  return local_conf()->name;
+}
+
+seastar::future<> LogClient::handle_log_ack(Ref<MLogAck> m)
+{
+  logger().debug("handle_log_ack {}", *m);
+
+  version_t last = m->last;
+
+  auto q = log_queue.begin();
+  while (q != log_queue.end()) {
+    const LogEntry &entry(*q);
+    if (entry.seq > last)
+      break;
+    logger().debug(" logged {}", entry);
+    q = log_queue.erase(q);
+  }
+  return seastar::now();
+}
+
+LogChannelRef LogClient::create_channel(const std::string& name) {
+  auto it = channels.find(name);
+  if (it == channels.end()) {
+    it = channels.insert(it,
+           {name, seastar::make_lw_shared<LogChannel>(this, name)});
+  }
+  return it->second;
+}
+
+seastar::future<> LogClient::set_fsid(const uuid_d& fsid) {
+  m_fsid = fsid;
+  return seastar::now();
+}
+