Merge branch 'dev' into pr371

Author: sy-c

.github/workflows/build.yml

@@ -41,14 +41,24 @@ jobs:
       run: cd build && make -j
     - name: Test
       run: cd build && ctest --output-on-failure
+  alma9:
+    runs-on: [self-hosted, alma9]
+    steps:
+    - uses: actions/checkout@v2
+    - name: CMake
+      run: mkdir build && cd build && source /opt/rh/gcc-toolset-12/enable && cmake ..
+    - name: Build
+      run: cd build && make -j 2
+    - name: Test
+      run: cd build && ctest --output-on-failure
   cs8:
     runs-on: [self-hosted, cs8]
     steps:
     - uses: actions/checkout@v2
     - name: CMake
-      run: mkdir build && cd build && source /opt/rh/gcc-toolset-10/enable && cmake ..
+      run: mkdir build && cd build && source /opt/rh/gcc-toolset-12/enable && cmake ..
     - name: Build
-      run: cd build && make -j
+      run: cd build && make -j 2
     - name: Test
       run: cd build && ctest --output-on-failure
   cc7:
@@ -58,6 +68,6 @@ jobs:
     - name: CMake
       run: mkdir build && cd build && source /opt/rh/devtoolset-9/enable  && cmake3 ..
     - name: Build
-      run: cd build && make -j
+      run: cd build && make -j 2
     - name: Test
       run: cd build && ctest3 --output-on-failure

.gitignore

@@ -4,3 +4,4 @@
 *~
 .idea
 examples/config.ini
+/cmake-build-debug/

CMakeLists.txt

@@ -1,4 +1,4 @@
-# Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+# Copyright 2019-2024 CERN and copyright holders of ALICE O2.
 # See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
 # All rights not expressly granted are reserved.
 #
@@ -16,22 +16,20 @@
 # General project definition
 ####################################
 
-CMAKE_MINIMUM_REQUIRED(VERSION 3.5.2 FATAL_ERROR)
+CMAKE_MINIMUM_REQUIRED(VERSION 3.26 FATAL_ERROR)
 set(Boost_USE_STATIC_RUNTIME OFF)
 set(Boost_USE_STATIC_LIBS OFF)
 set(BUILD_SHARED_LIBS ON)
 set(Boost_USE_MULTITHREADED ON)
+set(O2_MONITORING_KAFKA_ENABLE ON)
+set(O2_MONITORING_CONTROL_ENABLE ON)
 
 # Set cmake policy by version: https://cmake.org/cmake/help/latest/manual/cmake-policies.7.html
-if(${CMAKE_VERSION} VERSION_LESS 3.12)
-    cmake_policy(VERSION ${CMAKE_MAJOR_VERSION}.${CMAKE_MINOR_VERSION})
-else()
-    cmake_policy(VERSION 3.12)
-endif()
+cmake_policy(VERSION 3.12)
 
 # Define project
 project(Monitoring
-  VERSION 3.17.5
+  VERSION 3.19.12
   DESCRIPTION "O2 Monitoring library"
   LANGUAGES CXX
 )
@@ -73,13 +71,34 @@ endif()
 
 list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_LIST_DIR}/cmake")
 
-find_package(Boost 1.70 REQUIRED COMPONENTS unit_test_framework program_options system)
+find_package(Boost 1.70 REQUIRED COMPONENTS unit_test_framework program_options)
 find_package(ApMon MODULE)
 find_package(CURL MODULE)
-find_package(RdKafka CONFIG)
 find_package(InfoLogger CONFIG)
-find_package(Protobuf)
-find_package(gRPC CONFIG)
+if(O2_MONITORING_KAFKA_ENABLE OR O2_MONITORING_CONTROL_ENABLE)
+  find_package(Protobuf CONFIG)
+endif()
+if(O2_MONITORING_CONTROL_ENABLE)
+  find_package(gRPC CONFIG)
+endif()
+
+if(RDKAFKA_ROOT)
+    message("RDKAFKA_ROOT set, we enable corresponding libs and binaries")
+    find_library(RDKAFKA_LIB1 "rdkafka++" REQUIRED PATHS ${RDKAFKA_ROOT}/lib)
+    find_library(RDKAFKA_LIB2 "rdkafka" REQUIRED PATHS ${RDKAFKA_ROOT}/lib)
+    set(RDKAFKA_LIBS  ${RDKAFKA_LIB1} ${RDKAFKA_LIB2})
+    set(RDKAFKA_INCLUDE "${RDKAFKA_ROOT}/include")
+    set(RdKafka_FOUND true)
+else()
+    message("RDKAFKA_ROOT not set, corresponding libs and binaries won't be built")
+endif()
+
+if(gRPC_FOUND)
+    message("gRPC found, we enable corresponding libs and binaries")
+else()
+    message("gRPC not found, corresponding libs and binaries won't be built")
+endif()
+
 
 ####################################
 # Set OUTPUT vars
@@ -142,6 +161,7 @@ target_include_directories(Monitoring
   PUBLIC
     $<INSTALL_INTERFACE:include>
     $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>
+    $<BUILD_INTERFACE:${RDKAFKA_INCLUDE}>
   PRIVATE
     ${CMAKE_CURRENT_SOURCE_DIR}/src
 )
@@ -153,13 +173,12 @@ set_target_properties(Monitoring PROPERTIES OUTPUT_NAME "O2Monitoring")
 target_link_libraries(Monitoring
   PUBLIC
     Boost::boost
-    $<$<BOOL:${RdKafka_FOUND}>:RdKafka::rdkafka++>
   PRIVATE
-    Boost::system
     pthread
     $<$<BOOL:${ApMon_FOUND}>:ApMon::ApMon>
     $<$<BOOL:${CURL_FOUND}>:CURL::libcurl>
     $<$<BOOL:${InfoLogger_FOUND}>:AliceO2::InfoLogger>
+    $<$<BOOL:${RdKafka_FOUND}>:${RDKAFKA_LIBS}>
 )
 
 # Handle ApMon optional dependency
@@ -247,7 +266,10 @@ foreach (example ${EXAMPLES})
   add_executable(${example_name} ${example})
   target_link_libraries(${example_name}
     PRIVATE
+      pthread
       Monitoring Boost::program_options
+      $<$<BOOL:${CURL_FOUND}>:CURL::libcurl>
+      $<$<BOOL:${RdKafka_FOUND}>:${RDKAFKA_LIBS}>
   )
 endforeach()
 
@@ -258,6 +280,17 @@ if(RdKafka_FOUND)
 endif()
 
 
+# executable: o2-monitoring-send
+add_executable(
+  o2-monitoring-send
+  src/sendMetric.cxx
+)
+target_link_libraries(
+  o2-monitoring-send
+  Monitoring
+)
+install(TARGETS o2-monitoring-send)
+
 ####################################
 # Generate protobuf
 ####################################
@@ -289,6 +322,7 @@ if(RdKafka_FOUND AND Protobuf_FOUND AND CURL_FOUND)
       Boost::program_options
       protobuf::libprotobuf
       $<$<BOOL:${InfoLogger_FOUND}>:AliceO2::InfoLogger>
+      $<$<BOOL:${RdKafka_FOUND}>:${RDKAFKA_LIBS}>
     )
     target_compile_definitions(${example_name} PRIVATE $<$<BOOL:${InfoLogger_FOUND}>:O2_MONITORING_WITH_INFOLOGGER>)
     target_include_directories(${example_name} PRIVATE ${CMAKE_CURRENT_BINARY_DIR})

README.md

@@ -110,6 +110,8 @@ send(Metric{"throughput"}.addValue(100, "tx").addValue(200, "rx"))
 
 See how it works in the example: [examples/1-Basic.cxx](examples/1-Basic.cxx).
 
+Metrics can also be injected from the command line using the o2-monitoring-send utility (self-documented).
+
 ## Advanced features
 
 ### Metric verbosity
@@ -206,13 +208,15 @@ This guide explains manual installation. For `ansible` deployment see [AliceO2Gr
 ## Receiving metrics from Monitoring system (development instructions)
 
 ### Requirements
-  - CS8 or CC7 with `devtoolset-9`
-  - Boost >= 1.70
+  - Ubuntu, RHEL9, RHEL8, CS8, macOS, or CC7 with `devtoolset-9`
+  - Boost >= 1.83, CMake
 
 ### Compile Monitoring library with Kafka backend
+
+#### Manually
 - Compile `librdkafka`
   ```bash
-  git clone -b v1.8.2 https://github.com/edenhill/librdkafka && cd librdkafka
+  git clone -b v2.3.0 https://github.com/edenhill/librdkafka && cd librdkafka
   cmake -H. -B./_cmake_build -DENABLE_LZ4_EXT=OFF -DCMAKE_INSTALL_LIBDIR=lib -DRDKAFKA_BUILD_TESTS=OFF -DRDKAFKA_BUILD_EXAMPLES=OFF -DCMAKE_INSTALL_PREFIX=~/librdkafka_install
   cmake --build ./_cmake_build --target install -j
   ```
@@ -223,6 +227,11 @@ This guide explains manual installation. For `ansible` deployment see [AliceO2Gr
   cmake --build ./_cmake_build --target install -j
   ```
 
+#### aliBuild
+- Modify `monitoring.sh`: add `- librdkafka` to "requires"
+- Compile Monitoring: `aliBuild build Monitoring --defaults o2-dataflow --always-prefer-system`
+- Add `Monitoring` as dependency of your project
+
 ### Look for Monitoring library in your CMake
 As `librdkafka` is optional dependency of Monitoring it is not handled by CMakeConfig, therefore you need:
 ```cmake
@@ -237,16 +246,24 @@ And then, link against `AliceO2::Monitoring` target.
 #include "Monitoring/MonitoringFactory.h"
 ...
 
-std::vector<std::string> topics = {"topic-to-subscribe"};
-auto client = MonitoringFactory::GetPullClient("kafka-server:9092", topics);
+std::vector<std::string> topics = {"<topic-to-subscribe>"};
+auto client = MonitoringFactory::GetPullClient("<kafka-server:9092>", topics, "<client-id>");
 for (;;) {
   auto metrics = client->pull();
   if (!metrics.empty()) {
     /// metric.first => topic name; metric.second => metric itself
+  } else {
+    // wait a bit if no data available
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
   }
-  std::this_thread::sleep_for(std::chrono::milliseconds(100));
-}
 ```
 
+Run-time parameters:
+- `<topic-to-subscribe>` - List of topics to subscribe
+- `<kafka-server:9092>` - Kafka broker (staging or production)
+- `<client_id>` - unique, self-explainable string describing the client, eg. `dcs-link-status` or `its-link-status`.
+
+Metrics are returned in batch of maximum 100 for each pull() call.
+
 ### Data format
 Native data format is [Influx Line Protocol](https://docs.influxdata.com/influxdb/latest/reference/syntax/line-protocol/) but metrics can be converted into any format listed in here: https://docs.influxdata.com/telegraf/latest/data_formats/output/

examples/12-KafkaToInfluxDb.cxx

@@ -18,6 +18,21 @@
 
 using namespace o2::monitoring;
 
+
+std::string getCreateBucketBody(const std::string& orgId, const int run) {
+  std::stringstream postPayload;
+  postPayload << R"({
+    "orgID": ")" + orgId + R"(",
+    "name": ")" + std::to_string(run) + R"(",
+    "retentionRules": [{
+      "type": "expire",
+      "everySeconds": 86400,
+      "shardGroupDurationSeconds": 86400
+    }]
+  })";
+  return postPayload.str();
+}
+
 int main(int argc, char* argv[])
 {
   boost::program_options::options_description desc("Program options");
@@ -26,7 +41,10 @@ int main(int argc, char* argv[])
     ("influxdb-url", boost::program_options::value<std::string>()->required(), "InfluxDB hostname")
     ("influxdb-token", boost::program_options::value<std::string>()->required(), "InfluxDB token")
     ("influxdb-org", boost::program_options::value<std::string>()->default_value("cern"), "InfluxDB organisation")
-    ("influxdb-bucket", boost::program_options::value<std::string>()->default_value("aliecs"), "InfluxDB bucket");
+    ("influxdb-bucket", boost::program_options::value<std::string>()->default_value("aliecs"), "InfluxDB bucket")
+    ("influxdb-dpl-url", boost::program_options::value<std::string>(), "InfluxDB DPL ID")
+    ("influxdb-dpl-orgid", boost::program_options::value<std::string>(), "InfluxDB DPL organization ID")
+    ("influxdb-dpl-token", boost::program_options::value<std::string>(), "InfluxDB DPL token");
   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);
@@ -41,6 +59,14 @@ int main(int argc, char* argv[])
   );
   httpTransport->addHeader("Authorization: Token " + vm["influxdb-token"].as<std::string>());
   auto influxdbBackend = std::make_unique<backends::InfluxDB>(std::move(httpTransport));
+
+  std::unique_ptr<transports::HTTP> influxBucketApi;
+  if (vm.count("influxdb-dpl-orgid") && vm.count("influxdb-dpl-url") && vm.count("influxdb-dpl-token")) {
+    MonLogger::Get() << "Creating bucket HTTP API for " << vm["influxdb-dpl-url"].as<std::string>() << MonLogger::End();
+    influxBucketApi.reset(new transports::HTTP(vm["influxdb-dpl-url"].as<std::string>() + "/api/v2/buckets"));
+    influxBucketApi->addHeader("Authorization: Token " + vm["influxdb-dpl-token"].as<std::string>());
+  }
+
   for (;;) {
     auto changes = kafkaConsumer->pull();
     if (!changes.empty()) {
@@ -50,15 +76,19 @@ int main(int argc, char* argv[])
         if (stateChange.envinfo().state().empty()) {
           continue;
         }
+        int run = stateChange.envinfo().runnumber();
         auto metric = Metric{"run_times"};
         if (change.first.find("leave") != std::string::npos) {
           metric.addValue(stateChange.timestamp(), "eor");
           MonLogger::Get() << stateChange.envinfo().environmentid() << "/" << stateChange.envinfo().runnumber() << "  " << change.first << " EOR: " << stateChange.timestamp() << MonLogger::End();
         } else {
           metric.addValue(stateChange.envinfo().runtype(), "type").addValue(stateChange.envinfo().enterstatetimestamp(), "sor");
           MonLogger::Get() << stateChange.envinfo().environmentid() << "/" << stateChange.envinfo().runnumber() << "  " << change.first << " SOR: " <<stateChange.envinfo().enterstatetimestamp() << MonLogger::End();
+          if (vm.count("influxdb-dpl-orgid") && run > 1) {
+            MonLogger::Get() << "Request sent to create bucket " << stateChange.envinfo().runnumber() << " on  " << vm["influxdb-dpl-url"].as<std::string>() << MonLogger::End();
+            influxBucketApi->send(getCreateBucketBody(vm["influxdb-dpl-orgid"].as<std::string>(), stateChange.envinfo().runnumber()));
+          }
         }
-        int run = stateChange.envinfo().runnumber();
         if (run > 1) {
           influxdbBackend->sendWithRun(metric, stateChange.envinfo().environmentid(), std::to_string(run));
         }

examples/13-PullClient.cxx

@@ -9,6 +9,7 @@ using namespace o2::monitoring;
 
 int main()
 {
+  //int ix=0;
   /// List of topics to subscribe
   std::vector<std::string> topics = {"cru.link_status"};
   /// Connect to server
@@ -21,7 +22,9 @@ int main()
         /// metric sample: link,serialId=983,endpoint=1,CRU=1,id=11,type=CRU pciAddress="b0:00.0",status=1i 1673940809860009855
         std::cout << metric.second << std::endl;
       }
+    } else {
+      // wait a bit if no data available
+      std::this_thread::sleep_for(std::chrono::milliseconds(100));
     }
-    std::this_thread::sleep_for(std::chrono::milliseconds(100));
   }
 }