--- /dev/null
+#include "crimson/common/logclient.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();
+}
+
--- /dev/null
+#ifndef CEPH_LOGCLIENT_H
+#define CEPH_LOGCLIENT_H
+
+#include "common/LogEntry.h"
+#include "common/ostream_temp.h"
+#include "common/ref.h"
+#include "include/health.h"
+#include "crimson/net/Fwd.h"
+
+#include <seastar/core/future.hh>
+#include <seastar/core/gate.hh>
+#include <seastar/core/lowres_clock.hh>
+#include <seastar/core/shared_ptr.hh>
+#include <seastar/core/timer.hh>
+
+class LogClient;
+class MLog;
+class MLogAck;
+class Message;
+struct uuid_d;
+struct Connection;
+
+class LogChannel;
+
+namespace ceph {
+namespace logging {
+ class Graylog;
+}
+}
+
+template<typename Message> using Ref = boost::intrusive_ptr<Message>;
+namespace crimson::net {
+ class Messenger;
+}
+
+enum class log_flushing_t {
+ NO_FLUSH,
+ FLUSH
+};
+
+int parse_log_client_options(CephContext *cct,
+ std::map<std::string,std::string> &log_to_monitors,
+ std::map<std::string,std::string> &log_to_syslog,
+ std::map<std::string,std::string> &log_channels,
+ std::map<std::string,std::string> &log_prios,
+ std::map<std::string,std::string> &log_to_graylog,
+ std::map<std::string,std::string> &log_to_graylog_host,
+ std::map<std::string,std::string> &log_to_graylog_port,
+ uuid_d &fsid,
+ std::string &host);
+
+/** Manage where we output to and at which priority
+ *
+ * Not to be confused with the LogClient, which is the almighty coordinator
+ * of channels. We just deal with the boring part of the logging: send to
+ * syslog, send to file, generate LogEntry and queue it for the LogClient.
+ *
+ * Past queueing the LogEntry, the LogChannel is done with the whole thing.
+ * LogClient will deal with sending and handling of LogEntries.
+ */
+class LogChannel : public OstreamTemp::OstreamTempSink
+{
+public:
+ LogChannel(LogClient *lc, const std::string &channel);
+ LogChannel(LogClient *lc, const std::string &channel,
+ const std::string &facility, const std::string &prio);
+
+ OstreamTemp debug() {
+ return OstreamTemp(CLOG_DEBUG, this);
+ }
+ void debug(std::stringstream &s) {
+ do_log(CLOG_DEBUG, s);
+ }
+ /**
+ * Convenience function mapping health status to
+ * the appropriate cluster log severity.
+ */
+ OstreamTemp health(health_status_t health) {
+ switch(health) {
+ case HEALTH_OK:
+ return info();
+ case HEALTH_WARN:
+ return warn();
+ case HEALTH_ERR:
+ return error();
+ default:
+ // Invalid health_status_t value
+ ceph_abort();
+ }
+ }
+ OstreamTemp info() {
+ return OstreamTemp(CLOG_INFO, this);
+ }
+ void info(std::stringstream &s) {
+ do_log(CLOG_INFO, s);
+ }
+ OstreamTemp warn() {
+ return OstreamTemp(CLOG_WARN, this);
+ }
+ void warn(std::stringstream &s) {
+ do_log(CLOG_WARN, s);
+ }
+ OstreamTemp error() {
+ return OstreamTemp(CLOG_ERROR, this);
+ }
+ void error(std::stringstream &s) {
+ do_log(CLOG_ERROR, s);
+ }
+ OstreamTemp sec() {
+ return OstreamTemp(CLOG_SEC, this);
+ }
+ void sec(std::stringstream &s) {
+ do_log(CLOG_SEC, s);
+ }
+
+ void set_log_to_monitors(bool v);
+ void set_log_to_syslog(bool v) {
+ log_to_syslog = v;
+ }
+ void set_log_channel(const std::string& v) {
+ log_channel = v;
+ }
+ void set_log_prio(const std::string& v) {
+ log_prio = v;
+ }
+ void set_syslog_facility(const std::string& v) {
+ syslog_facility = v;
+ }
+ const std::string& get_log_prio() const { return log_prio; }
+ const std::string& get_log_channel() const { return log_channel; }
+ const std::string& get_syslog_facility() const { return syslog_facility; }
+ bool must_log_to_syslog() const { return log_to_syslog; }
+ /**
+ * Do we want to log to syslog?
+ *
+ * @return true if log_to_syslog is true and both channel and prio
+ * are not empty; false otherwise.
+ */
+ bool do_log_to_syslog() {
+ return must_log_to_syslog() &&
+ !log_prio.empty() && !log_channel.empty();
+ }
+ bool must_log_to_monitors() { return log_to_monitors; }
+
+ bool do_log_to_graylog() {
+ return (graylog != nullptr);
+ }
+
+ using Ref = seastar::lw_shared_ptr<LogChannel>;
+
+ /**
+ * update config values from parsed k/v std::map for each config option
+ *
+ * Pick out the relevant value based on our channel.
+ */
+ void update_config(std::map<std::string,std::string> &log_to_monitors,
+ std::map<std::string,std::string> &log_to_syslog,
+ std::map<std::string,std::string> &log_channels,
+ std::map<std::string,std::string> &log_prios,
+ std::map<std::string,std::string> &log_to_graylog,
+ std::map<std::string,std::string> &log_to_graylog_host,
+ std::map<std::string,std::string> &log_to_graylog_port,
+ uuid_d &fsid,
+ std::string &host);
+
+ void do_log(clog_type prio, std::stringstream& ss);
+ void do_log(clog_type prio, const std::string& s);
+
+private:
+ LogClient *parent;
+ std::string log_channel;
+ std::string log_prio;
+ std::string syslog_facility;
+ bool log_to_syslog;
+ bool log_to_monitors;
+ seastar::shared_ptr<ceph::logging::Graylog> graylog;
+};
+
+using LogChannelRef = LogChannel::Ref;
+
+class LogClient
+{
+public:
+ enum logclient_flag_t {
+ NO_FLAGS = 0,
+ FLAG_MON = 0x1,
+ };
+
+ LogClient(crimson::net::Messenger *m, logclient_flag_t flags);
+
+ virtual ~LogClient() = default;
+
+ seastar::future<> handle_log_ack(Ref<MLogAck> m);
+ MessageURef get_mon_log_message(log_flushing_t flush_flag);
+ bool are_pending() const;
+
+ LogChannelRef create_channel() {
+ return create_channel(CLOG_CHANNEL_DEFAULT);
+ }
+
+ LogChannelRef create_channel(const std::string& name);
+
+ void destroy_channel(const std::string& name) {
+ channels.erase(name);
+ }
+
+ void shutdown() {
+ channels.clear();
+ }
+
+ uint64_t get_next_seq();
+ entity_addrvec_t get_myaddrs() const;
+ const EntityName& get_myname() const;
+ entity_name_t get_myrank();
+ version_t queue(LogEntry &entry);
+ void reset();
+ seastar::future<> set_fsid(const uuid_d& fsid);
+
+private:
+ MessageURef _get_mon_log_message();
+
+ crimson::net::Messenger *messenger;
+ bool is_mon;
+ version_t last_log_sent;
+ version_t last_log;
+ std::deque<LogEntry> log_queue;
+
+ std::map<std::string, LogChannelRef> channels;
+ uuid_d m_fsid;
+};
+#endif
+
projects / ceph.git / commitdiff