Ctpdev: counters to QCDB (#13350)

* dev:scalers to qcdb

* clang

Author: lietava

DataFormats/Detectors/CTP/src/Configuration.cxx

@@ -510,8 +510,8 @@ int CTPConfiguration::processConfigurationLineRun3(std::string& line, int& level
 }
 int CTPConfiguration::processConfigurationLineRun3v2(std::string& line, int& level, std::map<int, std::vector<int>>& descInputsIndex)
 {
-  LOG(info) << "Processing line";
-  LOG(info) << "line:" << line << " lev:" << level;
+  LOG(debug) << "Processing line";
+  LOG(debug) << "line:" << line << " lev:" << level;
   //
   std::vector<std::string> tokens = o2::utils::Str::tokenize(line, ' ');
   size_t ntokens = tokens.size();
@@ -557,7 +557,7 @@ int CTPConfiguration::processConfigurationLineRun3v2(std::string& line, int& lev
       level = UNKNOWN;
     }
   }
-  LOG(info) << "Level:" << level;
+  LOG(debug) << "Level:" << level;
   switch (level) {
     case VERSION: {
       break;
@@ -585,7 +585,7 @@ int CTPConfiguration::processConfigurationLineRun3v2(std::string& line, int& lev
       uint32_t index = std::stoul(tokens[2]);
       ctpinp.inputMask = (1ull << (index - 1));
       mInputs.push_back(ctpinp);
-      LOG(info) << "Input:" << ctpinp.name << " index:" << index;
+      LOG(debug) << "Input:" << ctpinp.name << " index:" << index;
       break;
     }
     case MASKS: {
@@ -596,15 +596,15 @@ int CTPConfiguration::processConfigurationLineRun3v2(std::string& line, int& lev
       }
       bcmask.setBCmask(tokens);
       mBCMasks.push_back(bcmask);
-      LOG(info) << "BC mask added:" << bcmask.name;
+      LOG(debug) << "BC mask added:" << bcmask.name;
       break;
     }
     case GENS: {
       CTPGenerator gen;
       gen.name = tokens[0];
       gen.frequency = tokens[1];
       mGenerators.push_back(gen);
-      LOG(info) << "Gen added:" << line;
+      LOG(debug) << "Gen added:" << line;
       break;
     }
     case DESCRIPTORS: {
@@ -630,9 +630,9 @@ int CTPConfiguration::processConfigurationLineRun3v2(std::string& line, int& lev
       o2::detectors::DetID det(detname.c_str());
       if (isDetector(det)) {
         ctpdet.detID = det.getID();
-        LOG(info) << "Detector found:" << det.getID() << " " << detname;
+        LOG(debug) << "Detector found:" << det.getID() << " " << detname;
       } else {
-        LOG(info) << "Unknown detectors:" << line;
+        LOG(error) << "Unknown detectors:" << line;
       }
       mDetectors.push_back(ctpdet);
       level = LTGitems;
@@ -642,18 +642,18 @@ int CTPConfiguration::processConfigurationLineRun3v2(std::string& line, int& lev
       if (ntokens == 1) {
         mDetectors.back().mode = tokens[0];
       }
-      LOG(info) << "LTGitem:" << line;
+      LOG(debug) << "LTGitem:" << line;
       break;
     }
     case CLUSTER: {
       CTPCluster cluster;
       try {
         cluster.hwMask = std::stoull(tokens[0]);
       } catch (...) {
-        LOG(info) << "Cluster syntax error:" << line;
+        LOG(error) << "Cluster syntax error:" << line;
         return level;
       }
-      LOG(info) << "Cluster:" << line;
+      LOG(debug) << "Cluster:" << line;
       cluster.name = tokens[2];
       o2::detectors::DetID::mask_t mask;
       for (uint32_t item = 3; item < ntokens; item++) {
@@ -680,10 +680,10 @@ int CTPConfiguration::processConfigurationLineRun3v2(std::string& line, int& lev
       try {
         index = std::stoull(tokens[1]);
       } catch (...) {
-        LOG(info) << "Class syntax error:" << line;
+        LOG(error) << "Class syntax error:" << line;
         return level;
       }
-      LOG(info) << "Class:" << line;
+      LOG(debug) << "Class:" << line;
       CTPClass cls;
       cls.classMask = 1ull << index;
       cls.name = tokens[0];
@@ -716,7 +716,7 @@ int CTPConfiguration::processConfigurationLineRun3v2(std::string& line, int& lev
       break;
     }
     default: {
-      LOG(info) << "unknown line:" << line;
+      LOG(warning) << "unknown line:" << line;
     }
   }
   return 0;

Detectors/CTP/workflowScalers/include/CTPWorkflowScalers/RunManager.h

@@ -38,6 +38,8 @@ struct CTPActiveRun {
   counters_t cntslast0; // last minus one read counters needed for overflow correction
   counters_t cntslast;  // last read counters
   counters64_t overflows;
+  // QC
+  int qcwpcount = 0;
 };
 class CTPRunManager : public ctpCCDBManager
 {
@@ -53,6 +55,7 @@ class CTPRunManager : public ctpCCDBManager
   int loadScalerNames();
   int getNRuns();
   void setBKHost(std::string host) { mBKHost = host; };
+  void setQCWritePeriod(int period) { mQCWritePeriod = period; };
   uint64_t checkOverflow(uint32_t lcnt0, uint32_t lcnt1, uint64_t lcntcor);
   void printCounters();
 
@@ -67,8 +70,8 @@ class CTPRunManager : public ctpCCDBManager
   std::unique_ptr<BkpClient> mBKClient;
   int mEOX = 0; // redundancy check
   int mNew = 1; // 1 - no CCDB: used for QC
-
-  ClassDefNV(CTPRunManager, 6);
+  int mQCWritePeriod = 3; // Time in 10secs between two writes to QCCD
+  ClassDefNV(CTPRunManager, 7);
 };
 } // namespace ctp
 } // namespace o2

Detectors/CTP/workflowScalers/include/CTPWorkflowScalers/ctpCCDBManager.h

@@ -25,22 +25,27 @@ class ctpCCDBManager
  public:
   ctpCCDBManager() = default;
   int saveRunScalersToCCDB(CTPRunScalers& scalers, long timeStart, long timeStop);
+  int saveRunScalersToQCDB(CTPRunScalers& scalers, long timeStart, long timeStop);
   int saveRunConfigToCCDB(CTPConfiguration* cfg, long timeStart);
   static CTPConfiguration getConfigFromCCDB(long timestamp, std::string run, bool& ok);
   static CTPConfiguration getConfigFromCCDB(long timestamp, std::string run);
   CTPRunScalers getScalersFromCCDB(long timestamp, std::string, bool& ok);
   void setCCDBPathConfig(std::string path) { mCCDBPathCTPConfig = path; };
   void setCCDBPathScalers(std::string path) { mCCDBPathCTPScalers = path; };
+  void setQCDBPathScalers(std::string path) { mQCDBPathCTPScalers = path; };
   static void setCCDBHost(std::string host) { mCCDBHost = host; };
+  static void setQCDBHost(std::string host) { mQCDBHost = host; };
 
  protected:
   /// Database constants
   // std::string mCCDBHost = "http://ccdb-test.cern.ch:8080";
+  // std::string mQCDBHost = "http://ali-qcdb.cern.ch:8083";
   static std::string mCCDBHost;
+  static std::string mQCDBHost;
   std::string mCCDBPathCTPScalers = "CTP/Calib/Scalers";
   std::string mCCDBPathCTPConfig = "CTP/Config/Config";
-
-  ClassDefNV(ctpCCDBManager, 0);
+  std::string mQCDBPathCTPScalers = "qc/CTP/Scalers";
+  ClassDefNV(ctpCCDBManager, 1);
 };
 } // namespace ctp
 } // namespace o2

Detectors/CTP/workflowScalers/src/RunManager.cxx

@@ -81,8 +81,11 @@ void CTPRunManager::init()
   } else {
     LOG(info) << "BK not sent";
   }
+  setQCDBHost(mQCDBHost);
+  LOG(info) << "QC host:" << mQCDBHost;
+  LOG(info) << "QCDB writing every:" << mQCWritePeriod << " 10 secs";
   LOG(info) << "CCDB host:" << mCCDBHost;
-  LOG(info) << "CTP vNew:" << mNew;
+  LOG(info) << "CTP vNew cfg:" << mNew;
   LOG(info) << "CTPRunManager initialised.";
 }
 int CTPRunManager::loadRun(const std::string& cfg)
@@ -136,6 +139,7 @@ int CTPRunManager::stopRun(uint32_t irun, long timeStamp)
   // const long timeStamp = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
   mActiveRuns[irun]->timeStop = timeStamp * 1000.;
   saveRunScalersToCCDB(mActiveRuns[irun]->scalers, mActiveRuns[irun]->timeStart, mActiveRuns[irun]->timeStop);
+  saveRunScalersToQCDB(mActiveRuns[irun]->scalers, mActiveRuns[irun]->timeStart, mActiveRuns[irun]->timeStop);
   delete mActiveRuns[irun];
   mActiveRuns[irun] = nullptr;
   return 0;
@@ -160,7 +164,7 @@ int CTPRunManager::addScalers(uint32_t irun, std::time_t time, bool start)
     std::string c1a = "cla1a" + std::to_string(cls + 1);
     CTPScalerRaw scalraw;
     scalraw.classIndex = (uint32_t)cls;
-    std::cout << "cls:" << cls << " " << scalraw.classIndex << std::endl;
+    // std::cout << "cls:" << cls << " " << scalraw.classIndex << std::endl;
     scalraw.lmBefore = mCounters[mScalerName2Position[cmb]];
     scalraw.lmAfter = mCounters[mScalerName2Position[cma]];
     scalraw.l0Before = mCounters[mScalerName2Position[c0b]];
@@ -276,6 +280,12 @@ int CTPRunManager::processMessage(std::string& topic, const std::string& message
       // active , do scalers
       LOG(info) << "Run continue:" << mCounters[i];
       addScalers(i, tt);
+      // LOG(info) << " QC period:" << mActiveRunNumbers[i] << " " << mActiveRuns[i]->qcwpcount << " " << mQCWritePeriod;
+      if (mActiveRuns[i]->qcwpcount > mQCWritePeriod) {
+        saveRunScalersToQCDB(mActiveRuns[i]->scalers, tt * 1000, tt * 1000);
+        mActiveRuns[i]->qcwpcount = 0;
+      }
+      mActiveRuns[i]->qcwpcount++;
     } else if ((mCounters[i] != 0) && (mActiveRunNumbers[i] == 0)) {
       LOG(info) << "Run started:" << mCounters[i];
       auto run = mRunsLoaded.find(mCounters[i]);
@@ -284,6 +294,7 @@ int CTPRunManager::processMessage(std::string& topic, const std::string& message
         mActiveRuns[i] = run->second;
         mRunsLoaded.erase(run);
         addScalers(i, tt, 1);
+        saveRunScalersToQCDB(mActiveRuns[i]->scalers, tt * 1000, tt * 1000);
       } else {
         LOG(error) << "Trying to start run which is not loaded:" << mCounters[i];
       }

Detectors/CTP/workflowScalers/src/ctp-proxy.cxx

@@ -46,11 +46,13 @@
 #include "BookkeepingApi/BkpClient.h"
 using namespace o2::framework;
 using DetID = o2::detectors::DetID;
-InjectorFunction dcs2dpl(std::string& ccdbhost, std::string& bkhost)
+InjectorFunction dcs2dpl(std::string& ccdbhost, std::string& bkhost, std::string& qchost, int qcwriteperiod)
 {
   auto runMgr = std::make_shared<o2::ctp::CTPRunManager>();
   runMgr->setCCDBHost(ccdbhost);
   runMgr->setBKHost(bkhost);
+  runMgr->setQCDBHost(qchost);
+  runMgr->setQCWritePeriod(qcwriteperiod);
   runMgr->init();
   // runMgr->setClient(client);
   return [runMgr](TimingInfo&, ServiceRegistryRef const& services, fair::mq::Parts& parts, ChannelRetriever channelRetriever, size_t newTimesliceId, bool& stop) -> bool {
@@ -75,6 +77,8 @@ void customize(std::vector<ConfigParamSpec>& workflowOptions)
   workflowOptions.push_back(ConfigParamSpec{"subscribe-to", VariantType::String, "type=sub,method=connect,address=tcp://188.184.30.57:5556,rateLogging=10,transport=zeromq", {"channel subscribe to"}});
   workflowOptions.push_back(ConfigParamSpec{"ccdb-host", VariantType::String, "http://o2-ccdb.internal:8080", {"ccdb host"}});
   workflowOptions.push_back(ConfigParamSpec{"bk-host", VariantType::String, "none", {"bk host"}});
+  workflowOptions.push_back(ConfigParamSpec{"qc-host", VariantType::String, "none", {"qc host"}});
+  workflowOptions.push_back(ConfigParamSpec{"qc-writeperiod", VariantType::Int, 30, {"Period of writing to QCDB in units of 10secs, default = 30 (5 mins)"}});
 }
 
 #include "Framework/runDataProcessing.h"
@@ -98,6 +102,8 @@ WorkflowSpec defineDataProcessing(ConfigContext const& config)
   auto chan = config.options().get<std::string>("subscribe-to");
   std::string ccdbhost = config.options().get<std::string>("ccdb-host");
   std::string bkhost = config.options().get<std::string>("bk-host");
+  std::string qchost = config.options().get<std::string>("qc-host");
+  int qcwriteperiod = config.options().get<int>("qc-writeperiod");
   if (chan.empty()) {
     throw std::runtime_error("input channel is not provided");
   }
@@ -112,7 +118,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const& config)
     std::move(ctpCountersOutputs),
     // this is just default, can be overriden by --ctp-config-proxy '--channel-config..'
     chan.c_str(),
-    dcs2dpl(ccdbhost, bkhost));
+    dcs2dpl(ccdbhost, bkhost, qchost, qcwriteperiod));
   ctpProxy.labels.emplace_back(DataProcessorLabel{"input-proxy"});
   LOG(info) << "===> Proxy done";
   WorkflowSpec workflow;

Detectors/CTP/workflowScalers/src/ctpCCDBManager.cxx

@@ -21,6 +21,8 @@
 #include <fairlogger/Logger.h>
 using namespace o2::ctp;
 std::string ctpCCDBManager::mCCDBHost = "http://o2-ccdb.internal";
+std::string ctpCCDBManager::mQCDBHost = "http://ali-qcdb.cern.ch:8083";
+// std::string ctpCCDBManager::mQCDBHost = "none";
 //
 int ctpCCDBManager::saveRunScalersToCCDB(CTPRunScalers& scalers, long timeStart, long timeStop)
 {
@@ -43,7 +45,39 @@ int ctpCCDBManager::saveRunScalersToCCDB(CTPRunScalers& scalers, long timeStart,
   api.init(mCCDBHost.c_str()); // or http://localhost:8080 for a local installation
   // store abitrary user object in strongly typed manner
   int ret = api.storeAsTFileAny(&(scalers), mCCDBPathCTPScalers, metadata, tmin, tmax);
-  LOG(info) << "CTP scalers saved in ccdb:" << mCCDBHost << " run:" << scalers.getRunNumber() << " tmin:" << tmin << " tmax:" << tmax;
+  if (ret == 0) {
+    LOG(info) << "CTP scalers saved in ccdb:" << mCCDBHost << " run:" << scalers.getRunNumber() << " tmin:" << tmin << " tmax:" << tmax;
+  } else {
+    LOG(FATAL) << "Problem writing to database ret:" << ret;
+  }
+  return ret;
+}
+int ctpCCDBManager::saveRunScalersToQCDB(CTPRunScalers& scalers, long timeStart, long timeStop)
+{
+  // data base
+  if (mQCDBHost == "none") {
+    LOG(info) << "Scalers not written to QCDB none";
+    return 0;
+  }
+  // CTPActiveRun* run = mActiveRuns[i];q
+  using namespace std::chrono_literals;
+  std::chrono::seconds days3 = 259200s;
+  std::chrono::seconds min10 = 600s;
+  long time3days = std::chrono::duration_cast<std::chrono::milliseconds>(days3).count();
+  long time10min = std::chrono::duration_cast<std::chrono::milliseconds>(min10).count();
+  long tmin = timeStart - time10min;
+  long tmax = timeStop + time3days;
+  o2::ccdb::CcdbApi api;
+  map<string, string> metadata; // can be empty
+  metadata["runNumber"] = std::to_string(scalers.getRunNumber());
+  api.init(mQCDBHost.c_str()); // or http://localhost:8080 for a local installation
+  // store abitrary user object in strongly typed manner
+  int ret = api.storeAsTFileAny(&(scalers), mQCDBPathCTPScalers, metadata, tmin, tmax);
+  if (ret == 0) {
+    LOG(info) << "CTP scalers saved in qcdb:" << mQCDBHost << " run:" << scalers.getRunNumber() << " tmin:" << tmin << " tmax:" << tmax;
+  } else {
+    LOG(FATAL) << "CTP scalers Problem writing to database qcdb ret:" << ret;
+  }
   return ret;
 }
 int ctpCCDBManager::saveRunConfigToCCDB(CTPConfiguration* cfg, long timeStart)
@@ -66,7 +100,11 @@ int ctpCCDBManager::saveRunConfigToCCDB(CTPConfiguration* cfg, long timeStart)
   api.init(mCCDBHost.c_str()); // or http://localhost:8080 for a local installation
   // store abitrary user object in strongly typed manner
   int ret = api.storeAsTFileAny(cfg, CCDBPathCTPConfig, metadata, tmin, tmax);
-  LOG(info) << "CTP config  saved in ccdb:" << mCCDBHost << " run:" << cfg->getRunNumber() << " tmin:" << tmin << " tmax:" << tmax;
+  if (ret == 0) {
+    LOG(info) << "CTP config  saved in ccdb:" << mCCDBHost << " run:" << cfg->getRunNumber() << " tmin:" << tmin << " tmax:" << tmax;
+  } else {
+    LOG(FATAL) << "CTPConfig: Problem writing to database ret:" << ret;
+  }
   return ret;
 }
 CTPConfiguration ctpCCDBManager::getConfigFromCCDB(long timestamp, std::string run, bool& ok)