Line data    Source code

       1              : /**
       2              :  * @file OpMonPublisher.cpp OpMonPublisher Class Implementation
       3              :  *
       4              :  * This is part of the DUNE DAQ Software Suite, copyright 2020.
       5              :  * Licensing/copyright details are in the COPYING file that you should have
       6              :  * received with this code.
       7              :  */
       8              : 
       9              : #include "kafkaopmon/OpMonPublisher.hpp"
      10              : 
      11              : using namespace dunedaq::kafkaopmon;
      12              : 
      13            3 : OpMonPublisher::OpMonPublisher(const nlohmann::json& conf)
      14              : {
      15              : 
      16            3 :   RdKafka::Conf* k_conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
      17            3 :   std::string errstr;
      18              : 
      19              :   // Good observations on threadsafety here
      20              :   // https://www.confluent.io/blog/modern-cpp-kafka-api-for-safe-easy-messaging/
      21              : 
      22            3 :   auto it = conf.find("bootstrap");
      23            3 :   if (it == conf.end()) {
      24            1 :     throw MissingParameter(ERS_HERE, "bootstrap", nlohmann::to_string(conf));
      25              :   }
      26              : 
      27            2 :   k_conf->set("bootstrap.servers", *it, errstr);
      28            2 :   if (!errstr.empty()) {
      29            0 :     throw FailedConfiguration(ERS_HERE, "bootstrap.servers", errstr);
      30              :   }
      31              : 
      32            2 :   std::string client_id;
      33            2 :   it = conf.find("cliend_id");
      34            2 :   if (it != conf.end())
      35            0 :     client_id = *it;
      36              :   else
      37            2 :     client_id = "kafkaopmon_default_producer";
      38              : 
      39            2 :   k_conf->set("client.id", client_id, errstr);
      40            2 :   if (!errstr.empty()) {
      41            0 :     ers::error(FailedConfiguration(ERS_HERE, "client.id", errstr));
      42              :   }
      43              : 
      44              :   // Create producer instance
      45            2 :   m_producer.reset(RdKafka::Producer::create(k_conf, errstr));
      46              : 
      47            2 :   if (!m_producer) {
      48            0 :     throw FailedProducerCreation(ERS_HERE, errstr);
      49              :   }
      50              : 
      51            2 :   it = conf.find("default_topic");
      52            2 :   if (it != conf.end())
      53            1 :     m_default_topic = "monitoring." + it->get<std::string>();
      54            5 : }
      55              : 
      56            2 : OpMonPublisher::~OpMonPublisher()
      57              : {
      58              : 
      59            2 :   int timeout_ms = 500;
      60            2 :   RdKafka::ErrorCode err = m_producer->flush(timeout_ms);
      61              : 
      62            2 :   if (err == RdKafka::ERR__TIMED_OUT) {
      63            0 :     ers::warning(TimeoutReachedWhileFlushing(ERS_HERE, timeout_ms));
      64              :   }
      65            2 : }
      66              : 
      67              : void
      68            0 : OpMonPublisher::publish(dunedaq::opmon::OpMonEntry&& entry) const
      69              : {
      70              : 
      71            0 :   std::string binary;
      72            0 :   entry.SerializeToString(&binary);
      73              : 
      74            0 :   auto topic = extract_topic(entry);
      75            0 :   auto key = extract_key(entry);
      76              : 
      77            0 :   RdKafka::ErrorCode err = m_producer->produce(topic,
      78              :                                                RdKafka::Topic::PARTITION_UA,
      79              :                                                RdKafka::Producer::RK_MSG_COPY,
      80            0 :                                                const_cast<char*>(binary.c_str()),
      81              :                                                binary.size(),
      82            0 :                                                key.c_str(),
      83              :                                                key.size(),
      84              :                                                0,
      85              :                                                nullptr);
      86              : 
      87            0 :   if (err == RdKafka::ERR_NO_ERROR)
      88            0 :     return;
      89              : 
      90            0 :   std::string err_cause;
      91              : 
      92            0 :   switch (err) {
      93            0 :     case RdKafka::ERR__QUEUE_FULL:
      94            0 :       err_cause = "maximum number of outstanding messages reached";
      95              :       break;
      96            0 :     case RdKafka::ERR_MSG_SIZE_TOO_LARGE:
      97            0 :       err_cause = "message too large";
      98              :       break;
      99            0 :     case RdKafka::ERR__UNKNOWN_PARTITION:
     100            0 :       err_cause = "Unknown partition";
     101              :       break;
     102            0 :     case RdKafka::ERR__UNKNOWN_TOPIC:
     103            0 :       err_cause = "Unknown topic (";
     104            0 :       err_cause += topic;
     105            0 :       err_cause += ')';
     106              :       break;
     107            0 :     default:
     108            0 :       err_cause = "unknown";
     109              :       break;
     110              :   }
     111              : 
     112            0 :   throw FailedProduce(ERS_HERE, key, err_cause);
     113            0 : }