synapse_log_sink.hpp

// Copyright Science Corporation 2025
#pragma once
#include <functional>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
 
#include <spdlog/spdlog.h>
#include <spdlog/sinks/base_sink.h>
 
namespace synapse::log {
 
template <typename Mutex>
class SynapseLogSink : public spdlog::sinks::base_sink<Mutex> {
 public:
  // Notify for new log messages
  using log_callback_t = std::function<void(const synapse::LogEntry)>;
 
  SynapseLogSink() {
    running_.store(true);
    worker_thread_ = std::thread([this]() { process_message_queue(); });
  }
 
  ~SynapseLogSink() {
    {
      std::lock_guard<std::mutex> lock(queue_mutex_);
      running_.store(false);
      queue_cv_.notify_all();
    }
    if (worker_thread_.joinable()) {
      worker_thread_.join();
    }
  }
 
  // Register a callback (subscriber) to new log messages
  // Returns the id of the subscriber if successful, otherwise returns nullopt
  std::optional<uint64_t> subscribe(log_callback_t callback) {
    std::lock_guard<std::mutex> lock(callback_mutex_);
    if (callbacks_.size() >= MAX_SUBSCRIBERS) {
      return std::nullopt;
    }
    uint64_t id = callback_id_++;
    callbacks_[id] = std::move(callback);
    return id;
  }
 
  // Unregister the subscriber
  bool unsubscribe(const uint64_t id) {
    std::lock_guard<std::mutex> lock(callback_mutex_);
    return callbacks_.erase(id) > 0;
  }
 
 protected:
  void sink_it_(const spdlog::details::log_msg& msg) override {
    // Check if the queue is full before pushing
    std::lock_guard<std::mutex> lock(queue_mutex_);
    if (queue_.size() >= MAX_QUEUE_SIZE) {
      return;
    }
 
    // Convert the message to a LogEntry and add it to the queue
    auto log_entry = convert_to_log_entry(msg);
    queue_.push(std::move(log_entry));
    queue_cv_.notify_one();
  }
 
  // Don't need to flush here
  void flush_() override {}
 
 private:
  // Just limit the number of subscribers to 5, I don't expect there to be more than that
  static constexpr uint64_t MAX_SUBSCRIBERS = 5;
 
  // Max size of the queue, if we are this far behind we should be dropping logs
  static constexpr uint64_t MAX_QUEUE_SIZE = 1000;
 
  std::atomic<bool> running_ = false;
 
  // Keep track of our subscribers
  std::mutex callback_mutex_;
  std::unordered_map<uint64_t, log_callback_t> callbacks_;
 
  // We don't actually fire any of the callbacks in the hot loop, we add them to a queue and process
  // them asynchronously This is to avoid blocking the hot loop and to avoid calling any spdlog
  // functions in the callback (which would cause a circular dependency and result in a deadlock)
  std::thread worker_thread_;
  std::mutex queue_mutex_;
  std::condition_variable queue_cv_;
  std::queue<synapse::LogEntry> queue_;
 
  std::atomic<uint64_t> callback_id_ = 0;
 
  void process_message_queue() {
    while (running_) {
      synapse::LogEntry log_entry;
      {
        std::unique_lock<std::mutex> lock(queue_mutex_);
        queue_cv_.wait(lock, [this]() { return !queue_.empty() || !running_; });
 
        if (!running_ && queue_.empty()) {
          break;
        }
 
        if (!queue_.empty()) {
          log_entry = queue_.front();
          queue_.pop();
        }
      }
 
      // Call our callbacks while trying to minimize the time that the lock is held
      {
        std::lock_guard<std::mutex> lock(callback_mutex_);
        for (const auto& [id, callback] : callbacks_) {
          try {
            callback(log_entry);
          } catch (const std::exception& e) {
            // Error here instead of log to avoid a circular dependency
            // (we shouldn't be calling spdlog in the sink)
            std::cerr << "Error processing callback in SynapseLogSink: " << e.what() << std::endl;
          }
        }
      }
    }
  }
 
  synapse::LogEntry convert_to_log_entry(const spdlog::details::log_msg& msg) {
    // Create it into a log message for grpc
    synapse::LogEntry entry;
    const uint64_t timestamp_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
                                      std::chrono::system_clock::now().time_since_epoch())
                                      .count();
    entry.set_timestamp_ns(timestamp_ns);
 
    switch (msg.level) {
      case spdlog::level::trace:
      case spdlog::level::debug:
        entry.set_level(synapse::LogLevel::LOG_LEVEL_DEBUG);
        break;
      case spdlog::level::info:
        entry.set_level(synapse::LogLevel::LOG_LEVEL_INFO);
        break;
      case spdlog::level::warn:
        entry.set_level(synapse::LogLevel::LOG_LEVEL_WARNING);
        break;
      case spdlog::level::err:
        entry.set_level(synapse::LogLevel::LOG_LEVEL_ERROR);
        break;
      case spdlog::level::critical:
        entry.set_level(synapse::LogLevel::LOG_LEVEL_CRITICAL);
        break;
      default:
        entry.set_level(synapse::LogLevel::LOG_LEVEL_DEBUG);
        break;
    }
    const std::string source = std::string(msg.logger_name.begin(), msg.logger_name.end());
    if (source.empty()) {
      entry.set_source("synapse-app");
    } else {
      entry.set_source(source);
    }
 
    // Try to convert the payload to a string, if it is binary data, we don't serialize it
    try {
      const std::string payload(msg.payload.begin(), msg.payload.end());
      entry.set_message(payload);
    } catch (const std::exception& e) {
      // If it is binary data, we don't serialize it
      entry.set_message("[BINARY DATA]");
    }
    return entry;
  }
};
}  // namespace synapse::log