34#include <TStreamerInfo.h>
38#include <fairlogger/Logger.h>
44#include <boost/algorithm/string.hpp>
47#include <boost/interprocess/sync/named_semaphore.hpp>
53#include <TAlienUserAgent.h>
54#include <unordered_set>
55#include "rapidjson/document.h"
56#include "rapidjson/writer.h"
57#include "rapidjson/stringbuffer.h"
65unique_ptr<TJAlienCredentials> CcdbApi::mJAlienCredentials =
nullptr;
73std::string_view trimHeaderValue(std::string_view
value)
75 constexpr std::string_view whitespace =
" \t\r\n";
76 const auto first =
value.find_first_not_of(whitespace);
77 return first == std::string_view::npos
88const std::vector<std::pair<std::string, std::string>>& gateTokenTable()
90 static const auto table = []() {
91 std::vector<std::pair<std::string, std::string>>
entries;
92 const char* spec = getenv(
"ALICEO2_CCDB_AUTH_TOKENS");
93 std::string_view rest = spec ? spec :
"";
94 while (!rest.empty()) {
95 const auto sep = rest.find(
';');
96 const auto entry = trimHeaderValue(rest.substr(0, sep));
97 rest = (sep == std::string_view::npos) ? std::string_view{} : rest.substr(sep + 1);
98 const auto eq =
entry.find(
'=');
99 if (eq == std::string_view::npos) {
102 auto url = trimHeaderValue(
entry.substr(0, eq));
105 const auto token = trimHeaderValue(
entry.substr(eq + 1));
106 while (
url.size() > 1 &&
url.back() ==
'/') {
107 url.remove_suffix(1);
109 if (!
url.empty() && !token.empty()) {
110 entries.emplace_back(std::string(
url), std::string(
"Authorization: Bearer ").append(token));
114 [](
const auto&
a,
const auto&
b) { return a.first.size() > b.first.size(); });
128curl_slist* appendGateToken(curl_slist* list, std::string_view
url)
130 for (
const auto& [prefix,
header] : gateTokenTable()) {
131 if (
url.substr(0, prefix.size()) == prefix &&
132 (
url.size() == prefix.size() ||
url[prefix.size()] ==
'/')) {
133 return curl_slist_append(list,
header.c_str());
153 boost::interprocess::named_semaphore* mSem =
nullptr;
154 std::string mSemName{};
169 std::unordered_set<CCDBSemaphore const*> mStore;
179 mIsCCDBDownloaderPreferred = 0;
180 if (deploymentMode == DeploymentMode::OnlineDDS && deploymentMode == DeploymentMode::OnlineECS && deploymentMode == DeploymentMode::OnlineAUX && deploymentMode == DeploymentMode::FST) {
181 mIsCCDBDownloaderPreferred = 1;
183 if (getenv(
"ALICEO2_ENABLE_MULTIHANDLE_CCDBAPI")) {
184 mIsCCDBDownloaderPreferred = atoi(getenv(
"ALICEO2_ENABLE_MULTIHANDLE_CCDBAPI"));
191 curl_global_cleanup();
195void CcdbApi::setUniqueAgentID()
197 mUniqueAgentID = TAlienUserAgent::BasedOnEnvironment().ToString();
203 LOG(
debug) <<
"On macOS we simply rely on TGrid::Connect(\"alien\").";
206 if (getenv(
"ALICEO2_CCDB_NOTOKENCHECK") && atoi(getenv(
"ALICEO2_CCDB_NOTOKENCHECK"))) {
209 if (getenv(
"JALIEN_TOKEN_CERT")) {
212 auto returncode = system(
"LD_PRELOAD= alien-token-info &> /dev/null");
213 if (returncode == -1) {
216 return returncode == 0;
219void CcdbApi::curlInit()
222 curl_global_init(CURL_GLOBAL_DEFAULT);
223 CcdbApi::mJAlienCredentials = std::make_unique<TJAlienCredentials>();
224 CcdbApi::mJAlienCredentials->loadCredentials();
225 CcdbApi::mJAlienCredentials->selectPreferedCredentials();
228 if (getenv(
"ALICEO2_CCDB_SOCKET_TIMEOUT")) {
229 auto timeoutMS = atoi(getenv(
"ALICEO2_CCDB_SOCKET_TIMEOUT"));
230 if (timeoutMS >= 0) {
231 LOG(info) <<
"Setting socket timeout to " << timeoutMS <<
" milliseconds";
240 throw std::invalid_argument(
"Empty url passed CcdbApi, cannot initialize. Aborting.");
245 constexpr const char* SNAPSHOTPREFIX =
"file://";
248 if (host.substr(0, 7).compare(SNAPSHOTPREFIX) == 0) {
249 auto path = host.substr(7);
250 initInSnapshotMode(
path);
271 std::string snapshotReport{};
272 const char* cachedir = getenv(
"ALICEO2_CCDB_LOCALCACHE");
273 namespace fs = std::filesystem;
275 if (cachedir[0] == 0) {
276 mSnapshotCachePath = fs::weakly_canonical(fs::absolute(
"."));
278 mSnapshotCachePath = fs::weakly_canonical(fs::absolute(cachedir));
280 snapshotReport = fmt::format(
"(cache snapshots to dir={}", mSnapshotCachePath);
283 mPreferSnapshotCache =
true;
284 if (mSnapshotCachePath.empty()) {
285 LOGP(fatal,
"IGNORE_VALIDITYCHECK_OF_CCDB_LOCALCACHE is defined but the ALICEO2_CCDB_LOCALCACHE is not");
287 snapshotReport +=
", prefer if available";
289 if (!snapshotReport.empty()) {
290 snapshotReport +=
')';
293 mNeedAlienToken = (host.find(
"https://") != std::string::npos) || (host.find(
"alice-ccdb.cern.ch") != std::string::npos) || (host.find(
"ccdb-test.cern.ch") != std::string::npos);
295 if (getenv(
"ALICEO2_CCDB_CURL_TIMEOUT_DOWNLOAD")) {
296 auto timeout = atoi(getenv(
"ALICEO2_CCDB_CURL_TIMEOUT_DOWNLOAD"));
298 mCurlTimeoutDownload =
timeout;
305 mCurlTimeoutDownload = 15;
308 mCurlTimeoutDownload = 15;
310 mCurlTimeoutDownload = 5;
314 if (getenv(
"ALICEO2_CCDB_CURL_TIMEOUT_UPLOAD")) {
315 auto timeout = atoi(getenv(
"ALICEO2_CCDB_CURL_TIMEOUT_UPLOAD"));
324 mCurlTimeoutUpload = 3;
327 mCurlTimeoutUpload = 20;
329 mCurlTimeoutUpload = 20;
336 LOGP(
debug,
"Curl timeouts are set to: download={:2}, upload={:2} seconds", mCurlTimeoutDownload, mCurlTimeoutUpload);
338 LOGP(info,
"Init CcdApi with UserAgentID: {}, Host: {}{}, Curl timeouts: upload:{} download:{}", mUniqueAgentID, host,
339 mInSnapshotMode ?
"(snapshot readonly mode)" : snapshotReport.c_str(), mCurlTimeoutUpload, mCurlTimeoutDownload);
348void CcdbApi::updateMetaInformationInLocalFile(std::string
const&
filename, std::map<std::string, std::string>
const* headers,
CCDBQuery const* querysummary)
350 std::lock_guard<std::mutex> guard(
gIOMutex);
351 auto oldlevel = gErrorIgnoreLevel;
352 gErrorIgnoreLevel = 6001;
353 TFile snapshotfile(
filename.c_str(),
"UPDATE");
355 if (!snapshotfile.IsZombie()) {
357 snapshotfile.WriteObjectAny(querysummary, TClass::GetClass(
typeid(*querysummary)),
CCDBQUERY_ENTRY);
360 snapshotfile.WriteObjectAny(headers, TClass::GetClass(
typeid(*headers)),
CCDBMETA_ENTRY);
362 snapshotfile.Write();
363 snapshotfile.Close();
365 gErrorIgnoreLevel = oldlevel;
375 std::string tmpObjectName = objectName;
376 tmpObjectName.erase(std::remove_if(tmpObjectName.begin(), tmpObjectName.end(),
377 [](
auto const&
c) ->
bool { return (!std::isalnum(c) && c !=
'_' && c !=
'/' && c !=
'.'); }),
378 tmpObjectName.end());
379 return tmpObjectName;
386 std::lock_guard<std::mutex> guard(
gIOMutex);
390 info->setFileName(tmpFileName);
391 info->setObjectType(className);
400 std::string className = rootObject->GetName();
403 info->setFileName(tmpFileName);
404 info->setObjectType(
"TObject");
406 std::lock_guard<std::mutex> guard(
gIOMutex);
411 std::map<std::string, std::string>
const& metadata,
412 long startValidityTimestamp,
long endValidityTimestamp,
413 std::vector<char>::size_type
maxSize)
const
417 LOGP(error,
"nullptr is provided for object {}/{}/{}",
path, startValidityTimestamp, endValidityTimestamp);
423 path, metadata, startValidityTimestamp, endValidityTimestamp,
maxSize);
427 const std::string&
path,
const std::map<std::string, std::string>& metadata,
428 long startValidityTimestamp,
long endValidityTimestamp, std::vector<char>::size_type
maxSize)
const
431 LOGP(alarm,
"Object will not be uploaded to {} since its size {} exceeds max allowed {}",
path,
size,
maxSize);
437 long sanitizedStartValidityTimestamp = startValidityTimestamp;
438 if (startValidityTimestamp == -1) {
439 LOGP(info,
"Start of Validity not set, current timestamp used.");
442 long sanitizedEndValidityTimestamp = endValidityTimestamp;
443 if (endValidityTimestamp == -1) {
444 LOGP(info,
"End of Validity not set, start of validity plus 1 day used.");
447 if (mInSnapshotMode) {
449 LOGP(alarm,
"Snapshot mode does not support headers-only upload");
452 auto pthLoc = getSnapshotDir(mSnapshotTopPath,
path);
454 auto flLoc = getSnapshotFile(mSnapshotTopPath,
path,
filename);
456 auto pent = flLoc.find_last_of(
'.');
457 if (pent == std::string::npos) {
460 flLoc.insert(pent, fmt::format(
"_{}_{}", startValidityTimestamp, endValidityTimestamp));
461 ofstream outf(flLoc.c_str(), ios::out | ios::binary);
465 throw std::runtime_error(fmt::format(
"Failed to write local CCDB file {}", flLoc));
467 std::map<std::string, std::string> metaheader(metadata);
469 metaheader[
"Valid-From"] =
std::to_string(startValidityTimestamp);
471 updateMetaInformationInLocalFile(flLoc.c_str(), &metaheader);
472 std::string metaStr{};
473 for (
const auto& mentry : metadata) {
474 metaStr += fmt::format(
"{}={};", mentry.first, mentry.second);
476 metaStr +=
"$USER_META;";
477 LOGP(info,
"Created local snapshot {}", flLoc);
478 LOGP(info, R
"(Upload with: o2-ccdb-upload --host "$ccdbhost" -p {} -f {} -k {} --starttimestamp {} --endtimestamp {} -m "{}")",
486 curl = curl_easy_init();
489 checkMetadataKeys(metadata);
491 if (curl !=
nullptr) {
492 auto mime = curl_mime_init(curl);
493 auto field = curl_mime_addpart(mime);
494 curl_mime_name(field,
"send");
496 curl_mime_filedata(field,
filename.c_str());
501 curl_mime_data(field,
"", 0);
506 curl_easy_setopt(curl, CURLOPT_MIMEPOST, mime);
507 curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);
508 curl_easy_setopt(curl, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
509 curl_easy_setopt(curl, CURLOPT_TIMEOUT, mCurlTimeoutUpload);
511 CURLcode
res = CURL_LAST;
513 for (
size_t hostIndex = 0; hostIndex < hostsPool.size() &&
res > 0; hostIndex++) {
514 std::string fullUrl = getFullUrlForStorage(curl,
path, objectType, metadata, sanitizedStartValidityTimestamp, sanitizedEndValidityTimestamp, hostIndex);
515 LOG(debug3) <<
"Full URL Encoded: " << fullUrl;
517 curl_easy_setopt(curl, CURLOPT_URL, fullUrl.c_str());
520 struct curl_slist* headerlist = curl_slist_append(
nullptr,
"Expect:");
521 headerlist = appendGateToken(headerlist, fullUrl);
522 curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headerlist);
525 res =
static_cast<CURLcode
>(CURL_perform(curl));
527 if (
res != CURLE_OK) {
528 if (
res == CURLE_OPERATION_TIMEDOUT) {
529 LOGP(alarm,
"curl_easy_perform() timed out. Consider increasing the timeout using the env var `ALICEO2_CCDB_CURL_TIMEOUT_UPLOAD` (seconds), current one is {}", mCurlTimeoutUpload);
531 LOGP(alarm,
"curl_easy_perform() failed: {}", curl_easy_strerror(
res));
535 curl_slist_free_all(headerlist);
539 curl_easy_cleanup(curl);
542 curl_mime_free(mime);
544 LOGP(alarm,
"curl initialization failure");
551 long startValidityTimestamp,
long endValidityTimestamp, std::vector<char>::size_type
maxSize)
const
555 LOGP(error,
"nullptr is provided for object {}/{}/{}",
path, startValidityTimestamp, endValidityTimestamp);
563std::string CcdbApi::getFullUrlForStorage(
CurlHandle* curl,
const std::string&
path,
const std::string& objtype,
564 const std::map<std::string, std::string>& metadata,
565 long startValidityTimestamp,
long endValidityTimestamp,
int hostIndex)
const
568 std::string startValidityString = getTimestampString(startValidityTimestamp < 0 ?
getCurrentTimestamp() : startValidityTimestamp);
569 std::string endValidityString = getTimestampString(endValidityTimestamp < 0 ?
getFutureTimestamp(60 * 60 * 24 * 1) : endValidityTimestamp);
571 std::string
url = getHostUrl(hostIndex);
573 std::string fullUrl =
url +
"/" +
path +
"/" + startValidityString +
"/" + endValidityString +
"/";
576 char* objtypeEncoded = curl_easy_escape(curl, objtype.c_str(), objtype.size());
577 fullUrl +=
"ObjectType=" + std::string(objtypeEncoded) +
"/";
578 curl_free(objtypeEncoded);
580 for (
auto& kv : metadata) {
581 std::string mfirst = kv.first;
582 std::string msecond = kv.second;
584 char* mfirstEncoded = curl_easy_escape(curl, mfirst.c_str(), mfirst.size());
585 char* msecondEncoded = curl_easy_escape(curl, msecond.c_str(), msecond.size());
586 fullUrl += std::string(mfirstEncoded) +
"=" + std::string(msecondEncoded) +
"/";
587 curl_free(mfirstEncoded);
588 curl_free(msecondEncoded);
594std::string CcdbApi::getFullUrlForRetrieval(
CurlHandle* curl,
const std::string&
path,
const std::map<std::string, std::string>& metadata,
long timestamp,
int hostIndex)
const
596 if (mInSnapshotMode) {
597 return getSnapshotFile(mSnapshotTopPath,
path);
603 std::string hostUrl = getHostUrl(hostIndex);
605 std::string fullUrl = hostUrl +
"/" +
path +
"/" + validityString +
"/";
607 for (
auto& kv : metadata) {
608 std::string mfirst = kv.first;
609 std::string msecond = kv.second;
611 char* mfirstEncoded = curl_easy_escape(curl, mfirst.c_str(), mfirst.size());
612 char* msecondEncoded = curl_easy_escape(curl, msecond.c_str(), msecond.size());
613 fullUrl += std::string(mfirstEncoded) +
"=" + std::string(msecondEncoded) +
"/";
614 curl_free(mfirstEncoded);
615 curl_free(msecondEncoded);
637static size_t WriteMemoryCallback(
void* contents,
size_t size,
size_t nmemb,
void* userp)
639 size_t realsize =
size * nmemb;
642 mem->memory = (
char*)realloc(mem->memory, mem->size + realsize + 1);
643 if (mem->memory ==
nullptr) {
644 printf(
"not enough memory (realloc returned NULL)\n");
648 memcpy(&(mem->memory[mem->size]), contents, realsize);
649 mem->size += realsize;
650 mem->memory[mem->size] = 0;
666static size_t WriteToFileCallback(
void*
ptr,
size_t size,
size_t nmemb, FILE*
stream)
679static CURLcode ssl_ctx_callback(
CurlHandle*,
void*,
void* parm)
681 std::string
msg((
const char*)parm);
684 if (
msg.length() > 0 &&
end == -1) {
686 }
else if (
end > 0) {
698 CredentialsKind cmk = mJAlienCredentials->getPreferedCredentials();
701 if (cmk == cNOT_FOUND) {
705 TJAlienCredentialsObject cmo = mJAlienCredentials->get(cmk);
707 char* CAPath = getenv(
"X509_CERT_DIR");
709 curl_easy_setopt(curl_handle, CURLOPT_CAPATH, CAPath);
711 curl_easy_setopt(curl_handle, CURLOPT_CAINFO,
nullptr);
712 curl_easy_setopt(curl_handle, CURLOPT_SSLCERT, cmo.certpath.c_str());
713 curl_easy_setopt(curl_handle, CURLOPT_SSLKEY, cmo.keypath.c_str());
716 curl_easy_setopt(curl_handle, CURLOPT_SSL_CTX_FUNCTION, ssl_ctx_callback);
717 curl_easy_setopt(curl_handle, CURLOPT_SSL_CTX_DATA, mJAlienCredentials->getMessages().c_str());
724void CcdbApi::initCurlOptionsForRetrieve(
CurlHandle* curlHandle,
void* chunk,
CurlWriteCallback writeCallback,
bool followRedirect)
const
726 curl_easy_setopt(curlHandle, CURLOPT_WRITEFUNCTION, writeCallback);
727 curl_easy_setopt(curlHandle, CURLOPT_WRITEDATA, chunk);
728 curl_easy_setopt(curlHandle, CURLOPT_FOLLOWLOCATION, followRedirect ? 1L : 0L);
733template <
typename MapType = std::map<std::
string, std::
string>>
734size_t header_map_callback(
char*
buffer,
size_t size,
size_t nitems,
void* userdata)
736 auto* headers =
static_cast<MapType*
>(userdata);
738 std::string::size_type
index =
header.find(
':', 0);
739 if (
index != std::string::npos) {
740 const auto key = boost::algorithm::trim_copy(
header.substr(0,
index));
742 LOGP(
debug,
"Adding #{} {} -> {}", headers->size(),
key,
value);
744 if (
key ==
"Content-Length") {
745 auto cl = headers->find(
"Content-Length");
746 if (cl != headers->end()) {
747 if (std::stol(cl->second) < stol(
value)) {
757 auto cl = headers->find(
"ETag");
758 if (cl != headers->end()) {
764 if (
key ==
"Content-Type") {
765 auto cl = headers->find(
"Content-Type");
766 if (cl != headers->end()) {
772 headers->insert(std::make_pair(
key,
value));
775 return size * nitems;
779void CcdbApi::initCurlHTTPHeaderOptionsForRetrieve(
CurlHandle* curlHandle, curl_slist*& option_list,
long timestamp, std::map<std::string, std::string>* headers, std::string
const&
etag,
784 option_list = curl_slist_append(option_list, (
"If-None-Match: " +
etag).c_str());
788 option_list = curl_slist_append(option_list, (
"If-Not-After: " +
createdNotAfter).c_str());
792 option_list = curl_slist_append(option_list, (
"If-Not-Before: " +
createdNotBefore).c_str());
795 if (headers !=
nullptr) {
796 option_list = curl_slist_append(option_list, (
"If-None-Match: " +
to_string(timestamp)).c_str());
797 curl_easy_setopt(curlHandle, CURLOPT_HEADERFUNCTION, header_map_callback<>);
798 curl_easy_setopt(curlHandle, CURLOPT_HEADERDATA, headers);
801 option_list = appendGateToken(option_list,
url);
805 curl_easy_setopt(curlHandle, CURLOPT_HTTPHEADER, option_list);
807 curl_easy_setopt(curlHandle, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
810bool CcdbApi::receiveToFile(FILE* fileHandle, std::string
const&
path, std::map<std::string, std::string>
const& metadata,
811 long timestamp, std::map<std::string, std::string>* headers, std::string
const&
etag,
817bool CcdbApi::receiveToMemory(
void* chunk, std::string
const&
path, std::map<std::string, std::string>
const& metadata,
818 long timestamp, std::map<std::string, std::string>* headers, std::string
const&
etag,
824bool CcdbApi::receiveObject(
void* dataHolder, std::string
const&
path, std::map<std::string, std::string>
const& metadata,
825 long timestamp, std::map<std::string, std::string>* headers, std::string
const&
etag,
830 curlHandle = curl_easy_init();
831 curl_easy_setopt(curlHandle, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
833 if (curlHandle !=
nullptr) {
836 initCurlOptionsForRetrieve(curlHandle, dataHolder, writeCallback, followRedirect);
837 long responseCode = 0;
838 CURLcode curlResultCode = CURL_LAST;
840 for (
size_t hostIndex = 0; hostIndex < hostsPool.size() && (responseCode >= 400 || curlResultCode > 0); hostIndex++) {
841 std::string fullUrl = getFullUrlForRetrieval(curlHandle,
path, metadata, timestamp, hostIndex);
842 curl_easy_setopt(curlHandle, CURLOPT_URL, fullUrl.c_str());
845 curl_slist* option_list =
nullptr;
848 curlResultCode =
static_cast<CURLcode
>(CURL_perform(curlHandle));
850 if (curlResultCode != CURLE_OK) {
851 LOGP(alarm,
"curl_easy_perform() failed: {}", curl_easy_strerror(curlResultCode));
853 curlResultCode = curl_easy_getinfo(curlHandle, CURLINFO_RESPONSE_CODE, &responseCode);
854 if ((curlResultCode == CURLE_OK) && (responseCode < 300)) {
855 curl_slist_free_all(option_list);
856 curl_easy_cleanup(curlHandle);
859 if (curlResultCode != CURLE_OK) {
860 LOGP(alarm,
"invalid URL {}", fullUrl);
862 LOGP(alarm,
"not found under link {}", fullUrl);
866 curl_slist_free_all(option_list);
869 curl_easy_cleanup(curlHandle);
875 long timestamp)
const
883 bool res = receiveToMemory((
void*)&chunk,
path, metadata, timestamp);
886 std::lock_guard<std::mutex> guard(
gIOMutex);
888 mess.SetBuffer(chunk.
memory, chunk.
size, kFALSE);
893 LOGP(info,
"couldn't retrieve the object {}",
path);
905 std::string
str = inp;
906 str.erase(std::remove_if(
str.begin(),
str.end(), ::isspace),
str.end());
907 str = std::regex_replace(
str, std::regex(
"::"),
"-");
913 long timestamp, std::map<std::string, std::string>* headers, std::string
const&
etag,
920 long timestamp,
bool preservePath, std::string
const& localFileName, std::string
const&
createdNotAfter, std::string
const&
createdNotBefore, std::map<std::string, std::string>* outHeaders)
const
924 std::string fulltargetdir = targetdir + (preservePath ? (
'/' +
path) :
"");
928 }
catch (std::exception e) {
929 LOGP(error,
"Could not create local snapshot cache directory {}, reason: {}", fulltargetdir, e.what());
934 std::map<std::string, std::string> headers;
937 if ((headers.count(
"Error") != 0) || (buff.empty())) {
938 LOGP(error,
"Unable to find object {}/{}, Aborting",
path, timestamp);
942 auto getFileName = [&headers]() {
943 auto& s = headers[
"Content-Disposition"];
945 std::regex re(
"(.*;)filename=\"(.*)\"");
947 if (std::regex_match(s.c_str(),
m, re)) {
951 std::string backupname(
"ccdb-blob.bin");
952 LOG(error) <<
"Cannot determine original filename from Content-Disposition ... falling back to " << backupname;
955 auto filename = localFileName.size() > 0 ? localFileName : getFileName();
956 std::string targetpath = fulltargetdir +
"/" +
filename;
958 std::ofstream objFile(targetpath, std::ios::out | std::ofstream::binary);
959 std::copy(buff.begin(), buff.end(), std::ostreambuf_iterator<char>(objFile));
960 if (!objFile.good()) {
961 LOGP(error,
"Unable to open local file {}, Aborting", targetpath);
967 updateMetaInformationInLocalFile(targetpath.c_str(), &headers, &querysummary);
969 *outHeaders = std::move(headers);
974void CcdbApi::snapshot(std::string
const& ccdbrootpath, std::string
const& localDir,
long timestamp)
const
978 std::map<std::string, std::string> metadata;
979 for (
auto& folder : allfolders) {
989 auto object = file.GetObjectChecked(what, cl);
993 std::string objectName(cl->GetName());
995 object = file.GetObjectChecked(objectName.c_str(), cl);
996 LOG(warn) <<
"Did not find object under expected name " << what;
1000 LOG(warn) <<
"Found object under deprecated name " << cl->GetName();
1005 if (cl->InheritsFrom(
"TObject")) {
1008 auto tree =
dynamic_cast<TTree*
>((
TObject*)
object);
1010 tree->LoadBaskets(0x1L << 32);
1011 tree->SetDirectory(
nullptr);
1014 auto h =
dynamic_cast<TH1*
>((
TObject*)
object);
1016 h->SetDirectory(
nullptr);
1024void* CcdbApi::extractFromLocalFile(std::string
const&
filename, std::type_info
const& tinfo, std::map<std::string, std::string>* headers)
const
1026 if (!std::filesystem::exists(
filename)) {
1027 LOG(error) <<
"Local snapshot " <<
filename <<
" not found \n";
1030 std::lock_guard<std::mutex> guard(
gIOMutex);
1031 auto tcl = tinfo2TClass(tinfo);
1036 *headers = *storedmeta;
1039 if ((
isSnapshotMode() || mPreferSnapshotCache) && headers->find(
"ETag") == headers->end()) {
1042 if (headers->find(
"fileSize") == headers->end()) {
1043 (*headers)[
"fileSize"] = fmt::format(
"{}",
f.GetEND());
1049bool CcdbApi::initTGrid()
const
1051 if (mNeedAlienToken && !gGrid) {
1052 static bool allowNoToken = getenv(
"ALICEO2_CCDB_NOTOKENCHECK") && atoi(getenv(
"ALICEO2_CCDB_NOTOKENCHECK"));
1054 LOG(fatal) <<
"Alien Token Check failed - Please get an alien token before running with https CCDB endpoint, or alice-ccdb.cern.ch!";
1056 TGrid::Connect(
"alien");
1057 static bool errorShown =
false;
1058 if (!gGrid && errorShown ==
false) {
1060 LOG(error) <<
"TGrid::Connect returned nullptr. May be due to missing alien token";
1062 LOG(fatal) <<
"TGrid::Connect returned nullptr. May be due to missing alien token";
1067 return gGrid !=
nullptr;
1070void* CcdbApi::downloadFilesystemContent(std::string
const&
url, std::type_info
const& tinfo, std::map<std::string, std::string>* headers)
const
1072 if ((
url.find(
"alien:/", 0) != std::string::npos) && !initTGrid()) {
1075 std::lock_guard<std::mutex> guard(
gIOMutex);
1076 auto memfile = TMemFile::Open(
url.c_str(),
"OPEN");
1078 auto cl = tinfo2TClass(tinfo);
1080 if (headers && headers->find(
"fileSize") == headers->end()) {
1081 (*headers)[
"fileSize"] = fmt::format(
"{}", memfile->GetEND());
1089void* CcdbApi::interpretAsTMemFileAndExtract(
char* contentptr,
size_t contentsize, std::type_info
const& tinfo)
1092 Int_t previousErrorLevel = gErrorIgnoreLevel;
1093 gErrorIgnoreLevel = kFatal;
1094 std::lock_guard<std::mutex> guard(
gIOMutex);
1095 TMemFile memFile(
"name", contentptr, contentsize,
"READ");
1096 gErrorIgnoreLevel = previousErrorLevel;
1097 if (!memFile.IsZombie()) {
1098 auto tcl = tinfo2TClass(tinfo);
1109void* CcdbApi::navigateURLsAndRetrieveContent(
CurlHandle* curl_handle, std::string
const&
url, std::type_info
const& tinfo, std::map<std::string, std::string>* headers)
const
1114 static thread_local std::multimap<std::string, std::string> headerData;
1117 if ((
url.find(
"alien:/", 0) != std::string::npos) || (
url.find(
"file:/", 0) != std::string::npos)) {
1118 return downloadFilesystemContent(
url, tinfo, headers);
1125 curl_easy_setopt(curl_handle, CURLOPT_URL,
url.c_str());
1127 MemoryStruct chunk{(
char*)malloc(1), 0};
1128 initCurlOptionsForRetrieve(curl_handle, (
void*)&chunk, WriteMemoryCallback,
false);
1130 curl_easy_setopt(curl_handle, CURLOPT_HEADERFUNCTION, header_map_callback<
decltype(headerData)>);
1132 curl_easy_setopt(curl_handle, CURLOPT_HEADERDATA, (
void*)&headerData);
1136 auto res =
static_cast<CURLcode
>(CURL_perform(curl_handle));
1137 long response_code = -1;
1138 void* content =
nullptr;
1140 if (
res == CURLE_OK && curl_easy_getinfo(curl_handle, CURLINFO_RESPONSE_CODE, &response_code) == CURLE_OK) {
1142 for (
auto& p : headerData) {
1143 (*headers)[
p.first] =
p.second;
1146 if (200 <= response_code && response_code < 300) {
1148 content = interpretAsTMemFileAndExtract(chunk.memory, chunk.size, tinfo);
1149 if (headers && headers->find(
"fileSize") == headers->end()) {
1150 (*headers)[
"fileSize"] = fmt::format(
"{}", chunk.size);
1152 }
else if (response_code == 304) {
1157 LOGP(
debug,
"Object exists but I am not serving it since it's already in your possession");
1160 else if (300 <= response_code && response_code < 400) {
1166 auto complement_Location = [
this](std::string
const& loc) {
1167 if (loc[0] ==
'/') {
1174 std::vector<std::string> locs;
1175 auto iter = headerData.find(
"Location");
1176 if (iter != headerData.end()) {
1177 locs.push_back(complement_Location(iter->second));
1180 auto iter2 = headerData.find(
"Content-Location");
1181 if (iter2 != headerData.end()) {
1182 auto range = headerData.equal_range(
"Content-Location");
1183 for (
auto it =
range.first; it !=
range.second; ++it) {
1184 if (std::find(locs.begin(), locs.end(), it->second) == locs.end()) {
1185 locs.push_back(complement_Location(it->second));
1189 for (
auto& l : locs) {
1191 LOG(
debug) <<
"Trying content location " << l;
1192 content = navigateURLsAndRetrieveContent(curl_handle, l, tinfo, headers);
1198 }
else if (response_code == 404) {
1199 LOG(error) <<
"Requested resource does not exist: " <<
url;
1202 LOG(error) <<
"Error in fetching object " <<
url <<
", curl response code:" << response_code;
1206 if (chunk.memory !=
nullptr) {
1210 LOGP(alarm,
"Curl request to {} failed with result {}, response code: {}",
url,
int(
res), response_code);
1215 (*headers)[
"Error"] =
"An error occurred during retrieval";
1221 std::map<std::string, std::string>
const& metadata,
long timestamp,
1222 std::map<std::string, std::string>* headers, std::string
const&
etag,
1225 if (!mSnapshotCachePath.empty()) {
1227 auto semaphore_barrier = std::make_unique<CCDBSemaphore>(mSnapshotCachePath,
path);
1228 std::string logfile = mSnapshotCachePath +
"/log";
1229 std::fstream out(logfile, ios_base::out | ios_base::app);
1230 if (out.is_open()) {
1231 out <<
"CCDB-access[" << getpid() <<
"] of " << mUniqueAgentID <<
" to " <<
path <<
" timestamp " << timestamp <<
"\n";
1233 auto snapshotfile = getSnapshotFile(mSnapshotCachePath,
path);
1234 bool snapshoting =
false;
1235 if (!std::filesystem::exists(snapshotfile)) {
1237 out <<
"CCDB-access[" << getpid() <<
"] ... " << mUniqueAgentID <<
" downloading to snapshot " << snapshotfile <<
"\n";
1240 out <<
"CCDB-access[" << getpid() <<
"] ... " << mUniqueAgentID <<
" failed to create directory for " << snapshotfile <<
"\n";
1243 out <<
"CCDB-access[" << getpid() <<
"] ... " << mUniqueAgentID <<
"serving from local snapshot " << snapshotfile <<
"\n";
1246 auto res = extractFromLocalFile(snapshotfile, tinfo, headers);
1248 logReading(
path, timestamp, headers,
"retrieve from snapshot");
1256 curl_easy_setopt(curl_handle, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1257 std::string fullUrl = getFullUrlForRetrieval(curl_handle,
path, metadata, timestamp);
1259 if (mInSnapshotMode) {
1260 auto res = extractFromLocalFile(fullUrl, tinfo, headers);
1262 logReading(
path, timestamp, headers,
"retrieve from snapshot");
1267 curl_slist* option_list =
nullptr;
1269 auto content = navigateURLsAndRetrieveContent(curl_handle, fullUrl, tinfo, headers);
1271 for (
size_t hostIndex = 1; hostIndex < hostsPool.size() && !(content); hostIndex++) {
1272 fullUrl = getFullUrlForRetrieval(curl_handle,
path, metadata, timestamp, hostIndex);
1274 curl_slist_free_all(option_list);
1275 option_list =
nullptr;
1277 content = navigateURLsAndRetrieveContent(curl_handle, fullUrl, tinfo, headers);
1280 logReading(
path, timestamp, headers,
"retrieve");
1282 curl_slist_free_all(option_list);
1283 curl_easy_cleanup(curl_handle);
1289 size_t newLength =
size * nmemb;
1290 size_t oldLength = s->size();
1292 s->resize(oldLength + newLength);
1293 }
catch (std::bad_alloc& e) {
1294 LOG(error) <<
"memory error when getting data from CCDB";
1298 std::copy((
char*)contents, (
char*)contents + newLength, s->begin() + oldLength);
1299 return size * nmemb;
1305 CURLcode
res = CURL_LAST;
1308 curl = curl_easy_init();
1309 if (curl !=
nullptr) {
1311 curl_easy_setopt(curl, CURLOPT_WRITEDATA, &
result);
1312 curl_easy_setopt(curl, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1316 std::string fullUrl;
1318 for (
size_t hostIndex = 0; hostIndex < hostsPool.size() &&
res != CURLE_OK; hostIndex++) {
1319 fullUrl = getHostUrl(hostIndex);
1320 fullUrl += latestOnly ?
"/latest/" :
"/browse/";
1322 curl_easy_setopt(curl, CURLOPT_URL, fullUrl.c_str());
1325 struct curl_slist* headers =
nullptr;
1326 headers = curl_slist_append(headers, (std::string(
"Accept: ") + returnFormat).c_str());
1327 headers = curl_slist_append(headers, (std::string(
"Content-Type: ") + returnFormat).c_str());
1334 headers = appendGateToken(headers, fullUrl);
1335 curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
1337 res =
static_cast<CURLcode
>(CURL_perform(curl));
1338 if (
res != CURLE_OK) {
1339 LOGP(alarm,
"CURL_perform() failed: {}", curl_easy_strerror(
res));
1341 curl_slist_free_all(headers);
1343 curl_easy_cleanup(curl);
1349std::string CcdbApi::getTimestampString(
long timestamp)
const
1362 curl = curl_easy_init();
1363 if (curl !=
nullptr) {
1364 curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST,
"DELETE");
1365 curl_easy_setopt(curl, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1366 curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);
1369 for (
size_t hostIndex = 0; hostIndex < hostsPool.size(); hostIndex++) {
1372 stringstream fullUrl;
1373 fullUrl << getHostUrl(hostIndex) <<
"/" <<
path <<
"/" << timestampLocal;
1374 curl_easy_setopt(curl, CURLOPT_URL, fullUrl.str().c_str());
1378 struct curl_slist*
list = appendGateToken(
nullptr, fullUrl.str());
1379 curl_easy_setopt(curl, CURLOPT_HTTPHEADER,
list);
1382 res =
static_cast<CURLcode
>(CURL_perform(curl));
1383 if (
res != CURLE_OK) {
1384 LOGP(alarm,
"CURL_perform() failed: {}", curl_easy_strerror(
res));
1386 curl_slist_free_all(
list);
1390 curl_easy_cleanup(curl);
1398 for (
size_t i = 0;
i < hostsPool.size();
i++) {
1402 stringstream fullUrl;
1403 std::string
url = getHostUrl(
i);
1404 fullUrl <<
url <<
"/truncate/" <<
path;
1406 curl = curl_easy_init();
1407 curl_easy_setopt(curl, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1408 if (curl !=
nullptr) {
1409 curl_easy_setopt(curl, CURLOPT_URL, fullUrl.str().c_str());
1415 struct curl_slist*
list = appendGateToken(
nullptr, fullUrl.str());
1416 curl_easy_setopt(curl, CURLOPT_HTTPHEADER,
list);
1417 curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);
1421 res =
static_cast<CURLcode
>(CURL_perform(curl));
1422 if (
res != CURLE_OK) {
1423 LOGP(alarm,
"CURL_perform() failed: {}", curl_easy_strerror(
res));
1425 curl_easy_cleanup(curl);
1426 curl_slist_free_all(
list);
1433 return size * nmemb;
1439 CURLcode
res = CURL_LAST;
1442 curl = curl_easy_init();
1443 curl_easy_setopt(curl, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1456 for (
size_t hostIndex = 0; hostIndex < hostsPool.size() &&
res != CURLE_OK; hostIndex++) {
1457 curl_easy_setopt(curl, CURLOPT_URL, mUrl.data());
1458 curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION,
write_data);
1460 res =
static_cast<CURLcode
>(CURL_perform(curl));
1465 curl_easy_cleanup(curl);
1474 std::stringstream ss(reply.c_str());
1476 std::vector<std::string> folders;
1478 size_t numberoflines = std::count(reply.begin(), reply.end(),
'\n');
1479 bool inSubFolderSection =
false;
1481 for (
size_t linenumber = 0; linenumber < numberoflines; ++linenumber) {
1482 std::getline(ss, line);
1483 if (inSubFolderSection && line.size() > 0) {
1488 if (line.compare(
"Subfolders:") == 0) {
1489 inSubFolderSection =
true;
1497size_t header_callback(
char*
buffer,
size_t size,
size_t nitems,
void* userdata)
1499 auto* headers =
static_cast<std::vector<std::string>*
>(userdata);
1501 headers->emplace_back(std::string(
header.data()));
1502 return size * nitems;
1510 auto p = std::filesystem::path(
filename).parent_path();
1511 if (!std::filesystem::exists(p)) {
1512 std::filesystem::create_directories(p);
1515 rapidjson::StringBuffer
buffer;
1516 rapidjson::Writer<rapidjson::StringBuffer> writer(
buffer);
1517 writer.StartObject();
1518 for (
const auto& pair : meta) {
1519 writer.Key(pair.first.c_str());
1520 writer.String(pair.second.c_str());
1526 if (file.is_open()) {
1527 file <<
buffer.GetString();
1539 if (!file.is_open()) {
1540 std::cerr <<
"Failed to open file for reading." << std::endl;
1544 std::string jsonStr((std::istreambuf_iterator<char>(file)), std::istreambuf_iterator<char>());
1547 rapidjson::Document document;
1548 document.Parse(jsonStr.c_str());
1550 if (document.HasParseError()) {
1551 std::cerr <<
"Error parsing JSON" << std::endl;
1556 for (
auto itr = document.MemberBegin(); itr != document.MemberEnd(); ++itr) {
1557 meta[itr->name.GetString()] = itr->value.GetString();
1562std::map<std::string, std::string>
CcdbApi::retrieveHeaders(std::string
const&
path, std::map<std::string, std::string>
const& metadata,
long timestamp)
const
1565 auto do_remote_header_call = [
this, &
path, &metadata, timestamp]() -> std::map<std::string, std::string> {
1567 CURLcode
res = CURL_LAST;
1568 std::string fullUrl = getFullUrlForRetrieval(curl,
path, metadata, timestamp);
1569 std::map<std::string, std::string> headers;
1571 if (curl !=
nullptr) {
1572 struct curl_slist*
list =
nullptr;
1574 list = appendGateToken(
list, fullUrl);
1576 curl_easy_setopt(curl, CURLOPT_HTTPHEADER,
list);
1579 curl_easy_setopt(curl, CURLOPT_NOBODY, 1L);
1580 curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);
1581 curl_easy_setopt(curl, CURLOPT_HEADERFUNCTION, header_map_callback<>);
1582 curl_easy_setopt(curl, CURLOPT_HEADERDATA, &headers);
1583 curl_easy_setopt(curl, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1588 long httpCode = 404;
1589 CURLcode getCodeRes = CURL_LAST;
1590 for (
size_t hostIndex = 0; hostIndex < hostsPool.size() && (httpCode >= 400 ||
res > 0 || getCodeRes > 0); hostIndex++) {
1591 curl_easy_setopt(curl, CURLOPT_URL, fullUrl.c_str());
1592 res =
static_cast<CURLcode
>(CURL_perform(curl));
1593 if (
res != CURLE_OK &&
res != CURLE_UNSUPPORTED_PROTOCOL) {
1597 LOG(error) <<
"CURL_perform() failed: " << curl_easy_strerror(
res);
1599 getCodeRes = curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &httpCode);
1601 if (httpCode == 404) {
1604 curl_easy_cleanup(curl);
1609 if (!mSnapshotCachePath.empty()) {
1611 auto semaphore_barrier = std::make_unique<CCDBSemaphore>(mSnapshotCachePath + std::string(
"_headers"),
path);
1613 std::string logfile = mSnapshotCachePath +
"/log";
1614 std::fstream out(logfile, ios_base::out | ios_base::app);
1615 if (out.is_open()) {
1616 out <<
"CCDB-header-access[" << getpid() <<
"] of " << mUniqueAgentID <<
" to " <<
path <<
" timestamp " << timestamp <<
"\n";
1618 auto snapshotfile = getSnapshotFile(mSnapshotCachePath,
path +
"/" +
std::to_string(timestamp),
"header.json");
1619 if (!std::filesystem::exists(snapshotfile)) {
1620 out <<
"CCDB-header-access[" << getpid() <<
"] ... " << mUniqueAgentID <<
" storing to snapshot " << snapshotfile <<
"\n";
1623 auto meta = do_remote_header_call();
1627 LOG(warn) <<
"Failed to cache the header information to disc";
1631 out <<
"CCDB-header-access[" << getpid() <<
"] ... " << mUniqueAgentID <<
"serving from local snapshot " << snapshotfile <<
"\n";
1632 std::map<std::string, std::string> meta;
1634 LOG(warn) <<
"Failed to read cached information from disc";
1635 return do_remote_header_call();
1640 return do_remote_header_call();
1645 auto curl = curl_easy_init();
1651 struct curl_slist*
list =
nullptr;
1652 list = curl_slist_append(
list, (
"If-None-Match: " +
etag).c_str());
1655 curl_easy_setopt(curl, CURLOPT_HTTPHEADER,
list);
1657 curl_easy_setopt(curl, CURLOPT_URL,
url.c_str());
1659 curl_easy_setopt(curl, CURLOPT_NOBODY, 1L);
1660 curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);
1661 curl_easy_setopt(curl, CURLOPT_HEADERFUNCTION, header_callback);
1662 curl_easy_setopt(curl, CURLOPT_HEADERDATA, &headers);
1663 if (!agentID.empty()) {
1664 curl_easy_setopt(curl, CURLOPT_USERAGENT, agentID.c_str());
1670 curl_easy_perform(curl);
1671 long http_code = 404;
1672 curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &http_code);
1673 if (http_code == 304) {
1681 static std::string etagHeader =
"ETag: ";
1682 static std::string locationHeader =
"Content-Location: ";
1685 for (
auto h : headers) {
1686 if (
h.find(etagHeader) == 0) {
1687 etag = trimHeaderValue(std::string_view(
h).substr(etagHeader.size()));
1688 }
else if (
h.find(locationHeader) == 0) {
1689 pfns.emplace_back(trimHeaderValue(std::string_view(
h).substr(locationHeader.size())));
1705 auto object = file.GetObjectChecked(
CCDBMETA_ENTRY, TClass::GetClass(
typeid(std::map<std::string, std::string>)));
1707 return static_cast<std::map<std::string, std::string>*
>(
object);
1714void traverseAndFillFolders(
CcdbApi const& api, std::string
const&
top, std::vector<std::string>& folders)
1718 folders.emplace_back(
top);
1721 if (subfolders.size() > 0) {
1723 for (
auto& sub : subfolders) {
1724 traverseAndFillFolders(api, sub, folders);
1734 std::vector<std::string> folders;
1735 traverseAndFillFolders(*
this,
top, folders);
1739TClass* CcdbApi::tinfo2TClass(std::type_info
const& tinfo)
1741 TClass* cl = TClass::GetClass(tinfo);
1743 throw std::runtime_error(fmt::format(
"Could not retrieve ROOT dictionary for type {}, aborting", tinfo.name()));
1749int CcdbApi::updateMetadata(std::string
const&
path, std::map<std::string, std::string>
const& metadata,
long timestamp, std::string
const&
id,
long newEOV)
1753 curl_easy_setopt(curl, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1754 if (curl !=
nullptr) {
1756 for (
size_t hostIndex = 0; hostIndex < hostsPool.size(); hostIndex++) {
1759 stringstream fullUrl;
1760 fullUrl << getHostUrl(hostIndex) <<
"/" <<
path <<
"/" << timestamp;
1762 fullUrl <<
"/" << newEOV;
1765 fullUrl <<
"/" <<
id;
1769 for (
auto& kv : metadata) {
1770 std::string mfirst = kv.first;
1771 std::string msecond = kv.second;
1773 char* mfirstEncoded = curl_easy_escape(curl, mfirst.c_str(), mfirst.size());
1774 char* msecondEncoded = curl_easy_escape(curl, msecond.c_str(), msecond.size());
1775 fullUrl << std::string(mfirstEncoded) +
"=" + std::string(msecondEncoded) +
"&";
1776 curl_free(mfirstEncoded);
1777 curl_free(msecondEncoded);
1780 if (curl !=
nullptr) {
1781 LOG(
debug) <<
"passing to curl: " << fullUrl.str();
1782 curl_easy_setopt(curl, CURLOPT_URL, fullUrl.str().c_str());
1783 curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST,
"PUT");
1784 curl_easy_setopt(curl, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1785 curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);
1788 struct curl_slist*
list = appendGateToken(
nullptr, fullUrl.str());
1789 curl_easy_setopt(curl, CURLOPT_HTTPHEADER,
list);
1793 res =
static_cast<CURLcode
>(CURL_perform(curl));
1794 if (
res != CURLE_OK) {
1795 LOGP(alarm,
"CURL_perform() failed: {}, code: {}", curl_easy_strerror(
res),
int(
res));
1800 curl_slist_free_all(
list);
1805 curl_easy_cleanup(curl);
1810void CcdbApi::initHostsPool(std::string hosts)
1813 auto splitted = hosts | std::views::transform([](
char c) {
return (
c ==
';') ?
',' :
c; }) | std::views::split(
',');
1814 for (
auto&& part : splitted) {
1815 hostsPool.emplace_back(part.begin(), part.end());
1819std::string CcdbApi::getHostUrl(
int hostIndex)
const
1821 return hostsPool.at(hostIndex);
1827 data->hoPair.object = &requestContext.
dest;
1829 std::function<bool(std::string)> localContentCallback = [
this, &requestContext](std::string
url) {
1833 auto writeCallback = [](
void* contents,
size_t size,
size_t nmemb,
void* chunkptr) {
1835 auto& chunk = *ho.
object;
1836 size_t realsize =
size * nmemb, sz = 0;
1839 if (chunk.capacity() < chunk.size() + realsize) {
1841 const char hannot[] =
"header";
1842 size_t hsize = getFlatHeaderSize(ho.header);
1843 auto cl = ho.header.find(
"Content-Length");
1844 if (cl != ho.header.end()) {
1845 size_t sizeFromHeader = std::stol(cl->second);
1846 sz = hsize + std::max(chunk.size() * (sizeFromHeader ? 1 : 2) + realsize, sizeFromHeader);
1848 sz = hsize + std::max(chunk.size() * 2, chunk.size() + realsize);
1853 char* contC = (
char*)contents;
1854 chunk.insert(chunk.end(), contC, contC + realsize);
1855 }
catch (std::exception e) {
1863 curl_easy_setopt(curl_handle, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1864 std::string fullUrl = getFullUrlForRetrieval(curl_handle, requestContext.
path, requestContext.
metadata, requestContext.
timestamp);
1867 data->hosts = hostsPool;
1870 data->localContentCallback = localContentCallback;
1871 data->userAgent = mUniqueAgentID;
1878 data->optionsLists.reserve(hostsPool.size());
1879 for (
size_t hostIndex = 0; hostIndex < hostsPool.size(); hostIndex++) {
1880 curl_slist* hostOptions =
nullptr;
1881 const std::string hostUrl = getFullUrlForRetrieval(curl_handle, requestContext.
path, requestContext.
metadata,
1883 initCurlHTTPHeaderOptionsForRetrieve(curl_handle, hostOptions, requestContext.
timestamp, &requestContext.
headers,
1886 data->optionsLists.push_back(hostOptions);
1891 if (!
data->optionsLists.empty()) {
1892 curl_easy_setopt(curl_handle, CURLOPT_HTTPHEADER,
data->optionsLists.front());
1895 curl_easy_setopt(curl_handle, CURLOPT_URL, fullUrl.c_str());
1896 initCurlOptionsForRetrieve(curl_handle, (
void*)(&
data->hoPair), writeCallback,
false);
1897 curl_easy_setopt(curl_handle, CURLOPT_HEADERFUNCTION, header_map_callback<
decltype(
data->hoPair.header)>);
1898 curl_easy_setopt(curl_handle, CURLOPT_HEADERDATA, (
void*)&(
data->hoPair.header));
1899 curl_easy_setopt(curl_handle, CURLOPT_PRIVATE, (
void*)
data);
1902 asynchPerform(curl_handle, requestCounter);
1907 std::hash<std::string> hasher;
1908 std::string semhashedstring =
"aliceccdb" +
std::to_string(hasher(basedir + ccdbpath)).substr(0, 16);
1909 return semhashedstring;
1917 return new boost::interprocess::named_semaphore(boost::interprocess::open_or_create_t{}, semhashedstring.c_str(), 1);
1918 }
catch (std::exception e) {
1919 LOG(warn) <<
"Exception occurred during CCDB (cache) semaphore setup; Continuing without";
1928 if (sem->try_wait()) {
1939 boost::interprocess::named_semaphore semaphore(boost::interprocess::open_only, semaname.c_str());
1940 std::cout <<
"Found CCDB semaphore: " << semaname <<
"\n";
1942 auto success = boost::interprocess::named_semaphore::remove(semaname.c_str());
1944 std::cout <<
"Removed CCDB semaphore: " << semaname <<
"\n";
1949 }
catch (std::exception
const& e) {
1960 namespace fs = std::filesystem;
1961 std::string fileName{
"snapshot.root"};
1963 auto absolutesnapshotdir = fs::weakly_canonical(fs::absolute(snapshotdir));
1964 for (
const auto&
entry : fs::recursive_directory_iterator(absolutesnapshotdir)) {
1965 if (
entry.is_directory()) {
1966 const fs::path& currentDir = fs::canonical(fs::absolute(
entry.path()));
1967 fs::path filePath = currentDir / fileName;
1968 if (fs::exists(filePath) && fs::is_regular_file(filePath)) {
1969 std::cout <<
"Directory with file '" << fileName <<
"': " << currentDir << std::endl;
1973 auto numtokens = pathtokens.size();
1974 if (numtokens < 3) {
1979 std::string
path = pathtokens[numtokens - 3] +
"/" + pathtokens[numtokens - 2] +
"/" + pathtokens[numtokens - 1];
1985 }
catch (std::exception
const& e) {
1986 LOG(info) <<
"Semaphore search had exception " << e.what();
1991 long timestamp, std::map<std::string, std::string>& headers,
1994 if (createSnapshot) {
1995 std::string logfile = mSnapshotCachePath +
"/log";
1996 std::fstream logStream = std::fstream(logfile, ios_base::out | ios_base::app);
1997 if (logStream.is_open()) {
1998 logStream <<
"CCDB-access[" << getpid() <<
"] of " << mUniqueAgentID <<
" to " <<
path <<
" timestamp " << timestamp <<
" for load to memory\n";
2001 if (mInSnapshotMode) {
2006 }
else if (mPreferSnapshotCache && std::filesystem::exists(snapshotpath)) {
2018 if (!mSnapshotCachePath.empty() && !(mInSnapshotMode && mSnapshotTopPath == mSnapshotCachePath)) {
2019 auto semaphore_barrier = std::make_unique<CCDBSemaphore>(mSnapshotCachePath, requestContext.
path);
2021 auto snapshotdir = getSnapshotDir(mSnapshotCachePath, requestContext.
path);
2022 std::string snapshotpath = getSnapshotFile(mSnapshotCachePath, requestContext.
path);
2024 std::fstream logStream;
2025 if (logStream.is_open()) {
2026 logStream <<
"CCDB-access[" << getpid() <<
"] ... " << mUniqueAgentID <<
" downloading to snapshot " << snapshotpath <<
" from memory\n";
2029 LOGP(
debug,
"creating snapshot {} -> {}", requestContext.
path, snapshotpath);
2032 std::ofstream objFile(snapshotpath, std::ios::out | std::ofstream::binary);
2033 std::copy(requestContext.
dest.begin(), requestContext.
dest.end(), std::ostreambuf_iterator<char>(objFile));
2036 updateMetaInformationInLocalFile(snapshotpath, &requestContext.
headers, &querysummary);
2042 std::map<std::string, std::string>
const& metadata,
long timestamp,
2043 std::map<std::string, std::string>* headers, std::string
const&
etag,
2047 destP.reserve(dest.size());
2050 dest.reserve(destP.size());
2051 for (
const auto c : destP) {
2057 std::map<std::string, std::string>
const& metadata,
long timestamp,
2058 std::map<std::string, std::string>* headers, std::string
const&
etag,
2070 std::vector<RequestContext> contexts = {requestContext};
2076 size_t hsize = getFlatHeaderSize(headers), cnt = dest.size();
2077 dest.resize(cnt + hsize);
2078 auto addString = [&dest, &cnt](
const std::string& s) {
2085 for (
auto&
h : headers) {
2087 addString(
h.second);
2089 *
reinterpret_cast<int*
>(&dest[cnt]) = hsize;
2090 std::memcpy(&dest[cnt +
sizeof(
int)], FlatHeaderAnnot,
sizeof(FlatHeaderAnnot));
2095 LOGP(
debug,
"loadFileToMemory {} ETag=[{}]", requestContext.
path, requestContext.
etag);
2096 bool createSnapshot = requestContext.
considerSnapshot && !mSnapshotCachePath.empty();
2098 std::string snapshotpath;
2099 if (mInSnapshotMode || std::filesystem::exists(snapshotpath = getSnapshotFile(mSnapshotCachePath, requestContext.
path))) {
2100 auto semaphore_barrier = std::make_unique<CCDBSemaphore>(mSnapshotCachePath, requestContext.
path);
2111 std::vector<int> fromSnapshots(requestContexts.size());
2112 size_t requestCounter = 0;
2115 for (
int i = 0;
i < requestContexts.size();
i++) {
2117 auto& requestContext = requestContexts.at(
i);
2122 while (requestCounter > 0) {
2127 for (
int i = 0;
i < requestContexts.size();
i++) {
2128 auto& requestContext = requestContexts.at(
i);
2129 if (!requestContext.dest.empty()) {
2130 logReading(requestContext.path, requestContext.timestamp, &requestContext.headers,
2131 fmt::format(
"{}{}", requestContext.considerSnapshot ?
"load to memory" :
"retrieve", fromSnapshots.at(
i) ?
" from snapshot" :
""));
2132 if (requestContext.considerSnapshot && fromSnapshots.at(
i) != 2) {
2141 if (
url.find(
"alien:/", 0) != std::string::npos) {
2142 std::map<std::string, std::string> localHeaders;
2144 auto it = localHeaders.find(
"Error");
2145 if (it != localHeaders.end() && it->second ==
"An error occurred during retrieval") {
2151 if ((
url.find(
"file:/", 0) != std::string::npos)) {
2152 std::string
path =
url.substr(7);
2153 if (std::filesystem::exists(
path)) {
2154 std::map<std::string, std::string> localHeaders;
2156 auto it = localHeaders.find(
"Error");
2157 if (it != localHeaders.end() && it->second ==
"An error occurred during retrieval") {
2170 constexpr size_t MaxCopySize = 0x1L << 25;
2171 auto signalError = [&dest, localHeaders]() {
2175 (*localHeaders)[
"Error"] =
"An error occurred during retrieval";
2178 if (
path.find(
"alien:/") == 0 && !initTGrid()) {
2182 std::string fname(
path);
2183 if (fname.find(
"?filetype=raw") == std::string::npos) {
2184 fname +=
"?filetype=raw";
2186 std::unique_ptr<TFile> sfile{TFile::Open(fname.c_str())};
2187 if (!sfile || sfile->IsZombie()) {
2188 LOG(error) <<
"Failed to open file " << fname;
2192 size_t totalread = 0, fsize = sfile->GetSize(), b00 = sfile->GetBytesRead();
2194 char* dptr = dest.data();
2198 size_t b0 = sfile->GetBytesRead(), b1 = b0 - b00;
2199 size_t readsize = fsize - b1 > MaxCopySize ? MaxCopySize : fsize - b1;
2200 if (readsize == 0) {
2203 sfile->Seek(totalread, TFile::kBeg);
2204 bool failed = sfile->ReadBuffer(dptr, (Int_t)readsize);
2205 nread = sfile->GetBytesRead() - b0;
2206 if (
failed || nread < 0) {
2207 LOG(error) <<
"failed to copy file " << fname <<
" to memory buffer";
2213 }
while (nread == (
long)MaxCopySize);
2215 if (localHeaders && fetchLocalMetaData) {
2216 TMemFile memFile(
"name",
const_cast<char*
>(dest.data()), dest.size(),
"READ");
2217 auto storedmeta = (std::map<std::string, std::string>*)
extractFromTFile(memFile, TClass::GetClass(
"std::map<std::string, std::string>"),
CCDBMETA_ENTRY);
2219 *localHeaders = *storedmeta;
2222 if ((
isSnapshotMode() || mPreferSnapshotCache) && localHeaders->find(
"ETag") == localHeaders->end()) {
2223 (*localHeaders)[
"ETag"] =
path;
2225 if (localHeaders->find(
"fileSize") == localHeaders->end()) {
2226 (*localHeaders)[
"fileSize"] = fmt::format(
"{}", memFile.GetEND());
2232void CcdbApi::checkMetadataKeys(std::map<std::string, std::string>
const& metadata)
const
2238 const std::regex regexPatternSearch(R
"([ :;.,\\/'?!\(\)\{\}\[\]@<>=+*#$&`|~^%])");
2239 bool isInvalid =
false;
2241 for (
auto& el : metadata) {
2242 auto keyMd = el.first;
2244 std::smatch searchRes;
2245 while (std::regex_search(keyMd, searchRes, regexPatternSearch)) {
2247 LOG(error) <<
"Invalid character found in metadata key '" << tmp <<
"\': '" << searchRes.str() <<
"\'";
2248 keyMd = searchRes.suffix();
2252 LOG(fatal) <<
"Some metadata keys have invalid characters, please fix!";
2257void CcdbApi::logReading(
const std::string&
path,
long ts,
const std::map<std::string, std::string>* headers,
const std::string& comment)
const
2259 std::string upath{
path};
2261 auto ent = headers->find(
"Valid-From");
2262 if (ent != headers->end()) {
2263 upath +=
"/" + ent->second;
2265 ent = headers->find(
"ETag");
2266 if (ent != headers->end()) {
2267 upath +=
"/" + ent->second;
2270 upath.erase(remove(upath.begin(), upath.end(),
'\"'), upath.end());
2271 LOGP(info,
"ccdb reads {}{}{} for {} ({}, agent_id: {}), ", mUrl, mUrl.back() ==
'/' ?
"" :
"/", upath, ts < 0 ?
getCurrentTimestamp() : ts, comment, mUniqueAgentID);
2274void CcdbApi::asynchPerform(
CurlHandle* handle,
size_t* requestCounter)
const
2279int CcdbApi::CURL_perform(
CurlHandle* handle)
const
2281 if (mIsCCDBDownloaderPreferred) {
2282 return mDownloader->
perform(handle);
2285 for (
int i = 1;
i <= mCurlRetries && (
result = curl_easy_perform(handle)) != CURLE_OK;
i++) {
2286 usleep(mCurlDelayRetries *
i);
2297 LOG(
debug) <<
"Entering semaphore barrier";
2300 mSem =
new boost::interprocess::named_semaphore(boost::interprocess::open_or_create_t{}, mSemName.c_str(), 1);
2301 }
catch (std::exception e) {
2302 LOG(warn) <<
"Exception occurred during CCDB (cache) semaphore setup; Continuing without";
2307 gSemaRegistry.
add(
this);
2314 LOG(
debug) <<
"Ending semaphore barrier";
2317 if (mSem->try_wait()) {
2319 boost::interprocess::named_semaphore::remove(mSemName.c_str());
2321 gSemaRegistry.
remove(
this);
2327 LOG(
debug) <<
"Cleaning up semaphore registry with count " << mStore.size();
2328 for (
auto& s : mStore) {
std::string createdNotBefore
std::string createdNotAfter
std::vector< std::string > header
std::vector< long > entries
Class for time synchronization of RawReader instances.
void setRequestTimeoutTime(int timeoutMS)
CURLcode perform(CURL *handle)
void asynchSchedule(CURL *handle, size_t *requestCounter)
void setKeepaliveTimeoutTime(int timeoutMS)
void runLoop(bool noWait)
CCDBSemaphore(std::string const &cachepath, std::string const &path)
static std::string generateFileName(const std::string &inp)
std::string list(std::string const &path="", bool latestOnly=false, std::string const &returnFormat="text/plain", long createdNotAfter=-1, long createdNotBefore=-1) 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) const
static bool checkAlienToken()
void runDownloaderLoop(bool noWait)
void releaseNamedSemaphore(boost::interprocess::named_semaphore *sem, std::string const &path) const
static std::map< std::string, std::string > * retrieveMetaInfo(TFile &)
void scheduleDownload(RequestContext &requestContext, size_t *requestCounter) const
TObject * retrieve(std::string const &path, std::map< std::string, std::string > const &metadata, long timestamp) const
void init(std::string const &hosts)
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
std::string const & getURL() 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
bool loadLocalContentToMemory(o2::pmr::vector< char > &dest, std::string &url) const
static void removeLeakingSemaphores(std::string const &basedir, bool remove=false)
void saveSnapshot(RequestContext &requestContext) const
static std::unique_ptr< std::vector< char > > createObjectImage(const T *obj, CcdbObjectInfo *info=nullptr)
static void * extractFromTFile(TFile &file, TClass const *cl, const char *what=CCDBOBJECT_ENTRY)
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) const
void snapshot(std::string const &ccdbrootpath, std::string const &localDir, long timestamp) const
bool isSnapshotMode() const
bool isHostReachable() 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
static std::string determineSemaphoreName(std::string const &basedir, std::string const &objectpath)
void deleteObject(std::string const &path, long timestamp=-1) const
static void appendFlatHeader(o2::pmr::vector< char > &dest, const std::map< std::string, std::string > &headers)
std::vector< std::string > getAllFolders(std::string const &top) const
void vectoredLoadFileToMemory(std::vector< RequestContext > &requestContext) const
static void curlSetSSLOptions(CurlHandle *curl)
boost::interprocess::named_semaphore * createNamedSemaphore(std::string const &path) const
static bool removeSemaphore(std::string const &name, bool remove=false)
std::map< std::string, std::string > retrieveHeaders(std::string const &path, std::map< std::string, std::string > const &metadata, long timestamp=-1) const
CcdbApi()
Default constructor.
static bool getCCDBEntryHeaders(std::string const &url, std::string const &etag, std::vector< std::string > &headers, const std::string &agentID="")
static CCDBQuery * retrieveQueryInfo(TFile &)
static constexpr const char * CCDBQUERY_ENTRY
void truncate(std::string const &path) const
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) const
static void parseCCDBHeaders(std::vector< std::string > const &headers, std::vector< std::string > &pfns, std::string &etag)
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
int updateMetadata(std::string const &path, std::map< std::string, std::string > const &metadata, long timestamp, std::string const &id="", long newEOV=0)
static constexpr const char * CCDBMETA_ENTRY
static constexpr const char * CCDBOBJECT_ENTRY
void navigateSourcesAndLoadFile(RequestContext &requestContext, int &fromSnapshot, size_t *requestCounter) const
std::vector< std::string > parseSubFolders(std::string const &reply) const
virtual ~CcdbApi()
Default destructor.
void add(CCDBSemaphore const *ptr)
void remove(CCDBSemaphore const *ptr)
SemaphoreRegistry()=default
GLdouble GLdouble GLdouble GLdouble top
GLboolean GLboolean GLboolean b
GLsizei const GLfloat * value
GLsizei const GLchar *const * path
GLboolean GLboolean GLboolean GLboolean a
GLbitfield GLuint64 timeout
information complementary to a CCDB object (path, metadata, startTimeValidity, endTimeValidity etc)
bool stdmap_to_jsonfile(std::map< std::string, std::string > const &meta, std::string const &filename)
size_t write_data(void *, size_t size, size_t nmemb, void *)
long getCurrentTimestamp()
returns the timestamp in long corresponding to "now"
size_t(*)(void *, size_t, size_t, void *) CurlWriteCallback
std::string sanitizeObjectName(const std::string &objectName)
bool jsonfile_to_stdmap(std::map< std::string, std::string > &meta, std::string const &filename)
long getFutureTimestamp(int secondsInFuture)
returns the timestamp in long corresponding to "now + secondsInFuture"
void CurlHandle
stands in for libcurl's typedef void CURL without including <curl/curl.h>
size_t CurlWrite_CallbackFunc_StdString2(void *contents, size_t size, size_t nmemb, std::string *s)
std::string timestamp() noexcept
Defining ITS Vertex explicitly as messageable.
std::vector< T, fair::mq::pmr::polymorphic_allocator< T > > vector
void createDirectoriesIfAbsent(std::string const &path)
std::string to_string(gsl::span< T, Size > span)
std::map< std::string, std::string > const & metadata
std::string createdNotAfter
std::map< std::string, std::string > & headers
o2::pmr::vector< char > & dest
std::string createdNotBefore
static DeploymentMode deploymentMode()
static std::string getClassName(const T &obj)
get the class name of the object
static std::unique_ptr< FileImage > createFileImage(const TObject &obj, const std::string &fileName, const std::string &objName)
static void trim(std::string &s)
static std::vector< std::string > tokenize(const std::string &src, char delim, bool trimToken=true, bool skipEmpty=true)
static std::string concat_string(Ts const &... ts)
static bool endsWith(const std::string &s, const std::string &ending)
LOG(info)<< "Compressed in "<< sw.CpuTime()<< " s"
std::unique_ptr< TTree > tree((TTree *) flIn.Get(std::string(o2::base::NameConf::CTFTREENAME).c_str()))
uint64_t const void const *restrict const msg