Interval flushing to ensure timely metric emission

Adds a flushIntervalMs parameter to WriterConfig (and Config) that
starts a background timer thread which drains all thread-local buffers
on a fixed period. This guarantees metrics are emitted within a bounded
timeframe even when write volume is low and the capacity-based flush
never triggers.

Author: mathetake

libs/config/config.h

@@ -23,6 +23,7 @@ class Config
     const std::string& GetWriterLocation() const noexcept { return m_writerConfig.GetLocation(); }
     const WriterType& GetWriterType() const noexcept { return m_writerConfig.GetType(); }
     const unsigned int GetWriterBufferSize() const noexcept { return m_writerConfig.GetBufferSize(); }
+    const unsigned int GetFlushIntervalMs() const noexcept { return m_writerConfig.GetFlushIntervalMs(); }
 
 
    private:

libs/writer/writer_config/writer_config.cpp

@@ -56,4 +56,11 @@ WriterConfig::WriterConfig(const std::string& type, const unsigned int bufferSiz
     Logger::info("WriterConfig buffering enabled with size: {}", m_bufferSize);
 }
 
+WriterConfig::WriterConfig(const std::string& type, const unsigned int bufferSize, const unsigned int flushIntervalMs)
+    : WriterConfig(type, bufferSize)  // Constructor delegation
+{
+    m_flushIntervalMs = flushIntervalMs;
+    Logger::info("WriterConfig flush interval enabled with interval: {}ms", m_flushIntervalMs);
+}
+
 }  // namespace spectator

libs/writer/writer_config/writer_config.h

@@ -12,15 +12,18 @@ class WriterConfig
    public:
     explicit WriterConfig(const std::string& type);
     WriterConfig(const std::string& type, unsigned int bufferSize);
+    WriterConfig(const std::string& type, unsigned int bufferSize, unsigned int flushIntervalMs);
 
     [[nodiscard]] const WriterType& GetType() const noexcept { return m_type; }
     [[nodiscard]] unsigned int GetBufferSize() const noexcept { return m_bufferSize; }
     [[nodiscard]] const std::string& GetLocation() const noexcept { return m_location; }
+    [[nodiscard]] unsigned int GetFlushIntervalMs() const noexcept { return m_flushIntervalMs; }
 
    private:
     WriterType m_type;
     std::string m_location;
     unsigned int m_bufferSize = 0;
+    unsigned int m_flushIntervalMs = 0;
 };
 
 }  // namespace spectator

libs/writer/writer_wrapper/test_writer.cpp

@@ -127,16 +127,56 @@ TEST_F(WriterWrapperUDSWriterTest, MultithreadedWrite)
     EXPECT_EQ(actualIncrements, expectedIncrements);
 }
 
+// Verify that metrics written infrequently are still flushed via the interval mechanism
+// even when the buffer is not full.
+TEST_F(WriterWrapperUDSWriterTest, IntervalFlushWithIdleThread)
+{
+    Logger::info("Starting interval flush test...");
+
+    // Use a large buffer so capacity-based flush never triggers,
+    // but set a short flush interval so metrics get sent in time.
+    const std::string unixUrl = "/tmp/test_uds_socket";
+    constexpr unsigned int largeBuffer = 4096;
+    constexpr unsigned int flushIntervalMs = 200;
+    WriterTestHelper::InitializeWriter(WriterType::Unix, unixUrl, 0, largeBuffer, flushIntervalMs);
+
+    MeterId meterId("interval.test.counter");
+    Counter counter(meterId);
+    counter.Increment();
+
+    // Wait for at least two flush intervals to pass
+    std::this_thread::sleep_for(std::chrono::milliseconds(flushIntervalMs * 3));
+
+    auto msgs = get_uds_messages();
+    EXPECT_FALSE(msgs.empty()) << "Expected metrics to be flushed by interval timer";
+
+    std::regex counter_regex(R"(c:interval\.test\.counter:1.000000)");
+    bool found = false;
+    for (const auto& msg : msgs)
+    {
+        std::stringstream ss(msg);
+        std::string line;
+        while (std::getline(ss, line))
+        {
+            if (std::regex_match(line, counter_regex))
+            {
+                found = true;
+            }
+        }
+    }
+    EXPECT_TRUE(found) << "Expected interval.test.counter metric to be received";
+}
+
 // Verify that multiple worker threads do not block each other: each thread uses its own
-// local buffer, flushing by capacity. Remaining data is drained by ThreadLocalBuffer's
-// destructor when each thread exits.
+// local buffer, flushing by capacity. The interval timer catches any tail data that
+// does not fill the buffer before the threads exit.
 TEST_F(WriterWrapperUDSWriterTest, ThreadLocalBufferNoMutexContention)
 {
     Logger::info("Starting thread-local buffer contention test...");
 
     const std::string unixUrl = "/tmp/test_uds_socket";
-    // Small buffer so capacity-based flush fires frequently
-    WriterTestHelper::InitializeWriter(WriterType::Unix, unixUrl, 0, 64);
+    // Small buffer so capacity-based flush fires frequently; interval timer catches the tail
+    WriterTestHelper::InitializeWriter(WriterType::Unix, unixUrl, 0, 64, 100);
 
     constexpr auto numThreads = 8;
     constexpr auto incrementsPerThread = 20;
@@ -162,7 +202,7 @@ TEST_F(WriterWrapperUDSWriterTest, ThreadLocalBufferNoMutexContention)
         t.join();
     }
 
-    // Allow time for thread destructor flushes to reach the UDS server
+    // Allow the interval timer to flush any remaining buffered data
     std::this_thread::sleep_for(std::chrono::milliseconds(300));
 
     auto msgs = get_uds_messages();

libs/writer/writer_wrapper/writer.cpp

@@ -36,12 +36,24 @@ Writer::ThreadLocalBuffer::~ThreadLocalBuffer()
 Writer::~Writer()
 {
     auto& instance = GetInstance();
-    instance.shutdown.store(true);
+
+    // Signal all background threads to stop
+    {
+        std::lock_guard<std::mutex> lock(instance.shutdownMutex);
+        instance.shutdown.store(true);
+    }
+    instance.cv_shutdown.notify_all();
+
+    if (instance.flushTimerThread.joinable())
+    {
+        instance.flushTimerThread.join();
+    }
+
     this->Close();
 }
 
 void Writer::Initialize(WriterType type, const std::string& param, int port,
-                        unsigned int bufferSize)
+                        unsigned int bufferSize, unsigned int flushIntervalMs)
 {
     auto& instance = GetInstance();
 
@@ -66,20 +78,37 @@ void Writer::Initialize(WriterType type, const std::string& param, int port,
         }
 
         instance.m_currentType = type;
-        instance.shutdown.store(false);
-        instance.bufferingEnabled = false;
 
-        // Clear thread-local buffer registry from any previous init
+        // Stop any pre-existing timer thread before re-initialising
+        {
+            std::lock_guard<std::mutex> lock(instance.shutdownMutex);
+            instance.shutdown.store(true);
+        }
+        instance.cv_shutdown.notify_all();
+        if (instance.flushTimerThread.joinable())
+        {
+            instance.flushTimerThread.join();
+        }
+        // Clear the thread-local buffer registry from any previous init
         {
             std::lock_guard<std::mutex> regLock(instance.registryMutex);
             instance.threadBufferRegistry.clear();
         }
+        instance.shutdown.store(false);
+        instance.bufferingEnabled = false;
 
-        if (bufferSize > 0)
+        if (bufferSize > 0 || flushIntervalMs > 0)
         {
             instance.bufferingEnabled = true;
             instance.bufferSize = bufferSize;
             instance.writeImpl = &Writer::ThreadLocalBufferedWrite;
+
+            if (flushIntervalMs > 0)
+            {
+                instance.flushInterval = std::chrono::milliseconds(flushIntervalMs);
+                instance.flushTimerThread = std::thread(&Writer::FlushTimerThread, &instance);
+                Logger::info("Writer interval flush enabled: {}ms", flushIntervalMs);
+            }
         }
         else
         {
@@ -99,6 +128,63 @@ void Writer::TryToSend(const std::string& message)
     instance.m_impl->Write(message);
 }
 
+void Writer::FlushTimerThread()
+{
+    auto& instance = GetInstance();
+
+    while (true)
+    {
+        // Sleep for the flush interval, or wake early on shutdown
+        {
+            std::unique_lock<std::mutex> lock(instance.shutdownMutex);
+            instance.cv_shutdown.wait_for(lock, instance.flushInterval,
+                [&instance] { return instance.shutdown.load(); });
+        }
+
+        if (instance.shutdown.load())
+        {
+            break;
+        }
+
+        // Collect live buffers (clean up expired weak_ptrs in the same pass)
+        std::vector<std::shared_ptr<ThreadLocalBuffer>> buffersToFlush;
+        {
+            std::lock_guard<std::mutex> regLock(instance.registryMutex);
+            for (auto it = instance.threadBufferRegistry.begin();
+                 it != instance.threadBufferRegistry.end();)
+            {
+                if (auto buf = it->lock())
+                {
+                    buffersToFlush.push_back(std::move(buf));
+                    ++it;
+                }
+                else
+                {
+                    it = instance.threadBufferRegistry.erase(it);
+                }
+            }
+        }
+
+        // Drain each thread-local buffer
+        for (auto& buf : buffersToFlush)
+        {
+            std::string toSend;
+            {
+                std::lock_guard<std::mutex> lock(buf->mutex);
+                if (!buf->data.empty())
+                {
+                    toSend = std::move(buf->data);
+                    buf->data.clear();
+                }
+            }
+            if (!toSend.empty())
+            {
+                instance.TryToSend(toSend);
+            }
+        }
+    }
+}
+
 void Writer::ThreadLocalBufferedWrite(const std::string& message)
 {
     auto& instance = GetInstance();
@@ -118,7 +204,7 @@ void Writer::ThreadLocalBufferedWrite(const std::string& message)
         tl_buffer->data.push_back(NEW_LINE);
 
         // Capacity-based flush: drain when the thread-local buffer is full
-        if (tl_buffer->data.size() >= instance.bufferSize)
+        if (instance.bufferSize > 0 && tl_buffer->data.size() >= instance.bufferSize)
         {
             toSend = std::move(tl_buffer->data);
             tl_buffer->data.clear();

libs/writer/writer_wrapper/writer.h

@@ -3,9 +3,12 @@
 #include <singleton.h>
 #include <writer_types.h>
 
+#include <chrono>
+#include <condition_variable>
 #include <memory>
 #include <mutex>
 #include <string>
+#include <thread>
 #include <vector>
 
 namespace spectator {
@@ -42,14 +45,17 @@ class Writer final : public Singleton<Writer>
     Writer() = default;
 
     static void Initialize(WriterType type, const std::string& param = "", int port = 0,
-                           unsigned int bufferSize = 0);
+                           unsigned int bufferSize = 0, unsigned int flushIntervalMs = 0);
 
     static void Write(const std::string& message);
 
     void ThreadLocalBufferedWrite(const std::string& message);
 
     void NonBufferedWrite(const std::string& message);
 
+    // Background thread: periodically drains all thread-local buffers
+    void FlushTimerThread();
+
     void TryToSend(const std::string& message);
 
     void Close();
@@ -71,7 +77,14 @@ class Writer final : public Singleton<Writer>
     std::mutex registryMutex;
     std::vector<std::weak_ptr<ThreadLocalBuffer>> threadBufferRegistry;
 
+    // Interval flush
+    std::chrono::milliseconds flushInterval{0};
+    std::thread flushTimerThread;
+
+    // Shutdown coordination
     std::atomic<bool> shutdown{false};
+    std::mutex shutdownMutex;
+    std::condition_variable cv_shutdown;
 };
 
 }  // namespace spectator

libs/writer/writer_wrapper/writer_test_helper.h

@@ -15,9 +15,9 @@ class WriterTestHelper
    public:
     // Initialize the Writer for testing purposes
     static void InitializeWriter(WriterType type, const std::string& param = "", int port = 0,
-                                  unsigned int bufferSize = 0)
+                                  unsigned int bufferSize = 0, unsigned int flushIntervalMs = 0)
     {
-        Writer::Initialize(type, param, port, bufferSize);
+        Writer::Initialize(type, param, port, bufferSize, flushIntervalMs);
     }
 
     // Get the Writer's implementation for testing purposes

spectator/registry.cpp

@@ -41,19 +41,25 @@ Registry::Registry(const Config& config) : m_config(config)
     if (config.GetWriterType() == WriterType::Memory)
     {
         Logger::info("Registry initializing Memory Writer");
-        Writer::Initialize(config.GetWriterType(), "", 0, this->m_config.GetWriterBufferSize());
+        Writer::Initialize(config.GetWriterType(), "", 0,
+                           this->m_config.GetWriterBufferSize(),
+                           this->m_config.GetFlushIntervalMs());
     }
     else if (config.GetWriterType() == WriterType::UDP)
     {
         auto [ip, port] = ParseUdpAddress(this->m_config.GetWriterLocation());
         Logger::info("Registry initializing UDP Writer at {}:{}", ip, port);
-        Writer::Initialize(config.GetWriterType(), ip, port, this->m_config.GetWriterBufferSize());
+        Writer::Initialize(config.GetWriterType(), ip, port,
+                           this->m_config.GetWriterBufferSize(),
+                           this->m_config.GetFlushIntervalMs());
     }
     else if (config.GetWriterType() == WriterType::Unix)
     {
         auto socketPath = ParseUnixAddress(this->m_config.GetWriterLocation());
         Logger::info("Registry initializing UDS Writer at {}", socketPath);
-        Writer::Initialize(config.GetWriterType(), socketPath, 0, this->m_config.GetWriterBufferSize());
+        Writer::Initialize(config.GetWriterType(), socketPath, 0,
+                           this->m_config.GetWriterBufferSize(),
+                           this->m_config.GetFlushIntervalMs());
     }    
 }