DPL: refactor dumping of performanceMetrics.json

ktf · ktf · commit b33b8bca2ecb · 2025-11-07T16:12:59.000+01:00
* Add test.
* Allow specifying the file on the command line
* Make the function take a std::stream rather than opening the file
  itself.
diff --git a/Framework/Core/CMakeLists.txt b/Framework/Core/CMakeLists.txt
@@ -240,6 +240,7 @@ add_executable(o2-test-framework-core
               test/test_OverrideLabels.cxx
               test/test_O2DataModelHelpers.cxx
               test/test_RootConfigParamHelpers.cxx
+              test/test_ResourcesMonitoringHelpers.cxx
               test/test_Services.cxx
               test/test_StringHelpers.cxx
               test/test_StaticFor.cxx
diff --git a/Framework/Core/include/Framework/DriverInfo.h b/Framework/Core/include/Framework/DriverInfo.h
@@ -149,6 +149,8 @@ struct DriverInfo {
   std::string uniqueWorkflowId = "";
   /// Metrics gathering interval
   unsigned short resourcesMonitoringInterval = 0;
+  /// Where to dump the metrics
+  std::string resourcesMonitoringFilename = "performanceMetrics.json";
   /// Metrics gathering dump to disk interval
   unsigned short resourcesMonitoringDumpInterval = 0;
   /// Port used by the websocket control. 0 means not initialised.
diff --git a/Framework/Core/src/ResourcesMonitoringHelper.cxx b/Framework/Core/src/ResourcesMonitoringHelper.cxx
@@ -55,7 +55,8 @@ boost::property_tree::ptree fillNodeWithValue(const DeviceMetricsInfo& deviceMet
 bool ResourcesMonitoringHelper::dumpMetricsToJSON(const std::vector<DeviceMetricsInfo>& metrics,
                                                   const DeviceMetricsInfo& driverMetrics,
                                                   const std::vector<DeviceSpec>& specs,
-                                                  std::vector<std::regex> const& performanceMetricsRegex) noexcept
+                                                  std::vector<std::regex> const& performanceMetricsRegex,
+                                                  std::ostream &out) noexcept
 {
 
   assert(metrics.size() == specs.size());
@@ -161,14 +162,7 @@ bool ResourcesMonitoringHelper::dumpMetricsToJSON(const std::vector<DeviceMetric
 
   root.add_child("driver", driverRoot);
 
-  std::ofstream file("performanceMetrics.json", std::ios::out);
-  if (file.is_open()) {
-    boost::property_tree::json_parser::write_json(file, root);
-  } else {
-    return false;
-  }
-
-  file.close();
+  boost::property_tree::json_parser::write_json(out, root);
 
   return true;
 }
diff --git a/Framework/Core/src/ResourcesMonitoringHelper.h b/Framework/Core/src/ResourcesMonitoringHelper.h
@@ -13,23 +13,22 @@
 #define O2_FRAMEWORK_RESOURCESMONITORINGHELPER_H_
 
 #include "Framework/DeviceMetricsInfo.h"
-#include "Monitoring/ProcessMonitor.h"
 #include "Framework/DeviceSpec.h"
 
 #include <vector>
-#include <type_traits>
 #include <regex>
+#include <iosfwd>
 
 namespace o2::framework
 {
-
 struct ResourcesMonitoringHelper {
   /// Dump the metrics in @a metrics which match the names specified in @a metricsToDump
   /// @a specs are the DeviceSpecs associated to the metrics.
   static bool dumpMetricsToJSON(std::vector<DeviceMetricsInfo> const& metrics,
                                 DeviceMetricsInfo const& driverMetrics,
                                 std::vector<DeviceSpec> const& specs,
-                                std::vector<std::regex> const& metricsToDump) noexcept;
+                                std::vector<std::regex> const& metricsToDump,
+                                std::ostream &out) noexcept;
   static bool isResourcesMonitoringEnabled(unsigned short interval) noexcept { return interval > 0; }
 };
 
diff --git a/Framework/Core/src/runDataProcessing.cxx b/Framework/Core/src/runDataProcessing.cxx
@@ -1247,8 +1247,10 @@ void dumpMetricsCallback(uv_timer_t* handle)
   auto* context = (DriverServerContext*)handle->data;
 
   static auto performanceMetrics = getDumpableMetrics();
+  std::ofstream file(context->driver->resourcesMonitoringFilename, std::ios::out);
   ResourcesMonitoringHelper::dumpMetricsToJSON(*(context->metrics),
-                                               context->driver->metrics, *(context->specs), performanceMetrics);
+                                               context->driver->metrics, *(context->specs), performanceMetrics,
+                                               file);
 }
 
 void dumpRunSummary(DriverServerContext& context, DriverInfo const& driverInfo, DeviceInfos const& infos, DeviceSpecs const& specs)
@@ -2035,6 +2037,7 @@ int runStateMachine(DataProcessorSpecs const& workflow,
             "--fairmq-ipc-prefix",
             "--readers",
             "--resources-monitoring",
+            "--resources-monitoring-file",
             "--resources-monitoring-dump-interval",
             "--time-limit",
           };
@@ -2268,7 +2271,7 @@ int runStateMachine(DataProcessorSpecs const& workflow,
           if (driverInfo.resourcesMonitoringDumpInterval) {
             uv_timer_stop(&metricDumpTimer);
           }
-          LOG(info) << "Dumping performance metrics to performanceMetrics.json file";
+          LOGP(info, "Dumping performance metrics to {}.json file", driverInfo.resourcesMonitoringFilename);
           dumpMetricsCallback(&metricDumpTimer);
         }
         dumpRunSummary(serverContext, driverInfo, infos, runningWorkflow.devices);
@@ -2916,6 +2919,7 @@ int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow,
     ("no-IPC", bpo::value<bool>()->zero_tokens()->default_value(false), "disable IPC topology optimization")                                                           //                                                                                                                                        //
     ("o2-control,o2", bpo::value<std::string>()->default_value(""), "dump O2 Control workflow configuration under the specified name")                                 //
     ("resources-monitoring", bpo::value<unsigned short>()->default_value(0), "enable cpu/memory monitoring for provided interval in seconds")                          //
+    ("resources-monitoring-file", bpo::value<std::string>()->default_value("performanceMetrics.json"), "file where to dump the metrics")                               //
     ("resources-monitoring-dump-interval", bpo::value<unsigned short>()->default_value(0), "dump monitoring information to disk every provided seconds");              //
   // some of the options must be forwarded by default to the device
   executorOptions.add(DeviceSpecHelpers::getForwardedDeviceOptions());
@@ -3186,6 +3190,7 @@ int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow,
   driverInfo.deployHostname = varmap["hostname"].as<std::string>();
   driverInfo.resources = varmap["resources"].as<std::string>();
   driverInfo.resourcesMonitoringInterval = varmap["resources-monitoring"].as<unsigned short>();
+  driverInfo.resourcesMonitoringFilename = varmap["resources-monitoring-file"].as<std::string>();
   driverInfo.resourcesMonitoringDumpInterval = varmap["resources-monitoring-dump-interval"].as<unsigned short>();
 
   // FIXME: should use the whole dataProcessorInfos, actually...
diff --git a/Framework/Core/test/test_ResourcesMonitoringHelpers.cxx b/Framework/Core/test/test_ResourcesMonitoringHelpers.cxx
@@ -0,0 +1,270 @@
+// Copyright 2019-2025 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.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+#include "../src/ResourcesMonitoringHelper.h"
+#include "Framework/DeviceMetricsInfo.h"
+#include "Framework/DeviceMetricsHelper.h"
+
+#include <catch_amalgamated.hpp>
+#include <regex>
+#include <sstream>
+#include <iostream>
+
+TEST_CASE("StreamMetrics")
+{
+  using namespace o2::framework;
+  std::vector<DeviceSpec> specs{
+    DeviceSpec{
+      .name = "someDevice",
+      .id = "someDevice",
+      .inputChannels = {},
+      .outputChannels = {},
+      .arguments = {},
+      .options = {},
+      .services = {},
+      .algorithm = AlgorithmSpec{},
+      .inputs = {},
+      .outputs = {},
+      .forwards = {},
+      .rank = 0,
+      .nSlots = 0,
+      .inputTimesliceId = 0,
+      .maxInputTimeslices = 0,
+      .completionPolicy = CompletionPolicy{},
+      .dispatchPolicy = DispatchPolicy{},
+      .callbacksPolicy = CallbacksPolicy{},
+      .sendingPolicy = SendingPolicy{},
+      .resourcePolicy = ResourcePolicy{},
+      .resource = {},
+      .resourceMonitoringInterval = 10,
+      .labels = {},
+      .metadata = {}},
+    DeviceSpec{
+      .name = "anotherDevice",
+      .id = "anotherDevice",
+      .inputChannels = {},
+      .outputChannels = {},
+      .arguments = {},
+      .options = {},
+      .services = {},
+      .algorithm = AlgorithmSpec{},
+      .inputs = {},
+      .outputs = {},
+      .forwards = {},
+      .rank = 0,
+      .nSlots = 0,
+      .inputTimesliceId = 0,
+      .maxInputTimeslices = 0,
+      .completionPolicy = CompletionPolicy{},
+      .dispatchPolicy = DispatchPolicy{},
+      .callbacksPolicy = CallbacksPolicy{},
+      .sendingPolicy = SendingPolicy{},
+      .resourcePolicy = ResourcePolicy{},
+      .resource = {},
+      .resourceMonitoringInterval = 10,
+      .labels = {},
+      .metadata = {}},
+
+  };
+
+  // This is the device metrics
+  std::vector<DeviceMetricsInfo> metrics;
+  metrics.resize(2);
+  {
+    DeviceMetricsInfo& info = metrics[0];
+    auto bkey = DeviceMetricsHelper::createNumericMetric<int>(info, "bkey");
+    REQUIRE(info.metricLabels.size() == 1);
+    REQUIRE(info.metricPrefixes.size() == 1);
+    auto akey = DeviceMetricsHelper::createNumericMetric<float>(info, "akey");
+    REQUIRE(info.metricLabels.size() == 2);
+    REQUIRE(info.metricPrefixes.size() == 2);
+    auto ckey = DeviceMetricsHelper::createNumericMetric<uint64_t>(info, "ckey");
+    REQUIRE(info.metricLabels.size() == 3);
+    REQUIRE(info.metricPrefixes.size() == 3);
+    REQUIRE(DeviceMetricsHelper::metricIdxByName("akey", info) == 1);
+    REQUIRE(DeviceMetricsHelper::metricIdxByName("bkey", info) == 0);
+    REQUIRE(DeviceMetricsHelper::metricIdxByName("ckey", info) == 2);
+    REQUIRE(info.changed.size() == 3);
+    REQUIRE(info.changed.at(0) == false);
+    size_t t = 1000;
+    ckey(info, 0, t++);
+    ckey(info, 1, t++);
+    ckey(info, 2, t++);
+    ckey(info, 3, t++);
+    ckey(info, 4, t++);
+    ckey(info, 5, t++);
+  }
+  // Metrics for the second device
+  {
+    DeviceMetricsInfo& info = metrics[1];
+    auto bkey = DeviceMetricsHelper::createNumericMetric<int>(info, "bkey");
+    REQUIRE(info.metricLabels.size() == 1);
+    REQUIRE(info.metricPrefixes.size() == 1);
+    auto akey = DeviceMetricsHelper::createNumericMetric<float>(info, "akey");
+    REQUIRE(info.metricLabels.size() == 2);
+    REQUIRE(info.metricPrefixes.size() == 2);
+    auto ckey = DeviceMetricsHelper::createNumericMetric<uint64_t>(info, "ckey");
+    REQUIRE(info.metricLabels.size() == 3);
+    REQUIRE(info.metricPrefixes.size() == 3);
+    REQUIRE(DeviceMetricsHelper::metricIdxByName("akey", info) == 1);
+    REQUIRE(DeviceMetricsHelper::metricIdxByName("bkey", info) == 0);
+    REQUIRE(DeviceMetricsHelper::metricIdxByName("ckey", info) == 2);
+    REQUIRE(info.changed.size() == 3);
+    REQUIRE(info.changed.at(0) == false);
+    size_t t = 1000;
+    bkey(info, 0, t++);
+    bkey(info, 1, t++);
+    bkey(info, 2, t++);
+    bkey(info, 3, t++);
+    bkey(info, 4, t++);
+    bkey(info, 5, t++);
+  }
+
+  // This is the driver metrics
+  DeviceMetricsInfo driverMetrics;
+  auto dbkey = DeviceMetricsHelper::createNumericMetric<int>(driverMetrics, "bkey");
+  REQUIRE(driverMetrics.metricLabels.size() == 1);
+  REQUIRE(driverMetrics.metricPrefixes.size() == 1);
+  auto dakey = DeviceMetricsHelper::createNumericMetric<float>(driverMetrics, "akey");
+  REQUIRE(driverMetrics.metricLabels.size() == 2);
+  REQUIRE(driverMetrics.metricPrefixes.size() == 2);
+  auto dckey = DeviceMetricsHelper::createNumericMetric<uint64_t>(driverMetrics, "ckey");
+  REQUIRE(driverMetrics.metricLabels.size() == 3);
+  REQUIRE(driverMetrics.metricPrefixes.size() == 3);
+  REQUIRE(DeviceMetricsHelper::metricIdxByName("akey", driverMetrics) == 1);
+  REQUIRE(DeviceMetricsHelper::metricIdxByName("bkey", driverMetrics) == 0);
+  REQUIRE(DeviceMetricsHelper::metricIdxByName("ckey", driverMetrics) == 2);
+  REQUIRE(driverMetrics.changed.size() == 3);
+  REQUIRE(driverMetrics.changed.at(0) == false);
+  size_t t = 2000;
+  dbkey(driverMetrics, 0, t++);
+  dbkey(driverMetrics, 1, t++);
+  dbkey(driverMetrics, 2, t++);
+  dbkey(driverMetrics, 3, t++);
+  dbkey(driverMetrics, 4, t++);
+  dbkey(driverMetrics, 5, t++);
+
+  dbkey(driverMetrics, 0, t++);
+  dbkey(driverMetrics, 1, t++);
+
+  dckey(driverMetrics, 0, t++);
+
+  std::stringstream streamer;
+  std::vector<std::regex> performanceMetrics{
+    std::regex("bkey"),
+    std::regex("ckey"),
+  };
+
+  ResourcesMonitoringHelper::dumpMetricsToJSON(metrics,
+                                               driverMetrics, specs, performanceMetrics,
+                                               streamer);
+  REQUIRE(streamer.str() == R"JSON({
+    "someDevice": {
+        "ckey": [
+            {
+                "timestamp": "1000",
+                "value": "0"
+            },
+            {
+                "timestamp": "1001",
+                "value": "1"
+            },
+            {
+                "timestamp": "1002",
+                "value": "2"
+            },
+            {
+                "timestamp": "1003",
+                "value": "3"
+            },
+            {
+                "timestamp": "1004",
+                "value": "4"
+            },
+            {
+                "timestamp": "1005",
+                "value": "5"
+            }
+        ]
+    },
+    "anotherDevice": {
+        "bkey": [
+            {
+                "timestamp": "1000",
+                "value": "0"
+            },
+            {
+                "timestamp": "1001",
+                "value": "1"
+            },
+            {
+                "timestamp": "1002",
+                "value": "2"
+            },
+            {
+                "timestamp": "1003",
+                "value": "3"
+            },
+            {
+                "timestamp": "1004",
+                "value": "4"
+            },
+            {
+                "timestamp": "1005",
+                "value": "5"
+            }
+        ]
+    },
+    "driver": {
+        "bkey": [
+            {
+                "timestamp": "2000",
+                "value": "0"
+            },
+            {
+                "timestamp": "2001",
+                "value": "1"
+            },
+            {
+                "timestamp": "2002",
+                "value": "2"
+            },
+            {
+                "timestamp": "2003",
+                "value": "3"
+            },
+            {
+                "timestamp": "2004",
+                "value": "4"
+            },
+            {
+                "timestamp": "2005",
+                "value": "5"
+            },
+            {
+                "timestamp": "2006",
+                "value": "0"
+            },
+            {
+                "timestamp": "2007",
+                "value": "1"
+            }
+        ],
+        "ckey": [
+            {
+                "timestamp": "2008",
+                "value": "0"
+            }
+        ]
+    }
+}
+)JSON");
+}
