kafka_opmon_consumer.cxx File Reference

#include "JsonInfluxConverter.hpp"
#include "boost/program_options.hpp"
#include "cpr/cpr.h"
#include "curl/curl.h"
#include "ers/ers.hpp"
#include "librdkafka/rdkafkacpp.h"
#include "nlohmann/json.hpp"
#include <csignal>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <iomanip>
#include <iostream>
#include <regex.h>
#include <sstream>
#include <string>
#include <sys/time.h>
#include <utility>
#include <vector>
#include <memory>

Include dependency graph for kafka_opmon_consumer.cxx:

Go to the source code of this file.

Namespaces
namespace	dunedaq
	Including Qt Headers.

Functions
	dunedaq::ERS_DECLARE_ISSUE (kafkaopmon, CannotPostToDb, "Cannot post to Influx DB : "<< error,((std::string) error)) ERS_DECLARE_ISSUE(kafkaopmon
Cannot create std::string fatal	dunedaq::ERS_DECLARE_ISSUE (kafkaopmon, CannotConsumeMessage, "Cannot consume message : "<< error,((std::string) error)) ERS_DECLARE_ISSUE(kafkaopmon
static int64_t	now ()
static std::vector< std::unique_ptr< RdKafka::Message > >	consume_batch (RdKafka::KafkaConsumer &consumer, size_t batch_size, int batch_tmout)
void	execution_command (const std::string &adress, const std::string &cmd)
void	consumerLoop (RdKafka::KafkaConsumer &consumer, int batch_size, int batch_tmout, std::string adress)
int	main (int argc, char *argv[])

Variables
	dunedaq::CannotCreateConsumer
Cannot create	dunedaq::consumer
Cannot create std::string fatal	dunedaq::IncorrectParameters
Cannot create std::string fatal Incorrect	dunedaq::parameters
static volatile sig_atomic_t	run = 1
static dunedaq::influxopmon::JsonConverter	m_json_converter
static std::vector< std::string >	inserts_vectors
static uint	seed = 0

Function Documentation

consume_batch()

static std::vector< std::unique_ptr< RdKafka::Message > > consume_batch ( RdKafka::KafkaConsumer & consumer, size_t batch_size, int batch_tmout )

static

Definition at line 52 of file kafka_opmon_consumer.cxx.

{
 
  std::vector<std::unique_ptr<RdKafka::Message>> msgs;
  msgs.reserve(batch_size);
 
  int64_t end = now() + batch_tmout;
  int remaining_timeout = batch_tmout;
 
  while (msgs.size() < batch_size) {
    auto msg = std::unique_ptr<RdKafka::Message>(consumer.consume(remaining_timeout));
 
    if (msg == nullptr)
      ers::error(dunedaq::kafkaopmon::CannotConsumeMessage(ERS_HERE, "%% Consumer error: Message is null"));
 
    switch (msg->err()) {
      case RdKafka::ERR__TIMED_OUT:
        return msgs;
 
      case RdKafka::ERR_NO_ERROR:
        msgs.push_back(std::move(msg));
        break;
 
      default:
        ers::error(
          dunedaq::kafkaopmon::CannotConsumeMessage(ERS_HERE, "%% Unhandled consumer error: " + msg->errstr()));
        run = 0;
        return msgs;
    }
 
    remaining_timeout = end - now();
    if (remaining_timeout < 0)
      break;
  }
 
  return msgs;
}

consumerLoop()

void consumerLoop	(	RdKafka::KafkaConsumer &	consumer,
		int	batch_size,
		int	batch_tmout,
		std::string	adress )

Definition at line 105 of file kafka_opmon_consumer.cxx.

{
 
  std::cout << "Using query: " << adress << std::endl;
 
  while (run) {
    auto msgs = consume_batch(consumer, batch_size, batch_tmout);
    for (auto& msg : msgs) {
 
      // execution_command(adress, message_text);
      std::string json_string(static_cast<char*>(msg->payload()), msg->len());
 
      // std::cout << json_string << std::endl;
      //  json message = json::parse( json_string );
 
      // std::string query;
      // query = message["type"].get<std::string>() + ',';
      // query += ("source_id="+message["source_id"].get<std::string>()+',');
      // query += ("partition_id="+message["partition_id"].get<std::string>()+' ');
      // const auto & data = message["__data"];
      // std::stringstream data_stream;
      // for ( auto it = data.begin() ; it != data.end() ; ++it ) {
      //   if ( it != data.begin() ) data_stream << ',' ;
      //   data_stream << it.key() << '=' << it.value() ;
      // }
      // query += data_stream.str();
      // query += ' ';
      // query += std::to_string(message["__time"].get<uint64_t>() * 1000000000);
 
      // execution_command(adress, query);
    }
  }
}

execution_command()

void execution_command	(	const std::string &	adress,
		const std::string &	cmd )

Definition at line 91 of file kafka_opmon_consumer.cxx.

{
 
  cpr::Response response = cpr::Post(cpr::Url{ adress }, cpr::Body{ cmd });
  // std::cout << cmd << std::endl;
  if (response.status_code >= 400) {
    ers::error(dunedaq::kafkaopmon::CannotPostToDb(
      ERS_HERE, "Error [" + std::to_string(response.status_code) + "] making request"));
  } else if (response.status_code == 0) {
    ers::error(dunedaq::kafkaopmon::CannotPostToDb(ERS_HERE, "Query returned 0"));
  }
}

main()

int main	(	int	argc,
		char *	argv[] )

Definition at line 140 of file kafka_opmon_consumer.cxx.

{
  std::string broker;
  std::string topic;
  std::string db_host;
  std::string db_port;
  std::string db_path;
  std::string db_dbname;
  std::string topic_str;
  std::vector<std::string> topics;
  // Bulk consume parameters
  int batch_size = 100;
  int batch_tmout = 1000;
  // Kafka server settings
  std::string errstr;
 
  // get parameters
 
  bpo::options_description desc{ "example: -broker 188.185.122.48:9092 -topic kafkaopmon-reporting -dbhost "
                                 "188.185.88.195 -dbport 80 -dbpath insert -dbname db1" };
  desc.add_options()("help,h", "Help screen")(
    "broker,b", bpo::value<std::string>()->default_value("monkafka.cern.ch:30092"), "Broker")(
    "topic,t", bpo::value<std::string>()->default_value("opmon"), "Topic")(
    "dbhost,ho", bpo::value<std::string>()->default_value("opmondb.cern.ch"), "Database host")(
    "dbport,po", bpo::value<std::string>()->default_value("31002"), "Database port")(
    "dbpath,pa", bpo::value<std::string>()->default_value("write"), "Database path")(
    "dbname,n", bpo::value<std::string>()->default_value("influxdb"), "Database name");
 
  bpo::variables_map vm;
 
  try {
    auto parsed = bpo::command_line_parser(argc, argv).options(desc).run();
    bpo::store(parsed, vm);
  } catch (bpo::error const& e) {
    ers::error(dunedaq::kafkaopmon::IncorrectParameters(ERS_HERE, e.what()));
  }
 
  if (vm.count("help")) {
    TLOG() << desc << std::endl;
    return 0;
  }
 
  broker = vm["broker"].as<std::string>();
  topic = vm["topic"].as<std::string>();
  db_host = vm["dbhost"].as<std::string>();
  db_port = vm["dbport"].as<std::string>();
  db_path = vm["dbpath"].as<std::string>();
  db_dbname = vm["dbname"].as<std::string>();
 
  // Broker parameters
 
  try {
    auto conf = std::unique_ptr<RdKafka::Conf>(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
 
    srand((unsigned)time(0));
    std::string group_id = "opmon_microservices";
 
    conf->set("bootstrap.servers", broker, errstr);
    if (errstr != "") {
      ers::fatal(dunedaq::kafkaopmon::CannotCreateConsumer(ERS_HERE, errstr));
    }
    conf->set("client.id", "opmon_microservice-0", errstr);
    if (errstr != "") {
      ers::fatal(dunedaq::kafkaopmon::CannotCreateConsumer(ERS_HERE, errstr));
    }
    conf->set("group.id", group_id, errstr);
    if (errstr != "") {
      ers::fatal(dunedaq::kafkaopmon::CannotCreateConsumer(ERS_HERE, errstr));
    }
    topics.push_back(topic);
 
    auto consumer = std::unique_ptr<RdKafka::KafkaConsumer>(RdKafka::KafkaConsumer::create(conf.get(), errstr));
 
    if (errstr != "") {
      ers::fatal(dunedaq::kafkaopmon::CannotCreateConsumer(ERS_HERE, errstr));
    }
 
    if (consumer)
      consumer->subscribe(topics);
    else
      ers::fatal(dunedaq::kafkaopmon::CannotCreateConsumer(ERS_HERE, errstr));
 
    consumerLoop(*consumer, batch_size, batch_tmout, db_host + ":" + db_port + "/" + db_path + "?db=" + db_dbname);
 
    consumer->close();
  } catch (ers::IssueCatcherAlreadySet& ex) {
    ers::error(ex);
  }
 
  return 0;
}

now()

static int64_t now ( )

static

Definition at line 44 of file kafka_opmon_consumer.cxx.

{
  struct timeval tv;
  gettimeofday(&tv, nullptr);
  return ((int64_t)tv.tv_sec * 1000) + (tv.tv_usec / 1000);
}

Variable Documentation

inserts_vectors

std::vector<std::string> inserts_vectors

static

Definition at line 40 of file kafka_opmon_consumer.cxx.

m_json_converter

dunedaq::influxopmon::JsonConverter m_json_converter

static

Definition at line 39 of file kafka_opmon_consumer.cxx.

run

volatile sig_atomic_t run = 1

static

Definition at line 38 of file kafka_opmon_consumer.cxx.

seed

uint seed = 0

static

Definition at line 41 of file kafka_opmon_consumer.cxx.