remote logger: Reconnect if writes have been stalled for too long

Signed-off-by: Remi Gacogne <remi.gacogne@powerdns.com>

diff --git a/pdns/remote_logger.cc b/pdns/remote_logger.cc
index 61248f6303fa1748afc1507a81407ed6e4fffa90..d9ef7fc8ce2975d18775ac7290eb979189e64be8 100644 (file)
--- a/pdns/remote_logger.cc
+++ b/pdns/remote_logger.cc
@@ -44,6 +44,16 @@ bool CircularWriteBuffer::tooBig(const std::string& str) const
   return str.size() > (d_framesize == 2 ? std::numeric_limits<uint16_t>::max() : std::numeric_limits<uint32_t>::max());
 }
 
+bool CircularWriteBuffer::isEmpty() const
+{
+  return d_buffer.empty();
+}
+
+void CircularWriteBuffer::clear()
+{
+  d_buffer.clear();
+}
+
 bool CircularWriteBuffer::write(const std::string& str)
 {
   if (tooBig(str) || !hasRoomFor(str)) {
@@ -192,8 +202,18 @@ RemoteLoggerInterface::Result RemoteLogger::queueData(const std::string& data)
       if (!runtime->d_writer.flush(runtime->d_socket->getHandle())) {
         /* but failed, let's just drop */
         ++runtime->d_stats.d_pipeFull;
+        if (connectionStalled()) {
+          /* we have not been able to write for far too long,
+             something is wrong. */
+          runtime->d_socket.reset();
+          /* we can't be sure we haven't sent a partial message,
+             and we don't want to send the remaining part after reconnecting */
+          runtime->d_writer.clear();
+          ++runtime->d_stats.d_otherError;
+        }
         return Result::PipeFull;
       }
+      d_tryingToWriteSince = 0;
 
       /* see if we freed enough data */
       if (!runtime->d_writer.hasRoomFor(data)) {
@@ -253,7 +273,20 @@ void RemoteLogger::maintenanceThread()
             /* if flush() returns false, it means that we couldn't flush anything yet
                either because there is nothing to flush, or because the outgoing TCP
                buffer is full. That's fine by us */
-            runtime->d_writer.flush(runtime->d_socket->getHandle());
+            auto flushed = runtime->d_writer.flush(runtime->d_socket->getHandle());
+            if (flushed) {
+              d_tryingToWriteSince = 0;
+            }
+            else if (!runtime->d_writer.isEmpty() && connectionStalled()) {
+              /* we have not been able to write for far too long,
+                 something is wrong. */
+              runtime->d_socket.reset();
+              /* we can't be sure we haven't sent a partial message,
+                 and we don't want to send the remaining part after reconnecting */
+              runtime->d_writer.clear();
+              connected = false;
+              ++runtime->d_stats.d_otherError;
+            }
           }
           else {
             connected = false;
@@ -298,3 +331,14 @@ RemoteLogger::~RemoteLogger()
 
   d_thread.join();
 }
+
+bool RemoteLogger::connectionStalled()
+{
+  auto now = time(nullptr);
+  if (d_tryingToWriteSince == 0) {
+    d_tryingToWriteSince = now;
+    return false;
+  }
+
+  return (d_tryingToWriteSince < now && (now - d_tryingToWriteSince) > s_stalledTimeoutSeconds);
+}

diff --git a/pdns/remote_logger.hh b/pdns/remote_logger.hh
index 51d3f8f2d77fce6d0e756df362a6e28e8a23ed66..732a9af481df61255fe87d00023a658478279eba 100644 (file)
--- a/pdns/remote_logger.hh
+++ b/pdns/remote_logger.hh
@@ -51,8 +51,10 @@ public:
 
   [[nodiscard]] bool hasRoomFor(const std::string& str) const;
   [[nodiscard]] bool tooBig(const std::string& str) const;
+  [[nodiscard]] bool isEmpty() const;
   bool write(const std::string& str);
   bool flush(int fileDesc);
+  void clear();
 
 private:
   boost::circular_buffer<char> d_buffer;
@@ -172,8 +174,13 @@ public:
   }
 
 private:
+  // if we have been trying to write for that long without any progress,
+  // the connection is dead
+  static constexpr time_t s_stalledTimeoutSeconds{5};
+
   bool reconnect();
   void maintenanceThread();
+  [[nodiscard]] bool connectionStalled();
 
   struct RuntimeData
   {
@@ -183,6 +190,7 @@ private:
   };
 
   ComboAddress d_remote;
+  time_t d_tryingToWriteSince{};
   uint16_t d_timeout;
   uint8_t d_reconnectWaitTime;
   std::atomic<bool> d_exiting{false};