pdns/remote_logger.cc

   1 #include <unistd.h>
   2 #include "threadname.hh"
   3 #include "remote_logger.hh"
   4 #ifdef HAVE_CONFIG_H
   5 #include "config.h"
   6 #endif
   7 #ifdef PDNS_CONFIG_ARGS
   8 #include "logger.hh"
   9 #define WE_ARE_RECURSOR
  10 #else
  11 #include "dolog.hh"
  12 #endif
  13
  14 bool RemoteLogger::reconnect()
  15 {
  16   if (d_socket >= 0) {
  17     close(d_socket);
  18     d_socket = -1;
  19   }
  20   d_connected = false;
  21   try {
  22     d_socket = SSocket(d_remote.sin4.sin_family, SOCK_STREAM, 0);
  23     setNonBlocking(d_socket);
  24     SConnectWithTimeout(d_socket, d_remote, d_timeout);
  25   }
  26   catch(const std::exception& e) {
  27 #ifdef WE_ARE_RECURSOR
  28     g_log<<Logger::Warning<<"Error connecting to remote logger "<<d_remote.toStringWithPort()<<": "<<e.what()<<std::endl;
  29 #else
  30     warnlog("Error connecting to remote logger %s: %s", d_remote.toStringWithPort(), e.what());
  31 #endif
  32     return false;
  33   }
  34   d_connected = true;
  35   return true;
  36 }
  37
  38 void RemoteLogger::busyReconnectLoop()
  39 {
  40   while (!reconnect()) {
  41     sleep(d_reconnectWaitTime);
  42   }
  43 }
  44
  45 void RemoteLogger::worker()
  46 {
  47 #ifdef WE_ARE_RECURSOR
  48   string threadName = "pdns-r/remLog";
  49 #else
  50   string threadName = "dnsdist/remLog";
  51 #endif
  52   setThreadName(threadName);
  53   while(true) {
  54     std::string data;
  55     {
  56       std::unique_lock<std::mutex> lock(d_writeMutex);
  57       d_queueCond.wait(lock, [this]{return (!d_writeQueue.empty()) || d_exiting;});
  58       if (d_exiting) {
  59         return;
  60       }
  61       data = d_writeQueue.front();
  62       d_writeQueue.pop();
  63     }
  64
  65     if (!d_connected) {
  66       busyReconnectLoop();
  67     }
  68
  69     try {
  70       uint16_t len = static_cast<uint16_t>(data.length());
  71       sendSizeAndMsgWithTimeout(d_socket, len, data.c_str(), static_cast<int>(d_timeout), nullptr, nullptr, 0, 0, 0);
  72     }
  73     catch(const std::runtime_error& e) {
  74 #ifdef WE_ARE_RECURSOR
  75       g_log<<Logger::Info<<"Error sending data to remote logger "<<d_remote.toStringWithPort()<<": "<< e.what()<<endl;
  76 #else
  77       vinfolog("Error sending data to remote logger (%s): %s", d_remote.toStringWithPort(), e.what());
  78 #endif
  79       busyReconnectLoop();
  80     }
  81   }
  82 }
  83
  84 void RemoteLogger::queueData(const std::string& data)
  85 {
  86   {
  87     std::lock_guard<std::mutex> lock(d_writeMutex);
  88     if (d_writeQueue.size() >= d_maxQueuedEntries) {
  89       d_writeQueue.pop();
  90     }
  91     d_writeQueue.push(data);
  92   }
  93   d_queueCond.notify_one();
  94 }
  95
  96 RemoteLogger::RemoteLogger(const ComboAddress& remote, uint16_t timeout, uint64_t maxQueuedEntries, uint8_t reconnectWaitTime, bool asyncConnect): d_remote(remote), d_maxQueuedEntries(maxQueuedEntries), d_timeout(timeout), d_reconnectWaitTime(reconnectWaitTime), d_asyncConnect(asyncConnect), d_thread(&RemoteLogger::worker, this)
  97 {
  98   if (!d_asyncConnect) {
  99     reconnect();
 100   }
 101 }
 102
 103 RemoteLogger::~RemoteLogger()
 104 {
 105   d_exiting = true;
 106   if (d_socket >= 0) {
 107     close(d_socket);
 108     d_socket = -1;
 109     d_connected = false;
 110   }
 111   d_queueCond.notify_one();
 112   d_thread.join();
 113 }