ceph/src/mon/LogMonitor.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 <boost/algorithm/string/predicate.hpp>
  16
  17 #include <sstream>
  18 #include <syslog.h>
  19
  20 #include "LogMonitor.h"
  21 #include "Monitor.h"
  22 #include "MonitorDBStore.h"
  23
  24 #include "messages/MMonCommand.h"
  25 #include "messages/MLog.h"
  26 #include "messages/MLogAck.h"
  27 #include "common/Graylog.h"
  28 #include "common/errno.h"
  29 #include "common/strtol.h"
  30 #include "include/assert.h"
  31 #include "include/str_list.h"
  32 #include "include/str_map.h"
  33 #include "include/compat.h"
  34
  35 #define dout_subsys ceph_subsys_mon
  36 #undef dout_prefix
  37 #define dout_prefix _prefix(_dout, mon, get_last_committed())
  38 static ostream& _prefix(std::ostream *_dout, Monitor *mon, version_t v) {
  39   return *_dout << "mon." << mon->name << "@" << mon->rank
  40                 << "(" << mon->get_state_name()
  41                 << ").log v" << v << " ";
  42 }
  43
  44 ostream& operator<<(ostream &out, const LogMonitor &pm)
  45 {
  46   return out << "log";
  47 }
  48
  49 /*
  50  Tick function to update the map based on performance every N seconds
  51 */
  52
  53 void LogMonitor::tick()
  54 {
  55   if (!is_active()) return;
  56
  57   dout(10) << *this << dendl;
  58
  59 }
  60
  61 void LogMonitor::create_initial()
  62 {
  63   dout(10) << "create_initial -- creating initial map" << dendl;
  64   LogEntry e;
  65   memset(&e.who, 0, sizeof(e.who));
  66   e.name = g_conf->name;
  67   e.stamp = ceph_clock_now();
  68   e.prio = CLOG_INFO;
  69   std::stringstream ss;
  70   ss << "mkfs " << mon->monmap->get_fsid();
  71   e.msg = ss.str();
  72   e.seq = 0;
  73   pending_log.insert(pair<utime_t,LogEntry>(e.stamp, e));
  74 }
  75
  76 void LogMonitor::update_from_paxos(bool *need_bootstrap)
  77 {
  78   dout(10) << __func__ << dendl;
  79   version_t version = get_last_committed();
  80   dout(10) << __func__ << " version " << version
  81            << " summary v " << summary.version << dendl;
  82   if (version == summary.version)
  83     return;
  84   assert(version >= summary.version);
  85
  86   map<string,bufferlist> channel_blog;
  87
  88   version_t latest_full = get_version_latest_full();
  89   dout(10) << __func__ << " latest full " << latest_full << dendl;
  90   if ((latest_full > 0) && (latest_full > summary.version)) {
  91     bufferlist latest_bl;
  92     get_version_full(latest_full, latest_bl);
  93     assert(latest_bl.length() != 0);
  94     dout(7) << __func__ << " loading summary e" << latest_full << dendl;
  95     bufferlist::iterator p = latest_bl.begin();
  96     ::decode(summary, p);
  97     dout(7) << __func__ << " loaded summary e" << summary.version << dendl;
  98   }
  99
 100   // walk through incrementals
 101   while (version > summary.version) {
 102     bufferlist bl;
 103     int err = get_version(summary.version+1, bl);
 104     assert(err == 0);
 105     assert(bl.length());
 106
 107     bufferlist::iterator p = bl.begin();
 108     __u8 v;
 109     ::decode(v, p);
 110     while (!p.end()) {
 111       LogEntry le;
 112       le.decode(p);
 113       dout(7) << "update_from_paxos applying incremental log " << summary.version+1 <<  " " << le << dendl;
 114
 115       string channel = le.channel;
 116       if (channel.empty()) // keep retrocompatibility
 117         channel = CLOG_CHANNEL_CLUSTER;
 118
 119       if (channels.do_log_to_syslog(channel)) {
 120         string level = channels.get_level(channel);
 121         string facility = channels.get_facility(channel);
 122         if (level.empty() || facility.empty()) {
 123           derr << __func__ << " unable to log to syslog -- level or facility"
 124                << " not defined (level: " << level << ", facility: "
 125                << facility << ")" << dendl;
 126           continue;
 127         }
 128         le.log_to_syslog(channels.get_level(channel),
 129                          channels.get_facility(channel));
 130       }
 131
 132       if (channels.do_log_to_graylog(channel)) {
 133         ceph::logging::Graylog::Ref graylog = channels.get_graylog(channel);
 134         if (graylog) {
 135           graylog->log_log_entry(&le);
 136         }
 137         dout(7) << "graylog: " << channel << " " << graylog
 138                 << " host:" << channels.log_to_graylog_host << dendl;
 139       }
 140
 141       string log_file = channels.get_log_file(channel);
 142       dout(20) << __func__ << " logging for channel '" << channel
 143                << "' to file '" << log_file << "'" << dendl;
 144
 145       if (!log_file.empty()) {
 146         string log_file_level = channels.get_log_file_level(channel);
 147         if (log_file_level.empty()) {
 148           dout(1) << __func__ << " warning: log file level not defined for"
 149                   << " channel '" << channel << "' yet a log file is --"
 150                   << " will assume lowest level possible" << dendl;
 151         }
 152
 153         int min = string_to_syslog_level(log_file_level);
 154         int l = clog_type_to_syslog_level(le.prio);
 155         if (l <= min) {
 156           stringstream ss;
 157           ss << le << "\n";
 158           // init entry if DNE
 159           bufferlist &blog = channel_blog[channel];
 160           blog.append(ss.str());
 161         }
 162       }
 163
 164       summary.add(le);
 165     }
 166
 167     summary.version++;
 168     summary.prune(g_conf->mon_log_max_summary);
 169   }
 170
 171   dout(15) << __func__ << " logging for "
 172            << channel_blog.size() << " channels" << dendl;
 173   for(map<string,bufferlist>::iterator p = channel_blog.begin();
 174       p != channel_blog.end(); ++p) {
 175     if (!p->second.length()) {
 176       dout(15) << __func__ << " channel '" << p->first
 177                << "': nothing to log" << dendl;
 178       continue;
 179     }
 180
 181     dout(15) << __func__ << " channel '" << p->first
 182              << "' logging " << p->second.length() << " bytes" << dendl;
 183     string log_file = channels.get_log_file(p->first);
 184
 185     int fd = ::open(log_file.c_str(), O_WRONLY|O_APPEND|O_CREAT, 0600);
 186     if (fd < 0) {
 187       int err = -errno;
 188       dout(1) << "unable to write to '" << log_file << "' for channel '"
 189               << p->first << "': " << cpp_strerror(err) << dendl;
 190     } else {
 191       int err = p->second.write_fd(fd);
 192       if (err < 0) {
 193         dout(1) << "error writing to '" << log_file << "' for channel '"
 194                 << p->first << ": " << cpp_strerror(err) << dendl;
 195       }
 196       VOID_TEMP_FAILURE_RETRY(::close(fd));
 197     }
 198   }
 199
 200   check_subs();
 201 }
 202
 203 void LogMonitor::create_pending()
 204 {
 205   pending_log.clear();
 206   pending_summary = summary;
 207   dout(10) << "create_pending v " << (get_last_committed() + 1) << dendl;
 208 }
 209
 210 void LogMonitor::encode_pending(MonitorDBStore::TransactionRef t)
 211 {
 212   version_t version = get_last_committed() + 1;
 213   bufferlist bl;
 214   dout(10) << __func__ << " v" << version << dendl;
 215   __u8 v = 1;
 216   ::encode(v, bl);
 217   multimap<utime_t,LogEntry>::iterator p;
 218   for (p = pending_log.begin(); p != pending_log.end(); ++p)
 219     p->second.encode(bl, mon->get_quorum_con_features());
 220
 221   put_version(t, version, bl);
 222   put_last_committed(t, version);
 223 }
 224
 225 void LogMonitor::encode_full(MonitorDBStore::TransactionRef t)
 226 {
 227   dout(10) << __func__ << " log v " << summary.version << dendl;
 228   assert(get_last_committed() == summary.version);
 229
 230   bufferlist summary_bl;
 231   ::encode(summary, summary_bl, mon->get_quorum_con_features());
 232
 233   put_version_full(t, summary.version, summary_bl);
 234   put_version_latest_full(t, summary.version);
 235 }
 236
 237 version_t LogMonitor::get_trim_to()
 238 {
 239   if (!mon->is_leader())
 240     return 0;
 241
 242   unsigned max = g_conf->mon_max_log_epochs;
 243   version_t version = get_last_committed();
 244   if (version > max)
 245     return version - max;
 246   return 0;
 247 }
 248
 249 bool LogMonitor::preprocess_query(MonOpRequestRef op)
 250 {
 251   op->mark_logmon_event("preprocess_query");
 252   PaxosServiceMessage *m = static_cast<PaxosServiceMessage*>(op->get_req());
 253   dout(10) << "preprocess_query " << *m << " from " << m->get_orig_source_inst() << dendl;
 254   switch (m->get_type()) {
 255   case MSG_MON_COMMAND:
 256     return preprocess_command(op);
 257
 258   case MSG_LOG:
 259     return preprocess_log(op);
 260
 261   default:
 262     ceph_abort();
 263     return true;
 264   }
 265 }
 266
 267 bool LogMonitor::prepare_update(MonOpRequestRef op)
 268 {
 269   op->mark_logmon_event("prepare_update");
 270   PaxosServiceMessage *m = static_cast<PaxosServiceMessage*>(op->get_req());
 271   dout(10) << "prepare_update " << *m << " from " << m->get_orig_source_inst() << dendl;
 272   switch (m->get_type()) {
 273   case MSG_MON_COMMAND:
 274     return prepare_command(op);
 275   case MSG_LOG:
 276     return prepare_log(op);
 277   default:
 278     ceph_abort();
 279     return false;
 280   }
 281 }
 282
 283 bool LogMonitor::preprocess_log(MonOpRequestRef op)
 284 {
 285   op->mark_logmon_event("preprocess_log");
 286   MLog *m = static_cast<MLog*>(op->get_req());
 287   dout(10) << "preprocess_log " << *m << " from " << m->get_orig_source() << dendl;
 288   int num_new = 0;
 289
 290   MonSession *session = m->get_session();
 291   if (!session)
 292     goto done;
 293   if (!session->is_capable("log", MON_CAP_W)) {
 294     dout(0) << "preprocess_log got MLog from entity with insufficient privileges "
 295             << session->caps << dendl;
 296     goto done;
 297   }
 298
 299   for (deque<LogEntry>::iterator p = m->entries.begin();
 300        p != m->entries.end();
 301        ++p) {
 302     if (!pending_summary.contains(p->key()))
 303       num_new++;
 304   }
 305   if (!num_new) {
 306     dout(10) << "  nothing new" << dendl;
 307     goto done;
 308   }
 309
 310   return false;
 311
 312  done:
 313   return true;
 314 }
 315
 316 struct LogMonitor::C_Log : public C_MonOp {
 317   LogMonitor *logmon;
 318   C_Log(LogMonitor *p, MonOpRequestRef o) :
 319     C_MonOp(o), logmon(p) {}
 320   void _finish(int r) override {
 321     if (r == -ECANCELED) {
 322       return;
 323     }
 324     logmon->_updated_log(op);
 325   }
 326 };
 327
 328 bool LogMonitor::prepare_log(MonOpRequestRef op)
 329 {
 330   op->mark_logmon_event("prepare_log");
 331   MLog *m = static_cast<MLog*>(op->get_req());
 332   dout(10) << "prepare_log " << *m << " from " << m->get_orig_source() << dendl;
 333
 334   if (m->fsid != mon->monmap->fsid) {
 335     dout(0) << "handle_log on fsid " << m->fsid << " != " << mon->monmap->fsid
 336             << dendl;
 337     return false;
 338   }
 339
 340   for (deque<LogEntry>::iterator p = m->entries.begin();
 341        p != m->entries.end();
 342        ++p) {
 343     dout(10) << " logging " << *p << dendl;
 344     if (!pending_summary.contains(p->key())) {
 345       pending_summary.add(*p);
 346       pending_log.insert(pair<utime_t,LogEntry>(p->stamp, *p));
 347     }
 348   }
 349   pending_summary.prune(g_conf->mon_log_max_summary);
 350   wait_for_finished_proposal(op, new C_Log(this, op));
 351   return true;
 352 }
 353
 354 void LogMonitor::_updated_log(MonOpRequestRef op)
 355 {
 356   MLog *m = static_cast<MLog*>(op->get_req());
 357   dout(7) << "_updated_log for " << m->get_orig_source_inst() << dendl;
 358   mon->send_reply(op, new MLogAck(m->fsid, m->entries.rbegin()->seq));
 359 }
 360
 361 bool LogMonitor::should_propose(double& delay)
 362 {
 363   // commit now if we have a lot of pending events
 364   if (g_conf->mon_max_log_entries_per_event > 0 &&
 365       pending_log.size() >= (unsigned)g_conf->mon_max_log_entries_per_event)
 366     return true;
 367
 368   // otherwise fall back to generic policy
 369   return PaxosService::should_propose(delay);
 370 }
 371
 372
 373 bool LogMonitor::preprocess_command(MonOpRequestRef op)
 374 {
 375   op->mark_logmon_event("preprocess_command");
 376   MMonCommand *m = static_cast<MMonCommand*>(op->get_req());
 377   int r = -EINVAL;
 378   bufferlist rdata;
 379   stringstream ss;
 380
 381   map<string, cmd_vartype> cmdmap;
 382   if (!cmdmap_from_json(m->cmd, &cmdmap, ss)) {
 383     string rs = ss.str();
 384     mon->reply_command(op, -EINVAL, rs, get_last_committed());
 385     return true;
 386   }
 387   MonSession *session = m->get_session();
 388   if (!session) {
 389     mon->reply_command(op, -EACCES, "access denied", get_last_committed());
 390     return true;
 391   }
 392
 393   string prefix;
 394   cmd_getval(g_ceph_context, cmdmap, "prefix", prefix);
 395
 396   string format;
 397   cmd_getval(g_ceph_context, cmdmap, "format", format, string("plain"));
 398   boost::scoped_ptr<Formatter> f(Formatter::create(format));
 399
 400   if (prefix == "log last") {
 401     int64_t num = 20;
 402     cmd_getval(g_ceph_context, cmdmap, "num", num);
 403     if (f) {
 404       f->open_array_section("tail");
 405     }
 406     auto p = summary.tail.end();
 407     while (num > 0 && p != summary.tail.begin()) {
 408       num--;
 409       --p;
 410     }
 411     ostringstream ss;
 412     for ( ; p != summary.tail.end(); ++p) {
 413       if (f) {
 414         f->dump_object("entry", *p);
 415       } else {
 416         ss << *p << "\n";
 417       }
 418     }
 419     if (f) {
 420       f->close_section();
 421       f->flush(rdata);
 422     } else {
 423       rdata.append(ss.str());
 424     }
 425     r = 0;
 426   } else {
 427     return false;
 428   }
 429
 430   string rs;
 431   getline(ss, rs);
 432   mon->reply_command(op, r, rs, rdata, get_last_committed());
 433   return true;
 434 }
 435
 436
 437 bool LogMonitor::prepare_command(MonOpRequestRef op)
 438 {
 439   op->mark_logmon_event("prepare_command");
 440   MMonCommand *m = static_cast<MMonCommand*>(op->get_req());
 441   stringstream ss;
 442   string rs;
 443   int err = -EINVAL;
 444
 445   map<string, cmd_vartype> cmdmap;
 446   if (!cmdmap_from_json(m->cmd, &cmdmap, ss)) {
 447     // ss has reason for failure
 448     string rs = ss.str();
 449     mon->reply_command(op, -EINVAL, rs, get_last_committed());
 450     return true;
 451   }
 452
 453   string prefix;
 454   cmd_getval(g_ceph_context, cmdmap, "prefix", prefix);
 455
 456   MonSession *session = m->get_session();
 457   if (!session) {
 458     mon->reply_command(op, -EACCES, "access denied", get_last_committed());
 459     return true;
 460   }
 461
 462   if (prefix == "log") {
 463     vector<string> logtext;
 464     cmd_getval(g_ceph_context, cmdmap, "logtext", logtext);
 465     LogEntry le;
 466     le.who = m->get_orig_source_inst();
 467     le.name = session->entity_name;
 468     le.stamp = m->get_recv_stamp();
 469     le.seq = 0;
 470     le.prio = CLOG_INFO;
 471     le.msg = str_join(logtext, " ");
 472     pending_summary.add(le);
 473     pending_summary.prune(g_conf->mon_log_max_summary);
 474     pending_log.insert(pair<utime_t,LogEntry>(le.stamp, le));
 475     wait_for_finished_proposal(op, new Monitor::C_Command(
 476           mon, op, 0, string(), get_last_committed() + 1));
 477     return true;
 478   }
 479
 480   getline(ss, rs);
 481   mon->reply_command(op, err, rs, get_last_committed());
 482   return false;
 483 }
 484
 485
 486 int LogMonitor::sub_name_to_id(const string& n)
 487 {
 488   if (n == "log-debug")
 489     return CLOG_DEBUG;
 490   if (n == "log-info")
 491     return CLOG_INFO;
 492   if (n == "log-sec")
 493     return CLOG_SEC;
 494   if (n == "log-warn")
 495     return CLOG_WARN;
 496   if (n == "log-error")
 497     return CLOG_ERROR;
 498   return CLOG_UNKNOWN;
 499 }
 500
 501 void LogMonitor::check_subs()
 502 {
 503   dout(10) << __func__ << dendl;
 504   for (map<string, xlist<Subscription*>*>::iterator i = mon->session_map.subs.begin();
 505        i != mon->session_map.subs.end();
 506        ++i) {
 507     for (xlist<Subscription*>::iterator j = i->second->begin(); !j.end(); ++j) {
 508       if (sub_name_to_id((*j)->type) >= 0)
 509         check_sub(*j);
 510     }
 511   }
 512 }
 513
 514 void LogMonitor::check_sub(Subscription *s)
 515 {
 516   dout(10) << __func__ << " client wants " << s->type << " ver " << s->next << dendl;
 517
 518   int sub_level = sub_name_to_id(s->type);
 519   assert(sub_level >= 0);
 520
 521   version_t summary_version = summary.version;
 522   if (s->next > summary_version) {
 523     dout(10) << __func__ << " client " << s->session->inst
 524             << " requested version (" << s->next << ") is greater than ours ("
 525             << summary_version << "), which means we already sent him"
 526             << " everything we have." << dendl;
 527     return;
 528   }
 529
 530   MLog *mlog = new MLog(mon->monmap->fsid);
 531
 532   if (s->next == 0) {
 533     /* First timer, heh? */
 534     bool ret = _create_sub_summary(mlog, sub_level);
 535     if (!ret) {
 536       dout(1) << __func__ << " ret = " << ret << dendl;
 537       mlog->put();
 538       return;
 539     }
 540   } else {
 541     /* let us send you an incremental log... */
 542     _create_sub_incremental(mlog, sub_level, s->next);
 543   }
 544
 545   dout(1) << __func__ << " sending message to " << s->session->inst
 546           << " with " << mlog->entries.size() << " entries"
 547           << " (version " << mlog->version << ")" << dendl;
 548
 549   if (!mlog->entries.empty()) {
 550     s->session->con->send_message(mlog);
 551   } else {
 552     mlog->put();
 553   }
 554   if (s->onetime)
 555     mon->session_map.remove_sub(s);
 556   else
 557     s->next = summary_version+1;
 558 }
 559
 560 /**
 561  * Create a log message containing only the last message in the summary.
 562  *
 563  * @param mlog  Log message we'll send to the client.
 564  * @param level Maximum log level the client is interested in.
 565  * @return      'true' if we consider we successfully populated @mlog;
 566  *              'false' otherwise.
 567  */
 568 bool LogMonitor::_create_sub_summary(MLog *mlog, int level)
 569 {
 570   dout(10) << __func__ << dendl;
 571
 572   assert(mlog != NULL);
 573
 574   if (!summary.tail.size())
 575     return false;
 576
 577   list<LogEntry>::reverse_iterator it = summary.tail.rbegin();
 578   for (; it != summary.tail.rend(); ++it) {
 579     LogEntry e = *it;
 580     if (e.prio < level)
 581       continue;
 582
 583     mlog->entries.push_back(e);
 584     mlog->version = summary.version;
 585     break;
 586   }
 587
 588   return true;
 589 }
 590
 591 /**
 592  * Create an incremental log message from version \p sv to \p summary.version
 593  *
 594  * @param mlog  Log message we'll send to the client with the messages received
 595  *              since version \p sv, inclusive.
 596  * @param level The max log level of the messages the client is interested in.
 597  * @param sv    The version the client is looking for.
 598  */
 599 void LogMonitor::_create_sub_incremental(MLog *mlog, int level, version_t sv)
 600 {
 601   dout(10) << __func__ << " level " << level << " ver " << sv
 602           << " cur summary ver " << summary.version << dendl;
 603
 604   if (sv < get_first_committed()) {
 605     dout(10) << __func__ << " skipped from " << sv
 606              << " to first_committed " << get_first_committed() << dendl;
 607     LogEntry le;
 608     le.stamp = ceph_clock_now();
 609     le.prio = CLOG_WARN;
 610     ostringstream ss;
 611     ss << "skipped log messages from " << sv << " to " << get_first_committed();
 612     le.msg = ss.str();
 613     mlog->entries.push_back(le);
 614     sv = get_first_committed();
 615   }
 616
 617   version_t summary_ver = summary.version;
 618   while (sv <= summary_ver) {
 619     bufferlist bl;
 620     int err = get_version(sv, bl);
 621     assert(err == 0);
 622     assert(bl.length());
 623     bufferlist::iterator p = bl.begin();
 624     __u8 v;
 625     ::decode(v,p);
 626     while (!p.end()) {
 627       LogEntry le;
 628       le.decode(p);
 629
 630       if (le.prio < level) {
 631         dout(20) << __func__ << " requested " << level
 632                  << " entry " << le.prio << dendl;
 633         continue;
 634       }
 635
 636       mlog->entries.push_back(le);
 637     }
 638     mlog->version = sv++;
 639   }
 640
 641   dout(10) << __func__ << " incremental message ready ("
 642            << mlog->entries.size() << " entries)" << dendl;
 643 }
 644
 645 void LogMonitor::update_log_channels()
 646 {
 647   ostringstream oss;
 648
 649   channels.clear();
 650
 651   int r = get_conf_str_map_helper(g_conf->mon_cluster_log_to_syslog,
 652                                   oss, &channels.log_to_syslog,
 653                                   CLOG_CONFIG_DEFAULT_KEY);
 654   if (r < 0) {
 655     derr << __func__ << " error parsing 'mon_cluster_log_to_syslog'" << dendl;
 656     return;
 657   }
 658
 659   r = get_conf_str_map_helper(g_conf->mon_cluster_log_to_syslog_level,
 660                               oss, &channels.syslog_level,
 661                               CLOG_CONFIG_DEFAULT_KEY);
 662   if (r < 0) {
 663     derr << __func__ << " error parsing 'mon_cluster_log_to_syslog_level'"
 664          << dendl;
 665     return;
 666   }
 667
 668   r = get_conf_str_map_helper(g_conf->mon_cluster_log_to_syslog_facility,
 669                               oss, &channels.syslog_facility,
 670                               CLOG_CONFIG_DEFAULT_KEY);
 671   if (r < 0) {
 672     derr << __func__ << " error parsing 'mon_cluster_log_to_syslog_facility'"
 673          << dendl;
 674     return;
 675   }
 676
 677   r = get_conf_str_map_helper(g_conf->mon_cluster_log_file, oss,
 678                               &channels.log_file,
 679                               CLOG_CONFIG_DEFAULT_KEY);
 680   if (r < 0) {
 681     derr << __func__ << " error parsing 'mon_cluster_log_file'" << dendl;
 682     return;
 683   }
 684
 685   r = get_conf_str_map_helper(g_conf->mon_cluster_log_file_level, oss,
 686                               &channels.log_file_level,
 687                               CLOG_CONFIG_DEFAULT_KEY);
 688   if (r < 0) {
 689     derr << __func__ << " error parsing 'mon_cluster_log_file_level'"
 690          << dendl;
 691     return;
 692   }
 693
 694   r = get_conf_str_map_helper(g_conf->mon_cluster_log_to_graylog, oss,
 695                               &channels.log_to_graylog,
 696                               CLOG_CONFIG_DEFAULT_KEY);
 697   if (r < 0) {
 698     derr << __func__ << " error parsing 'mon_cluster_log_to_graylog'"
 699          << dendl;
 700     return;
 701   }
 702
 703   r = get_conf_str_map_helper(g_conf->mon_cluster_log_to_graylog_host, oss,
 704                               &channels.log_to_graylog_host,
 705                               CLOG_CONFIG_DEFAULT_KEY);
 706   if (r < 0) {
 707     derr << __func__ << " error parsing 'mon_cluster_log_to_graylog_host'"
 708          << dendl;
 709     return;
 710   }
 711
 712   r = get_conf_str_map_helper(g_conf->mon_cluster_log_to_graylog_port, oss,
 713                               &channels.log_to_graylog_port,
 714                               CLOG_CONFIG_DEFAULT_KEY);
 715   if (r < 0) {
 716     derr << __func__ << " error parsing 'mon_cluster_log_to_graylog_port'"
 717          << dendl;
 718     return;
 719   }
 720
 721   channels.expand_channel_meta();
 722 }
 723
 724 void LogMonitor::log_channel_info::expand_channel_meta(map<string,string> &m)
 725 {
 726   generic_dout(20) << __func__ << " expand map: " << m << dendl;
 727   for (map<string,string>::iterator p = m.begin(); p != m.end(); ++p) {
 728     m[p->first] = expand_channel_meta(p->second, p->first);
 729   }
 730   generic_dout(20) << __func__ << " expanded map: " << m << dendl;
 731 }
 732
 733 string LogMonitor::log_channel_info::expand_channel_meta(
 734     const string &input,
 735     const string &change_to)
 736 {
 737   size_t pos = string::npos;
 738   string s(input);
 739   while ((pos = s.find(LOG_META_CHANNEL)) != string::npos) {
 740     string tmp = s.substr(0, pos) + change_to;
 741     if (pos+LOG_META_CHANNEL.length() < s.length())
 742       tmp += s.substr(pos+LOG_META_CHANNEL.length());
 743     s = tmp;
 744   }
 745   generic_dout(20) << __func__ << " from '" << input
 746                    << "' to '" << s << "'" << dendl;
 747
 748   return s;
 749 }
 750
 751 bool LogMonitor::log_channel_info::do_log_to_syslog(const string &channel) {
 752   string v = get_str_map_key(log_to_syslog, channel,
 753                              &CLOG_CONFIG_DEFAULT_KEY);
 754   // We expect booleans, but they are in k/v pairs, kept
 755   // as strings, in 'log_to_syslog'. We must ensure
 756   // compatibility with existing boolean handling, and so
 757   // we are here using a modified version of how
 758   // md_config_t::set_val_raw() handles booleans. We will
 759   // accept both 'true' and 'false', but will also check for
 760   // '1' and '0'. The main distiction between this and the
 761   // original code is that we will assume everything not '1',
 762   // '0', 'true' or 'false' to be 'false'.
 763   bool ret = false;
 764
 765   if (boost::iequals(v, "false")) {
 766     ret = false;
 767   } else if (boost::iequals(v, "true")) {
 768     ret = true;
 769   } else {
 770     std::string err;
 771     int b = strict_strtol(v.c_str(), 10, &err);
 772     ret = (err.empty() && b == 1);
 773   }
 774
 775   return ret;
 776 }
 777
 778 ceph::logging::Graylog::Ref LogMonitor::log_channel_info::get_graylog(
 779     const string &channel)
 780 {
 781   generic_dout(25) << __func__ << " for channel '"
 782                    << channel << "'" << dendl;
 783
 784   if (graylogs.count(channel) == 0) {
 785     auto graylog(std::make_shared<ceph::logging::Graylog>("mon"));
 786
 787     graylog->set_fsid(g_conf->fsid);
 788     graylog->set_hostname(g_conf->host);
 789     graylog->set_destination(get_str_map_key(log_to_graylog_host, channel,
 790                                              &CLOG_CONFIG_DEFAULT_KEY),
 791                              atoi(get_str_map_key(log_to_graylog_port, channel,
 792                                                   &CLOG_CONFIG_DEFAULT_KEY).c_str()));
 793
 794     graylogs[channel] = graylog;
 795     generic_dout(20) << __func__ << " for channel '"
 796                      << channel << "' to graylog host '"
 797                      << log_to_graylog_host[channel] << ":"
 798                      << log_to_graylog_port[channel]
 799                      << "'" << dendl;
 800   }
 801   return graylogs[channel];
 802 }
 803
 804 void LogMonitor::handle_conf_change(const struct md_config_t *conf,
 805                                     const std::set<std::string> &changed)
 806 {
 807   if (changed.count("mon_cluster_log_to_syslog") ||
 808       changed.count("mon_cluster_log_to_syslog_level") ||
 809       changed.count("mon_cluster_log_to_syslog_facility") ||
 810       changed.count("mon_cluster_log_file") ||
 811       changed.count("mon_cluster_log_file_level") ||
 812       changed.count("mon_cluster_log_to_graylog") ||
 813       changed.count("mon_cluster_log_to_graylog_host") ||
 814       changed.count("mon_cluster_log_to_graylog_port")) {
 815     update_log_channels();
 816   }
 817 }