erskafka::KafkaStream Class Reference

ES stream implementation. More...

#include <KafkaStream.hpp>

Inheritance diagram for erskafka::KafkaStream:

Collaboration diagram for erskafka::KafkaStream:

Public Member Functions
	KafkaStream (const std::string &param)
void	write (const ers::Issue &issue) override
Public Member Functions inherited from ers::OutputStream
virtual	~OutputStream ()
	Sends the issue into this stream.

Private Member Functions
void	ers_to_json (const ers::Issue &issue, size_t chain, std::vector< nlohmann::json > &j_objs)
void	kafka_exporter (std::string input, std::string topic)

Private Attributes
std::string	m_partition
RdKafka::Producer *	m_producer

Additional Inherited Members
Protected Member Functions inherited from ers::OutputStream
	OutputStream ()
OutputStream &	chained ()
virtual bool	isNull () const

Detailed Description

ES stream implementation.

This stream offers capability of publishing Issues to elastic search. A stream configuration is composed of the stream name, that is "erskafka".

Definition at line 27 of file KafkaStream.hpp.

Constructor & Destructor Documentation

KafkaStream()

erskafka::KafkaStream::KafkaStream ( const std::string & param )

explicit

Definition at line 23 of file KafkaStream.cpp.

{
 
  if (const char* env_p = std::getenv("DUNEDAQ_PARTITION"))
    m_partition = env_p;
 
  // Kafka server settings
  std::string brokers = param;
  std::string errstr;
 
  RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
  conf->set("bootstrap.servers", brokers, errstr);
  if (errstr != "") {
    std::cout << "Bootstrap server error : " << errstr << '\n';
  }
  if (const char* env_p = std::getenv("DUNEDAQ_APPLICATION_NAME"))
    conf->set("client.id", env_p, errstr);
  else
    conf->set("client.id", "erskafkaproducerdefault", errstr);
  if (errstr != "") {
    std::cout << "Producer configuration error : " << errstr << '\n';
  }
  // Create producer instance
  m_producer = RdKafka::Producer::create(conf, errstr);
 
  if (errstr != "") {
    std::cout << "Producer creation error : " << errstr << '\n';
  }
}

Member Function Documentation

ers_to_json()

void erskafka::KafkaStream::ers_to_json ( const ers::Issue & issue, size_t chain, std::vector< nlohmann::json > & j_objs )

private

Definition at line 54 of file KafkaStream.cpp.

{
  try {
    nlohmann::json message;
    message["partition"] = m_partition.c_str();
    message["issue_name"] = issue.get_class_name();
    message["message"] = issue.message().c_str();
    message["severity"] = ers::to_string(issue.severity());
    message["usecs_since_epoch"] =
      std::chrono::duration_cast<std::chrono::microseconds>(issue.ptime().time_since_epoch()).count();
    message["time"] = std::chrono::duration_cast<std::chrono::milliseconds>(issue.ptime().time_since_epoch()).count();
 
    message["qualifiers"] = issue.qualifiers();
    message["params"] = nlohmann::json::array({});
    for (auto p : issue.parameters()) {
      message["params"].push_back(p.first + ": " + p.second);
    }
    message["cwd"] = issue.context().cwd();
    message["file_name"] = issue.context().file_name();
    message["function_name"] = issue.context().function_name();
    message["host_name"] = issue.context().host_name();
    message["package_name"] = issue.context().package_name();
    message["user_name"] = issue.context().user_name();
    message["application_name"] = issue.context().application_name();
    message["user_id"] = issue.context().user_id();
    message["process_id"] = issue.context().process_id();
    message["thread_id"] = issue.context().thread_id();
    message["line_number"] = issue.context().line_number();
    message["chain"] = chain;
 
    if (issue.cause()) {
      ers_to_json(*issue.cause(), 2, j_objs);
    }
    j_objs.push_back(message);
  } catch (const std::exception& e) {
    std::cout << "Conversion from json error : " << e.what() << '\n';
  }
}

kafka_exporter()

void erskafka::KafkaStream::kafka_exporter ( std::string input, std::string topic )

private

Definition at line 94 of file KafkaStream.cpp.

{
  try {
    // RdKafka::Producer::RK_MSG_COPY to be investigated
    RdKafka::ErrorCode err = m_producer->produce(topic,
                                                 RdKafka::Topic::PARTITION_UA,
                                                 RdKafka::Producer::RK_MSG_COPY,
                                                 const_cast<char*>(input.c_str()),
                                                 input.size(),
                                                 nullptr,
                                                 0,
                                                 0,
                                                 nullptr,
                                                 nullptr);
    if (err != RdKafka::ERR_NO_ERROR) {
      std::cout << "% Failed to produce to topic " << topic << ": " << RdKafka::err2str(err) << std::endl;
    }
  } catch (const std::exception& e) {
    std::cout << "Producer error : " << e.what() << '\n';
  }
}

write()

void erskafka::KafkaStream::write ( const ers::Issue & issue )

overridevirtual

Write method

Parameters

issue

issue to be sent.

Implements ers::OutputStream.

Definition at line 120 of file KafkaStream.cpp.

{
  try {
    std::vector<nlohmann::json> j_objs;
    if (issue.cause() == nullptr) {
      ers_to_json(issue, 0, j_objs);
    } else {
      ers_to_json(issue, 1, j_objs);
    }
 
    // build a unique hash for a group of nested issues
    std::ostringstream tmpstream(issue.message());
    tmpstream << issue.context().process_id() << issue.time_t() << issue.context().application_name()
              << issue.context().host_name() << rand();
    std::string tmp = tmpstream.str();
    boost::crc_32_type crc32;
    crc32.process_bytes(tmp.c_str(), tmp.length());
 
    for (auto j : j_objs) {
      j["group_hash"] = crc32.checksum();
 
      erskafka::KafkaStream::kafka_exporter(j.dump(), "erskafka-reporting");
    }
    chained().write(issue);
  } catch (const std::exception& e) {
    std::cout << "Producer error : " << e.what() << '\n';
  }
}

Member Data Documentation

m_partition

std::string erskafka::KafkaStream::m_partition

private

Definition at line 34 of file KafkaStream.hpp.

m_producer

RdKafka::Producer* erskafka::KafkaStream::m_producer

private

Definition at line 35 of file KafkaStream.hpp.

---

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

• /github/workspace/dunedaq/sourcecode/erskafka/src/KafkaStream.hpp
• /github/workspace/dunedaq/sourcecode/erskafka/src/KafkaStream.cpp