1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
4 * Ceph - scalable distributed file system
6 * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
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.
15 #include "common/LogClient.h"
16 #include "include/str_map.h"
17 #include "messages/MLog.h"
18 #include "messages/MLogAck.h"
19 #include "mon/MonMap.h"
20 #include "common/Graylog.h"
23 #include <sys/param.h>
24 #include <sys/mount.h>
27 #define dout_subsys ceph_subsys_monc
29 int parse_log_client_options(CephContext *cct,
30 map<string,string> &log_to_monitors,
31 map<string,string> &log_to_syslog,
32 map<string,string> &log_channels,
33 map<string,string> &log_prios,
34 map<string,string> &log_to_graylog,
35 map<string,string> &log_to_graylog_host,
36 map<string,string> &log_to_graylog_port,
42 int r = get_conf_str_map_helper(cct->_conf->clog_to_monitors, oss,
43 &log_to_monitors, CLOG_CONFIG_DEFAULT_KEY);
45 lderr(cct) << __func__ << " error parsing 'clog_to_monitors'" << dendl;
49 r = get_conf_str_map_helper(cct->_conf->clog_to_syslog, oss,
50 &log_to_syslog, CLOG_CONFIG_DEFAULT_KEY);
52 lderr(cct) << __func__ << " error parsing 'clog_to_syslog'" << dendl;
56 r = get_conf_str_map_helper(cct->_conf->clog_to_syslog_facility, oss,
57 &log_channels, CLOG_CONFIG_DEFAULT_KEY);
59 lderr(cct) << __func__ << " error parsing 'clog_to_syslog_facility'" << dendl;
63 r = get_conf_str_map_helper(cct->_conf->clog_to_syslog_level, oss,
64 &log_prios, CLOG_CONFIG_DEFAULT_KEY);
66 lderr(cct) << __func__ << " error parsing 'clog_to_syslog_level'" << dendl;
70 r = get_conf_str_map_helper(cct->_conf->clog_to_graylog, oss,
71 &log_to_graylog, CLOG_CONFIG_DEFAULT_KEY);
73 lderr(cct) << __func__ << " error parsing 'clog_to_graylog'" << dendl;
77 r = get_conf_str_map_helper(cct->_conf->clog_to_graylog_host, oss,
78 &log_to_graylog_host, CLOG_CONFIG_DEFAULT_KEY);
80 lderr(cct) << __func__ << " error parsing 'clog_to_graylog_host'" << dendl;
84 r = get_conf_str_map_helper(cct->_conf->clog_to_graylog_port, oss,
85 &log_to_graylog_port, CLOG_CONFIG_DEFAULT_KEY);
87 lderr(cct) << __func__ << " error parsing 'clog_to_graylog_port'" << dendl;
91 fsid = cct->_conf->get_val<uuid_d>("fsid");
92 host = cct->_conf->host;
97 #define dout_prefix _prefix(_dout, this)
98 static ostream& _prefix(std::ostream *_dout, LogClient *logc) {
99 return *_dout << "log_client ";
102 static ostream& _prefix(std::ostream *_dout, LogChannel *lc) {
103 return *_dout << "log_channel(" << lc->get_log_channel() << ") ";
106 LogChannel::LogChannel(CephContext *cct, LogClient *lc, const string &channel)
107 : cct(cct), parent(lc), channel_lock("LogChannel::channel_lock"),
108 log_channel(channel), log_to_syslog(false), log_to_monitors(false)
112 LogChannel::LogChannel(CephContext *cct, LogClient *lc,
113 const string &channel, const string &facility,
115 : cct(cct), parent(lc), channel_lock("LogChannel::channel_lock"),
116 log_channel(channel), log_prio(prio), syslog_facility(facility),
117 log_to_syslog(false), log_to_monitors(false)
121 LogClient::LogClient(CephContext *cct, Messenger *m, MonMap *mm,
122 enum logclient_flag_t flags)
123 : cct(cct), messenger(m), monmap(mm), is_mon(flags & FLAG_MON),
124 log_lock("LogClient::log_lock"), last_log_sent(0), last_log(0)
128 LogClientTemp::LogClientTemp(clog_type type_, LogChannel &parent_)
129 : type(type_), parent(parent_)
133 LogClientTemp::LogClientTemp(const LogClientTemp &rhs)
134 : type(rhs.type), parent(rhs.parent)
136 // don't want to-- nor can we-- copy the ostringstream
139 LogClientTemp::~LogClientTemp()
141 if (ss.peek() != EOF)
142 parent.do_log(type, ss);
145 void LogChannel::update_config(map<string,string> &log_to_monitors,
146 map<string,string> &log_to_syslog,
147 map<string,string> &log_channels,
148 map<string,string> &log_prios,
149 map<string,string> &log_to_graylog,
150 map<string,string> &log_to_graylog_host,
151 map<string,string> &log_to_graylog_port,
155 ldout(cct, 20) << __func__ << " log_to_monitors " << log_to_monitors
156 << " log_to_syslog " << log_to_syslog
157 << " log_channels " << log_channels
158 << " log_prios " << log_prios
160 bool to_monitors = (get_str_map_key(log_to_monitors, log_channel,
161 &CLOG_CONFIG_DEFAULT_KEY) == "true");
162 bool to_syslog = (get_str_map_key(log_to_syslog, log_channel,
163 &CLOG_CONFIG_DEFAULT_KEY) == "true");
164 string syslog_facility = get_str_map_key(log_channels, log_channel,
165 &CLOG_CONFIG_DEFAULT_KEY);
166 string prio = get_str_map_key(log_prios, log_channel,
167 &CLOG_CONFIG_DEFAULT_KEY);
168 bool to_graylog = (get_str_map_key(log_to_graylog, log_channel,
169 &CLOG_CONFIG_DEFAULT_KEY) == "true");
170 string graylog_host = get_str_map_key(log_to_graylog_host, log_channel,
171 &CLOG_CONFIG_DEFAULT_KEY);
172 string graylog_port_str = get_str_map_key(log_to_graylog_port, log_channel,
173 &CLOG_CONFIG_DEFAULT_KEY);
174 int graylog_port = atoi(graylog_port_str.c_str());
176 set_log_to_monitors(to_monitors);
177 set_log_to_syslog(to_syslog);
178 set_syslog_facility(syslog_facility);
181 if (to_graylog && !graylog) { /* should but isn't */
182 graylog = std::make_shared<ceph::logging::Graylog>("clog");
183 } else if (!to_graylog && graylog) { /* shouldn't but is */
187 if (to_graylog && graylog) {
188 graylog->set_fsid(fsid);
189 graylog->set_hostname(host);
192 if (graylog && (!graylog_host.empty()) && (graylog_port != 0)) {
193 graylog->set_destination(graylog_host, graylog_port);
196 ldout(cct, 10) << __func__
197 << " to_monitors: " << (to_monitors ? "true" : "false")
198 << " to_syslog: " << (to_syslog ? "true" : "false")
199 << " syslog_facility: " << syslog_facility
201 << " to_graylog: " << (to_graylog ? "true" : "false")
202 << " graylog_host: " << graylog_host
203 << " graylog_port: " << graylog_port
207 void LogChannel::do_log(clog_type prio, std::stringstream& ss)
217 void LogChannel::do_log(clog_type prio, const std::string& s)
219 Mutex::Locker l(channel_lock);
220 int lvl = (prio == CLOG_ERROR ? -1 : 0);
221 ldout(cct,lvl) << "log " << prio << " : " << s << dendl;
223 e.stamp = ceph_clock_now();
224 // seq and who should be set for syslog/graylog/log_to_mon
225 e.who = parent->get_myinst();
226 e.name = parent->get_myname();
229 e.channel = get_log_channel();
232 if (log_to_monitors) {
233 e.seq = parent->queue(e);
235 e.seq = parent->get_next_seq();
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());
245 if (do_log_to_graylog()) {
246 ldout(cct,0) << __func__ << " log to graylog" << dendl;
247 graylog->log_log_entry(&e);
251 Message *LogClient::get_mon_log_message(bool flush)
253 Mutex::Locker l(log_lock);
255 if (log_queue.empty())
258 last_log_sent = log_queue.front().seq;
260 return _get_mon_log_message();
263 bool LogClient::are_pending()
265 Mutex::Locker l(log_lock);
266 return last_log > last_log_sent;
269 Message *LogClient::_get_mon_log_message()
271 assert(log_lock.is_locked());
272 if (log_queue.empty())
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.
279 if (last_log_sent == last_log)
282 // limit entries per message
283 unsigned num_unsent = last_log - last_log_sent;
285 if (cct->_conf->mon_client_max_log_entries_per_message > 0)
286 num_send = MIN(num_unsent, (unsigned)cct->_conf->mon_client_max_log_entries_per_message);
288 num_send = num_unsent;
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 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) {
299 assert(p != log_queue.end());
302 assert(p != log_queue.end());
304 last_log_sent = p->seq;
305 ldout(cct,10) << " will send " << *p << dendl;
309 MLog *log = new MLog(monmap->get_fsid());
310 log->entries.swap(o);
315 void LogClient::_send_to_mon()
317 assert(log_lock.is_locked());
319 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);
325 version_t LogClient::queue(LogEntry &entry)
327 Mutex::Locker l(log_lock);
328 entry.seq = ++last_log;
329 log_queue.push_back(entry);
338 uint64_t LogClient::get_next_seq()
340 Mutex::Locker l(log_lock);
344 const entity_inst_t& LogClient::get_myinst()
346 return messenger->get_myinst();
349 const EntityName& LogClient::get_myname()
351 return cct->_conf->name;
354 bool LogClient::handle_log_ack(MLogAck *m)
356 Mutex::Locker l(log_lock);
357 ldout(cct,10) << "handle_log_ack " << *m << dendl;
359 version_t last = m->last;
361 deque<LogEntry>::iterator q = log_queue.begin();
362 while (q != log_queue.end()) {
363 const LogEntry &entry(*q);
364 if (entry.seq > last)
366 ldout(cct,10) << " logged " << entry << dendl;
367 q = log_queue.erase(q);