[OMON-539] Add example how to deserialise protobuf from Kafka messages (#283)

Author: awegrzyn

CMakeLists.txt

@@ -78,6 +78,7 @@ find_package(ApMon MODULE)
 find_package(CURL MODULE)
 find_package(RdKafka CONFIG)
 find_package(InfoLogger CONFIG)
+find_package(Protobuf CONFIG)
 
 ####################################
 # Set OUTPUT vars
@@ -183,6 +184,10 @@ if (APPLE)
   message(STATUS "Detected macOS: Process monitor disabled")
 endif()
 
+if(Protobuf_FOUND AND RdKafka_FOUND)
+   message(STATUS "Compiling Kafka consumer with protobuf deserializer")
+endif()
+
 # Handle custom compile definitions
 target_compile_definitions(Monitoring
   PRIVATE
@@ -234,6 +239,37 @@ set_target_properties(8-DbFiller PROPERTIES OUTPUT_NAME "o2-monitoring-dbfiller"
 if(RdKafka_FOUND)
   set_target_properties(11-KafkaToWebsocket PROPERTIES OUTPUT_NAME "o2-monitoring-kafka-to-grafana")
 endif()
+####################################
+# Kafka protobuf deserializer
+####################################
+if(RdKafka_FOUND AND Protobuf_FOUND)
+  set(PROTO_FILE ${CMAKE_CURRENT_SOURCE_DIR}/proto/envs.proto)
+  get_filename_component(PROTO_OUTPUT_NAME ${PROTO_FILE} NAME_WE)
+  get_filename_component(PROTO_FILE_PREFIX ${PROTO_FILE} PATH)
+  set(PROTO_CPP_OUTPUT ${CMAKE_CURRENT_BINARY_DIR}/${PROTO_OUTPUT_NAME}.pb.cc)
+
+  add_custom_command(
+      OUTPUT "${PROTO_CPP_OUTPUT}"
+      COMMAND protobuf::protoc
+      ARGS --proto_path ${PROTO_FILE_PREFIX} --cpp_out ${CMAKE_CURRENT_SOURCE_DIR}/build ${PROTO_FILE}
+      DEPENDS ${PROTO_FILE}
+      COMMENT "Running protoc on ${PROTO_FILE}"
+      VERBATIM)
+
+  add_executable(12-KafkvaToInfluxDb
+    examples/12-KafkaToInfluxDb.cxx
+    ${PROTO_CPP_OUTPUT})
+
+  target_include_directories(12-KafkvaToInfluxDb
+    PRIVATE
+    ${CMAKE_CURRENT_BINARY_DIR})
+
+  target_link_libraries(12-KafkvaToInfluxDb
+    PRIVATE
+    Monitoring
+    Boost::program_options
+    protobuf::libprotobuf)
+endif()
 
 ####################################
 # Tests

examples/12-KafkaToInfluxDb.cxx

@@ -0,0 +1,44 @@
+///
+/// \file 12-KafkaToInfluxDb.cxx
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#include "../src/Transports/KafkaConsumer.h"
+#include "../src/Transports/HTTP.h"
+#include "../src/Transports/WebSocket.h"
+
+#include <iostream>
+#include <memory>
+#include <thread>
+#include <boost/algorithm/string/join.hpp>
+#include <boost/program_options.hpp>
+
+#include "envs.pb.h"
+
+using namespace o2::monitoring;
+
+int main(int argc, char* argv[])
+{
+  boost::program_options::options_description desc("Program options");
+  desc.add_options()
+    ("kafka-host", boost::program_options::value<std::string>()->required(), "Kafka broker hostname")
+    ("kafka-topic", boost::program_options::value<std::string>()->required(), "Kafka topic");
+  boost::program_options::variables_map vm;
+  boost::program_options::store(boost::program_options::parse_command_line(argc, argv, desc), vm);
+  boost::program_options::notify(vm);
+
+  auto kafkaConsumer = std::make_unique<transports::KafkaConsumer>(vm["kafka-host"].as<std::string>(), 9092, vm["kafka-topic"].as<std::string>());
+  for (;;) {
+    auto changes = kafkaConsumer->receive();
+    if (!changes.empty()) {
+      for (auto& change : changes) {
+        aliceo2::envs::NewStateNotification stateChange;
+        stateChange.ParseFromString(change);
+        std::cout << "EnvID: " << stateChange.envinfo().environmentid() << std::endl
+                  << "State: " << stateChange.envinfo().state() << std::endl
+                  << "First detector: " << stateChange.envinfo().detectors().Get(0) << std::endl;
+      }
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
+  }
+}

proto/envs.proto

@@ -0,0 +1,21 @@
+syntax = "proto3";
+package aliceo2.envs;
+option go_package = "protos;kafka";
+
+message NewStateNotification {
+  EnvInfo envInfo = 1;
+  uint64 timestamp = 2; // ms since epoch
+}
+
+message ActiveRunsList {
+  repeated EnvInfo activeRuns = 1;
+  uint64 timestamp = 2; // ms since epoch
+}
+
+message EnvInfo {
+  string environmentId = 1;
+  uint32 runNumber = 2;
+  string runType = 3;
+  string state = 4;
+  repeated string detectors = 5;
+}