Allow for fstrmlib queue options to be set; switch for logging queries and/or answers...

diff --git a/pdns/fstrm_logger.cc b/pdns/fstrm_logger.cc
index 617aa73d5cfd76df42c687f3c7d52b2500c39e0a..7df889b049209d1eb48d1332bf2a27e13aec62d5 100644 (file)
--- a/pdns/fstrm_logger.cc
+++ b/pdns/fstrm_logger.cc
@@ -9,7 +9,8 @@
 
 #ifdef HAVE_FSTRM
 
-FrameStreamLogger::FrameStreamLogger(const int family, const std::string& address, bool connect): d_family(family), d_address(address)
+FrameStreamLogger::FrameStreamLogger(const int family, const std::string& address, bool connect,
+    const std::unordered_map<string,unsigned>& options): d_family(family), d_address(address)
 {
   fstrm_res res;
 
@@ -78,6 +79,44 @@ FrameStreamLogger::FrameStreamLogger(const int family, const std::string& addres
       throw std::runtime_error("FrameStreamLogger: fstrm_iothr_options_set_queue_model failed: " + std::to_string(res));
     }
 
+    if (options.find("bufferHint") != options.end() && options.at("bufferHint")) {
+      res = fstrm_iothr_options_set_buffer_hint(d_iothropt, options.at("bufferHint"));
+      if (res != fstrm_res_success) {
+        throw std::runtime_error("FrameStreamLogger: fstrm_iothr_options_set_buffer_hint failed: " + std::to_string(res));
+      }
+    }
+    if (options.find("flushTimeout") != options.end() && options.at("flushTimeout")) {
+      res = fstrm_iothr_options_set_flush_timeout(d_iothropt, options.at("flushTimeout"));
+      if (res != fstrm_res_success) {
+        throw std::runtime_error("FrameStreamLogger: fstrm_iothr_options_set_flush_timeout failed: " + std::to_string(res));
+      }
+    }
+    if (options.find("inputQueueSize") != options.end() && options.at("inputQueueSize")) {
+      res = fstrm_iothr_options_set_input_queue_size(d_iothropt, options.at("inputQueueSize"));
+      if (res != fstrm_res_success) {
+        throw std::runtime_error("FrameStreamLogger: fstrm_iothr_options_set_input_queue_size failed: " + std::to_string(res));
+      }
+    }
+    if (options.find("outputQueueSize") != options.end() && options.at("outputQueueSize")) {
+      res = fstrm_iothr_options_set_output_queue_size(d_iothropt, options.at("outputQueueSize"));
+      if (res != fstrm_res_success) {
+        throw std::runtime_error("FrameStreamLogger: fstrm_iothr_options_set_output_queue_size failed: " + std::to_string(res));
+      }
+    }
+    if (options.find("queueNotifyThreshold") != options.end() && options.at("queueNotifyThreshold")) {
+      res = fstrm_iothr_options_set_queue_notify_threshold(d_iothropt, options.at("queueNotifyThreshold"));
+      if (res != fstrm_res_success) {
+        throw std::runtime_error("FrameStreamLogger: fstrm_iothr_options_set_queue_notify_threshold failed: " + std::to_string(res));
+      }
+    }
+    if (options.find("setReopenInterval") != options.end() && options.at("setReopenInterval")) {
+      res = fstrm_iothr_options_set_reopen_interval(d_iothropt, options.at("setReopenInterval"));
+      if (res != fstrm_res_success) {
+        throw std::runtime_error("FrameStreamLogger: fstrm_iothr_options_set_reopen_interval failed: " + std::to_string(res));
+      }
+    }
+
+
     if (connect) {
       d_iothr = fstrm_iothr_init(d_iothropt, &d_writer);
       if (!d_iothr) {

diff --git a/pdns/fstrm_logger.hh b/pdns/fstrm_logger.hh
index 240b3d6adde12ce807df71f0723f2bf1e818da25..ceb1e800905b149403bfd8b74a159d8890a4e999 100644 (file)
--- a/pdns/fstrm_logger.hh
+++ b/pdns/fstrm_logger.hh
@@ -25,6 +25,7 @@
 
 #ifdef HAVE_FSTRM
 
+#include <unordered_map>
 #include <fstrm.h>
 #include <fstrm/iothr.h>
 #include <fstrm/unix_writer.h>
@@ -35,14 +36,20 @@
 class FrameStreamLogger : public RemoteLoggerInterface, boost::noncopyable
 {
 public:
-  FrameStreamLogger(int family, const std::string& address, bool connect);
+  FrameStreamLogger(int family, const std::string& address, bool connect, const std::unordered_map<string,unsigned>& options = std::unordered_map<string,unsigned>());
   virtual ~FrameStreamLogger();
   virtual void queueData(const std::string& data) override;
   virtual std::string toString() const override
   {
     return "FrameStreamLogger to " + d_address;
   }
+  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:
+
   const int d_family;
   const std::string d_address;
   struct fstrm_iothr_queue *d_ioqueue{nullptr};
@@ -56,6 +63,9 @@ private:
   struct fstrm_iothr *d_iothr{nullptr};
 
   void cleanup();
+
+  bool d_logQueries{true};
+  bool d_logResponses{true};
 };
 
 #endif /* HAVE_FSTRM */

diff --git a/pdns/lwres.cc b/pdns/lwres.cc
index 6dd3a20c8ad874f0f0710f2c35f0d9a108a1b250..5c5b80256554be4c4fdc6ee587589d3fdeb57f14 100644 (file)
--- a/pdns/lwres.cc
+++ b/pdns/lwres.cc
@@ -57,6 +57,21 @@
 #include "fstrm_logger.hh"
 bool g_syslog;
 
+static bool isEnabledForQueries(const std::shared_ptr<std::vector<std::unique_ptr<RemoteLoggerInterface>>>& fstreamLoggers)
+{
+  if (fstreamLoggers == nullptr) {
+    return false;
+  }
+  for (auto& logger : *fstreamLoggers) {
+    // We know this is safe since fstreamLoggers is filled with FrameStreamLogger objects only
+    auto fsl = static_cast<const FrameStreamLogger*>(logger.get());
+    if (fsl->logQueries()) {
+      return true;
+    }
+  }
+  return false;
+}
+
 static void logFstreamQuery(const std::shared_ptr<std::vector<std::unique_ptr<RemoteLoggerInterface>>>& fstreamLoggers, const struct timeval &queryTime, const ComboAddress& ip, bool doTCP, const vector<uint8_t>& packet)
 {
   if (fstreamLoggers == nullptr)
@@ -73,6 +88,21 @@ static void logFstreamQuery(const std::shared_ptr<std::vector<std::unique_ptr<Re
   }
 }
 
+static bool isEnabledForResponses(const std::shared_ptr<std::vector<std::unique_ptr<RemoteLoggerInterface>>>& fstreamLoggers)
+{
+  if (fstreamLoggers == nullptr) {
+    return false;
+  }
+  for (auto& logger : *fstreamLoggers) {
+    // We know this is safe since fstreamLoggers is filled with FrameStreamLogger objects only
+    auto fsl = static_cast<const FrameStreamLogger*>(logger.get());
+    if (fsl->logResponses()) {
+      return true;
+    }
+  }
+  return false;
+}
+
 static void logFstreamResponse(const std::shared_ptr<std::vector<std::unique_ptr<RemoteLoggerInterface>>>& fstreamLoggers, const ComboAddress& ip, bool doTCP, const std::string& packet, const struct timeval& queryTime, const struct timeval& replyTime)
 {
   if (fstreamLoggers == nullptr)
@@ -207,7 +237,7 @@ int asyncresolve(const ComboAddress& ip, const DNSName& domain, int type, bool d
   }
 #endif /* HAVE_PROTOBUF */
 #ifdef HAVE_FSTRM
-  if (fstrmLoggers) {
+  if (isEnabledForQueries(fstrmLoggers)) {
     logFstreamQuery(fstrmLoggers, queryTime, ip, doTCP, vpacket);
   }
 #endif /* HAVE_FSTRM */
@@ -286,7 +316,7 @@ int asyncresolve(const ComboAddress& ip, const DNSName& domain, int type, bool d
   buf.resize(len);
 
 #ifdef HAVE_FSTRM
-  if (fstrmLoggers) {
+  if (isEnabledForResponses(fstrmLoggers)) {
     logFstreamResponse(fstrmLoggers, ip, doTCP, buf, queryTime, *now);
   }
 #endif /* HAVE_FSTRM */

diff --git a/pdns/pdns_recursor.cc b/pdns/pdns_recursor.cc
index 3b54c7f8def766799312754d47988fb7ae594d2c..a2d10196ed20064269df9c2f0c0e1cc95a9c951b 100644 (file)
--- a/pdns/pdns_recursor.cc
+++ b/pdns/pdns_recursor.cc
@@ -968,7 +968,17 @@ static std::shared_ptr<std::vector<std::unique_ptr<RemoteLoggerInterface>>> star
 
   for (const auto& server : config.servers) {
     try {
-      result->emplace_back(new FrameStreamLogger(server.sin4.sin_family, server.toStringWithPort(), true));
+      std::unordered_map<string,unsigned> options;
+      options["bufferHint"] = config.bufferHint;
+      options["flushTimeout"] = config.flushTimeout;
+      options["inputQueueSize"] = config.inputQueueSize;
+      options["outputQueueSize"] = config.outputQueueSize;
+      options["queueNotifyThreshold"] = config.queueNotifyThreshold;
+      options["reopenInterval"] = config.reopenInterval;
+      auto fsl = new FrameStreamLogger(server.sin4.sin_family, server.toStringWithPort(), true, options);
+      fsl->setLogQueries(config.logQueries);
+      fsl->setLogResponses(config.logResponses);
+      result->emplace_back(fsl);
     }
     catch(const std::exception& e) {
       g_log<<Logger::Error<<"Error while starting dnstap framestream logger to '"<<server<<": "<<e.what()<<endl;

diff --git a/pdns/rec-lua-conf.cc b/pdns/rec-lua-conf.cc
index 106e6140cb911696ba498379e9c5c1164d8785a4..bd077250cbe8c4d3d5ef77e649f4f8e25b08604a 100644 (file)
--- a/pdns/rec-lua-conf.cc
+++ b/pdns/rec-lua-conf.cc
@@ -153,6 +153,32 @@ static void parseFrameStreamOptions(boost::optional<frameStreamOptions_t> vars,
   if (!vars) {
     return;
   }
+
+  if (vars->count("logQueries")) {
+    config.logQueries = boost::get<bool>((*vars)["logQueries"]);
+  }
+  if (vars->count("logResponses")) {
+    config.logResponses = boost::get<bool>((*vars)["logResponses"]);
+  }
+
+  if (vars->count("bufferHint")) {
+    config.bufferHint = boost::get<uint64_t>((*vars)["bufferHint"]);
+  }
+  if (vars->count("flushTimeout")) {
+    config.flushTimeout = boost::get<uint64_t>((*vars)["flushTimeout"]);
+  }
+  if (vars->count("inputQueueSize")) {
+    config.inputQueueSize = boost::get<uint64_t>((*vars)["inputQueueSize"]);
+  }
+  if (vars->count("outputQueueSize")) {
+    config.outputQueueSize = boost::get<uint64_t>((*vars)["outputQueueSize"]);
+  }
+  if (vars->count("queueNotifyThreshold")) {
+    config.queueNotifyThreshold = boost::get<uint64_t>((*vars)["queueNotifyThreshold"]);
+  }
+  if (vars->count("reopenInterval")) {
+    config.reopenInterval = boost::get<uint64_t>((*vars)["reopenInterval"]);
+  }
 }
 #endif /* HAVE_FSTRM */
 

diff --git a/pdns/rec-lua-conf.hh b/pdns/rec-lua-conf.hh
index 52ead91dab3e984465828cf92a15ec5f33f987d1..8060864c9a12307df650aa472c5a5943f65bea88 100644 (file)
--- a/pdns/rec-lua-conf.hh
+++ b/pdns/rec-lua-conf.hh
@@ -45,6 +45,14 @@ struct FrameStreamExportConfig
 {
   std::vector<ComboAddress> servers;
   bool enabled{false};
+  bool logQueries{true};
+  bool logResponses{true};
+  unsigned bufferHint{0};
+  unsigned flushTimeout{0};
+  unsigned inputQueueSize{0};
+  unsigned outputQueueSize{0};
+  unsigned queueNotifyThreshold{0};
+  unsigned reopenInterval{0};
 };
 
 struct TrustAnchorFileInfo {