monitoring: enable metrics from StfSender and TfBuilder

Author: ironMann

cmake/FindAliceO2.cmake

@@ -25,7 +25,6 @@ find_library(AliceO2_LIBRARY_HEADERS NAMES O2Headers HINTS ${O2_ROOT}/lib ENV LD
 find_library(AliceO2_LIBRARY_FRAMEWORK NAMES O2Framework HINTS ${O2_ROOT}/lib ENV LD_LIBRARY_PATH)
 
 set(AliceO2_LIBRARIES
-  ${AliceO2_LIBRARY_O2DEVICE}
   ${AliceO2_LIBRARY_HEADERS}
   ${AliceO2_LIBRARY_FRAMEWORK}
 )

script/start_Discovery-3FLP-3EPN.sh.in

@@ -145,11 +145,15 @@ STF_SENDER="StfSender"
 STF_SENDER+=" --discovery-net-if=$FLP_NETIF"
 STF_SENDER+=" --discovery-partition=$PARTITION"
 STF_SENDER+=" --mq-config $chainConfig"
+STF_SENDER+=" --monitoring-rate=5.0"
+STF_SENDER+=" --monitoring-log"
 
 TF_BUILDER="TfBuilder"
 TF_BUILDER+=" --discovery-net-if=$EPN_NETIF"
 TF_BUILDER+=" --discovery-partition=$PARTITION"
 TF_BUILDER+=" --mq-config $chainConfig"
+TF_BUILDER+=" --monitoring-rate=1.0"
+TF_BUILDER+=" --monitoring-log"
 
 if [[ ! -z $TF_BUILDER_DPL_CHAN ]]; then
   TF_BUILDER+=" --dpl-channel-name=$TF_BUILDER_DPL_CHAN"

src/StfSender/CMakeLists.txt

@@ -9,7 +9,7 @@ set(EXE_STFS_SOURCES
 )
 
 add_library(StfSender_lib OBJECT ${EXE_STFS_SOURCES})
-target_link_libraries(StfSender_lib  base fmqtools common discovery)
+target_link_libraries(StfSender_lib  base fmqtools common discovery monitoring)
 
 add_executable(StfSender)
 
@@ -22,7 +22,7 @@ endif()
 target_link_libraries(StfSender
   PRIVATE
     StfSender_lib
-    base fmqtools common discovery
+    base fmqtools common discovery monitoring
 )
 
 install(TARGETS StfSender RUNTIME DESTINATION bin)

src/StfSender/StfSenderDevice.cxx

@@ -65,6 +65,11 @@ void StfSenderDevice::InitTask()
   I().mInputChannelName = GetConfig()->GetValue<std::string>(OptionKeyInputChannelName);
   I().mStandalone = GetConfig()->GetValue<bool>(OptionKeyStandalone);
 
+  // start monitoring
+  DataDistMonitor::start_datadist(ProcessType::StfSender, GetConfig()->GetValue<std::string>("monitoring-backend"));
+  DataDistMonitor::set_rate(GetConfig()->GetValue<float>("monitoring-rate"));
+  DataDistMonitor::set_log(GetConfig()->GetValue<bool>("monitoring-log"));
+
   if (!standalone()) {
     // Discovery
     I().mDiscoveryConfig = std::make_shared<ConsulStfSender>(ProcessType::StfSender, Config::getEndpointOption(*GetConfig()));
@@ -77,11 +82,17 @@ void StfSenderDevice::InitTask()
 
     // wait for "partition-id"
     while (!Config::getPartitionOption(*GetConfig())) {
-      WDDLOG("TfBuilder waiting on 'discovery-partition' config parameter.");
+      WDDLOG("StfSender waiting on 'discovery-partition' config parameter.");
       std::this_thread::sleep_for(1s);
     }
 
-    lStatus.mutable_partition()->set_partition_id(*Config::getPartitionOption(*GetConfig()));
+    I().mPartitionId = Config::getPartitionOption(*GetConfig()).value_or("");
+    if (I().mPartitionId.empty()) {
+      WDDLOG("StfSender 'discovery-partition' parameter not set.");
+      std::this_thread::sleep_for(1s); exit(-1);
+    }
+
+    lStatus.mutable_partition()->set_partition_id(I().mPartitionId);
     I().mDiscoveryConfig->write();
   }
 
@@ -144,11 +155,19 @@ void StfSenderDevice::PreRun()
     I().mDiscoveryConfig->write();
   }
 
+  // reset counters
+  if (I().mOutputHandler) {
+    I().mOutputHandler->resetCounters();
+  }
+
   // make directory for file sink
   if (I().mFileSink) {
     I().mFileSink->makeDirectory();
   }
 
+  // enable monitoring
+  DataDistMonitor::enable_datadist(DataDistLogger::sRunNumber, I().mPartitionId);
+
   // start accepting data
   I().mAcceptingData = true;
 
@@ -160,13 +179,21 @@ void StfSenderDevice::PostRun()
   // stop accepting data
   I().mAcceptingData = false;
 
+  // disable monitoring
+  DataDistMonitor::disable_datadist();
+
   // update running state
   if (!standalone() && I().mDiscoveryConfig) {
     auto& lStatus = I().mDiscoveryConfig->status();
     lStatus.mutable_info()->set_process_state(BasicInfo::NOT_RUNNING);
     I().mDiscoveryConfig->write();
   }
 
+  // reset counters
+  if (I().mOutputHandler) {
+    I().mOutputHandler->resetCounters();
+  }
+
   IDDLOG("Exiting running state. RunNumber: {}", DataDistLogger::sRunNumberStr);
 }
 
@@ -204,6 +231,9 @@ void StfSenderDevice::ResetTask()
     I().mTfSchedulerRpcClient.stop();
   }
 
+  // stop monitoring
+  DataDistMonitor::stop_datadist();
+
   DDDLOG("ResetTask() done.");
 }
 
@@ -218,7 +248,7 @@ void StfSenderDevice::StfReceiverThread()
   DplToStfAdapter  lStfReceiver;
   std::unique_ptr<SubTimeFrame> lStf;
 
-  auto lStfStartTime = hres_clock::now();
+  decltype(hres_clock::now()) lStfStartTime = hres_clock::now();
 
   while (running()) {
     try {
@@ -235,18 +265,20 @@ void StfSenderDevice::StfReceiverThread()
       if (lStf) {
         WDDLOG_RL(1000, "StfSender: received STF but not in the running state.");
       }
-      std::this_thread::sleep_for(10ms);
+      DDMON("stfsender", "stf_input.rate", 0.0);
+      DDMON("stfsender", "stf_input.size", 0.0);
+      std::this_thread::sleep_for(20ms);
       continue;
     }
 
     { // Input STF frequency
       const auto lNow = hres_clock::now();
       const auto lStfDur = std::chrono::duration<double>(lNow - lStfStartTime);
       lStfStartTime = lNow;
-      I().mStfTimeMean += (lStfDur.count()/100.0 - I().mStfTimeMean/100.0);
 
-      // get data size
-      I().mStfSizeMean += (lStf->getDataSize()/128 - I().mStfSizeMean/128);
+      DDMON("stfsender", "stf_input.rate", (1.0 / lStfDur.count()));
+      DDMON("stfsender", "stf_input.size", lStf->getDataSize());
+      DDMON("stfsender", "stf_input.id", (uint64_t)lStf->header().mId);
     }
 
     ++lReceivedStfs;
@@ -265,11 +297,10 @@ void StfSenderDevice::StfReceiverThread()
 void StfSenderDevice::InfoThread()
 {
   while (running()) {
-    IDDLOG("StfSender: SubTimeFrame size_mean={} in_frequency_mean={:.4}", I().mStfSizeMean, (1.0 / I().mStfTimeMean));
     if (!standalone()) {
       const auto lCounters = I().mOutputHandler->getCounters();
 
-      IDDLOG("StfSender: SubTimeFrame queued_stf_num={} queued_stf_size={} sending_stf_num={} sending_stf_size={} ",
+      DDDLOG("StfSender: SubTimeFrame queued_stf_num={} queued_stf_size={} sending_stf_num={} sending_stf_size={} ",
           lCounters.mBuffered.mCnt, lCounters.mBuffered.mSize,
           lCounters.mInSending.mCnt, lCounters.mInSending.mSize);
     }

src/StfSender/StfSenderDevice.h

@@ -23,6 +23,7 @@
 #include <SubTimeFrameFileSink.h>
 #include <Utilities.h>
 #include <FmqUtilities.h>
+#include <DataDistMonitoring.h>
 
 #include <thread>
 #include <vector>
@@ -87,11 +88,12 @@ class StfSenderDevice : public DataDistDevice
     /// Configuration
     std::string mInputChannelName;
     bool mStandalone = false;
+    std::string mPartitionId;
 
     /// Discovery configuration
     std::shared_ptr<ConsulStfSender> mDiscoveryConfig;
 
-    /// Scheculer RPC client
+    /// Scheduler RPC client
     TfSchedulerRpcClient mTfSchedulerRpcClient;
 
     /// Receiver threads
@@ -110,8 +112,6 @@ class StfSenderDevice : public DataDistDevice
 
     /// Info thread
     std::thread mInfoThread;
-    std::uint64_t mStfSizeMean = 0;
-    double mStfTimeMean = 50.0;
 
     unsigned getNextPipelineStage(unsigned pStage) final
     {

src/StfSender/StfSenderOutput.cxx

@@ -250,7 +250,7 @@ void StfSenderOutput::StfSchedulerThread()
       auto [it, ins] = mScheduledStfMap.try_emplace(lStfId, std::move(lStf));
       if (!ins) {
         (void)it;
-        EDDLOG_RL(500, "StfSchedulerThread: Stf already scheduled! Skipping the duplicate. stf_id={}", lStfId);
+        EDDLOG_RL(500, "StfSchedulerThread: Stf is already scheduled! Skipping the duplicate. stf_id={}", lStfId);
         continue;
       }
 
@@ -303,15 +303,10 @@ void StfSenderOutput::sendStfToTfBuilder(const std::uint64_t pStfId, const std::
     } else {
       pRes.set_status(StfDataResponse::DATA_DROPPED_SCHEDULER);
     }
-    return;
-  }
-
-  // check if it is drop request from the scheduler
-  if (pTfBuilderId == "-1") {
+  } else if (pTfBuilderId == "-1") { // check if it is drop request from the scheduler
     pRes.set_status(StfDataResponse::DATA_DROPPED_SCHEDULER);
     mDropQueue.push(std::move(lStfIter->second));
     mScheduledStfMap.erase(lStfIter);
-    return;
   } else {
     auto lTfBuilderIter = mOutputMap.find(pTfBuilderId);
     if (lTfBuilderIter == mOutputMap.end()) {
@@ -326,12 +321,25 @@ void StfSenderOutput::sendStfToTfBuilder(const std::uint64_t pStfId, const std::
     // we clean the buffer when data is sent
     pRes.set_status(StfDataResponse::OK);
 
-    mCounters.mInSending.mSize += lStfIter->second->getDataSize();
+    const auto lStfSize = lStfIter->second->getDataSize();
+    const auto lStfId = lStfIter->second->id();
+    mCounters.mInSending.mSize += lStfSize;
     mCounters.mInSending.mCnt += 1;
 
     auto lStfNode = mScheduledStfMap.extract(lStfIter);
     lTfBuilderIter->second.mStfQueue->push(std::move(lStfNode.mapped()));
 
+    // monitoring
+    {
+      using hres_clock = std::chrono::high_resolution_clock;
+      static decltype(hres_clock::now()) sStfStartTime = hres_clock::now();
+      const auto lDuration = std::chrono::duration<double>(hres_clock::now() - sStfStartTime);
+      DDMON("stfsender", "stf_output.stf_id", lStfId);
+      DDMON("stfsender", "stf_output.stf_rate", (1.0 / lDuration.count()));
+      DDMON("stfsender", "stf_output.stf_size", lStfSize);
+      sStfStartTime = hres_clock::now();
+    }
+
     if (lTfBuilderIter->second.mStfQueue->size() > 50) {
       WDDLOG_RL(1000, "SendToTfBuilder: STF queue backlog. queue_size={}", lTfBuilderIter->second.mStfQueue->size());
     }
@@ -390,11 +398,18 @@ void StfSenderOutput::DataHandlerThread(const std::string pTfBuilderId)
       mCounters.mBuffered.mCnt -= 1;
       mCounters.mInSending.mSize -= lStfSize;
       mCounters.mInSending.mCnt -= 1;
+      mCounters.mTotalSent.mSize += lStfSize;
+      mCounters.mTotalSent.mCnt += 1;
 
       if (mCounters.mInSending.mCnt > 100) {
         DDDLOG_RL(2000, "DataHandlerThread: Number of buffered STFs. tfb_id={} num_stfs={} num_stf_total={} size_stf_total={}",
           pTfBuilderId, lInputStfQueue->size(), mCounters.mInSending.mCnt, mCounters.mInSending.mSize);
       }
+
+      DDMON("stfsender", "stf_output.sent_count", mCounters.mTotalSent.mCnt);
+      DDMON("stfsender", "stf_output.sent_size", mCounters.mTotalSent.mSize);
+      DDMON("stfsender", "buffered.stf_cnt", mCounters.mBuffered.mCnt);
+      DDMON("stfsender", "buffered.stf_size", mCounters.mBuffered.mSize);
     }
   }
 
@@ -426,6 +441,9 @@ void StfSenderOutput::StfDropThread()
       std::scoped_lock lLock(mScheduledStfMapLock);
       mCounters.mBuffered.mSize -= lStfSize;
       mCounters.mBuffered.mCnt -= 1;
+
+      DDMON("stfsender", "buffered.stf_size", mCounters.mBuffered.mSize);
+      DDMON("stfsender", "buffered.stf_cnt", mCounters.mBuffered.mCnt);
     }
   }
 

src/StfSender/StfSenderOutput.h

@@ -44,6 +44,12 @@ class StfSenderOutput
       std::uint64_t mSize = 0;
       std::uint32_t mCnt = 0;
     } mInSending;
+
+    // total sent
+    struct alignas(128) {
+      std::uint64_t mSize = 0;
+      std::uint64_t mCnt = 0;
+    } mTotalSent;
   };
 
   StfSenderOutput() = delete;
@@ -74,6 +80,13 @@ class StfSenderOutput
     return mCounters;
   }
 
+  StdSenderOutputCounters resetCounters() {
+    std::scoped_lock lLock(mScheduledStfMapLock);
+    StdSenderOutputCounters lRet = mCounters;
+    mCounters = StdSenderOutputCounters();
+    return lRet;
+  }
+
  private:
   /// Ref to the main SubTimeBuilder O2 device
   StfSenderDevice& mDevice;
@@ -100,7 +113,7 @@ class StfSenderOutput
   mutable std::mutex mOutputMapLock;
     std::map<std::string, OutputChannelObjects> mOutputMap;
 
-  // Buffer mainenance
+  // Buffer maintenance
   std::uint64_t mBufferSize = std::uint64_t(32) << 30;
   ConcurrentFifo<std::unique_ptr<SubTimeFrame>> mDropQueue;
   std::thread mStfDropThread;

src/StfSender/runStfSenderDevice.cxx

@@ -13,6 +13,7 @@
 #include <SubTimeFrameFileSink.h>
 #include <Config.h>
 #include <FmqUtilities.h>
+#include <DataDistMonitoring.h>
 
 #include <options/FairMQProgOptions.h>
 
@@ -36,6 +37,8 @@ int main(int argc, char* argv[])
 
       // Add InfoLogger Options
       r.fConfig.AddToCmdLineOptions(impl::DataDistLoggerCtx::getProgramOptions());
+      // Add Monitoring Options
+      r.fConfig.AddToCmdLineOptions(DataDistMonitor::getProgramOptions());
 
       // StfSender options
       bpo::options_description lStfSenderOptions("StfSender options", 120);
@@ -57,7 +60,6 @@ int main(int argc, char* argv[])
 
       // Add options for STF file sink
       r.fConfig.AddToCmdLineOptions(o2::DataDistribution::SubTimeFrameFileSink::getProgramOptions());
-
     });
 
     runner.AddHook<InstantiateDevice>([](DeviceRunner& r){

src/TfBuilder/CMakeLists.txt

@@ -9,7 +9,7 @@ set(EXE_TFB_SOURCES
 )
 
 add_library(TfBuilder_lib OBJECT ${EXE_TFB_SOURCES})
-target_link_libraries(TfBuilder_lib base fmqtools common discovery)
+target_link_libraries(TfBuilder_lib base fmqtools common discovery monitoring)
 
 add_executable(TfBuilder)
 
@@ -22,7 +22,7 @@ endif()
 target_link_libraries(TfBuilder
   PRIVATE
     TfBuilder_lib
-    base fmqtools common discovery
+    base fmqtools common discovery monitoring
 )
 
 install(TARGETS TfBuilder RUNTIME DESTINATION bin)