Merge pull request #4 from fvallenilla/freddy/timed-flush

Add condition to flush based on time

spectator/config.h

@@ -9,6 +9,7 @@ struct Config {
   std::string endpoint;
   std::unordered_map<std::string, std::string> common_tags;
   uint32_t bytes_to_buffer;
+  const std::chrono::milliseconds flush_interval;
 };
 
 }  // namespace spectator

spectator/publisher.cc

@@ -8,10 +8,14 @@ static const char NEW_LINE = '\n';
 
 SpectatordPublisher::SpectatordPublisher(absl::string_view endpoint,
                                          uint32_t bytes_to_buffer,
+                                         std::chrono::milliseconds flush_interval,
                                          std::shared_ptr<spdlog::logger> logger)
     : logger_(std::move(logger)),
       udp_socket_(io_context_),
-      local_socket_(io_context_), bytes_to_buffer_(bytes_to_buffer) {
+      local_socket_(io_context_),
+      bytes_to_buffer_(bytes_to_buffer),
+      last_flush_time_(std::chrono::steady_clock::now()),
+      flush_interval_(flush_interval) {
   buffer_.reserve(bytes_to_buffer_ + 1024);     
   if (absl::StartsWith(endpoint, "unix:")) {
     setup_unix_domain(endpoint.substr(5));
@@ -55,11 +59,16 @@ void SpectatordPublisher::setup_unix_domain(absl::string_view path) {
   std::string local_path{path};
   sender_ = [local_path, this](std::string_view msg) {
     buffer_.append(msg);
-    if (buffer_.length() >= bytes_to_buffer_) {
+    const auto now = std::chrono::steady_clock::now();
+    const bool should_flush = buffer_.length() >= bytes_to_buffer_ ||
+                        now - last_flush_time_ >= flush_interval_;
+
+    if (should_flush) {
       for (auto i = 0; i < 3; ++i) {
         try {
           auto sent_bytes = local_socket_.send(asio::buffer(buffer_));
           logger_->trace("Sent (local): {} bytes, in total had {}", sent_bytes, buffer_.length());
+          last_flush_time_ = now;
           break;
         } catch (std::exception& e) {
           local_reconnect(local_path);

spectator/publisher.h

@@ -12,6 +12,7 @@ class SpectatordPublisher {
   explicit SpectatordPublisher(
       absl::string_view endpoint,
       uint32_t bytes_to_buffer = 0,
+      std::chrono::milliseconds flush_interval = std::chrono::milliseconds(60000),
       std::shared_ptr<spdlog::logger> logger = DefaultLogger());
   SpectatordPublisher(const SpectatordPublisher&) = delete;
 
@@ -34,6 +35,8 @@ class SpectatordPublisher {
   asio::local::datagram_protocol::socket local_socket_;
   std::string buffer_;
   uint32_t bytes_to_buffer_;
+  std::chrono::steady_clock::time_point last_flush_time_;
+  const std::chrono::milliseconds flush_interval_;
 };
 
 }  // namespace spectator

spectator/publisher_test.cc

@@ -83,6 +83,37 @@ TEST(Publisher, UnixBuffer) {
   unlink(path.c_str());
 }
 
+TEST(Publisher, UnixBufferTimeFlush) {
+  auto logger = spectator::DefaultLogger();
+  const auto* dir = first_not_null(std::getenv("TMPDIR"), "/tmp");
+  auto path = fmt::format("{}/testserver.{}", dir, getpid());
+  TestUnixServer server{path};
+  server.Start();
+  logger->info("Unix Server started on path {}", path);
+
+  // Set buffer size to a large value so that flushing is based on time
+  SpectatordPublisher publisher{fmt::format("unix:{}", path), 10000, std::chrono::milliseconds(500)};
+  Counter c{std::make_shared<Id>("counter", Tags{}), &publisher};
+
+  // Wait for 300ms, increment, and the counter should not be flushed (300ms is less than the 500ms flush interval)
+  std::this_thread::sleep_for(std::chrono::milliseconds(300));
+  c.Increment();
+  std::this_thread::sleep_for(std::chrono::milliseconds(50));
+  auto msgs = server.GetMessages();
+  EXPECT_TRUE(msgs.empty());
+
+  // Wait for another 300ms, increment, and the counter should be flushed (600ms is greater than 500ms flush interval)
+  std::this_thread::sleep_for(std::chrono::milliseconds(300));
+  c.Increment();
+  std::this_thread::sleep_for(std::chrono::milliseconds(50));
+  msgs = server.GetMessages();
+  std::vector<std::string> first_flush{"c:counter:1\nc:counter:1"};
+  EXPECT_EQ(msgs, first_flush);
+
+  server.Stop();
+  unlink(path.c_str());
+}
+
 TEST(Publisher, Nop) {
   SpectatordPublisher publisher{"", 0};
   Counter c{std::make_shared<Id>("counter", Tags{}), &publisher};

spectator/registry.h

@@ -324,7 +324,7 @@ class SpectatordRegistry
             std::move(logger)) {
     extra_tags_ = Tags::from(config.common_tags);
     state_.publisher =
-        std::make_unique<SpectatordPublisher>(config.endpoint, config.bytes_to_buffer, logger_);
+        std::make_unique<SpectatordPublisher>(config.endpoint, config.bytes_to_buffer, config.flush_interval, logger_);
   }
 };