[OMON-540] Provide selected metrics from Kafka over Webserver (#285)

Author: awegrzyn

CMakeLists.txt

@@ -215,7 +215,6 @@ set(EXAMPLES
   examples/5-Benchmark.cxx
   examples/6-Increment.cxx
   examples/7-InternalBenchamrk.cxx
-  examples/8-DbFiller.cxx
   examples/9-RunNumber.cxx
   examples/10-Buffering.cxx
 )
@@ -235,12 +234,14 @@ foreach (example ${EXAMPLES})
 endforeach()
 
 set_target_properties(5-Benchmark PROPERTIES OUTPUT_NAME "o2-monitoring-benchmark")
-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
+# Generate protobuf
 ####################################
 if(RdKafka_FOUND AND Protobuf_FOUND AND CURL_FOUND)
   set(PROTO_FILE ${CMAKE_CURRENT_SOURCE_DIR}/proto/envs.proto)
@@ -256,19 +257,24 @@ if(RdKafka_FOUND AND Protobuf_FOUND AND CURL_FOUND)
       COMMENT "Running protoc on ${PROTO_FILE}"
       VERBATIM)
 
-  add_executable(12-KafkvaToInfluxDb
+  set(PROTO_EXAMPLES
     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)
+    examples/8-KafkaToHttpServer.cxx
+  )
+  foreach (example ${PROTO_EXAMPLES})
+    get_filename_component(example_name ${example} NAME)
+    string(REGEX REPLACE ".cxx" "" example_name ${example_name})
+    add_executable(${example_name} ${example} ${PROTO_CPP_OUTPUT})
+    target_link_libraries(${example_name} PRIVATE
+      Monitoring
+      Boost::program_options
+      protobuf::libprotobuf
+    )
+    target_include_directories(${example_name} PRIVATE ${CMAKE_CURRENT_BINARY_DIR})
+    install(TARGETS ${example_name})
+  endforeach()
+  set_target_properties(8-KafkaToHttpServer PROPERTIES OUTPUT_NAME "o2-monitoring-env-webserver")
+  set_target_properties(12-KafkaToInfluxDb PROPERTIES OUTPUT_NAME "o2-monitoring-kafka-to-influxdb")
 endif()
 
 ####################################
@@ -311,7 +317,7 @@ endforeach()
 ####################################
 
 # Install library
-install(TARGETS Monitoring 5-Benchmark 8-DbFiller
+install(TARGETS Monitoring 5-Benchmark
   EXPORT MonitoringTargets
   LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}
   ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR}

examples/8-DbFiller.cxx

@@ -0,0 +1,215 @@
+#define BOOST_BEAST_USE_STD_STRING_VIEW
+
+#include <boost/beast/core.hpp>
+#include <boost/beast/http.hpp>
+#include <boost/beast/version.hpp>
+#include <boost/asio.hpp>
+#include <boost/program_options.hpp>
+#include <boost/algorithm/string/join.hpp>
+#include <chrono>
+#include <iostream>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <thread>
+
+#include "../src/Transports/KafkaConsumer.h"
+#include "../src/MonLogger.h"
+#include "envs.pb.h"
+
+namespace beast = boost::beast;
+namespace http = beast::http;
+using tcp = boost::asio::ip::tcp;
+using o2::monitoring::MonLogger;
+using namespace std::literals::string_literals;
+
+
+aliceo2::envs::ActiveRunsList gActiveEnvs;
+std::mutex gEnvAccess;
+
+
+class httpConnection : public std::enable_shared_from_this<httpConnection>
+{
+public:
+  httpConnection(tcp::socket socket) : mSocket(std::move(socket)) {}
+
+  // Initiate the asynchronous operations associated with the connection.
+  void start() {
+    readRequest();
+    checkDeadline();
+  }
+
+  void addCallback(std::string_view path, std::function<void(http::request<http::dynamic_body>& request, http::response<http::dynamic_body>& response)> callback) {
+    mCallbacks.insert({path, callback});
+  }
+
+private:
+  tcp::socket mSocket;
+  beast::flat_buffer mBuffer{8192};
+  http::request<http::dynamic_body> mRequest;
+  http::response<http::dynamic_body> mResponse;
+  boost::asio::steady_timer mDeadline{mSocket.get_executor(), std::chrono::seconds(5)};
+  std::map<std::string_view, std::function<void(http::request<http::dynamic_body>& request, http::response<http::dynamic_body>& response)>> mCallbacks;
+
+  // Asynchronously receive a complete request message.
+  void readRequest() {
+    auto self = shared_from_this();
+    http::async_read(mSocket, mBuffer, mRequest, [self](beast::error_code ec, std::size_t bytes_transferred) {
+      boost::ignore_unused(bytes_transferred);
+      if (!ec) self->processRequest();
+    });
+  }
+
+  // Determine what needs to be done with the request message.
+  void processRequest() {
+     mResponse.version(mRequest.version());
+     mResponse.keep_alive(false);
+     mResponse.result(http::status::ok);
+     createResponse();
+     writeResponse();
+  }
+
+  // Construct a response message based on the program state.
+  void createResponse() {
+    for (const auto& [key, value] : mCallbacks) {
+      if (mRequest.target().find(key) != std::string_view::npos) {
+        value(mRequest, mResponse);
+        return;
+      }
+    }
+    mResponse.result(http::status::not_found);
+    mResponse.set(http::field::content_type, "text/plain");
+    beast::ostream(mResponse.body()) << "Not found\r\n";
+  }
+
+  // Asynchronously transmit the response message.
+  void writeResponse() {
+    auto self = shared_from_this();
+    mResponse.content_length(mResponse.body().size());
+    http::async_write(mSocket, mResponse, [self](beast::error_code ec, std::size_t) {
+      self->mSocket.shutdown(tcp::socket::shutdown_send, ec);
+      self->mDeadline.cancel();
+    });
+  }
+
+  // Check whether we have spent enough time on this connection.
+  void checkDeadline() {
+    auto self = shared_from_this();
+    mDeadline.async_wait(
+    [self](beast::error_code ec) {
+      if(!ec) {
+        self->mSocket.close(ec);
+      }
+    });
+  }
+};
+
+// "Loop" forever accepting new connections.
+void httpServer(tcp::acceptor& acceptor, tcp::socket& socket) {
+  acceptor.async_accept(socket, [&](beast::error_code ec) {
+    if (!ec) {
+     auto connection = std::make_shared<httpConnection>(std::move(socket));
+     connection->addCallback("SHOW+RETENTION+POLICIES",
+     [](http::request<http::dynamic_body>& request, http::response<http::dynamic_body>& response) {
+       response.set(http::field::content_type, "application/json");
+       beast::ostream(response.body()) << "{}\n";
+     });
+     connection->addCallback("SHOW+TAG+VALUES+FROM",
+     [](http::request<http::dynamic_body>& request, http::response<http::dynamic_body>& response) {
+       std::string jsonPrefix = R"({"results": [{"statement_id": 0, "series": [{"name": "env_active", "columns": ["key", "value"], "values": [)";
+       std::string jsonSuffix = R"(]}]}]})";
+       response.set(http::field::content_type, "application/json");
+       const std::lock_guard<std::mutex> lock(gEnvAccess);
+       std::string envsJson;
+       for (int i = 0; i < gActiveEnvs.activeruns_size(); i++) {
+         envsJson += "[\"run\", \"" + std::to_string(gActiveEnvs.activeruns(i).runnumber()) + "\"],";
+       }
+       if (!envsJson.empty()) {
+         envsJson.pop_back();
+       }
+       beast::ostream(response.body()) << jsonPrefix << envsJson << jsonSuffix << '\n';
+     });
+     connection->addCallback("active_runs+WHERE+run",
+     [](http::request<http::dynamic_body>& request, http::response<http::dynamic_body>& response) {
+      std::string jsonPrefix = R"({"results":[{"statement_id":0,"series":[{"name":"env","columns":["time","Env ID","Run number","Detectors","State"],"values":[)";
+      std::string jsonSuffix = R"(]}]}]})";
+      response.set(http::field::content_type, "application/json");
+      std::string runString = std::string(request.target().substr(request.target().find("WHERE+run+%3D+") + 14));
+      uint32_t run;
+      try {
+        run  = static_cast<uint32_t>(std::stoul(runString));
+      } catch(...) {
+        response.set(http::field::content_type, "application/json");
+        beast::ostream(response.body()) << "{}\r\n";
+        return;
+      }
+      const std::lock_guard<std::mutex> lock(gEnvAccess);
+      std::string envsJson;
+      for (int i = 0; i < gActiveEnvs.activeruns_size(); i++) {
+        if (run != gActiveEnvs.activeruns(i).runnumber()) {
+          continue;
+        }
+        auto detectorsProto = gActiveEnvs.activeruns(i).detectors();
+        std::vector<std::string> detectors(detectorsProto.begin(), detectorsProto.end());
+        envsJson += "[" + std::to_string(gActiveEnvs.timestamp()) + ", \""
+                 + gActiveEnvs.activeruns(i).environmentid() + "\", "
+                 + std::to_string(gActiveEnvs.activeruns(i).runnumber()) + ", \""
+                 + boost::algorithm::join(detectors, " ") + "\", \""
+                 + gActiveEnvs.activeruns(i).state() + "\"]";
+      }
+      beast::ostream(response.body()) << jsonPrefix << envsJson << jsonSuffix << '\n';
+     });
+     connection->start();
+    }
+    httpServer(acceptor, socket);
+  });
+}
+
+// Deserialize active runs
+void deserializeActiveRuns(const std::string& lastActiveRunMessage)
+{
+  aliceo2::envs::ActiveRunsList activeRuns;
+  activeRuns.ParseFromString(lastActiveRunMessage);
+  if (activeRuns.activeruns_size() == 0) {
+    MonLogger::Get() << "Empty active runs" << MonLogger::End();;
+  } else {
+    for (int i = 0; i < activeRuns.activeruns_size(); i++) {
+      MonLogger::Get() << "Active run: " << activeRuns.activeruns(i).runnumber() << " ("
+                       << activeRuns.activeruns(i).environmentid() << ")" << MonLogger::End();
+    }
+  }
+  const std::lock_guard<std::mutex> lock(gEnvAccess);
+  gActiveEnvs = activeRuns;
+}
+
+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 host")
+    ("kafka-topics", boost::program_options::value<std::vector<std::string>>()->multitoken()->required(), "Kafka topics");
+  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);
+
+  MonLogger::mLoggerSeverity = o2::monitoring::Severity::Debug; 
+
+  std::thread webServerThread([](){
+    auto const address = boost::asio::ip::make_address("0.0.0.0");
+    boost::asio::io_context ioc{1};
+    tcp::acceptor acceptor{ioc, {address, 8086}};
+    tcp::socket socket{ioc};
+    httpServer(acceptor, socket);
+    ioc.run();
+  });
+
+  auto kafkaConsumer = std::make_unique<o2::monitoring::transports::KafkaConsumer>(
+    vm["kafka-host"].as<std::string>(), 9092, std::vector<std::string>{vm["kafka-topics"].as<std::vector<std::string>>()}
+  );
+  for (;;) {
+    auto serializedRuns = kafkaConsumer->receive();
+    if (!serializedRuns.empty()) {
+      deserializeActiveRuns(serializedRuns.back());
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(500));
+  }
+}