ceph/src/common/LogClient.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #include "common/LogClient.h"
  16 #include "include/str_map.h"
  17 #include "messages/MLog.h"
  18 #include "messages/MLogAck.h"
  19 #include "msg/Messenger.h"
  20 #include "mon/MonMap.h"
  21 #include "common/Graylog.h"
  22
  23 #define dout_subsys ceph_subsys_monc
  24
  25 int parse_log_client_options(CephContext *cct,
  26                              map<string,string> &log_to_monitors,
  27                              map<string,string> &log_to_syslog,
  28                              map<string,string> &log_channels,
  29                              map<string,string> &log_prios,
  30                              map<string,string> &log_to_graylog,
  31                              map<string,string> &log_to_graylog_host,
  32                              map<string,string> &log_to_graylog_port,
  33                              uuid_d &fsid,
  34                              string &host)
  35 {
  36   ostringstream oss;
  37
  38   int r = get_conf_str_map_helper(cct->_conf->clog_to_monitors, oss,
  39                                   &log_to_monitors, CLOG_CONFIG_DEFAULT_KEY);
  40   if (r < 0) {
  41     lderr(cct) << __func__ << " error parsing 'clog_to_monitors'" << dendl;
  42     return r;
  43   }
  44
  45   r = get_conf_str_map_helper(cct->_conf->clog_to_syslog, oss,
  46                               &log_to_syslog, CLOG_CONFIG_DEFAULT_KEY);
  47   if (r < 0) {
  48     lderr(cct) << __func__ << " error parsing 'clog_to_syslog'" << dendl;
  49     return r;
  50   }
  51
  52   r = get_conf_str_map_helper(cct->_conf->clog_to_syslog_facility, oss,
  53                               &log_channels, CLOG_CONFIG_DEFAULT_KEY);
  54   if (r < 0) {
  55     lderr(cct) << __func__ << " error parsing 'clog_to_syslog_facility'" << dendl;
  56     return r;
  57   }
  58
  59   r = get_conf_str_map_helper(cct->_conf->clog_to_syslog_level, oss,
  60                               &log_prios, CLOG_CONFIG_DEFAULT_KEY);
  61   if (r < 0) {
  62     lderr(cct) << __func__ << " error parsing 'clog_to_syslog_level'" << dendl;
  63     return r;
  64   }
  65
  66   r = get_conf_str_map_helper(cct->_conf->clog_to_graylog, oss,
  67                               &log_to_graylog, CLOG_CONFIG_DEFAULT_KEY);
  68   if (r < 0) {
  69     lderr(cct) << __func__ << " error parsing 'clog_to_graylog'" << dendl;
  70     return r;
  71   }
  72
  73   r = get_conf_str_map_helper(cct->_conf->clog_to_graylog_host, oss,
  74                               &log_to_graylog_host, CLOG_CONFIG_DEFAULT_KEY);
  75   if (r < 0) {
  76     lderr(cct) << __func__ << " error parsing 'clog_to_graylog_host'" << dendl;
  77     return r;
  78   }
  79
  80   r = get_conf_str_map_helper(cct->_conf->clog_to_graylog_port, oss,
  81                               &log_to_graylog_port, CLOG_CONFIG_DEFAULT_KEY);
  82   if (r < 0) {
  83     lderr(cct) << __func__ << " error parsing 'clog_to_graylog_port'" << dendl;
  84     return r;
  85   }
  86
  87   fsid = cct->_conf.get_val<uuid_d>("fsid");
  88   host = cct->_conf->host;
  89   return 0;
  90 }
  91
  92 #undef dout_prefix
  93 #define dout_prefix _prefix(_dout, this)
  94 static ostream& _prefix(std::ostream *_dout, LogClient *logc) {
  95   return *_dout << "log_client ";
  96 }
  97
  98 static ostream& _prefix(std::ostream *_dout, LogChannel *lc) {
  99   return *_dout << "log_channel(" << lc->get_log_channel() << ") ";
 100 }
 101
 102 LogChannel::LogChannel(CephContext *cct, LogClient *lc, const string &channel)
 103   : cct(cct), parent(lc),
 104     log_channel(channel), log_to_syslog(false), log_to_monitors(false)
 105 {
 106 }
 107
 108 LogChannel::LogChannel(CephContext *cct, LogClient *lc,
 109                        const string &channel, const string &facility,
 110                        const string &prio)
 111   : cct(cct), parent(lc),
 112     log_channel(channel), log_prio(prio), syslog_facility(facility),
 113     log_to_syslog(false), log_to_monitors(false)
 114 {
 115 }
 116
 117 LogClient::LogClient(CephContext *cct, Messenger *m, MonMap *mm,
 118                      enum logclient_flag_t flags)
 119   : cct(cct), messenger(m), monmap(mm), is_mon(flags & FLAG_MON),
 120     last_log_sent(0), last_log(0)
 121 {
 122 }
 123
 124 LogClientTemp::LogClientTemp(clog_type type_, LogChannel &parent_)
 125   : type(type_), parent(parent_)
 126 {
 127 }
 128
 129 LogClientTemp::LogClientTemp(const LogClientTemp &rhs)
 130   : type(rhs.type), parent(rhs.parent)
 131 {
 132   // don't want to-- nor can we-- copy the ostringstream
 133 }
 134
 135 LogClientTemp::~LogClientTemp()
 136 {
 137   if (ss.peek() != EOF)
 138     parent.do_log(type, ss);
 139 }
 140
 141 void LogChannel::update_config(map<string,string> &log_to_monitors,
 142                                map<string,string> &log_to_syslog,
 143                                map<string,string> &log_channels,
 144                                map<string,string> &log_prios,
 145                                map<string,string> &log_to_graylog,
 146                                map<string,string> &log_to_graylog_host,
 147                                map<string,string> &log_to_graylog_port,
 148                                uuid_d &fsid,
 149                                string &host)
 150 {
 151   ldout(cct, 20) << __func__ << " log_to_monitors " << log_to_monitors
 152                  << " log_to_syslog " << log_to_syslog
 153                  << " log_channels " << log_channels
 154                  << " log_prios " << log_prios
 155                  << dendl;
 156   bool to_monitors = (get_str_map_key(log_to_monitors, log_channel,
 157                                       &CLOG_CONFIG_DEFAULT_KEY) == "true");
 158   bool to_syslog = (get_str_map_key(log_to_syslog, log_channel,
 159                                     &CLOG_CONFIG_DEFAULT_KEY) == "true");
 160   string syslog_facility = get_str_map_key(log_channels, log_channel,
 161                                            &CLOG_CONFIG_DEFAULT_KEY);
 162   string prio = get_str_map_key(log_prios, log_channel,
 163                                 &CLOG_CONFIG_DEFAULT_KEY);
 164   bool to_graylog = (get_str_map_key(log_to_graylog, log_channel,
 165                                      &CLOG_CONFIG_DEFAULT_KEY) == "true");
 166   string graylog_host = get_str_map_key(log_to_graylog_host, log_channel,
 167                                        &CLOG_CONFIG_DEFAULT_KEY);
 168   string graylog_port_str = get_str_map_key(log_to_graylog_port, log_channel,
 169                                             &CLOG_CONFIG_DEFAULT_KEY);
 170   int graylog_port = atoi(graylog_port_str.c_str());
 171
 172   set_log_to_monitors(to_monitors);
 173   set_log_to_syslog(to_syslog);
 174   set_syslog_facility(syslog_facility);
 175   set_log_prio(prio);
 176
 177   if (to_graylog && !graylog) { /* should but isn't */
 178     graylog = std::make_shared<ceph::logging::Graylog>("clog");
 179   } else if (!to_graylog && graylog) { /* shouldn't but is */
 180     graylog.reset();
 181   }
 182
 183   if (to_graylog && graylog) {
 184     graylog->set_fsid(fsid);
 185     graylog->set_hostname(host);
 186   }
 187
 188   if (graylog && (!graylog_host.empty()) && (graylog_port != 0)) {
 189     graylog->set_destination(graylog_host, graylog_port);
 190   }
 191
 192   ldout(cct, 10) << __func__
 193                  << " to_monitors: " << (to_monitors ? "true" : "false")
 194                  << " to_syslog: " << (to_syslog ? "true" : "false")
 195                  << " syslog_facility: " << syslog_facility
 196                  << " prio: " << prio
 197                  << " to_graylog: " << (to_graylog ? "true" : "false")
 198                  << " graylog_host: " << graylog_host
 199                  << " graylog_port: " << graylog_port
 200                  << ")" << dendl;
 201 }
 202
 203 void LogChannel::do_log(clog_type prio, std::stringstream& ss)
 204 {
 205   while (!ss.eof()) {
 206     string s;
 207     getline(ss, s);
 208     if (!s.empty())
 209       do_log(prio, s);
 210   }
 211 }
 212
 213 void LogChannel::do_log(clog_type prio, const std::string& s)
 214 {
 215   std::lock_guard l(channel_lock);
 216   if (CLOG_ERROR == prio) {
 217     ldout(cct,-1) << "log " << prio << " : " << s << dendl;
 218   } else {
 219     ldout(cct,0) << "log " << prio << " : " << s << dendl;
 220   }
 221   LogEntry e;
 222   e.stamp = ceph_clock_now();
 223   // seq and who should be set for syslog/graylog/log_to_mon
 224   e.addrs = parent->get_myaddrs();
 225   e.name = parent->get_myname();
 226   e.rank = parent->get_myrank();
 227   e.prio = prio;
 228   e.msg = s;
 229   e.channel = get_log_channel();
 230
 231   // log to monitor?
 232   if (log_to_monitors) {
 233     e.seq = parent->queue(e);
 234   } else {
 235     e.seq = parent->get_next_seq();
 236   }
 237
 238   // log to syslog?
 239   if (do_log_to_syslog()) {
 240     ldout(cct,0) << __func__ << " log to syslog"  << dendl;
 241     e.log_to_syslog(get_log_prio(), get_syslog_facility());
 242   }
 243
 244   // log to graylog?
 245   if (do_log_to_graylog()) {
 246     ldout(cct,0) << __func__ << " log to graylog"  << dendl;
 247     graylog->log_log_entry(&e);
 248   }
 249 }
 250
 251 Message *LogClient::get_mon_log_message(bool flush)
 252 {
 253   std::lock_guard l(log_lock);
 254   if (flush) {
 255     if (log_queue.empty())
 256       return nullptr;
 257     // reset session
 258     last_log_sent = log_queue.front().seq;
 259   }
 260   return _get_mon_log_message();
 261 }
 262
 263 bool LogClient::are_pending()
 264 {
 265   std::lock_guard l(log_lock);
 266   return last_log > last_log_sent;
 267 }
 268
 269 Message *LogClient::_get_mon_log_message()
 270 {
 271   ceph_assert(ceph_mutex_is_locked(log_lock));
 272   if (log_queue.empty())
 273     return NULL;
 274
 275   // only send entries that haven't been sent yet during this mon
 276   // session!  monclient needs to call reset_session() on mon session
 277   // reset for this to work right.
 278
 279   if (last_log_sent == last_log)
 280     return NULL;
 281
 282   // limit entries per message
 283   unsigned num_unsent = last_log - last_log_sent;
 284   unsigned num_send;
 285   if (cct->_conf->mon_client_max_log_entries_per_message > 0)
 286     num_send = std::min(num_unsent, (unsigned)cct->_conf->mon_client_max_log_entries_per_message);
 287   else
 288     num_send = num_unsent;
 289
 290   ldout(cct,10) << " log_queue is " << log_queue.size() << " last_log " << last_log << " sent " << last_log_sent
 291                 << " num " << log_queue.size()
 292                 << " unsent " << num_unsent
 293                 << " sending " << num_send << dendl;
 294   ceph_assert(num_unsent <= log_queue.size());
 295   std::deque<LogEntry>::iterator p = log_queue.begin();
 296   std::deque<LogEntry> o;
 297   while (p->seq <= last_log_sent) {
 298     ++p;
 299     ceph_assert(p != log_queue.end());
 300   }
 301   while (num_send--) {
 302     ceph_assert(p != log_queue.end());
 303     o.push_back(*p);
 304     last_log_sent = p->seq;
 305     ldout(cct,10) << " will send " << *p << dendl;
 306     ++p;
 307   }
 308
 309   MLog *log = new MLog(monmap->get_fsid());
 310   log->entries.swap(o);
 311
 312   return log;
 313 }
 314
 315 void LogClient::_send_to_mon()
 316 {
 317   ceph_assert(ceph_mutex_is_locked(log_lock));
 318   ceph_assert(is_mon);
 319   ceph_assert(messenger->get_myname().is_mon());
 320   ldout(cct,10) << __func__ << " log to self" << dendl;
 321   Message *log = _get_mon_log_message();
 322   messenger->get_loopback_connection()->send_message(log);
 323 }
 324
 325 version_t LogClient::queue(LogEntry &entry)
 326 {
 327   std::lock_guard l(log_lock);
 328   entry.seq = ++last_log;
 329   log_queue.push_back(entry);
 330
 331   if (is_mon) {
 332     _send_to_mon();
 333   }
 334
 335   return entry.seq;
 336 }
 337
 338 uint64_t LogClient::get_next_seq()
 339 {
 340   std::lock_guard l(log_lock);
 341   return ++last_log;
 342 }
 343
 344 entity_addrvec_t LogClient::get_myaddrs()
 345 {
 346   return messenger->get_myaddrs();
 347 }
 348
 349 entity_name_t LogClient::get_myrank()
 350 {
 351   return messenger->get_myname();
 352 }
 353
 354 const EntityName& LogClient::get_myname()
 355 {
 356   return cct->_conf->name;
 357 }
 358
 359 bool LogClient::handle_log_ack(MLogAck *m)
 360 {
 361   std::lock_guard l(log_lock);
 362   ldout(cct,10) << "handle_log_ack " << *m << dendl;
 363
 364   version_t last = m->last;
 365
 366   deque<LogEntry>::iterator q = log_queue.begin();
 367   while (q != log_queue.end()) {
 368     const LogEntry &entry(*q);
 369     if (entry.seq > last)
 370       break;
 371     ldout(cct,10) << " logged " << entry << dendl;
 372     q = log_queue.erase(q);
 373   }
 374   return true;
 375 }
 376