src/ceph/src/log/Log.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "Log.h"
   5
   6 #include <errno.h>
   7 #include <syslog.h>
   8
   9 #include "common/errno.h"
  10 #include "common/safe_io.h"
  11 #include "common/Clock.h"
  12 #include "common/Graylog.h"
  13 #include "common/valgrind.h"
  14
  15 #include "include/assert.h"
  16 #include "include/compat.h"
  17 #include "include/on_exit.h"
  18
  19 #include "Entry.h"
  20 #include "SubsystemMap.h"
  21
  22 #define DEFAULT_MAX_NEW    100
  23 #define DEFAULT_MAX_RECENT 10000
  24
  25 #define PREALLOC 1000000
  26
  27
  28 namespace ceph {
  29 namespace logging {
  30
  31 static OnExitManager exit_callbacks;
  32
  33 static void log_on_exit(void *p)
  34 {
  35   Log *l = *(Log **)p;
  36   if (l)
  37     l->flush();
  38   delete (Log **)p;// Delete allocated pointer (not Log object, the pointer only!)
  39 }
  40
  41 Log::Log(SubsystemMap *s)
  42   : m_indirect_this(NULL),
  43     m_subs(s),
  44     m_queue_mutex_holder(0),
  45     m_flush_mutex_holder(0),
  46     m_new(), m_recent(),
  47     m_fd(-1),
  48     m_uid(0),
  49     m_gid(0),
  50     m_fd_last_error(0),
  51     m_syslog_log(-2), m_syslog_crash(-2),
  52     m_stderr_log(1), m_stderr_crash(-1),
  53     m_graylog_log(-3), m_graylog_crash(-3),
  54     m_stop(false),
  55     m_max_new(DEFAULT_MAX_NEW),
  56     m_max_recent(DEFAULT_MAX_RECENT),
  57     m_inject_segv(false)
  58 {
  59   int ret;
  60
  61   ret = pthread_mutex_init(&m_flush_mutex, NULL);
  62   assert(ret == 0);
  63
  64   ret = pthread_mutex_init(&m_queue_mutex, NULL);
  65   assert(ret == 0);
  66
  67   ret = pthread_cond_init(&m_cond_loggers, NULL);
  68   assert(ret == 0);
  69
  70   ret = pthread_cond_init(&m_cond_flusher, NULL);
  71   assert(ret == 0);
  72
  73   // kludge for prealloc testing
  74   if (false)
  75     for (int i=0; i < PREALLOC; i++)
  76       m_recent.enqueue(new Entry);
  77 }
  78
  79 Log::~Log()
  80 {
  81   if (m_indirect_this) {
  82     *m_indirect_this = NULL;
  83   }
  84
  85   assert(!is_started());
  86   if (m_fd >= 0)
  87     VOID_TEMP_FAILURE_RETRY(::close(m_fd));
  88
  89   pthread_mutex_destroy(&m_queue_mutex);
  90   pthread_mutex_destroy(&m_flush_mutex);
  91   pthread_cond_destroy(&m_cond_loggers);
  92   pthread_cond_destroy(&m_cond_flusher);
  93 }
  94
  95
  96 ///
  97
  98 void Log::set_flush_on_exit()
  99 {
 100   // Make sure we flush on shutdown.  We do this by deliberately
 101   // leaking an indirect pointer to ourselves (on_exit() can't
 102   // unregister a callback).  This is not racy only becuase we
 103   // assume that exit() won't race with ~Log().
 104   if (m_indirect_this == NULL) {
 105     m_indirect_this = new (Log*)(this);
 106     exit_callbacks.add_callback(log_on_exit, m_indirect_this);
 107   }
 108 }
 109
 110 void Log::set_max_new(int n)
 111 {
 112   m_max_new = n;
 113 }
 114
 115 void Log::set_max_recent(int n)
 116 {
 117   pthread_mutex_lock(&m_flush_mutex);
 118   m_flush_mutex_holder = pthread_self();
 119   m_max_recent = n;
 120   m_flush_mutex_holder = 0;
 121   pthread_mutex_unlock(&m_flush_mutex);
 122 }
 123
 124 void Log::set_log_file(string fn)
 125 {
 126   m_log_file = fn;
 127 }
 128
 129 void Log::reopen_log_file()
 130 {
 131   pthread_mutex_lock(&m_flush_mutex);
 132   m_flush_mutex_holder = pthread_self();
 133   if (m_fd >= 0)
 134     VOID_TEMP_FAILURE_RETRY(::close(m_fd));
 135   if (m_log_file.length()) {
 136     m_fd = ::open(m_log_file.c_str(), O_CREAT|O_WRONLY|O_APPEND, 0644);
 137     if (m_fd >= 0 && (m_uid || m_gid)) {
 138       int r = ::fchown(m_fd, m_uid, m_gid);
 139       if (r < 0) {
 140         r = -errno;
 141         cerr << "failed to chown " << m_log_file << ": " << cpp_strerror(r)
 142              << std::endl;
 143       }
 144     }
 145   } else {
 146     m_fd = -1;
 147   }
 148   m_flush_mutex_holder = 0;
 149   pthread_mutex_unlock(&m_flush_mutex);
 150 }
 151
 152 void Log::chown_log_file(uid_t uid, gid_t gid)
 153 {
 154   pthread_mutex_lock(&m_flush_mutex);
 155   if (m_fd >= 0) {
 156     int r = ::fchown(m_fd, uid, gid);
 157     if (r < 0) {
 158       r = -errno;
 159       cerr << "failed to chown " << m_log_file << ": " << cpp_strerror(r)
 160            << std::endl;
 161     }
 162   }
 163   pthread_mutex_unlock(&m_flush_mutex);
 164 }
 165
 166 void Log::set_syslog_level(int log, int crash)
 167 {
 168   pthread_mutex_lock(&m_flush_mutex);
 169   m_syslog_log = log;
 170   m_syslog_crash = crash;
 171   pthread_mutex_unlock(&m_flush_mutex);
 172 }
 173
 174 void Log::set_stderr_level(int log, int crash)
 175 {
 176   pthread_mutex_lock(&m_flush_mutex);
 177   m_stderr_log = log;
 178   m_stderr_crash = crash;
 179   pthread_mutex_unlock(&m_flush_mutex);
 180 }
 181
 182 void Log::set_graylog_level(int log, int crash)
 183 {
 184   pthread_mutex_lock(&m_flush_mutex);
 185   m_graylog_log = log;
 186   m_graylog_crash = crash;
 187   pthread_mutex_unlock(&m_flush_mutex);
 188 }
 189
 190 void Log::start_graylog()
 191 {
 192   pthread_mutex_lock(&m_flush_mutex);
 193   if (! m_graylog.get())
 194     m_graylog = std::make_shared<Graylog>(m_subs, "dlog");
 195   pthread_mutex_unlock(&m_flush_mutex);
 196 }
 197
 198
 199 void Log::stop_graylog()
 200 {
 201   pthread_mutex_lock(&m_flush_mutex);
 202   m_graylog.reset();
 203   pthread_mutex_unlock(&m_flush_mutex);
 204 }
 205
 206 void Log::submit_entry(Entry *e)
 207 {
 208   pthread_mutex_lock(&m_queue_mutex);
 209   m_queue_mutex_holder = pthread_self();
 210
 211   if (m_inject_segv)
 212     *(volatile int *)(0) = 0xdead;
 213
 214   // wait for flush to catch up
 215   while (m_new.m_len > m_max_new)
 216     pthread_cond_wait(&m_cond_loggers, &m_queue_mutex);
 217
 218   m_new.enqueue(e);
 219   pthread_cond_signal(&m_cond_flusher);
 220   m_queue_mutex_holder = 0;
 221   pthread_mutex_unlock(&m_queue_mutex);
 222 }
 223
 224
 225 Entry *Log::create_entry(int level, int subsys)
 226 {
 227   if (true) {
 228     return new Entry(ceph_clock_now(),
 229                      pthread_self(),
 230                      level, subsys);
 231   } else {
 232     // kludge for perf testing
 233     Entry *e = m_recent.dequeue();
 234     e->m_stamp = ceph_clock_now();
 235     e->m_thread = pthread_self();
 236     e->m_prio = level;
 237     e->m_subsys = subsys;
 238     return e;
 239   }
 240 }
 241
 242 Entry *Log::create_entry(int level, int subsys, size_t* expected_size)
 243 {
 244   if (true) {
 245     ANNOTATE_BENIGN_RACE_SIZED(expected_size, sizeof(*expected_size),
 246                                "Log hint");
 247     size_t size = __atomic_load_n(expected_size, __ATOMIC_RELAXED);
 248     void *ptr = ::operator new(sizeof(Entry) + size);
 249     return new(ptr) Entry(ceph_clock_now(),
 250        pthread_self(), level, subsys,
 251        reinterpret_cast<char*>(ptr) + sizeof(Entry), size, expected_size);
 252   } else {
 253     // kludge for perf testing
 254     Entry *e = m_recent.dequeue();
 255     e->m_stamp = ceph_clock_now();
 256     e->m_thread = pthread_self();
 257     e->m_prio = level;
 258     e->m_subsys = subsys;
 259     return e;
 260   }
 261 }
 262
 263 void Log::flush()
 264 {
 265   pthread_mutex_lock(&m_flush_mutex);
 266   m_flush_mutex_holder = pthread_self();
 267   pthread_mutex_lock(&m_queue_mutex);
 268   m_queue_mutex_holder = pthread_self();
 269   EntryQueue t;
 270   t.swap(m_new);
 271   pthread_cond_broadcast(&m_cond_loggers);
 272   m_queue_mutex_holder = 0;
 273   pthread_mutex_unlock(&m_queue_mutex);
 274   _flush(&t, &m_recent, false);
 275
 276   // trim
 277   while (m_recent.m_len > m_max_recent) {
 278     delete m_recent.dequeue();
 279   }
 280
 281   m_flush_mutex_holder = 0;
 282   pthread_mutex_unlock(&m_flush_mutex);
 283 }
 284
 285 void Log::_flush(EntryQueue *t, EntryQueue *requeue, bool crash)
 286 {
 287   Entry *e;
 288   while ((e = t->dequeue()) != NULL) {
 289     unsigned sub = e->m_subsys;
 290
 291     bool should_log = crash || m_subs->get_log_level(sub) >= e->m_prio;
 292     bool do_fd = m_fd >= 0 && should_log;
 293     bool do_syslog = m_syslog_crash >= e->m_prio && should_log;
 294     bool do_stderr = m_stderr_crash >= e->m_prio && should_log;
 295     bool do_graylog2 = m_graylog_crash >= e->m_prio && should_log;
 296
 297     e->hint_size();
 298     if (do_fd || do_syslog || do_stderr) {
 299       size_t buflen = 0;
 300
 301       char *buf;
 302       size_t buf_size = 80 + e->size();
 303       bool need_dynamic = buf_size >= 0x10000; //avoids >64K buffers
 304                                                //allocation at stack
 305       char buf0[need_dynamic ? 1 : buf_size];
 306       if (need_dynamic) {
 307         buf = new char[buf_size];
 308       } else {
 309         buf = buf0;
 310       }
 311
 312       if (crash)
 313         buflen += snprintf(buf, buf_size, "%6d> ", -t->m_len);
 314       buflen += e->m_stamp.sprintf(buf + buflen, buf_size-buflen);
 315       buflen += snprintf(buf + buflen, buf_size-buflen, " %lx %2d ",
 316                         (unsigned long)e->m_thread, e->m_prio);
 317
 318       buflen += e->snprintf(buf + buflen, buf_size - buflen - 1);
 319       if (buflen > buf_size - 1) { //paranoid check, buf was declared
 320                                    //to hold everything
 321         buflen = buf_size - 1;
 322         buf[buflen] = 0;
 323       }
 324
 325       if (do_syslog) {
 326         syslog(LOG_USER|LOG_INFO, "%s", buf);
 327       }
 328
 329       if (do_stderr) {
 330         cerr << buf << std::endl;
 331       }
 332       if (do_fd) {
 333         buf[buflen] = '\n';
 334         int r = safe_write(m_fd, buf, buflen+1);
 335         if (r != m_fd_last_error) {
 336           if (r < 0)
 337             cerr << "problem writing to " << m_log_file
 338                  << ": " << cpp_strerror(r)
 339                  << std::endl;
 340           m_fd_last_error = r;
 341         }
 342       }
 343       if (need_dynamic)
 344         delete[] buf;
 345     }
 346     if (do_graylog2 && m_graylog) {
 347       m_graylog->log_entry(e);
 348     }
 349
 350     requeue->enqueue(e);
 351   }
 352 }
 353
 354 void Log::_log_message(const char *s, bool crash)
 355 {
 356   if (m_fd >= 0) {
 357     size_t len = strlen(s);
 358     std::string b;
 359     b.reserve(len + 1);
 360     b.append(s, len);
 361     b += '\n';
 362     int r = safe_write(m_fd, b.c_str(), b.size());
 363     if (r < 0)
 364       cerr << "problem writing to " << m_log_file << ": " << cpp_strerror(r) << std::endl;
 365   }
 366   if ((crash ? m_syslog_crash : m_syslog_log) >= 0) {
 367     syslog(LOG_USER|LOG_INFO, "%s", s);
 368   }
 369
 370   if ((crash ? m_stderr_crash : m_stderr_log) >= 0) {
 371     cerr << s << std::endl;
 372   }
 373 }
 374
 375 void Log::dump_recent()
 376 {
 377   pthread_mutex_lock(&m_flush_mutex);
 378   m_flush_mutex_holder = pthread_self();
 379
 380   pthread_mutex_lock(&m_queue_mutex);
 381   m_queue_mutex_holder = pthread_self();
 382
 383   EntryQueue t;
 384   t.swap(m_new);
 385
 386   m_queue_mutex_holder = 0;
 387   pthread_mutex_unlock(&m_queue_mutex);
 388   _flush(&t, &m_recent, false);
 389
 390   EntryQueue old;
 391   _log_message("--- begin dump of recent events ---", true);
 392   _flush(&m_recent, &old, true);
 393
 394   char buf[4096];
 395   _log_message("--- logging levels ---", true);
 396   for (vector<Subsystem>::iterator p = m_subs->m_subsys.begin();
 397        p != m_subs->m_subsys.end();
 398        ++p) {
 399     snprintf(buf, sizeof(buf), "  %2d/%2d %s", p->log_level, p->gather_level, p->name.c_str());
 400     _log_message(buf, true);
 401   }
 402
 403   sprintf(buf, "  %2d/%2d (syslog threshold)", m_syslog_log, m_syslog_crash);
 404   _log_message(buf, true);
 405   sprintf(buf, "  %2d/%2d (stderr threshold)", m_stderr_log, m_stderr_crash);
 406   _log_message(buf, true);
 407   sprintf(buf, "  max_recent %9d", m_max_recent);
 408   _log_message(buf, true);
 409   sprintf(buf, "  max_new    %9d", m_max_new);
 410   _log_message(buf, true);
 411   sprintf(buf, "  log_file %s", m_log_file.c_str());
 412   _log_message(buf, true);
 413
 414   _log_message("--- end dump of recent events ---", true);
 415
 416   m_flush_mutex_holder = 0;
 417   pthread_mutex_unlock(&m_flush_mutex);
 418 }
 419
 420 void Log::start()
 421 {
 422   assert(!is_started());
 423   pthread_mutex_lock(&m_queue_mutex);
 424   m_stop = false;
 425   pthread_mutex_unlock(&m_queue_mutex);
 426   create("log");
 427 }
 428
 429 void Log::stop()
 430 {
 431   assert(is_started());
 432   pthread_mutex_lock(&m_queue_mutex);
 433   m_stop = true;
 434   pthread_cond_signal(&m_cond_flusher);
 435   pthread_cond_broadcast(&m_cond_loggers);
 436   pthread_mutex_unlock(&m_queue_mutex);
 437   join();
 438 }
 439
 440 void *Log::entry()
 441 {
 442   pthread_mutex_lock(&m_queue_mutex);
 443   m_queue_mutex_holder = pthread_self();
 444   while (!m_stop) {
 445     if (!m_new.empty()) {
 446       m_queue_mutex_holder = 0;
 447       pthread_mutex_unlock(&m_queue_mutex);
 448       flush();
 449       pthread_mutex_lock(&m_queue_mutex);
 450       m_queue_mutex_holder = pthread_self();
 451       continue;
 452     }
 453
 454     pthread_cond_wait(&m_cond_flusher, &m_queue_mutex);
 455   }
 456   m_queue_mutex_holder = 0;
 457   pthread_mutex_unlock(&m_queue_mutex);
 458   flush();
 459   return NULL;
 460 }
 461
 462 bool Log::is_inside_log_lock()
 463 {
 464   return
 465     pthread_self() == m_queue_mutex_holder ||
 466     pthread_self() == m_flush_mutex_holder;
 467 }
 468
 469 void Log::inject_segv()
 470 {
 471   m_inject_segv = true;
 472 }
 473
 474 void Log::reset_segv()
 475 {
 476   m_inject_segv = false;
 477 }
 478
 479 } // ceph::logging::
 480 } // ceph::