dunedaq::kafkaopmon::OpMonPublisher Class Reference

#include <OpMonPublisher.hpp>

Public Member Functions
	OpMonPublisher (const nlohmann::json &conf)
	OpMonPublisher ()=delete
	OpMonPublisher (const OpMonPublisher &)=delete
OpMonPublisher &	operator= (const OpMonPublisher &)=delete
	OpMonPublisher (OpMonPublisher &&)=delete
OpMonPublisher &	operator= (OpMonPublisher &&)=delete
	~OpMonPublisher ()
void	publish (dunedaq::opmon::OpMonEntry &&) const

Protected Member Functions
std::string	extract_topic (const dunedaq::opmon::OpMonEntry &) const noexcept
std::string	extract_key (const dunedaq::opmon::OpMonEntry &e) const noexcept

Private Attributes
std::unique_ptr< RdKafka::Producer >	m_producer
std::string	m_default_topic = "monitoring.opmon_stream"

Detailed Description

Definition at line 55 of file OpMonPublisher.hpp.

Constructor & Destructor Documentation

OpMonPublisher() [1/4]

OpMonPublisher::OpMonPublisher

(

const nlohmann::json &

conf

)

Definition at line 13 of file OpMonPublisher.cpp.

{
 
  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() [2/4]

dunedaq::kafkaopmon::OpMonPublisher::OpMonPublisher ( )

delete

OpMonPublisher() [3/4]

dunedaq::kafkaopmon::OpMonPublisher::OpMonPublisher ( const OpMonPublisher & )

delete

OpMonPublisher() [4/4]

dunedaq::kafkaopmon::OpMonPublisher::OpMonPublisher ( OpMonPublisher && )

delete

~OpMonPublisher()

OpMonPublisher::~OpMonPublisher

(

)

Definition at line 56 of file OpMonPublisher.cpp.

{
 
  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));
  }
}

Member Function Documentation

extract_key()

std::string dunedaq::kafkaopmon::OpMonPublisher::extract_key ( const dunedaq::opmon::OpMonEntry & e ) const

inlineprotectednoexcept

Definition at line 73 of file OpMonPublisher.hpp.

  {
    return dunedaq::opmonlib::to_string(e.origin()) + '/' + e.measurement();
  }

extract_topic()

std::string dunedaq::kafkaopmon::OpMonPublisher::extract_topic ( const dunedaq::opmon::OpMonEntry & ) const

inlineprotectednoexcept

Definition at line 72 of file OpMonPublisher.hpp.

{ return m_default_topic; }

operator=() [1/2]

OpMonPublisher & dunedaq::kafkaopmon::OpMonPublisher::operator= ( const OpMonPublisher & )

delete

operator=() [2/2]

OpMonPublisher & dunedaq::kafkaopmon::OpMonPublisher::operator= ( OpMonPublisher && )

delete

publish()

void OpMonPublisher::publish

(

dunedaq::opmon::OpMonEntry &&

entry

)

const

Definition at line 68 of file OpMonPublisher.cpp.

{
 
  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);
}

Member Data Documentation

m_default_topic

std::string dunedaq::kafkaopmon::OpMonPublisher::m_default_topic = "monitoring.opmon_stream"

private

Definition at line 80 of file OpMonPublisher.hpp.

m_producer

std::unique_ptr<RdKafka::Producer> dunedaq::kafkaopmon::OpMonPublisher::m_producer

private

Definition at line 79 of file OpMonPublisher.hpp.

---

The documentation for this class was generated from the following files:

• /github/workspace/dunedaq/sourcecode/kafkaopmon/include/kafkaopmon/OpMonPublisher.hpp
• /github/workspace/dunedaq/sourcecode/kafkaopmon/src/OpMonPublisher.cpp