CXXCBC-346: allow to log network communication (#425)

Author: avsej

core/io/dns_client.cxx

@@ -72,16 +72,25 @@ class dns_srv_command : public std::enable_shared_from_this<dns_srv_command>
 
     void execute(std::chrono::milliseconds total_timeout, std::chrono::milliseconds udp_timeout)
     {
-        CB_LOG_TRACE("Query DNS-SRV (UDP) address=\"{}:{}\", udp_timeout={}, total_timeout={}{:a}",
+        CB_LOG_TRACE("Query DNS-SRV (UDP) address=\"{}:{}\", udp_timeout={}, total_timeout={}",
                      address_.to_string(),
                      port_,
                      udp_timeout,
-                     total_timeout,
-                     spdlog::to_hex(send_buf_));
+                     total_timeout);
         asio::ip::udp::endpoint endpoint(address_, port_);
         udp_.open(endpoint.protocol());
+        CB_LOG_PROTOCOL("[DNS, UDP, OUT] host=\"{}\", port={}, buffer_size={}{:a}",
+                        address_.to_string(),
+                        port_,
+                        send_buf_.size(),
+                        spdlog::to_hex(send_buf_));
         udp_.async_send_to(
-          asio::buffer(send_buf_), endpoint, [self = shared_from_this()](std::error_code ec1, std::size_t /* bytes_transferred */) mutable {
+          asio::buffer(send_buf_), endpoint, [self = shared_from_this()](std::error_code ec1, std::size_t bytes_transferred1) mutable {
+              CB_LOG_PROTOCOL("[DNS, UDP, OUT] host=\"{}\", port={}, rc={}, bytes_sent={}",
+                              self->address_.to_string(),
+                              self->port_,
+                              ec1 ? ec1.message() : "ok",
+                              bytes_transferred1);
               if (ec1) {
                   self->udp_deadline_.cancel();
                   CB_LOG_DEBUG("DNS UDP write operation has got error, retrying with TCP, address=\"{}:{}\", ec={}",
@@ -94,6 +103,13 @@ class dns_srv_command : public std::enable_shared_from_this<dns_srv_command>
               self->recv_buf_.resize(512);
               self->udp_.async_receive_from(
                 asio::buffer(self->recv_buf_), self->udp_sender_, [self](std::error_code ec2, std::size_t bytes_transferred) mutable {
+                    CB_LOG_PROTOCOL("[DNS, UDP, IN] host=\"{}\", port={}, rc={}, bytes_received={}{:a}",
+                                    self->address_.to_string(),
+                                    self->port_,
+                                    ec2 ? ec2.message() : "ok",
+                                    bytes_transferred,
+                                    spdlog::to_hex(self->recv_buf_.data(), self->recv_buf_.data() + bytes_transferred));
+
                     self->udp_deadline_.cancel();
                     if (ec2) {
                         CB_LOG_DEBUG("DNS UDP read operation has got error, retrying with TCP, address=\"{}:{}\", ec={}",
@@ -168,10 +184,18 @@ class dns_srv_command : public std::enable_shared_from_this<dns_srv_command>
             auto send_size = static_cast<std::uint16_t>(self->send_buf_.size());
             self->send_buf_.insert(self->send_buf_.begin(), static_cast<std::uint8_t>(send_size & 0xffU));
             self->send_buf_.insert(self->send_buf_.begin(), static_cast<std::uint8_t>(send_size >> 8U));
-            CB_LOG_TRACE(
-              "Query DNS-SRV (TCP) address=\"{}:{}\"{:a}", self->address_.to_string(), self->port_, spdlog::to_hex(self->send_buf_));
+            CB_LOG_PROTOCOL("[DNS, TCP, OUT] host=\"{}\", port={}, buffer_size={}{:a}",
+                            self->address_.to_string(),
+                            self->port_,
+                            self->send_buf_.size(),
+                            spdlog::to_hex(self->send_buf_));
             asio::async_write(
-              self->tcp_, asio::buffer(self->send_buf_), [self](std::error_code ec2, std::size_t /* bytes_transferred */) mutable {
+              self->tcp_, asio::buffer(self->send_buf_), [self](std::error_code ec2, std::size_t bytes_transferred2) mutable {
+                  CB_LOG_PROTOCOL("[DNS, TCP, OUT] host=\"{}\", port={}, rc={}, bytes_sent={}",
+                                  self->address_.to_string(),
+                                  self->port_,
+                                  ec2 ? ec2.message() : "ok",
+                                  bytes_transferred2);
                   if (ec2) {
                       CB_LOG_DEBUG("DNS TCP write operation has been aborted, address=\"{}:{}\", ec={}",
                                    self->address_.to_string(),
@@ -186,7 +210,14 @@ class dns_srv_command : public std::enable_shared_from_this<dns_srv_command>
                   asio::async_read(
                     self->tcp_,
                     asio::buffer(&self->recv_buf_size_, sizeof(std::uint16_t)),
-                    [self](std::error_code ec3, std::size_t /* bytes_transferred */) mutable {
+                    [self](std::error_code ec3, std::size_t bytes_transferred3) mutable {
+                        CB_LOG_PROTOCOL("[DNS, TCP, IN] host=\"{}\", port={}, rc={}, bytes_received={}{:a}",
+                                        self->address_.to_string(),
+                                        self->port_,
+                                        ec3 ? ec3.message() : "ok",
+                                        bytes_transferred3,
+                                        spdlog::to_hex(reinterpret_cast<std::uint8_t*>(&self->recv_buf_size_),
+                                                       reinterpret_cast<std::uint8_t*>(&self->recv_buf_size_) + bytes_transferred3));
                         if (ec3) {
                             CB_LOG_DEBUG("DNS TCP buf size read operation has been aborted, address=\"{}:{}\", ec={}",
                                          self->address_.to_string(),
@@ -199,16 +230,23 @@ class dns_srv_command : public std::enable_shared_from_this<dns_srv_command>
                         self->recv_buf_.resize(self->recv_buf_size_);
                         CB_LOG_DEBUG("DNS TCP schedule read of {} bytes", self->recv_buf_size_);
                         asio::async_read(
-                          self->tcp_, asio::buffer(self->recv_buf_), [self](std::error_code ec4, std::size_t bytes_transferred) mutable {
+                          self->tcp_, asio::buffer(self->recv_buf_), [self](std::error_code ec4, std::size_t bytes_transferred4) mutable {
                               self->deadline_.cancel();
+                              CB_LOG_PROTOCOL("[DNS, TCP, IN] host=\"{}\", port={}, rc={}, bytes_received={}{:a}",
+                                              self->address_.to_string(),
+                                              self->port_,
+                                              ec4 ? ec4.message() : "ok",
+                                              bytes_transferred4,
+                                              spdlog::to_hex(self->recv_buf_.data(), self->recv_buf_.data() + bytes_transferred4));
+
                               if (ec4) {
                                   CB_LOG_DEBUG("DNS TCP read operation has been aborted, address=\"{}:{}\", ec={}",
                                                self->address_.to_string(),
                                                self->port_,
                                                ec4.message());
                                   return self->handler_({ ec4 });
                               }
-                              self->recv_buf_.resize(bytes_transferred);
+                              self->recv_buf_.resize(bytes_transferred4);
                               const dns_message message = dns_codec::decode(self->recv_buf_);
                               dns_srv_response resp{ ec4 };
                               resp.targets.reserve(message.answers.size());

core/io/http_session.hxx

@@ -31,7 +31,10 @@
 
 #include <couchbase/error_codes.hxx>
 
+#include <spdlog/fmt/bin_to_hex.h>
+
 #include <asio.hpp>
+
 #include <list>
 #include <memory>
 #include <utility>
@@ -448,7 +451,20 @@ class http_session : public std::enable_shared_from_this<http_session>
         stream_->async_read_some(
           asio::buffer(input_buffer_), [self = shared_from_this()](std::error_code ec, std::size_t bytes_transferred) {
               if (ec == asio::error::operation_aborted || self->stopped_) {
+                  CB_LOG_PROTOCOL("[HTTP, IN] type={}, host=\"{}\", rc={}, bytes_received={}",
+                                  self->type_,
+                                  self->info_.remote_address(),
+                                  ec ? ec.message() : "ok",
+                                  bytes_transferred);
                   return;
+              } else {
+                  CB_LOG_PROTOCOL("[HTTP, IN] type={}, host=\"{}\", rc={}, bytes_received={}{:a}",
+                                  self->type_,
+                                  self->info_.remote_address(),
+                                  ec ? ec.message() : "ok",
+                                  bytes_transferred,
+                                  spdlog::to_hex(self->input_buffer_.data(),
+                                                 self->input_buffer_.data() + static_cast<std::ptrdiff_t>(bytes_transferred)));
               }
               self->last_active_ = std::chrono::steady_clock::now();
               if (ec) {
@@ -495,9 +511,16 @@ class http_session : public std::enable_shared_from_this<http_session>
         std::vector<asio::const_buffer> buffers;
         buffers.reserve(writing_buffer_.size());
         for (auto& buf : writing_buffer_) {
+            CB_LOG_PROTOCOL(
+              "[HTTP, OUT] type={}, host=\"{}\", buffer_size={}{:a}", type_, info_.remote_address(), buf.size(), spdlog::to_hex(buf));
             buffers.emplace_back(asio::buffer(buf));
         }
-        stream_->async_write(buffers, [self = shared_from_this()](std::error_code ec, std::size_t /* bytes_transferred */) {
+        stream_->async_write(buffers, [self = shared_from_this()](std::error_code ec, std::size_t bytes_transferred) {
+            CB_LOG_PROTOCOL("[HTTP, OUT] type={}, host=\"{}\", rc={}, bytes_sent={}",
+                            self->type_,
+                            self->info_.remote_address(),
+                            ec ? ec.message() : "ok",
+                            bytes_transferred);
             if (ec == asio::error::operation_aborted || self->stopped_) {
                 return;
             }

core/io/mcbp_session.cxx

@@ -1365,7 +1365,20 @@ class mcbp_session_impl
           asio::buffer(input_buffer_),
           [self = shared_from_this(), stream_id = stream_->id()](std::error_code ec, std::size_t bytes_transferred) {
               if (ec == asio::error::operation_aborted || self->stopped_) {
+                  CB_LOG_PROTOCOL("[MCBP, IN] host=\"{}\", port={}, rc={}, bytes_received={}",
+                                  self->endpoint_address_,
+                                  self->endpoint_.port(),
+                                  ec ? ec.message() : "ok",
+                                  bytes_transferred);
                   return;
+              } else {
+                  CB_LOG_PROTOCOL("[MCBP, IN] host=\"{}\", port={}, rc={}, bytes_received={}{:a}",
+                                  self->endpoint_address_,
+                                  self->endpoint_.port(),
+                                  ec ? ec.message() : "ok",
+                                  bytes_transferred,
+                                  spdlog::to_hex(self->input_buffer_.data(),
+                                                 self->input_buffer_.data() + static_cast<std::ptrdiff_t>(bytes_transferred)));
               }
               self->last_active_ = std::chrono::steady_clock::now();
               if (ec) {
@@ -1433,13 +1446,21 @@ class mcbp_session_impl
         std::vector<asio::const_buffer> buffers;
         buffers.reserve(writing_buffer_.size());
         for (auto& buf : writing_buffer_) {
+            CB_LOG_PROTOCOL(
+              "[MCBP, OUT] host=\"{}\", port={}, buffer_size={}{:a}", endpoint_address_, endpoint_.port(), buf.size(), spdlog::to_hex(buf));
             buffers.emplace_back(asio::buffer(buf));
         }
-        stream_->async_write(buffers, [self = shared_from_this()](std::error_code ec, std::size_t /*unused*/) {
+        stream_->async_write(buffers, [self = shared_from_this()](std::error_code ec, std::size_t bytes_transferred) {
+            CB_LOG_PROTOCOL("[MCBP, OUT] host=\"{}\", port={}, rc={}, bytes_sent={}",
+                            self->endpoint_address_,
+                            self->endpoint_.port(),
+                            ec ? ec.message() : "ok",
+                            bytes_transferred);
             if (ec == asio::error::operation_aborted || self->stopped_) {
                 return;
             }
             self->last_active_ = std::chrono::steady_clock::now();
+
             if (ec) {
                 CB_LOG_ERROR(R"({} IO error while writing to the socket("{}"): {} ({}))",
                              self->log_prefix_,

core/logger/custom_rotating_file_sink.cxx

@@ -127,7 +127,7 @@ custom_rotating_file_sink<Mutex>::add_hook(const std::string& hook)
 
     // Payload shouldn't contain anything yet, overwrite it
     Expects(msg.payload.size() == 0);
-    msg.payload = hook;
+    msg.payload = hookToAdd;
 
     spdlog::memory_buf_t formatted;
     formatter->format(msg, formatted);

core/logger/logger.cxx

@@ -22,7 +22,8 @@
 #include <spdlog/sinks/stdout_color_sinks.h>
 #include <spdlog/spdlog.h>
 
-static const std::string logger_name{ "couchbase_cxx_client_file_logger" };
+static const std::string file_logger_name{ "couchbase_cxx_client_file_logger" };
+static const std::string protocol_logger_name{ "couchbase_cxx_client_protocol_logger" };
 
 /**
  * Custom log pattern which the loggers will use.
@@ -39,7 +40,12 @@ static const std::string log_pattern{ "[%Y-%m-%d %T.%e] [%P,%t] [%^%l%$] %oms, %
  * messages and send them to the sinks, which do the actual writing (to file,
  * to stream etc.) or further processing.
  */
-static std::shared_ptr<spdlog::logger> file_logger;
+static std::shared_ptr<spdlog::logger> file_logger{};
+
+/**
+ * Instance of the protocol logger.
+ */
+static std::shared_ptr<spdlog::logger> protocol_logger{};
 
 namespace couchbase::core::logger
 {
@@ -146,13 +152,11 @@ is_initialized()
     return file_logger != nullptr;
 }
 
-/**
- * Initialises the loggers. Called if the logger configuration is
- * specified in a separate settings object.
- */
-std::optional<std::string>
-create_file_logger(const configuration& logger_settings)
+std::pair<std::optional<std::string>, std::shared_ptr<spdlog::logger>>
+create_file_logger_impl(const std::string logger_name, const configuration& logger_settings)
 {
+    std::shared_ptr<spdlog::logger> logger{};
+
     auto fname = logger_settings.filename;
     auto buffersz = logger_settings.buffer_size;
     auto cyclesz = logger_settings.cycle_size;
@@ -218,31 +222,79 @@ create_file_logger(const configuration& logger_settings)
         spdlog::drop(logger_name);
 
         if (logger_settings.unit_test) {
-            file_logger = std::make_shared<spdlog::logger>(logger_name, sink);
+            logger = std::make_shared<spdlog::logger>(logger_name, sink);
         } else {
             // Create the default thread pool for async logging
             spdlog::init_thread_pool(buffersz, 1);
 
             // Get the thread pool so that we can actually construct the
             // object with already created sinks...
             auto tp = spdlog::thread_pool();
-            file_logger = std::make_shared<spdlog::async_logger>(logger_name, sink, tp, spdlog::async_overflow_policy::block);
+            logger = std::make_shared<spdlog::async_logger>(logger_name, sink, tp, spdlog::async_overflow_policy::block);
         }
 
-        file_logger->set_pattern(log_pattern);
-        file_logger->set_level(translate_level(logger_settings.log_level));
+        logger->set_pattern(log_pattern);
+        logger->set_level(translate_level(logger_settings.log_level));
 
         // Set the flushing interval policy
         spdlog::flush_every(std::chrono::seconds(1));
 
-        spdlog::register_logger(file_logger);
+        spdlog::register_logger(logger);
     } catch (const spdlog::spdlog_ex& ex) {
         std::string msg = std::string{ "Log initialization failed: " } + ex.what();
-        return std::optional<std::string>{ msg };
+        return { msg, {} };
+    }
+    return { {}, logger };
+}
+
+/**
+ * Initialises the loggers. Called if the logger configuration is
+ * specified in a separate settings object.
+ */
+std::optional<std::string>
+create_file_logger(const configuration& logger_settings)
+{
+    auto [error, logger] = create_file_logger_impl(file_logger_name, logger_settings);
+    if (error) {
+        return error;
+    }
+    file_logger = std::move(logger);
+    return {};
+}
+
+std::optional<std::string>
+create_protocol_logger(const configuration& logger_settings)
+{
+    if (logger_settings.filename.empty()) {
+        return "File name is missing";
+    }
+    auto config = logger_settings;
+    config.log_level = couchbase::core::logger::level::trace;
+    auto [error, logger] = create_file_logger_impl(protocol_logger_name, config);
+    if (error) {
+        return error;
     }
+    protocol_logger = std::move(logger);
     return {};
 }
 
+bool
+should_log_protocol()
+{
+    return protocol_logger != nullptr;
+}
+
+namespace detail
+{
+void
+log_protocol(const char* file, int line, const char* function, std::string_view msg)
+{
+    if (should_log_protocol()) {
+        return protocol_logger->log(spdlog::source_loc{ file, line, function }, spdlog::level::level_enum::trace, msg);
+    }
+}
+} // namespace detail
+
 spdlog::logger*
 get()
 {
@@ -252,17 +304,20 @@ get()
 void
 reset()
 {
-    spdlog::drop(logger_name);
+    spdlog::drop(file_logger_name);
     file_logger.reset();
+
+    spdlog::drop(protocol_logger_name);
+    protocol_logger.reset();
 }
 
 void
 create_blackhole_logger()
 {
     // delete if already exists
-    spdlog::drop(logger_name);
+    spdlog::drop(file_logger_name);
 
-    file_logger = std::make_shared<spdlog::logger>(logger_name, std::make_shared<spdlog::sinks::null_sink_mt>());
+    file_logger = std::make_shared<spdlog::logger>(file_logger_name, std::make_shared<spdlog::sinks::null_sink_mt>());
 
     file_logger->set_level(spdlog::level::off);
     file_logger->set_pattern(log_pattern);
@@ -274,11 +329,11 @@ void
 create_console_logger()
 {
     // delete if already exists
-    spdlog::drop(logger_name);
+    spdlog::drop(file_logger_name);
 
     auto stderrsink = std::make_shared<spdlog::sinks::stderr_color_sink_mt>();
 
-    file_logger = std::make_shared<spdlog::logger>(logger_name, stderrsink);
+    file_logger = std::make_shared<spdlog::logger>(file_logger_name, stderrsink);
     file_logger->set_level(spdlog::level::info);
     file_logger->set_pattern(log_pattern);
 
@@ -322,8 +377,13 @@ void
 set_log_levels(level lvl)
 {
     auto level = translate_level(lvl);
-    // Apply the function to each registered spdlog::logger
+    // Apply the function to each registered spdlog::logger except protocol logger
     spdlog::apply_all([level](std::shared_ptr<spdlog::logger> l) {
+        if (l->name() == protocol_logger_name) {
+            l->set_level(spdlog::level::trace);
+            return;
+        }
+
         try {
             l->set_level(level);
         } catch (const spdlog::spdlog_ex& e) {