Merge pull request #8223 from PowerDNS/omoerbeek-patch-1

[thirdparty/pdns.git] / pdns / remote_logger.hh

diff --git a/pdns/remote_logger.hh b/pdns/remote_logger.hh
index 3eb2aa575711f0f23fa4ef645ce706a0171f8378..d5356c2d3e2b232f8f8c9286b9d0c042a67ba4be 100644 (file)
--- a/pdns/remote_logger.hh
+++ b/pdns/remote_logger.hh
@@ -30,6 +30,31 @@
 #include <thread>
 
 #include "iputils.hh"
+#include "circular_buffer.hh"
+
+/* Writes can be submitted and they are atomically accepted. Either the whole write
+   ends up in the buffer or nothing ends up in the buffer.
+   In case nothing ends up in the buffer, an exception is thrown.
+   Similarly, EOF leads to this treatment
+
+   The filedescriptor can be in non-blocking mode.
+
+   This class is not threadsafe.
+*/
+
+class CircularWriteBuffer
+{
+public:
+  explicit CircularWriteBuffer(int fd, size_t size) : d_fd(fd), d_buffer(size)
+  {
+  }
+
+  void write(const std::string& str);
+  void flush();
+private:
+  int d_fd;
+  boost::circular_buffer<char> d_buffer;
+};
 
 class RemoteLoggerInterface
 {
@@ -37,37 +62,54 @@ public:
   virtual ~RemoteLoggerInterface() {};
   virtual void queueData(const std::string& data) = 0;
   virtual std::string toString() const = 0;
+
+  bool logQueries(void) const { return d_logQueries; }
+  bool logResponses(void) const { return d_logResponses; }
+  void setLogQueries(bool flag) { d_logQueries = flag; }
+  void setLogResponses(bool flag) { d_logResponses = flag; }
+
+private:
+  bool d_logQueries{true};
+  bool d_logResponses{true};
 };
 
+/* Thread safe. Will connect asynchronously on request.
+   Runs a reconnection thread that also periodicall flushes.
+   Note that the buffer only runs as long as there is a connection.
+   If there is no connection we don't buffer a thing
+*/
 class RemoteLogger : public RemoteLoggerInterface
 {
 public:
-  RemoteLogger(const ComboAddress& remote, uint16_t timeout=2, uint64_t maxQueuedEntries=100, uint8_t reconnectWaitTime=1, bool asyncConnect=false);
-  virtual ~RemoteLogger();
-  virtual void queueData(const std::string& data) override;
-  virtual std::string toString() const override
+  RemoteLogger(const ComboAddress& remote, uint16_t timeout=2,
+               uint64_t maxQueuedBytes=100000,
+               uint8_t reconnectWaitTime=1,
+               bool asyncConnect=false);
+  ~RemoteLogger();
+  void queueData(const std::string& data) override;
+  std::string toString() const override
   {
-    return "RemoteLogger to " + d_remote.toStringWithPort();
+    return d_remote.toStringWithPort();
   }
   void stop()
   {
     d_exiting = true;
   }
+  std::atomic<uint32_t> d_drops{0};
+
 private:
-  void busyReconnectLoop();
   bool reconnect();
-  void worker();
+  void maintenanceThread();
 
-  std::queue<std::string> d_writeQueue;
-  std::mutex d_writeMutex;
-  std::condition_variable d_queueCond;
   ComboAddress d_remote;
-  uint64_t d_maxQueuedEntries;
+  uint64_t d_maxQueuedBytes;
   int d_socket{-1};
+  std::unique_ptr<CircularWriteBuffer> d_writer;
   uint16_t d_timeout;
   uint8_t d_reconnectWaitTime;
   std::atomic<bool> d_exiting{false};
   bool d_asyncConnect{false};
-  bool d_connected{false};
+
+  std::mutex d_mutex;
   std::thread d_thread;
 };