Merge branch 'AliceO2Group:dev' into new-detector4

Author: pkurash

.github/workflows/reports.yml

@@ -17,12 +17,12 @@ jobs:
     if: github.repository == 'AliceO2Group/AliceO2'
 
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
     - name: Set up Python 3.10
       uses: actions/setup-python@v5
       with:
         python-version: '3.10'
-    - uses: actions/cache@v2
+    - uses: actions/cache@v4
       name: Configure pip caching
       with:
         path: ~/.cache/pip

CCDB/include/CCDB/CCDBDownloader.h

@@ -47,6 +47,7 @@ struct HeaderObjectPair_t {
 
 typedef struct DownloaderRequestData {
   std::vector<std::string> hosts;
+  std::vector<std::string> locations;
   std::string path;
   long timestamp;
   HeaderObjectPair_t hoPair;
@@ -231,12 +232,13 @@ class CCDBDownloader
   std::string prepareRedirectedURL(std::string address, std::string potentialHost) const;
 
   /**
-   * Returns a vector of possible content locations based on the redirect headers.
+   * Updates the locations vector with the the locations.
    *
-   * @param baseUrl Content path.
    * @param headerMap Map containing response headers.
+   * @param locations Location list to be updated.
+   * @param locIndex Index of the next locaiton to be tried.
    */
-  std::vector<std::string> getLocations(std::multimap<std::string, std::string>* headerMap) const;
+  void updateLocations(std::multimap<std::string, std::string>* headerMap, std::vector<std::string>* locations, int* locIndex) const;
 
   std::string mUserAgentId = "CCDBDownloader";
   /**

CCDB/include/CCDB/CcdbApi.h

@@ -388,7 +388,7 @@ class CcdbApi //: public DatabaseInterface
   static bool removeSemaphore(std::string const& name, bool remove = false);
   static void removeLeakingSemaphores(std::string const& basedir, bool remove = false);
 
-  void loadFileToMemory(o2::pmr::vector<char>& dest, const std::string& path, std::map<std::string, std::string>* localHeaders = nullptr) const;
+  void loadFileToMemory(o2::pmr::vector<char>& dest, const std::string& path, std::map<std::string, std::string>* localHeaders = nullptr, bool fetchLocalMetaData = true) const;
   void loadFileToMemory(o2::pmr::vector<char>& dest, std::string const& path,
                         std::map<std::string, std::string> const& metadata, long timestamp,
                         std::map<std::string, std::string>* headers, std::string const& etag,

CCDB/src/CCDBDownloader.cxx

@@ -362,7 +362,7 @@ void CCDBDownloader::tryNewHost(PerformData* performData, CURL* easy_handle)
 {
   auto requestData = performData->requestData;
   std::string newUrl = requestData->hosts.at(performData->hostInd) + "/" + requestData->path + "/" + std::to_string(requestData->timestamp);
-  LOG(debug) << "Connecting to another host " << newUrl;
+  LOG(debug) << "Connecting to another host " << newUrl << "\n";
   requestData->hoPair.header.clear();
   curl_easy_setopt(easy_handle, CURLOPT_URL, newUrl.c_str());
   mHandlesToBeAdded.push_back(easy_handle);
@@ -374,9 +374,11 @@ void CCDBDownloader::getLocalContent(PerformData* performData, std::string& newL
   LOG(debug) << "Redirecting to local content " << newLocation << "\n";
   if (requestData->localContentCallback(newLocation)) {
     contentRetrieved = true;
+    LOG(debug) << "Local content retrieved succesfully: " << newLocation << " n";
   } else {
     // Prepare next redirect url
     newLocation = getNewLocation(performData, locations);
+    LOG(debug) << "Failed to retrieve local content: " << newLocation << "\n";
   }
 }
 
@@ -396,15 +398,15 @@ std::string CCDBDownloader::getNewLocation(PerformData* performData, std::vector
 void CCDBDownloader::httpRedirect(PerformData* performData, std::string& newLocation, CURL* easy_handle)
 {
   auto requestData = performData->requestData;
-  LOG(debug) << "Trying content location " << newLocation;
+  LOG(debug) << "Trying content location " << newLocation << "\n";
   curl_easy_setopt(easy_handle, CURLOPT_URL, newLocation.c_str());
   mHandlesToBeAdded.push_back(easy_handle);
 }
 
 void CCDBDownloader::followRedirect(PerformData* performData, CURL* easy_handle, std::vector<std::string>& locations, bool& rescheduled, bool& contentRetrieved)
 {
   std::string newLocation = getNewLocation(performData, locations);
-  if (newLocation.find("alien:/", 0) != std::string::npos || newLocation.find("file:/", 0) != std::string::npos) {
+  while (!contentRetrieved && (newLocation.find("alien:/", 0) != std::string::npos || newLocation.find("file:/", 0) != std::string::npos)) {
     getLocalContent(performData, newLocation, contentRetrieved, locations);
   }
   if (!contentRetrieved && newLocation != "") {
@@ -508,17 +510,17 @@ void CCDBDownloader::transferFinished(CURL* easy_handle, CURLcode curlCode)
       std::string currentHost = requestData->hosts[performData->hostInd];
       std::string loggingMessage = prepareLogMessage(currentHost, requestData->userAgent, requestData->path, requestData->timestamp, requestData->headers, httpCode);
 
-      // Get alternative locations for the same host
-      auto locations = getLocations(&(requestData->hoPair.header));
+      // Get new locations based on received headers
+      updateLocations(&(requestData->hoPair.header), &requestData->locations, &performData->locInd);
 
       // React to received http code
       if (200 <= httpCode && httpCode < 400) {
         LOG(debug) << loggingMessage;
         if (304 == httpCode) {
           LOGP(debug, "Object exists but I am not serving it since it's already in your possession");
           contentRetrieved = true;
-        } else if (300 <= httpCode && httpCode < 400 && performData->locInd < locations.size()) {
-          followRedirect(performData, easy_handle, locations, rescheduled, contentRetrieved);
+        } else if (300 <= httpCode && httpCode < 400 && performData->locInd < requestData->locations.size()) {
+          followRedirect(performData, easy_handle, requestData->locations, rescheduled, contentRetrieved);
         } else if (200 <= httpCode && httpCode < 300) {
           contentRetrieved = true; // Can be overruled by following error check
         }
@@ -531,8 +533,16 @@ void CCDBDownloader::transferFinished(CURL* easy_handle, CURLcode curlCode)
         contentRetrieved = false;
       }
 
-      // Check if content was retrieved, or scheduled to be retrieved
-      if (!rescheduled && !contentRetrieved && performData->locInd == locations.size()) {
+      // Check if content was retrieved or scheduled to be retrieved
+      if (!rescheduled && !contentRetrieved) {
+        // Current location failed without providing 3xx http code, try next redirect for the same host
+        if (performData->locInd < requestData->locations.size()) {
+          followRedirect(performData, easy_handle, requestData->locations, rescheduled, contentRetrieved);
+        }
+      }
+
+      // Check again because content might have been retrieved or rescheduled via a redirect
+      if (!rescheduled && !contentRetrieved) {
         // Ran out of locations to redirect, try new host
         if (++performData->hostInd < requestData->hosts.size()) {
           tryNewHost(performData, easy_handle);
@@ -650,24 +660,37 @@ CURLcode CCDBDownloader::perform(CURL* handle)
   return batchBlockingPerform(handleVector).back();
 }
 
-std::vector<std::string> CCDBDownloader::getLocations(std::multimap<std::string, std::string>* headerMap) const
+void CCDBDownloader::updateLocations(std::multimap<std::string, std::string>* headerMap, std::vector<std::string>* locations, int* locIndex) const
 {
-  std::vector<std::string> locs;
+  std::vector<std::string> newLocations;
+
   auto iter = headerMap->find("Location");
   if (iter != headerMap->end()) {
-    locs.push_back(iter->second);
+    auto range = headerMap->equal_range("Location");
+    for (auto it = range.first; it != range.second; ++it) {
+      if (std::find(locations->begin(), locations->end(), it->second) == locations->end()) {
+        if (std::find(newLocations.begin(), newLocations.end(), it->second) == newLocations.end()) {
+          newLocations.push_back(it->second);
+        }
+      }
+    }
   }
+
   // add alternative locations (not yet included)
   auto iter2 = headerMap->find("Content-Location");
   if (iter2 != headerMap->end()) {
     auto range = headerMap->equal_range("Content-Location");
     for (auto it = range.first; it != range.second; ++it) {
-      if (std::find(locs.begin(), locs.end(), it->second) == locs.end()) {
-        locs.push_back(it->second);
+      if (std::find(locations->begin(), locations->end(), it->second) == locations->end()) {
+        if (std::find(newLocations.begin(), newLocations.end(), it->second) == newLocations.end()) {
+          newLocations.push_back(it->second);
+        }
       }
     }
   }
-  return locs;
+
+  // Insert location list at the current location index. This assures that the provided locations will be tried first.
+  locations->insert(locations->begin() + (*locIndex), newLocations.begin(), newLocations.end());
 }
 
 std::vector<CURLcode> CCDBDownloader::batchBlockingPerform(std::vector<CURL*> const& handleVector)

CCDB/src/CcdbApi.cxx

@@ -667,6 +667,23 @@ size_t header_map_callback(char* buffer, size_t size, size_t nitems, void* userd
         }
       }
     }
+
+    // Keep only the first ETag encountered
+    if (key == "ETag") {
+      auto cl = headers->find("ETag");
+      if (cl != headers->end()) {
+        insert = false;
+      }
+    }
+
+    // Keep only the first Content-Type encountered
+    if (key == "Content-Type") {
+      auto cl = headers->find("Content-Type");
+      if (cl != headers->end()) {
+        insert = false;
+      }
+    }
+
     if (insert) {
       headers->insert(std::make_pair(key, value));
     }
@@ -1971,20 +1988,32 @@ void CcdbApi::vectoredLoadFileToMemory(std::vector<RequestContext>& requestConte
 bool CcdbApi::loadLocalContentToMemory(o2::pmr::vector<char>& dest, std::string& url) const
 {
   if (url.find("alien:/", 0) != std::string::npos) {
-    loadFileToMemory(dest, url, nullptr); // headers loaded from the file in case of the snapshot reading only
-    return true;
+    std::map<std::string, std::string> localHeaders;
+    loadFileToMemory(dest, url, &localHeaders, false);
+    auto it = localHeaders.find("Error");
+    if (it != localHeaders.end() && it->second == "An error occurred during retrieval") {
+      return false;
+    } else {
+      return true;
+    }
   }
   if ((url.find("file:/", 0) != std::string::npos)) {
     std::string path = url.substr(7);
     if (std::filesystem::exists(path)) {
-      loadFileToMemory(dest, path, nullptr);
-      return true;
+      std::map<std::string, std::string> localHeaders;
+      loadFileToMemory(dest, url, &localHeaders, o2::utils::Str::endsWith(path, ".root"));
+      auto it = localHeaders.find("Error");
+      if (it != localHeaders.end() && it->second == "An error occurred during retrieval") {
+        return false;
+      } else {
+        return true;
+      }
     }
   }
   return false;
 }
 
-void CcdbApi::loadFileToMemory(o2::pmr::vector<char>& dest, const std::string& path, std::map<std::string, std::string>* localHeaders) const
+void CcdbApi::loadFileToMemory(o2::pmr::vector<char>& dest, const std::string& path, std::map<std::string, std::string>* localHeaders, bool fetchLocalMetaData) const
 {
   // Read file to memory as vector. For special case of the locally cached file retriev metadata stored directly in the file
   constexpr size_t MaxCopySize = 0x1L << 25;
@@ -2032,7 +2061,7 @@ void CcdbApi::loadFileToMemory(o2::pmr::vector<char>& dest, const std::string& p
     totalread += nread;
   } while (nread == (long)MaxCopySize);
 
-  if (localHeaders) {
+  if (localHeaders && fetchLocalMetaData) {
     TMemFile memFile("name", const_cast<char*>(dest.data()), dest.size(), "READ");
     auto storedmeta = (std::map<std::string, std::string>*)extractFromTFile(memFile, TClass::GetClass("std::map<std::string, std::string>"), CCDBMETA_ENTRY);
     if (storedmeta) {

CODEOWNERS

@@ -29,7 +29,7 @@
 /DataFormats/Detectors/CPV                     @peressounko @kharlov
 /DataFormats/Detectors/CTP                     @lietava
 /DataFormats/Detectors/EMCAL                   @mfasDa @jokonig
-/DataFormats/Detectors/FIT                     @jotwinow @afurs @andreasmolander @arvindkhuntia @mslupeck
+/DataFormats/Detectors/FIT                     @jotwinow @afurs @andreasmolander @sahilupadhyaya92
 /DataFormats/Detectors/FOCAL                   @maxrauch @mfasDa @iarsene @matthiasrichter
 /DataFormats/Detectors/GlobalTracking          @shahor02
 /DataFormats/Detectors/GlobalTrackingWorkflow  @shahor02
@@ -59,7 +59,7 @@
 /Detectors/Calibration             @chiarazampolli @shahor02
 /Detectors/CPV                     @peressounko @kharlov
 /Detectors/EMCAL                   @mfasDa @jokonig
-/Detectors/FIT                     @jotwinow @afurs @andreasmolander @arvindkhuntia @mslupeck
+/Detectors/FIT                     @jotwinow @afurs @andreasmolander @sahilupadhyaya92
 /Detectors/FOCAL                   @maxrauch @mfasDa @iarsene @matthiasrichter
 /Detectors/Geometry                @sawenzel @shahor02
 /Detectors/GlobalTracking          @shahor02

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

@@ -47,7 +47,7 @@ class CTPRunManager : public ctpCCDBManager
   CTPRunManager() = default;
   void init();
   int loadRun(const std::string& cfg);
-  int startRun(const std::string& cfg);
+  int setRunConfigBK(uint32_t runNumber, const std::string& cfg);
   int stopRun(uint32_t irun, long timeStamp);
   int addScalers(uint32_t irun, std::time_t time, bool start = 0);
   int processMessage(std::string& topic, const std::string& message);

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

@@ -27,12 +27,11 @@ class ctpCCDBManager
   int saveRunScalersToCCDB(CTPRunScalers& scalers, long timeStart, long timeStop);
   int saveRunScalersToQCDB(CTPRunScalers& scalers, long timeStart, long timeStop);
   int saveRunConfigToCCDB(CTPConfiguration* cfg, long timeStart);
+  int saveSoxOrbit(uint32_t runNumber, uint32_t soxOrbit, long timeStart);
+  int saveOrbitReset(long timeStamp);
   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; };
 
@@ -42,9 +41,11 @@ class ctpCCDBManager
   // 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";
-  std::string mQCDBPathCTPScalers = "qc/CTP/Scalers";
+  const std::string mCCDBPathCTPScalers = "CTP/Calib/Scalers";
+  // std::string mCCDBPathCTPConfig = "CTP/Config/Config";  - in Configuration.h
+  const std::string mQCDBPathCTPScalers = "qc/CTP/Scalers";
+  const std::string mCCDBPathSoxOrbit = "CTP/Calib/FirstRunOrbit";
+  const std::string mCCDBPathOrbitReset = "CTP/Calib/OrbitReset";
   ClassDefNV(ctpCCDBManager, 1);
 };
 } // namespace ctp

Detectors/CTP/workflowScalers/src/RunManager.cxx

@@ -57,7 +57,7 @@ int CTPActiveRun::send2BK(std::unique_ptr<BkpClient>& BKClient, size_t ts, bool
     std::string clsname = cfg.getClassNameFromHWIndex(cls.first);
     // clsname = std::to_string(runOri) + "_" + clsname;
     try {
-      BKClient->triggerCounters()->createOrUpdateForRun(runNumber, clsname, ts, cntsbk[0], cntsbk[1], cntsbk[2], cntsbk[3], cntsbk[4], cntsbk[5]);
+      BKClient->ctpTriggerCounters()->createOrUpdateForRun(runNumber, clsname, ts, cntsbk[0], cntsbk[1], cntsbk[2], cntsbk[3], cntsbk[4], cntsbk[5]);
     } catch (std::runtime_error& error) {
       std::cerr << "An error occurred: " << error.what() << std::endl;
       return 1;
@@ -124,8 +124,19 @@ int CTPRunManager::loadRun(const std::string& cfg)
 
   return 0;
 }
-int CTPRunManager::startRun(const std::string& cfg)
+int CTPRunManager::setRunConfigBK(uint32_t runNumber, const std::string& cfg)
 {
+  std::cout << "Printing cfg:" << cfg << std::endl;
+  if (mBKClient) {
+    try {
+      uint32_t runNumber = 1;
+      mBKClient->run()->setRawCtpTriggerConfiguration(runNumber, cfg);
+    } catch (std::runtime_error& error) {
+      std::cerr << "An error occurred: " << error.what() << std::endl;
+      return 1;
+    }
+    LOG(info) << "Run BK:" << runNumber << " CFG:" << cfg;
+  }
   return 0;
 }
 int CTPRunManager::stopRun(uint32_t irun, long timeStamp)
@@ -221,6 +232,13 @@ int CTPRunManager::processMessage(std::string& topic, const std::string& message
     loadRun(message);
     return 0;
   }
+  if (topic.find("soxorbit") != std::string::npos) {
+    return 0;
+  }
+  if (topic.find("orbitreset") != std::string::npos) {
+    return 0;
+  }
+  static int nerror = 0;
   if (topic.find("sox") != std::string::npos) {
     // get config
     size_t irun = message.find("run");
@@ -230,17 +248,15 @@ int CTPRunManager::processMessage(std::string& topic, const std::string& message
     }
     LOG(info) << "SOX received, Run keyword position:" << irun;
     std::string cfg = message.substr(irun, message.size() - irun);
-    startRun(cfg);
     firstcounters = message.substr(0, irun);
-  }
-  if (topic.find("eox") != std::string::npos) {
+  } else if (topic.find("eox") != std::string::npos) {
     LOG(info) << "EOX received";
     mEOX = 1;
-  }
-  static int nerror = 0;
-  if (topic == "rocnts") {
-    if (nerror < 1) {
-      LOG(warning) << "Skipping topic rocnts";
+  } else if (topic.find("cnts") != std::string::npos) {
+    // just continue
+  } else {
+    if (nerror < 10) {
+      LOG(warning) << "Skipping topic:" << topic;
       nerror++;
     }
     return 0;
@@ -293,6 +309,7 @@ int CTPRunManager::processMessage(std::string& topic, const std::string& message
         mActiveRunNumbers[i] = mCounters[i];
         mActiveRuns[i] = run->second;
         mRunsLoaded.erase(run);
+        setRunConfigBK(mActiveRuns[i]->cfg.getRunNumber(), mActiveRuns[i]->cfg.getConfigString());
         addScalers(i, tt, 1);
         saveRunScalersToQCDB(mActiveRuns[i]->scalers, tt * 1000, tt * 1000);
       } else {