CcdbApi.h

Go to the documentation of this file.

// Copyright 2019-2020 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.
 
 
#ifndef PROJECT_CCDBAPI_H
#define PROJECT_CCDBAPI_H
 
#include <string>
#include <string_view>
#include <memory>
#include <map>
#include <TObject.h>
#include <TMessage.h>
#include "CCDB/CcdbObjectInfo.h"
#include <CommonUtils/ConfigurableParam.h>
#include <type_traits>
#include <vector>
 
#if !defined(__CINT__) && !defined(__MAKECINT__) && !defined(__ROOTCLING__) && !defined(__CLING__)
#include "MemoryResources/MemoryResources.h"
#include <boost/interprocess/sync/named_semaphore.hpp>
#include <TJAlienCredentials.h>
#else
class TJAlienCredentials;
#endif
 
// libcurl and the downloader are implementation details of CcdbApi.cxx;
// only opaque handles appear below, so neither header is needed here.
struct curl_slist;
 
class TFile;
#include <TGrid.h>
 
namespace o2
{
namespace ccdb
{
 
class CCDBQuery;
class CCDBDownloader;
 
using CurlHandle = void;
 
class CcdbApi //: public DatabaseInterface
{
 public:
  CcdbApi();
  virtual ~CcdbApi();
 
  // Delete copy and copy assignment constructor
  CcdbApi(const CcdbApi&) = delete;
  void operator=(const CcdbApi&) = delete;
 
  const std::string getUniqueAgentID() const { return mUniqueAgentID; }
 
  static bool checkAlienToken();
 
  void init(std::string const& hosts);
 
  std::string const& getURL() const { return mUrl; }
 
  bool isSnapshotMode() const { return mInSnapshotMode; }
 
  template <typename T>
  inline static std::unique_ptr<std::vector<char>> createObjectImage(const T* obj, CcdbObjectInfo* info = nullptr)
  {
    return createObjectImage(reinterpret_cast<const void*>(obj), typeid(T), info);
  }
 
  static std::unique_ptr<std::vector<char>> createObjectImage(const TObject* obj, CcdbObjectInfo* info = nullptr);
 
  static std::unique_ptr<std::vector<char>> createObjectImage(const void* obj, std::type_info const& tinfo, CcdbObjectInfo* info = nullptr);
 
  int storeAsTFile(const TObject* rootObject, std::string const& path, std::map<std::string, std::string> const& metadata,
                   long startValidityTimestamp = -1, long endValidityTimestamp = -1, std::vector<char>::size_type maxSize = 0 /*bytes*/) const;
 
  template <typename T>
  int storeAsTFileAny(const T* obj, std::string const& path, std::map<std::string, std::string> const& metadata,
                      long startValidityTimestamp = -1, long endValidityTimestamp = -1, std::vector<char>::size_type maxSize = 0 /*bytes*/) const
  {
    return storeAsTFile_impl(reinterpret_cast<const void*>(obj), typeid(T), path, metadata, startValidityTimestamp, endValidityTimestamp, maxSize);
  }
 
  // interface for storing TObject via storeAsTFileAny
  int storeAsTFileAny(const TObject* rootobj, std::string const& path, std::map<std::string, std::string> const& metadata,
                      long startValidityTimestamp = -1, long endValidityTimestamp = -1, std::vector<char>::size_type maxSize = 0 /*bytes*/) const
  {
    return storeAsTFile(rootobj, path, metadata, startValidityTimestamp, endValidityTimestamp, maxSize);
  }
 
  template <typename T>
  typename std::enable_if<!std::is_base_of<o2::conf::ConfigurableParam, T>::value, T*>::type
    retrieveFromTFileAny(std::string const& path, std::map<std::string, std::string> const& metadata,
                         long timestamp = -1, std::map<std::string, std::string>* headers = nullptr, std::string const& etag = "",
                         const std::string& createdNotAfter = "", const std::string& createdNotBefore = "") const;
 
  template <typename T>
  typename std::enable_if<std::is_base_of<o2::conf::ConfigurableParam, T>::value, T*>::type
    retrieveFromTFileAny(std::string const& path, std::map<std::string, std::string> const& metadata,
                         long timestamp = -1, std::map<std::string, std::string>* headers = nullptr, std::string const& etag = "",
                         const std::string& createdNotAfter = "", const std::string& createdNotBefore = "") const;
 
  void truncate(std::string const& path) const;
 
  void deleteObject(std::string const& path, long timestamp = -1) const;
 
  int updateMetadata(std::string const& path, std::map<std::string, std::string> const& metadata, long timestamp, std::string const& id = "", long newEOV = 0);
 
  std::string list(std::string const& path = "", bool latestOnly = false, std::string const& returnFormat = "text/plain", long createdNotAfter = -1, long createdNotBefore = -1) const;
 
  void snapshot(std::string const& ccdbrootpath, std::string const& localDir, long timestamp) const;
 
  bool isHostReachable() const;
 
  std::vector<std::string> parseSubFolders(std::string const& reply) const;
 
  std::vector<std::string> getAllFolders(std::string const& top) const;
 
  bool retrieveBlob(std::string const& path, std::string const& targetdir, std::map<std::string, std::string> const& metadata, long timestamp,
                    bool preservePathStructure = true, std::string const& localFileName = "snapshot.root", std::string const& createdNotAfter = "", std::string const& createdNotBefore = "", std::map<std::string, std::string>* headers = nullptr) const;
 
  std::map<std::string, std::string> retrieveHeaders(std::string const& path, std::map<std::string, std::string> const& metadata, long timestamp = -1) const;
 
  static void* extractFromTFile(TFile& file, TClass const* cl, const char* what = CCDBOBJECT_ENTRY);
 
  static bool getCCDBEntryHeaders(std::string const& url, std::string const& etag, std::vector<std::string>& headers, const std::string& agentID = "");
 
  static void parseCCDBHeaders(std::vector<std::string> const& headers, std::vector<std::string>& pfns, std::string& etag);
 
  static CCDBQuery* retrieveQueryInfo(TFile&);
 
  static std::map<std::string, std::string>* retrieveMetaInfo(TFile&);
 
  static std::string generateFileName(const std::string& inp);
 
  constexpr static const char* CCDBQUERY_ENTRY = "ccdb_query";
  constexpr static const char* CCDBMETA_ENTRY = "ccdb_meta";
  constexpr static const char* CCDBOBJECT_ENTRY = "ccdb_object";
 
  static void curlSetSSLOptions(CurlHandle* curl);
 
  TObject* retrieve(std::string const& path, std::map<std::string, std::string> const& metadata, long timestamp) const;
 
  TObject* retrieveFromTFile(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,
                             const std::string& createdNotAfter, const std::string& createdNotBefore) const;
  void loadFileToMemory(std::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,
                        const std::string& createdNotAfter, const std::string& createdNotBefore, bool considerSnapshot = true) const;
 
#if !defined(__CINT__) && !defined(__MAKECINT__) && !defined(__ROOTCLING__) && !defined(__CLING__)
  typedef struct RequestContext {
    o2::pmr::vector<char>& dest;
    std::string path;
    std::map<std::string, std::string> const& metadata;
    long timestamp;
    std::map<std::string, std::string>& headers;
    std::string etag;
    std::string createdNotAfter;
    std::string createdNotBefore;
    bool considerSnapshot;
 
    RequestContext(o2::pmr::vector<char>& d,
                   std::map<std::string, std::string> const& m,
                   std::map<std::string, std::string>& h)
      : dest(d), metadata(m), headers(h) {}
  } RequestContext;
 
  // Stores file associated with requestContext as a snapshot.
  void saveSnapshot(RequestContext& requestContext) const;
 
  // Schedules download via CCDBDownloader, but doesn't perform it until mUVLoop is ran.
  void scheduleDownload(RequestContext& requestContext, size_t* requestCounter) const;
 
  void getFromSnapshot(bool createSnapshot, std::string const& path,
                       long timestamp, std::map<std::string, std::string>& headers,
                       std::string& snapshotpath, o2::pmr::vector<char>& dest, int& fromSnapshot, std::string const& etag) const;
  void releaseNamedSemaphore(boost::interprocess::named_semaphore* sem, std::string const& path) const;
  boost::interprocess::named_semaphore* createNamedSemaphore(std::string const& path) const;
  static std::string determineSemaphoreName(std::string const& basedir, std::string const& objectpath);
  // queries and optionally removes a named semaphore from the system
  // returns true when successful (either found or found + removed)
  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, 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,
                        const std::string& createdNotAfter, const std::string& createdNotBefore, bool considerSnapshot = true) const;
 
  // Loads files from alien and cvmfs into given destination.
  bool loadLocalContentToMemory(o2::pmr::vector<char>& dest, std::string& url) const;
 
  // add annotated flattened headers in the end of the blob
  static void appendFlatHeader(o2::pmr::vector<char>& dest, const std::map<std::string, std::string>& headers);
 
  // the failure to load the file to memory is signaled by 0 size and non-0 capacity
  static bool isMemoryFileInvalid(const o2::pmr::vector<char>& v) { return v.size() == 0 && v.capacity() > 0; }
  template <typename T>
  static T* extractFromMemoryBlob(o2::pmr::vector<char>& blob)
  {
    auto obj = static_cast<T*>(interpretAsTMemFileAndExtract(blob.data(), blob.size(), typeid(T)));
    if constexpr (std::is_base_of<o2::conf::ConfigurableParam, T>::value) {
      auto& param = const_cast<typename std::remove_const<T&>::type>(T::Instance());
      param.syncCCDBandRegistry(obj);
      obj = &param;
    }
    return obj;
  }
 
  void navigateSourcesAndLoadFile(RequestContext& requestContext, int& fromSnapshot, size_t* requestCounter) const;
 
  void vectoredLoadFileToMemory(std::vector<RequestContext>& requestContext) const;
#endif
 
 private:
  // Sets the unique agent ID
  void setUniqueAgentID();
 
  void asynchPerform(CurlHandle* handle, size_t* requestCounter) const;
 
  // internal helper function to update a CCDB file with meta information
  static void updateMetaInformationInLocalFile(std::string const& filename, std::map<std::string, std::string> const* headers, CCDBQuery const* querysummary = nullptr);
 
  // report what file is read and for which purpose
  void logReading(const std::string& path, long ts, const std::map<std::string, std::string>* headers, const std::string& comment) const;
 
  void initInSnapshotMode(std::string const& snapshotpath)
  {
    mSnapshotTopPath = snapshotpath.empty() ? "." : snapshotpath;
    mInSnapshotMode = true;
  }
 
  std::string getTimestampString(long timestamp) const;
 
  std::string getFullUrlForStorage(CurlHandle* curl, const std::string& path, const std::string& objtype,
                                   const std::map<std::string, std::string>& metadata,
                                   long startValidityTimestamp = -1, long endValidityTimestamp = -1, int hostIndex = 0) const;
 
  std::string getFullUrlForRetrieval(CurlHandle* curl, const std::string& path, const std::map<std::string, std::string>& metadata,
                                     long timestamp = -1, int hostIndex = 0) const;
 
 public:
  int storeAsBinaryFile(const char* buffer, size_t size, const std::string& fileName, const std::string& objectType,
                        const std::string& path, const std::map<std::string, std::string>& metadata,
                        long startValidityTimestamp, long endValidityTimestamp, std::vector<char>::size_type maxSize = 0 /*in bytes*/) const;
 
  int storeAsTFile_impl(const void* obj1, std::type_info const& info, std::string const& path, std::map<std::string, std::string> const& metadata,
                        long startValidityTimestamp = -1, long endValidityTimestamp = -1, std::vector<char>::size_type maxSize = 0 /*in bytes*/) const;
 
  void* retrieveFromTFile(std::type_info const&, std::string const& path, std::map<std::string, std::string> const& metadata,
                          long timestamp = -1, std::map<std::string, std::string>* headers = nullptr, std::string const& etag = "",
                          const std::string& createdNotAfter = "", const std::string& createdNotBefore = "") const;
 
  void runDownloaderLoop(bool noWait);
  void setCurlRetriesParameters(int numberRetries, int delay = 100000 /* microseconds */)
  {
    mCurlRetries = numberRetries;
    mCurlDelayRetries = delay;
  }
 
 private:
  void* extractFromLocalFile(std::string const& filename, std::type_info const& tinfo, std::map<std::string, std::string>* headers) const;
 
  void* downloadFilesystemContent(std::string const& fullUrl, std::type_info const& tinfo, std::map<std::string, std::string>* headers) const;
 
  // initialize the TGrid (Alien connection)
  bool initTGrid() const;
 
  void* navigateURLsAndRetrieveContent(CurlHandle*, std::string const& url, std::type_info const& tinfo, std::map<std::string, std::string>* headers) const;
 
  // helper that interprets a content chunk as TMemFile and extracts the object therefrom
  static void* interpretAsTMemFileAndExtract(char* contentptr, size_t contentsize, std::type_info const& tinfo);
 
  void curlInit();
 
  // convert type_info to TClass, throw on failure
  static TClass* tinfo2TClass(std::type_info const& tinfo);
 
  typedef size_t (*CurlWriteCallback)(void*, size_t, size_t, void*);
 
  void initCurlOptionsForRetrieve(CurlHandle* curlHandle, void* pointer, CurlWriteCallback writeCallback, bool followRedirect = true) const;
 
  void initCurlHTTPHeaderOptionsForRetrieve(CurlHandle* curlHandle, curl_slist*& option_list, long timestamp, std::map<std::string, std::string>* headers, std::string const& etag, const std::string& createdNotAfter, const std::string& createdNotBefore, std::string_view url) const;
 
  bool receiveToFile(FILE* fileHandle, std::string const& path, std::map<std::string, std::string> const& metadata,
                     long timestamp, std::map<std::string, std::string>* headers = nullptr, std::string const& etag = "",
                     const std::string& createdNotAfter = "", const std::string& createdNotBefore = "", bool followRedirect = true) const;
 
  bool receiveToMemory(void* chunk, std::string const& path, std::map<std::string, std::string> const& metadata,
                       long timestamp, std::map<std::string, std::string>* headers = nullptr, std::string const& etag = "",
                       const std::string& createdNotAfter = "", const std::string& createdNotBefore = "", bool followRedirect = true) const;
 
  bool receiveObject(void* dataHolder, 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,
                     const std::string& createdNotAfter, const std::string& createdNotBefore, bool followRedirect, CurlWriteCallback writeCallback) const;
 
  void initHostsPool(std::string hosts);
 
  std::string getHostUrl(int hostIndex) const;
 
  void checkMetadataKeys(std::map<std::string, std::string> const& metadata) const;
 
  std::string getSnapshotDir(const std::string& topdir, const std::string& path) const { return topdir + "/" + path; }
  std::string getSnapshotFile(const std::string& topdir, const std::string& path, const std::string& sfile = "snapshot.root") const
  {
    return getSnapshotDir(topdir, path) + '/' + sfile;
  }
 
  template <typename MAP> // can be either std::map or std::multimap
  static size_t getFlatHeaderSize(const MAP& Headers)
  {
    size_t hsize = sizeof(int) + sizeof(FlatHeaderAnnot); // annotation size
    for (auto& h : Headers) {
      hsize += h.first.length() + h.second.length() + 2; // 2*(string_buffer + terminating null character)
    }
    return hsize;
  }
 
  // tmp helper and single point of entry for a CURL perform call
  // helps to switch between easy handle perform and multi handles in a single place
  int CURL_perform(CurlHandle* handle) const; // returns a CURLcode
 
  mutable CCDBDownloader* mDownloader = nullptr; 
  bool mIsCCDBDownloaderPreferred = false;
  std::string mUniqueAgentID{}; // Unique User-Agent ID communicated to server for logging
  std::string mUrl{};
  std::vector<std::string> hostsPool{};
  std::string mSnapshotTopPath{};    // root of the snaphot in the snapshot backend mode, i.e. with init("file://<dir>) call
  std::string mSnapshotCachePath{};  // root of the local snapshot (to fill or impose, even if not in the snapshot backend mode)
  bool mPreferSnapshotCache = false; // if snapshot is available, don't try to query its validity even in non-snapshot backend mode
  bool mInSnapshotMode = false;
  mutable TGrid* mAlienInstance = nullptr;                       // a cached connection to TGrid (needed for Alien locations)
  bool mNeedAlienToken = true;                                   // On EPN and FLP we use a local cache and don't need the alien token
  static std::unique_ptr<TJAlienCredentials> mJAlienCredentials; // access JAliEn credentials
  int mCurlRetries = 3;
  int mCurlDelayRetries = 100000; // in microseconds
  size_t mCurlTimeoutDownload = 15; // download timeout in seconds, can be configured via ALICEO2_CCDB_CURL_TIMEOUT_DOWNLOAD, updated according to the deployment mode
  size_t mCurlTimeoutUpload = 15;   // upload timeout in seconds, can be configured via ALICEO2_CCDB_CURL_TIMEOUT_UPLOAD, updated according to the deployment mode
 
  static constexpr char FlatHeaderAnnot[] = "$HEADER$"; // annotation for flat header
 
  ClassDefNV(CcdbApi, 1);
};
 
template <typename T>
typename std::enable_if<!std::is_base_of<o2::conf::ConfigurableParam, T>::value, T*>::type
  CcdbApi::retrieveFromTFileAny(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,
                                const std::string& createdNotAfter, const std::string& createdNotBefore) const
{
  return static_cast<T*>(retrieveFromTFile(typeid(T), path, metadata, timestamp, headers, etag, createdNotAfter, createdNotBefore));
}
 
template <typename T>
typename std::enable_if<std::is_base_of<o2::conf::ConfigurableParam, T>::value, T*>::type
  CcdbApi::retrieveFromTFileAny(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,
                                const std::string& createdNotAfter, const std::string& createdNotBefore) const
{
  auto obj = retrieveFromTFile(typeid(T), path, metadata, timestamp, headers, etag, createdNotAfter, createdNotBefore);
  if (obj) {
    auto& param = const_cast<typename std::remove_const<T&>::type>(T::Instance());
    param.syncCCDBandRegistry(obj);
    return &param;
  }
  return static_cast<T*>(obj);
}
 
} // namespace ccdb
} // namespace o2
 
#endif // PROJECT_CCDBAPI_H