OpMonPublisher.cpp

Go to the documentation of this file.

 
#include "kafkaopmon/OpMonPublisher.hpp"
 
using namespace dunedaq::kafkaopmon;
 
OpMonPublisher::OpMonPublisher(const nlohmann::json& conf)
{
 
  RdKafka::Conf* k_conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
  std::string errstr;
 
  // Good observations on threadsafety here
  // https://www.confluent.io/blog/modern-cpp-kafka-api-for-safe-easy-messaging/
 
  auto it = conf.find("bootstrap");
  if (it == conf.end()) {
    throw MissingParameter(ERS_HERE, "bootstrap", nlohmann::to_string(conf));
  }
 
  k_conf->set("bootstrap.servers", *it, errstr);
  if (!errstr.empty()) {
    throw FailedConfiguration(ERS_HERE, "bootstrap.servers", errstr);
  }
 
  std::string client_id;
  it = conf.find("cliend_id");
  if (it != conf.end())
    client_id = *it;
  else
    client_id = "kafkaopmon_default_producer";
 
  k_conf->set("client.id", client_id, errstr);
  if (!errstr.empty()) {
    ers::error(FailedConfiguration(ERS_HERE, "client.id", errstr));
  }
 
  // Create producer instance
  m_producer.reset(RdKafka::Producer::create(k_conf, errstr));
 
  if (!m_producer) {
    throw FailedProducerCreation(ERS_HERE, errstr);
  }
 
  it = conf.find("default_topic");
  if (it != conf.end())
    m_default_topic = "monitoring." + it->get<std::string>();
}
 
OpMonPublisher::~OpMonPublisher()
{
 
  int timeout_ms = 500;
  RdKafka::ErrorCode err = m_producer->flush(timeout_ms);
 
  if (err == RdKafka::ERR__TIMED_OUT) {
    ers::warning(TimeoutReachedWhileFlushing(ERS_HERE, timeout_ms));
  }
}
 
void
OpMonPublisher::publish(dunedaq::opmon::OpMonEntry&& entry) const
{
 
  std::string binary;
  entry.SerializeToString(&binary);
 
  auto topic = extract_topic(entry);
  auto key = extract_key(entry);
 
  RdKafka::ErrorCode err = m_producer->produce(topic,
                                               RdKafka::Topic::PARTITION_UA,
                                               RdKafka::Producer::RK_MSG_COPY,
                                               const_cast<char*>(binary.c_str()),
                                               binary.size(),
                                               key.c_str(),
                                               key.size(),
                                               0,
                                               nullptr);
 
  if (err == RdKafka::ERR_NO_ERROR)
    return;
 
  std::string err_cause;
 
  switch (err) {
    case RdKafka::ERR__QUEUE_FULL:
      err_cause = "maximum number of outstanding messages reached";
      break;
    case RdKafka::ERR_MSG_SIZE_TOO_LARGE:
      err_cause = "message too large";
      break;
    case RdKafka::ERR__UNKNOWN_PARTITION:
      err_cause = "Unknown partition";
      break;
    case RdKafka::ERR__UNKNOWN_TOPIC:
      err_cause = "Unknown topic (";
      err_cause += topic;
      err_cause += ')';
      break;
    default:
      err_cause = "unknown";
      break;
  }
 
  throw FailedProduce(ERS_HERE, key, err_cause);
}