Project
Loading...
Searching...
No Matches
CcdbApi.cxx
Go to the documentation of this file.
1// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
2// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
3// All rights not expressly granted are reserved.
4//
5// This software is distributed under the terms of the GNU General Public
6// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
7//
8// In applying this license CERN does not waive the privileges and immunities
9// granted to it by virtue of its status as an Intergovernmental Organization
10// or submit itself to any jurisdiction.
11
16
17#include "CCDB/CcdbApi.h"
18#include "CCDB/CCDBDownloader.h"
19#include <curl/curl.h>
20#include "CCDB/CCDBQuery.h"
21
27#include <chrono>
28#include <memory>
29#include <ranges>
30#include <sstream>
31#include <TFile.h>
32#include <TGrid.h>
33#include <TSystem.h>
34#include <TStreamerInfo.h>
35#include <TMemFile.h>
36#include <TH1F.h>
37#include <TTree.h>
38#include <fairlogger/Logger.h>
39#include <TError.h>
40#include <TClass.h>
42#include <algorithm>
43#include <filesystem>
44#include <boost/algorithm/string.hpp>
45#include <iostream>
46#include <mutex>
47#include <boost/interprocess/sync/named_semaphore.hpp>
48#include <regex>
49#include <cstdio>
50#include <string>
51#include <string_view>
52#include <utility>
53#include <TAlienUserAgent.h>
54#include <unordered_set>
55#include "rapidjson/document.h"
56#include "rapidjson/writer.h"
57#include "rapidjson/stringbuffer.h"
58
59namespace o2::ccdb
60{
61
62using namespace std;
63
64std::mutex gIOMutex; // to protect TMemFile IO operations
65unique_ptr<TJAlienCredentials> CcdbApi::mJAlienCredentials = nullptr;
66
67namespace
68{
73std::string_view trimHeaderValue(std::string_view value)
74{
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
78 ? std::string_view{}
79 : value.substr(first, value.find_last_not_of(whitespace) - first + 1);
80}
81
88const std::vector<std::pair<std::string, std::string>>& gateTokenTable()
89{
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) {
100 continue;
101 }
102 auto url = trimHeaderValue(entry.substr(0, eq));
103 // Trimmed: a stray newline in a token makes the request malformed, which
104 // a strict broker rejects with an opaque 400 rather than an auth error.
105 const auto token = trimHeaderValue(entry.substr(eq + 1));
106 while (url.size() > 1 && url.back() == '/') { // normalise, so the boundary test below is exact
107 url.remove_suffix(1);
108 }
109 if (!url.empty() && !token.empty()) {
110 entries.emplace_back(std::string(url), std::string("Authorization: Bearer ").append(token));
111 }
112 }
113 std::sort(entries.begin(), entries.end(),
114 [](const auto& a, const auto& b) { return a.first.size() > b.first.size(); });
115 return entries;
116 }();
117 return table;
118}
119
128curl_slist* appendGateToken(curl_slist* list, std::string_view url)
129{
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());
134 }
135 }
136 return list;
137}
138} // namespace
139
147{
148 public:
149 CCDBSemaphore(std::string const& cachepath, std::string const& path);
151
152 private:
153 boost::interprocess::named_semaphore* mSem = nullptr;
154 std::string mSemName{}; // name under which semaphore is kept by the OS kernel
155};
156
157// Small registry class with the purpose that a static object
158// ensures cleanup of registered semaphores even when programs
159// "crash".
161{
162 public:
163 SemaphoreRegistry() = default;
165 void add(CCDBSemaphore const* ptr);
166 void remove(CCDBSemaphore const* ptr);
167
168 private:
169 std::unordered_set<CCDBSemaphore const*> mStore;
170};
171static SemaphoreRegistry gSemaRegistry;
172
174{
175 using namespace o2::framework;
176 setUniqueAgentID();
177
179 mIsCCDBDownloaderPreferred = 0;
180 if (deploymentMode == DeploymentMode::OnlineDDS && deploymentMode == DeploymentMode::OnlineECS && deploymentMode == DeploymentMode::OnlineAUX && deploymentMode == DeploymentMode::FST) {
181 mIsCCDBDownloaderPreferred = 1;
182 }
183 if (getenv("ALICEO2_ENABLE_MULTIHANDLE_CCDBAPI")) { // todo rename ALICEO2_ENABLE_MULTIHANDLE_CCDBAPI to ALICEO2_PREFER_MULTIHANDLE_CCDBAPI
184 mIsCCDBDownloaderPreferred = atoi(getenv("ALICEO2_ENABLE_MULTIHANDLE_CCDBAPI"));
185 }
186 mDownloader = new CCDBDownloader();
187}
188
190{
191 curl_global_cleanup();
192 delete mDownloader;
193}
194
195void CcdbApi::setUniqueAgentID()
196{
197 mUniqueAgentID = TAlienUserAgent::BasedOnEnvironment().ToString();
198}
199
201{
202#ifdef __APPLE__
203 LOG(debug) << "On macOS we simply rely on TGrid::Connect(\"alien\").";
204 return true;
205#endif
206 if (getenv("ALICEO2_CCDB_NOTOKENCHECK") && atoi(getenv("ALICEO2_CCDB_NOTOKENCHECK"))) {
207 return true;
208 }
209 if (getenv("JALIEN_TOKEN_CERT")) {
210 return true;
211 }
212 auto returncode = system("LD_PRELOAD= alien-token-info &> /dev/null");
213 if (returncode == -1) {
214 LOG(error) << "...";
215 }
216 return returncode == 0;
217}
218
219void CcdbApi::curlInit()
220{
221 // todo : are there other things to initialize globally for curl ?
222 curl_global_init(CURL_GLOBAL_DEFAULT);
223 CcdbApi::mJAlienCredentials = std::make_unique<TJAlienCredentials>();
224 CcdbApi::mJAlienCredentials->loadCredentials();
225 CcdbApi::mJAlienCredentials->selectPreferedCredentials();
226
227 // allow to configure the socket timeout of CCDBDownloader (for some tuning studies)
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";
232 mDownloader->setKeepaliveTimeoutTime(timeoutMS);
233 }
234 }
235}
236
237void CcdbApi::init(std::string const& host)
238{
239 if (host.empty()) {
240 throw std::invalid_argument("Empty url passed CcdbApi, cannot initialize. Aborting.");
241 }
242
243 // if host is prefixed with "file://" this is a local snapshot
244 // in this case we init the API in snapshot (readonly) mode
245 constexpr const char* SNAPSHOTPREFIX = "file://";
246 mUrl = host;
247
248 if (host.substr(0, 7).compare(SNAPSHOTPREFIX) == 0) {
249 auto path = host.substr(7);
250 initInSnapshotMode(path);
251 } else {
252 initHostsPool(host);
253 curlInit();
254 }
255 // The environment option ALICEO2_CCDB_LOCALCACHE allows
256 // to reduce the number of queries to the server, by collecting the objects in a local
257 // cache folder, and serving from this folder for repeated queries.
258 // This is useful for instance for MC GRID productions in which we spawn
259 // many isolated processes, all querying the CCDB (for potentially the same objects and same timestamp).
260 // In addition, we can monitor exactly which objects are fetched and what is their content.
261 // One can also distribute so obtained caches to sites without network access.
262 //
263 // THE INFORMATION BELOW IS TEMPORARILY WRONG: the functionality of checking the validity if IGNORE_VALIDITYCHECK_OF_CCDB_LOCALCACHE
264 // is NOT set is broken. At the moment the code is modified to behave as if the IGNORE_VALIDITYCHECK_OF_CCDB_LOCALCACHE is always set
265 // whenever the ALICEO2_CCDB_LOCALCACHE is defined.
266 //
267 // When used with the DPL CCDB fetcher (i.e. loadFileToMemory is called), in order to prefer the available snapshot w/o its validity
268 // check an extra variable IGNORE_VALIDITYCHECK_OF_CCDB_LOCALCACHE must be defined, otherwhise the object will be fetched from the
269 // server after the validity check and new snapshot will be created if needed
270
271 std::string snapshotReport{};
272 const char* cachedir = getenv("ALICEO2_CCDB_LOCALCACHE");
273 namespace fs = std::filesystem;
274 if (cachedir) {
275 if (cachedir[0] == 0) {
276 mSnapshotCachePath = fs::weakly_canonical(fs::absolute("."));
277 } else {
278 mSnapshotCachePath = fs::weakly_canonical(fs::absolute(cachedir));
279 }
280 snapshotReport = fmt::format("(cache snapshots to dir={}", mSnapshotCachePath);
281 }
282 if (cachedir) { // || getenv("IGNORE_VALIDITYCHECK_OF_CCDB_LOCALCACHE")) {
283 mPreferSnapshotCache = true;
284 if (mSnapshotCachePath.empty()) {
285 LOGP(fatal, "IGNORE_VALIDITYCHECK_OF_CCDB_LOCALCACHE is defined but the ALICEO2_CCDB_LOCALCACHE is not");
286 }
287 snapshotReport += ", prefer if available";
288 }
289 if (!snapshotReport.empty()) {
290 snapshotReport += ')';
291 }
292
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);
294 // Set the curl timeout. It can be forced with an env var or it has different defaults based on the deployment mode.
295 if (getenv("ALICEO2_CCDB_CURL_TIMEOUT_DOWNLOAD")) {
296 auto timeout = atoi(getenv("ALICEO2_CCDB_CURL_TIMEOUT_DOWNLOAD"));
297 if (timeout >= 0) { // if valid int
298 mCurlTimeoutDownload = timeout;
299 }
300 } else { // set a default depending on the deployment mode
302 if (deploymentMode == o2::framework::DeploymentMode::OnlineDDS ||
305 mCurlTimeoutDownload = 15;
306 } else if (deploymentMode == o2::framework::DeploymentMode::Grid ||
307 deploymentMode == o2::framework::DeploymentMode::FST) {
308 mCurlTimeoutDownload = 15;
309 } else if (deploymentMode == o2::framework::DeploymentMode::Local) {
310 mCurlTimeoutDownload = 5;
311 }
312 }
313
314 if (getenv("ALICEO2_CCDB_CURL_TIMEOUT_UPLOAD")) {
315 auto timeout = atoi(getenv("ALICEO2_CCDB_CURL_TIMEOUT_UPLOAD"));
316 if (timeout >= 0) { // if valid int
317 mCurlTimeoutUpload = timeout;
318 }
319 } else { // set a default depending on the deployment mode
321 if (deploymentMode == o2::framework::DeploymentMode::OnlineDDS ||
324 mCurlTimeoutUpload = 3;
325 } else if (deploymentMode == o2::framework::DeploymentMode::Grid ||
326 deploymentMode == o2::framework::DeploymentMode::FST) {
327 mCurlTimeoutUpload = 20;
328 } else if (deploymentMode == o2::framework::DeploymentMode::Local) {
329 mCurlTimeoutUpload = 20;
330 }
331 }
332 if (mDownloader) {
333 mDownloader->setRequestTimeoutTime(mCurlTimeoutDownload * 1000L);
334 }
335
336 LOGP(debug, "Curl timeouts are set to: download={:2}, upload={:2} seconds", mCurlTimeoutDownload, mCurlTimeoutUpload);
337
338 LOGP(info, "Init CcdApi with UserAgentID: {}, Host: {}{}, Curl timeouts: upload:{} download:{}", mUniqueAgentID, host,
339 mInSnapshotMode ? "(snapshot readonly mode)" : snapshotReport.c_str(), mCurlTimeoutUpload, mCurlTimeoutDownload);
340}
341
343{
344 mDownloader->runLoop(noWait);
345}
346
347// A helper function used in a few places. Updates a ROOT file with meta/header information.
348void CcdbApi::updateMetaInformationInLocalFile(std::string const& filename, std::map<std::string, std::string> const* headers, CCDBQuery const* querysummary)
349{
350 std::lock_guard<std::mutex> guard(gIOMutex);
351 auto oldlevel = gErrorIgnoreLevel;
352 gErrorIgnoreLevel = 6001; // ignoring error messages here (since we catch with IsZombie)
353 TFile snapshotfile(filename.c_str(), "UPDATE");
354 // The assumption is that the blob is a ROOT file
355 if (!snapshotfile.IsZombie()) {
356 if (querysummary && !snapshotfile.Get(CCDBQUERY_ENTRY)) {
357 snapshotfile.WriteObjectAny(querysummary, TClass::GetClass(typeid(*querysummary)), CCDBQUERY_ENTRY);
358 }
359 if (headers && !snapshotfile.Get(CCDBMETA_ENTRY)) {
360 snapshotfile.WriteObjectAny(headers, TClass::GetClass(typeid(*headers)), CCDBMETA_ENTRY);
361 }
362 snapshotfile.Write();
363 snapshotfile.Close();
364 }
365 gErrorIgnoreLevel = oldlevel;
366}
367
373std::string sanitizeObjectName(const std::string& objectName)
374{
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;
380}
381
382std::unique_ptr<std::vector<char>> CcdbApi::createObjectImage(const void* obj, std::type_info const& tinfo, CcdbObjectInfo* info)
383{
384 // Create a binary image of the object, if CcdbObjectInfo pointer is provided, register there
385 // the assigned object class name and the filename
386 std::lock_guard<std::mutex> guard(gIOMutex);
387 std::string className = o2::utils::MemFileHelper::getClassName(tinfo);
388 std::string tmpFileName = generateFileName(className);
389 if (info) {
390 info->setFileName(tmpFileName);
391 info->setObjectType(className);
392 }
393 return o2::utils::MemFileHelper::createFileImage(obj, tinfo, tmpFileName, CCDBOBJECT_ENTRY);
394}
395
396std::unique_ptr<std::vector<char>> CcdbApi::createObjectImage(const TObject* rootObject, CcdbObjectInfo* info)
397{
398 // Create a binary image of the object, if CcdbObjectInfo pointer is provided, register there
399 // the assigned object class name and the filename
400 std::string className = rootObject->GetName();
401 std::string tmpFileName = generateFileName(className);
402 if (info) {
403 info->setFileName(tmpFileName);
404 info->setObjectType("TObject"); // why TObject and not the actual name?
405 }
406 std::lock_guard<std::mutex> guard(gIOMutex);
407 return o2::utils::MemFileHelper::createFileImage(*rootObject, tmpFileName, CCDBOBJECT_ENTRY);
408}
409
410int CcdbApi::storeAsTFile_impl(const void* obj, std::type_info const& tinfo, std::string const& path,
411 std::map<std::string, std::string> const& metadata,
412 long startValidityTimestamp, long endValidityTimestamp,
413 std::vector<char>::size_type maxSize) const
414{
415 // We need the TClass for this type; will verify if dictionary exists
416 if (!obj) {
417 LOGP(error, "nullptr is provided for object {}/{}/{}", path, startValidityTimestamp, endValidityTimestamp);
418 return -1;
419 }
420 CcdbObjectInfo info;
421 auto img = createObjectImage(obj, tinfo, &info);
422 return storeAsBinaryFile(img->data(), img->size(), info.getFileName(), info.getObjectType(),
423 path, metadata, startValidityTimestamp, endValidityTimestamp, maxSize);
424}
425
426int CcdbApi::storeAsBinaryFile(const char* buffer, size_t size, const std::string& filename, const std::string& objectType,
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
429{
430 if (maxSize > 0 && size > maxSize) {
431 LOGP(alarm, "Object will not be uploaded to {} since its size {} exceeds max allowed {}", path, size, maxSize);
432 return -1;
433 }
434 int returnValue = 0;
435
436 // Prepare URL
437 long sanitizedStartValidityTimestamp = startValidityTimestamp;
438 if (startValidityTimestamp == -1) {
439 LOGP(info, "Start of Validity not set, current timestamp used.");
440 sanitizedStartValidityTimestamp = getCurrentTimestamp();
441 }
442 long sanitizedEndValidityTimestamp = endValidityTimestamp;
443 if (endValidityTimestamp == -1) {
444 LOGP(info, "End of Validity not set, start of validity plus 1 day used.");
445 sanitizedEndValidityTimestamp = getFutureTimestamp(60 * 60 * 24 * 1);
446 }
447 if (mInSnapshotMode) { // write local file
448 if (filename.empty() || buffer == nullptr || size == 0) {
449 LOGP(alarm, "Snapshot mode does not support headers-only upload");
450 return -3;
451 }
452 auto pthLoc = getSnapshotDir(mSnapshotTopPath, path);
454 auto flLoc = getSnapshotFile(mSnapshotTopPath, path, filename);
455 // add the timestamps to the end
456 auto pent = flLoc.find_last_of('.');
457 if (pent == std::string::npos) {
458 pent = flLoc.size();
459 }
460 flLoc.insert(pent, fmt::format("_{}_{}", startValidityTimestamp, endValidityTimestamp));
461 ofstream outf(flLoc.c_str(), ios::out | ios::binary);
462 outf.write(buffer, size);
463 outf.close();
464 if (!outf.good()) {
465 throw std::runtime_error(fmt::format("Failed to write local CCDB file {}", flLoc));
466 } else {
467 std::map<std::string, std::string> metaheader(metadata);
468 // add time validity information
469 metaheader["Valid-From"] = std::to_string(startValidityTimestamp);
470 metaheader["Valid-Until"] = std::to_string(endValidityTimestamp);
471 updateMetaInformationInLocalFile(flLoc.c_str(), &metaheader);
472 std::string metaStr{};
473 for (const auto& mentry : metadata) {
474 metaStr += fmt::format("{}={};", mentry.first, mentry.second);
475 }
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 "{}")",
479 path, flLoc, CCDBOBJECT_ENTRY, startValidityTimestamp, endValidityTimestamp, metaStr);
480 }
481 return returnValue;
482 }
483
484 // Curl preparation
485 CurlHandle* curl = nullptr;
486 curl = curl_easy_init();
487
488 // checking that all metadata keys do not contain invalid characters
489 checkMetadataKeys(metadata);
490
491 if (curl != nullptr) {
492 auto mime = curl_mime_init(curl);
493 auto field = curl_mime_addpart(mime);
494 curl_mime_name(field, "send");
495 if (!filename.empty()) {
496 curl_mime_filedata(field, filename.c_str());
497 }
498 if (buffer != nullptr && size > 0) {
499 curl_mime_data(field, buffer, size);
500 } else {
501 curl_mime_data(field, "", 0);
502 }
503
504 curlSetSSLOptions(curl);
505
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);
510
511 CURLcode res = CURL_LAST;
512
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;
516 /* what URL that receives this POST */
517 curl_easy_setopt(curl, CURLOPT_URL, fullUrl.c_str());
518
519 // Per host: the gate token is per endpoint (see appendGateToken).
520 struct curl_slist* headerlist = curl_slist_append(nullptr, "Expect:");
521 headerlist = appendGateToken(headerlist, fullUrl);
522 curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headerlist);
523
524 /* Perform the request, res will get the return code */
525 res = static_cast<CURLcode>(CURL_perform(curl));
526 /* Check for errors */
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);
530 } else { // generic message
531 LOGP(alarm, "curl_easy_perform() failed: {}", curl_easy_strerror(res));
532 }
533 returnValue = res;
534 }
535 curl_slist_free_all(headerlist);
536 }
537
538 /* always cleanup */
539 curl_easy_cleanup(curl);
540
541 /* free mime */
542 curl_mime_free(mime);
543 } else {
544 LOGP(alarm, "curl initialization failure");
545 returnValue = -2;
546 }
547 return returnValue;
548}
549
550int CcdbApi::storeAsTFile(const TObject* rootObject, std::string const& path, std::map<std::string, std::string> const& metadata,
551 long startValidityTimestamp, long endValidityTimestamp, std::vector<char>::size_type maxSize) const
552{
553 // Prepare file
554 if (!rootObject) {
555 LOGP(error, "nullptr is provided for object {}/{}/{}", path, startValidityTimestamp, endValidityTimestamp);
556 return -1;
557 }
558 CcdbObjectInfo info;
559 auto img = createObjectImage(rootObject, &info);
560 return storeAsBinaryFile(img->data(), img->size(), info.getFileName(), info.getObjectType(), path, metadata, startValidityTimestamp, endValidityTimestamp, maxSize);
561}
562
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
566{
567 // Prepare timestamps
568 std::string startValidityString = getTimestampString(startValidityTimestamp < 0 ? getCurrentTimestamp() : startValidityTimestamp);
569 std::string endValidityString = getTimestampString(endValidityTimestamp < 0 ? getFutureTimestamp(60 * 60 * 24 * 1) : endValidityTimestamp);
570 // Get url
571 std::string url = getHostUrl(hostIndex);
572 // Build URL
573 std::string fullUrl = url + "/" + path + "/" + startValidityString + "/" + endValidityString + "/";
574 // Add type as part of metadata
575 // we need to URL encode the object type, since in case it has special characters (like the "<", ">" for templated classes) it won't work otherwise
576 char* objtypeEncoded = curl_easy_escape(curl, objtype.c_str(), objtype.size());
577 fullUrl += "ObjectType=" + std::string(objtypeEncoded) + "/";
578 curl_free(objtypeEncoded);
579 // Add general metadata
580 for (auto& kv : metadata) {
581 std::string mfirst = kv.first;
582 std::string msecond = kv.second;
583 // same trick for the metadata as for the object type
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);
589 }
590 return fullUrl;
591}
592
593// todo make a single method of the one above and below
594std::string CcdbApi::getFullUrlForRetrieval(CurlHandle* curl, const std::string& path, const std::map<std::string, std::string>& metadata, long timestamp, int hostIndex) const
595{
596 if (mInSnapshotMode) {
597 return getSnapshotFile(mSnapshotTopPath, path);
598 }
599
600 // Prepare timestamps
601 std::string validityString = getTimestampString(timestamp < 0 ? getCurrentTimestamp() : timestamp);
602 // Get host url
603 std::string hostUrl = getHostUrl(hostIndex);
604 // Build URL
605 std::string fullUrl = hostUrl + "/" + path + "/" + validityString + "/";
606 // Add metadata
607 for (auto& kv : metadata) {
608 std::string mfirst = kv.first;
609 std::string msecond = kv.second;
610 // trick for the metadata in case it contains special characters
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);
616 }
617 return fullUrl;
618}
619
624 char* memory;
625 unsigned int size;
626};
627
637static size_t WriteMemoryCallback(void* contents, size_t size, size_t nmemb, void* userp)
638{
639 size_t realsize = size * nmemb;
640 auto* mem = (struct MemoryStruct*)userp;
641
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");
645 return 0;
646 }
647
648 memcpy(&(mem->memory[mem->size]), contents, realsize);
649 mem->size += realsize;
650 mem->memory[mem->size] = 0;
651
652 return realsize;
653}
654
666static size_t WriteToFileCallback(void* ptr, size_t size, size_t nmemb, FILE* stream)
667{
668 size_t written = fwrite(ptr, size, nmemb, stream);
669 return written;
670}
671
679static CURLcode ssl_ctx_callback(CurlHandle*, void*, void* parm)
680{
681 std::string msg((const char*)parm);
682 int start = 0, end = msg.find('\n');
683
684 if (msg.length() > 0 && end == -1) {
685 LOG(warn) << msg;
686 } else if (end > 0) {
687 while (end > 0) {
688 LOG(warn) << msg.substr(start, end - start);
689 start = end + 1;
690 end = msg.find('\n', start);
691 }
692 }
693 return CURLE_OK;
694}
695
697{
698 CredentialsKind cmk = mJAlienCredentials->getPreferedCredentials();
699
700 /* NOTE: return early, the warning should be printed on SSL callback if needed */
701 if (cmk == cNOT_FOUND) {
702 return;
703 }
704
705 TJAlienCredentialsObject cmo = mJAlienCredentials->get(cmk);
706
707 char* CAPath = getenv("X509_CERT_DIR");
708 if (CAPath) {
709 curl_easy_setopt(curl_handle, CURLOPT_CAPATH, CAPath);
710 }
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());
714
715 // NOTE: for lazy logging only
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());
718
719 // CURLcode ret = curl_easy_setopt(curl_handle, CURLOPT_SSL_CTX_FUNCTION, *ssl_ctx_callback);
720}
721
722using CurlWriteCallback = size_t (*)(void*, size_t, size_t, void*);
723
724void CcdbApi::initCurlOptionsForRetrieve(CurlHandle* curlHandle, void* chunk, CurlWriteCallback writeCallback, bool followRedirect) const
725{
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);
729}
730
731namespace
732{
733template <typename MapType = std::map<std::string, std::string>>
734size_t header_map_callback(char* buffer, size_t size, size_t nitems, void* userdata)
735{
736 auto* headers = static_cast<MapType*>(userdata);
737 auto header = std::string(buffer, size * nitems);
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));
741 const auto value = boost::algorithm::trim_copy(header.substr(index + 1));
742 LOGP(debug, "Adding #{} {} -> {}", headers->size(), key, value);
743 bool insert = true;
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)) {
748 headers->erase(key);
749 } else {
750 insert = false;
751 }
752 }
753 }
754
755 // Keep only the first ETag encountered
756 if (key == "ETag") {
757 auto cl = headers->find("ETag");
758 if (cl != headers->end()) {
759 insert = false;
760 }
761 }
762
763 // Keep only the first Content-Type encountered
764 if (key == "Content-Type") {
765 auto cl = headers->find("Content-Type");
766 if (cl != headers->end()) {
767 insert = false;
768 }
769 }
770
771 if (insert) {
772 headers->insert(std::make_pair(key, value));
773 }
774 }
775 return size * nitems;
776}
777} // namespace
778
779void CcdbApi::initCurlHTTPHeaderOptionsForRetrieve(CurlHandle* curlHandle, curl_slist*& option_list, long timestamp, std::map<std::string, std::string>* headers, std::string const& etag,
780 const std::string& createdNotAfter, const std::string& createdNotBefore, std::string_view url) const
781{
782 // struct curl_slist* list = nullptr;
783 if (!etag.empty()) {
784 option_list = curl_slist_append(option_list, ("If-None-Match: " + etag).c_str());
785 }
786
787 if (!createdNotAfter.empty()) {
788 option_list = curl_slist_append(option_list, ("If-Not-After: " + createdNotAfter).c_str());
789 }
790
791 if (!createdNotBefore.empty()) {
792 option_list = curl_slist_append(option_list, ("If-Not-Before: " + createdNotBefore).c_str());
793 }
794
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);
799 }
800
801 option_list = appendGateToken(option_list, url);
802
803 // Unconditionally, nullptr included: the handle is reused across hosts, and
804 // skipping the set would leave a previous host's freed list installed.
805 curl_easy_setopt(curlHandle, CURLOPT_HTTPHEADER, option_list);
806
807 curl_easy_setopt(curlHandle, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
808}
809
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,
812 const std::string& createdNotAfter, const std::string& createdNotBefore, bool followRedirect) const
813{
814 return receiveObject((void*)fileHandle, path, metadata, timestamp, headers, etag, createdNotAfter, createdNotBefore, followRedirect, (CurlWriteCallback)&WriteToFileCallback);
815}
816
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,
819 const std::string& createdNotAfter, const std::string& createdNotBefore, bool followRedirect) const
820{
821 return receiveObject((void*)chunk, path, metadata, timestamp, headers, etag, createdNotAfter, createdNotBefore, followRedirect, (CurlWriteCallback)&WriteMemoryCallback);
822}
823
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,
826 const std::string& createdNotAfter, const std::string& createdNotBefore, bool followRedirect, CurlWriteCallback writeCallback) const
827{
828 CurlHandle* curlHandle;
829
830 curlHandle = curl_easy_init();
831 curl_easy_setopt(curlHandle, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
832
833 if (curlHandle != nullptr) {
834
835 curlSetSSLOptions(curlHandle);
836 initCurlOptionsForRetrieve(curlHandle, dataHolder, writeCallback, followRedirect);
837 long responseCode = 0;
838 CURLcode curlResultCode = CURL_LAST;
839
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());
843
844 // Per host: the gate token is per endpoint (see appendGateToken).
845 curl_slist* option_list = nullptr;
846 initCurlHTTPHeaderOptionsForRetrieve(curlHandle, option_list, timestamp, headers, etag, createdNotAfter, createdNotBefore, fullUrl);
847
848 curlResultCode = static_cast<CURLcode>(CURL_perform(curlHandle));
849
850 if (curlResultCode != CURLE_OK) {
851 LOGP(alarm, "curl_easy_perform() failed: {}", curl_easy_strerror(curlResultCode));
852 } else {
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);
857 return true;
858 } else {
859 if (curlResultCode != CURLE_OK) {
860 LOGP(alarm, "invalid URL {}", fullUrl);
861 } else {
862 LOGP(alarm, "not found under link {}", fullUrl);
863 }
864 }
865 }
866 curl_slist_free_all(option_list);
867 }
868
869 curl_easy_cleanup(curlHandle);
870 }
871 return false;
872}
873
874TObject* CcdbApi::retrieve(std::string const& path, std::map<std::string, std::string> const& metadata,
875 long timestamp) const
876{
877 struct MemoryStruct chunk{
878 (char*)malloc(1) /*memory*/, 0 /*size*/
879 };
880
881 TObject* result = nullptr;
882
883 bool res = receiveToMemory((void*)&chunk, path, metadata, timestamp);
884
885 if (res) {
886 std::lock_guard<std::mutex> guard(gIOMutex);
887 TMessage mess(kMESS_OBJECT);
888 mess.SetBuffer(chunk.memory, chunk.size, kFALSE);
889 mess.SetReadMode();
890 mess.Reset();
891 result = (TObject*)(mess.ReadObjectAny(mess.GetClass()));
892 if (result == nullptr) {
893 LOGP(info, "couldn't retrieve the object {}", path);
894 }
895 }
896
897 free(chunk.memory);
898
899 return result;
900}
901
902std::string CcdbApi::generateFileName(const std::string& inp)
903{
904 // generate file name for the CCDB object (for now augment the input string by the timestamp)
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("::"), "-");
909 return str;
910}
911
912TObject* CcdbApi::retrieveFromTFile(std::string const& path, std::map<std::string, std::string> const& metadata,
913 long timestamp, std::map<std::string, std::string>* headers, std::string const& etag,
914 const std::string& createdNotAfter, const std::string& createdNotBefore) const
915{
916 return (TObject*)retrieveFromTFile(typeid(TObject), path, metadata, timestamp, headers, etag, createdNotAfter, createdNotBefore);
917}
918
919bool CcdbApi::retrieveBlob(std::string const& path, std::string const& targetdir, std::map<std::string, std::string> const& metadata,
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
921{
922
923 // we setup the target path for this blob
924 std::string fulltargetdir = targetdir + (preservePath ? ('/' + path) : "");
925
926 try {
928 } catch (std::exception e) {
929 LOGP(error, "Could not create local snapshot cache directory {}, reason: {}", fulltargetdir, e.what());
930 return false;
931 }
932
934 std::map<std::string, std::string> headers;
935 // avoid creating snapshot via loadFileToMemory itself
936 loadFileToMemory(buff, path, metadata, timestamp, &headers, "", createdNotAfter, createdNotBefore, false);
937 if ((headers.count("Error") != 0) || (buff.empty())) {
938 LOGP(error, "Unable to find object {}/{}, Aborting", path, timestamp);
939 return false;
940 }
941 // determine local filename --> use user given one / default -- or if empty string determine from content
942 auto getFileName = [&headers]() {
943 auto& s = headers["Content-Disposition"];
944 if (s != "") {
945 std::regex re("(.*;)filename=\"(.*)\"");
946 std::cmatch m;
947 if (std::regex_match(s.c_str(), m, re)) {
948 return m[2].str();
949 }
950 }
951 std::string backupname("ccdb-blob.bin");
952 LOG(error) << "Cannot determine original filename from Content-Disposition ... falling back to " << backupname;
953 return backupname;
954 };
955 auto filename = localFileName.size() > 0 ? localFileName : getFileName();
956 std::string targetpath = fulltargetdir + "/" + filename;
957 {
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);
962 return false;
963 }
964 }
965 CCDBQuery querysummary(path, metadata, timestamp);
966
967 updateMetaInformationInLocalFile(targetpath.c_str(), &headers, &querysummary);
968 if (outHeaders) {
969 *outHeaders = std::move(headers);
970 }
971 return true;
972}
973
974void CcdbApi::snapshot(std::string const& ccdbrootpath, std::string const& localDir, long timestamp) const
975{
976 // query all subpaths to ccdbrootpath
977 const auto allfolders = getAllFolders(ccdbrootpath);
978 std::map<std::string, std::string> metadata;
979 for (auto& folder : allfolders) {
980 retrieveBlob(folder, localDir, metadata, timestamp);
981 }
982}
983
984void* CcdbApi::extractFromTFile(TFile& file, TClass const* cl, const char* what)
985{
986 if (!cl) {
987 return nullptr;
988 }
989 auto object = file.GetObjectChecked(what, cl);
990 if (!object) {
991 // it could be that object was stored with previous convention
992 // where the classname was taken as key
993 std::string objectName(cl->GetName());
994 o2::utils::Str::trim(objectName);
995 object = file.GetObjectChecked(objectName.c_str(), cl);
996 LOG(warn) << "Did not find object under expected name " << what;
997 if (!object) {
998 return nullptr;
999 }
1000 LOG(warn) << "Found object under deprecated name " << cl->GetName();
1001 }
1002 auto result = object;
1003 // We need to handle some specific cases as ROOT ties them deeply
1004 // to the file they are contained in
1005 if (cl->InheritsFrom("TObject")) {
1006 // make a clone
1007 // detach from the file
1008 auto tree = dynamic_cast<TTree*>((TObject*)object);
1009 if (tree) {
1010 tree->LoadBaskets(0x1L << 32); // make tree memory based
1011 tree->SetDirectory(nullptr);
1012 result = tree;
1013 } else {
1014 auto h = dynamic_cast<TH1*>((TObject*)object);
1015 if (h) {
1016 h->SetDirectory(nullptr);
1017 result = h;
1018 }
1019 }
1020 }
1021 return result;
1022}
1023
1024void* CcdbApi::extractFromLocalFile(std::string const& filename, std::type_info const& tinfo, std::map<std::string, std::string>* headers) const
1025{
1026 if (!std::filesystem::exists(filename)) {
1027 LOG(error) << "Local snapshot " << filename << " not found \n";
1028 return nullptr;
1029 }
1030 std::lock_guard<std::mutex> guard(gIOMutex);
1031 auto tcl = tinfo2TClass(tinfo);
1032 TFile f(filename.c_str(), "READ");
1033 if (headers) {
1034 auto storedmeta = retrieveMetaInfo(f);
1035 if (storedmeta) {
1036 *headers = *storedmeta; // do a simple deep copy
1037 delete storedmeta;
1038 }
1039 if ((isSnapshotMode() || mPreferSnapshotCache) && headers->find("ETag") == headers->end()) { // generate dummy ETag to profit from the caching
1040 (*headers)["ETag"] = filename;
1041 }
1042 if (headers->find("fileSize") == headers->end()) {
1043 (*headers)["fileSize"] = fmt::format("{}", f.GetEND());
1044 }
1045 }
1046 return extractFromTFile(f, tcl);
1047}
1048
1049bool CcdbApi::initTGrid() const
1050{
1051 if (mNeedAlienToken && !gGrid) {
1052 static bool allowNoToken = getenv("ALICEO2_CCDB_NOTOKENCHECK") && atoi(getenv("ALICEO2_CCDB_NOTOKENCHECK"));
1053 if (!allowNoToken && !checkAlienToken()) {
1054 LOG(fatal) << "Alien Token Check failed - Please get an alien token before running with https CCDB endpoint, or alice-ccdb.cern.ch!";
1055 }
1056 TGrid::Connect("alien");
1057 static bool errorShown = false;
1058 if (!gGrid && errorShown == false) {
1059 if (allowNoToken) {
1060 LOG(error) << "TGrid::Connect returned nullptr. May be due to missing alien token";
1061 } else {
1062 LOG(fatal) << "TGrid::Connect returned nullptr. May be due to missing alien token";
1063 }
1064 errorShown = true;
1065 }
1066 }
1067 return gGrid != nullptr;
1068}
1069
1070void* CcdbApi::downloadFilesystemContent(std::string const& url, std::type_info const& tinfo, std::map<std::string, std::string>* headers) const
1071{
1072 if ((url.find("alien:/", 0) != std::string::npos) && !initTGrid()) {
1073 return nullptr;
1074 }
1075 std::lock_guard<std::mutex> guard(gIOMutex);
1076 auto memfile = TMemFile::Open(url.c_str(), "OPEN");
1077 if (memfile) {
1078 auto cl = tinfo2TClass(tinfo);
1079 auto content = extractFromTFile(*memfile, cl);
1080 if (headers && headers->find("fileSize") == headers->end()) {
1081 (*headers)["fileSize"] = fmt::format("{}", memfile->GetEND());
1082 }
1083 delete memfile;
1084 return content;
1085 }
1086 return nullptr;
1087}
1088
1089void* CcdbApi::interpretAsTMemFileAndExtract(char* contentptr, size_t contentsize, std::type_info const& tinfo)
1090{
1091 void* result = nullptr;
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);
1099 result = extractFromTFile(memFile, tcl);
1100 if (!result) {
1101 LOG(error) << o2::utils::Str::concat_string("Couldn't retrieve object corresponding to ", tcl->GetName(), " from TFile");
1102 }
1103 memFile.Close();
1104 }
1105 return result;
1106}
1107
1108// navigate sequence of URLs until TFile content is found; object is extracted and returned
1109void* CcdbApi::navigateURLsAndRetrieveContent(CurlHandle* curl_handle, std::string const& url, std::type_info const& tinfo, std::map<std::string, std::string>* headers) const
1110{
1111 // a global internal data structure that can be filled with HTTP header information
1112 // static --> to avoid frequent alloc/dealloc as optimization
1113 // not sure if thread_local takes away that benefit
1114 static thread_local std::multimap<std::string, std::string> headerData;
1115
1116 // let's see first of all if the url is something specific that curl cannot handle
1117 if ((url.find("alien:/", 0) != std::string::npos) || (url.find("file:/", 0) != std::string::npos)) {
1118 return downloadFilesystemContent(url, tinfo, headers);
1119 }
1120 // add other final cases here
1121 // example root://
1122
1123 // otherwise make an HTTP/CURL request
1124 // specify URL to get
1125 curl_easy_setopt(curl_handle, CURLOPT_URL, url.c_str());
1126
1127 MemoryStruct chunk{(char*)malloc(1), 0};
1128 initCurlOptionsForRetrieve(curl_handle, (void*)&chunk, WriteMemoryCallback, false);
1129
1130 curl_easy_setopt(curl_handle, CURLOPT_HEADERFUNCTION, header_map_callback<decltype(headerData)>);
1131 headerData.clear();
1132 curl_easy_setopt(curl_handle, CURLOPT_HEADERDATA, (void*)&headerData);
1133
1134 curlSetSSLOptions(curl_handle);
1135
1136 auto res = static_cast<CURLcode>(CURL_perform(curl_handle));
1137 long response_code = -1;
1138 void* content = nullptr;
1139 bool errorflag = false;
1140 if (res == CURLE_OK && curl_easy_getinfo(curl_handle, CURLINFO_RESPONSE_CODE, &response_code) == CURLE_OK) {
1141 if (headers) {
1142 for (auto& p : headerData) {
1143 (*headers)[p.first] = p.second;
1144 }
1145 }
1146 if (200 <= response_code && response_code < 300) {
1147 // good response and the content is directly provided and should have been dumped into "chunk"
1148 content = interpretAsTMemFileAndExtract(chunk.memory, chunk.size, tinfo);
1149 if (headers && headers->find("fileSize") == headers->end()) {
1150 (*headers)["fileSize"] = fmt::format("{}", chunk.size);
1151 }
1152 } else if (response_code == 304) {
1153 // this means the object exist but I am not serving
1154 // it since it's already in your possession
1155
1156 // there is nothing to be done here
1157 LOGP(debug, "Object exists but I am not serving it since it's already in your possession");
1158 }
1159 // this is a more general redirection
1160 else if (300 <= response_code && response_code < 400) {
1161 // we try content locations in order of appearance until one succeeds
1162 // 1st: The "Location" field
1163 // 2nd: Possible "Content-Location" fields - Location field
1164
1165 // some locations are relative to the main server so we need to fix/complement them
1166 auto complement_Location = [this](std::string const& loc) {
1167 if (loc[0] == '/') {
1168 // if it's just a path (noticed by trailing '/' we prepend the server url
1169 return getURL() + loc;
1170 }
1171 return loc;
1172 };
1173
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));
1178 }
1179 // add alternative locations (not yet included)
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));
1186 }
1187 }
1188 }
1189 for (auto& l : locs) {
1190 if (l.size() > 0) {
1191 LOG(debug) << "Trying content location " << l;
1192 content = navigateURLsAndRetrieveContent(curl_handle, l, tinfo, headers);
1193 if (content /* or other success marker in future */) {
1194 break;
1195 }
1196 }
1197 }
1198 } else if (response_code == 404) {
1199 LOG(error) << "Requested resource does not exist: " << url;
1200 errorflag = true;
1201 } else {
1202 LOG(error) << "Error in fetching object " << url << ", curl response code:" << response_code;
1203 errorflag = true;
1204 }
1205 // cleanup
1206 if (chunk.memory != nullptr) {
1207 free(chunk.memory);
1208 }
1209 } else {
1210 LOGP(alarm, "Curl request to {} failed with result {}, response code: {}", url, int(res), response_code);
1211 errorflag = true;
1212 }
1213 // indicate that an error occurred ---> used by caching layers (such as CCDBManager)
1214 if (errorflag && headers) {
1215 (*headers)["Error"] = "An error occurred during retrieval";
1216 }
1217 return content;
1218}
1219
1220void* CcdbApi::retrieveFromTFile(std::type_info const& tinfo, std::string const& path,
1221 std::map<std::string, std::string> const& metadata, long timestamp,
1222 std::map<std::string, std::string>* headers, std::string const& etag,
1223 const std::string& createdNotAfter, const std::string& createdNotBefore) const
1224{
1225 if (!mSnapshotCachePath.empty()) {
1226 // protect this sensitive section by a multi-process named semaphore
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";
1232 }
1233 auto snapshotfile = getSnapshotFile(mSnapshotCachePath, path);
1234 bool snapshoting = false;
1235 if (!std::filesystem::exists(snapshotfile)) {
1236 snapshoting = true;
1237 out << "CCDB-access[" << getpid() << "] ... " << mUniqueAgentID << " downloading to snapshot " << snapshotfile << "\n";
1238 // if file not already here and valid --> snapshot it
1239 if (!retrieveBlob(path, mSnapshotCachePath, metadata, timestamp)) {
1240 out << "CCDB-access[" << getpid() << "] ... " << mUniqueAgentID << " failed to create directory for " << snapshotfile << "\n";
1241 }
1242 } else {
1243 out << "CCDB-access[" << getpid() << "] ... " << mUniqueAgentID << "serving from local snapshot " << snapshotfile << "\n";
1244 }
1245
1246 auto res = extractFromLocalFile(snapshotfile, tinfo, headers);
1247 if (!snapshoting) { // if snapshot was created at this call, the log was already done
1248 logReading(path, timestamp, headers, "retrieve from snapshot");
1249 }
1250 return res;
1251 }
1252
1253 // normal mode follows
1254
1255 CurlHandle* curl_handle = curl_easy_init();
1256 curl_easy_setopt(curl_handle, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1257 std::string fullUrl = getFullUrlForRetrieval(curl_handle, path, metadata, timestamp); // todo check if function still works correctly in case mInSnapshotMode
1258 // if we are in snapshot mode we can simply open the file; extract the object and return
1259 if (mInSnapshotMode) {
1260 auto res = extractFromLocalFile(fullUrl, tinfo, headers);
1261 if (res) {
1262 logReading(path, timestamp, headers, "retrieve from snapshot");
1263 }
1264 return res;
1265 }
1266
1267 curl_slist* option_list = nullptr;
1268 initCurlHTTPHeaderOptionsForRetrieve(curl_handle, option_list, timestamp, headers, etag, createdNotAfter, createdNotBefore, fullUrl);
1269 auto content = navigateURLsAndRetrieveContent(curl_handle, fullUrl, tinfo, headers);
1270
1271 for (size_t hostIndex = 1; hostIndex < hostsPool.size() && !(content); hostIndex++) {
1272 fullUrl = getFullUrlForRetrieval(curl_handle, path, metadata, timestamp, hostIndex);
1273 // Per host: the gate token is per endpoint (see appendGateToken).
1274 curl_slist_free_all(option_list);
1275 option_list = nullptr;
1276 initCurlHTTPHeaderOptionsForRetrieve(curl_handle, option_list, timestamp, headers, etag, createdNotAfter, createdNotBefore, fullUrl);
1277 content = navigateURLsAndRetrieveContent(curl_handle, fullUrl, tinfo, headers);
1278 }
1279 if (content) {
1280 logReading(path, timestamp, headers, "retrieve");
1281 }
1282 curl_slist_free_all(option_list);
1283 curl_easy_cleanup(curl_handle);
1284 return content;
1285}
1286
1287size_t CurlWrite_CallbackFunc_StdString2(void* contents, size_t size, size_t nmemb, std::string* s)
1288{
1289 size_t newLength = size * nmemb;
1290 size_t oldLength = s->size();
1291 try {
1292 s->resize(oldLength + newLength);
1293 } catch (std::bad_alloc& e) {
1294 LOG(error) << "memory error when getting data from CCDB";
1295 return 0;
1296 }
1297
1298 std::copy((char*)contents, (char*)contents + newLength, s->begin() + oldLength);
1299 return size * nmemb;
1300}
1301
1302std::string CcdbApi::list(std::string const& path, bool latestOnly, std::string const& returnFormat, long createdNotAfter, long createdNotBefore) const
1303{
1304 CurlHandle* curl;
1305 CURLcode res = CURL_LAST;
1306 std::string result;
1307
1308 curl = curl_easy_init();
1309 if (curl != nullptr) {
1310 curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, CurlWrite_CallbackFunc_StdString2);
1311 curl_easy_setopt(curl, CURLOPT_WRITEDATA, &result);
1312 curl_easy_setopt(curl, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1313
1314 curlSetSSLOptions(curl);
1315
1316 std::string fullUrl;
1317 // Perform the request, res will get the return code
1318 for (size_t hostIndex = 0; hostIndex < hostsPool.size() && res != CURLE_OK; hostIndex++) {
1319 fullUrl = getHostUrl(hostIndex);
1320 fullUrl += latestOnly ? "/latest/" : "/browse/";
1321 fullUrl += path;
1322 curl_easy_setopt(curl, CURLOPT_URL, fullUrl.c_str());
1323
1324 // Per host: the gate token is per endpoint (see appendGateToken).
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());
1328 if (createdNotAfter >= 0) {
1329 headers = curl_slist_append(headers, ("If-Not-After: " + std::to_string(createdNotAfter)).c_str());
1330 }
1331 if (createdNotBefore >= 0) {
1332 headers = curl_slist_append(headers, ("If-Not-Before: " + std::to_string(createdNotBefore)).c_str());
1333 }
1334 headers = appendGateToken(headers, fullUrl);
1335 curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
1336
1337 res = static_cast<CURLcode>(CURL_perform(curl));
1338 if (res != CURLE_OK) {
1339 LOGP(alarm, "CURL_perform() failed: {}", curl_easy_strerror(res));
1340 }
1341 curl_slist_free_all(headers);
1342 }
1343 curl_easy_cleanup(curl);
1344 }
1345
1346 return result;
1347}
1348
1349std::string CcdbApi::getTimestampString(long timestamp) const
1350{
1351 stringstream ss;
1352 ss << timestamp;
1353 return ss.str();
1354}
1355
1356void CcdbApi::deleteObject(std::string const& path, long timestamp) const
1357{
1358 CurlHandle* curl;
1359 CURLcode res;
1360 long timestampLocal = timestamp == -1 ? getCurrentTimestamp() : timestamp;
1361
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);
1367 curlSetSSLOptions(curl);
1368
1369 for (size_t hostIndex = 0; hostIndex < hostsPool.size(); hostIndex++) {
1370 // Inside the loop: hoisted out, the stream accumulates and the second
1371 // host's URL is the first with the second appended.
1372 stringstream fullUrl;
1373 fullUrl << getHostUrl(hostIndex) << "/" << path << "/" << timestampLocal;
1374 curl_easy_setopt(curl, CURLOPT_URL, fullUrl.str().c_str());
1375
1376 // A DELETE is a write, so it needs the gate token as storing does -- per
1377 // host, since the token is per endpoint (see appendGateToken).
1378 struct curl_slist* list = appendGateToken(nullptr, fullUrl.str());
1379 curl_easy_setopt(curl, CURLOPT_HTTPHEADER, list);
1380
1381 // Perform the request, res will get the return code
1382 res = static_cast<CURLcode>(CURL_perform(curl));
1383 if (res != CURLE_OK) {
1384 LOGP(alarm, "CURL_perform() failed: {}", curl_easy_strerror(res));
1385 }
1386 curl_slist_free_all(list);
1387 }
1388 // After the loop, not inside it: cleaning up per host left every later
1389 // iteration using a freed handle.
1390 curl_easy_cleanup(curl);
1391 }
1392}
1393
1394void CcdbApi::truncate(std::string const& path) const
1395{
1396 CurlHandle* curl;
1397 CURLcode res;
1398 for (size_t i = 0; i < hostsPool.size(); i++) {
1399 // Declared inside the loop: a stringstream hoisted out of it accumulates,
1400 // so the second host's URL would be the first one with the second appended
1401 // to it. Latent until now -- every caller used a single-host pool.
1402 stringstream fullUrl;
1403 std::string url = getHostUrl(i);
1404 fullUrl << url << "/truncate/" << path;
1405
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());
1410
1411 // Truncating is a write, so it needs the gate token exactly as storing
1412 // does. This was the one write path left without it, which a broker
1413 // answers 401 -- failing every CCDB suite in their teardown, since each
1414 // one truncates the path it just wrote.
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);
1418 curlSetSSLOptions(curl);
1419
1420 // Perform the request, res will get the return code
1421 res = static_cast<CURLcode>(CURL_perform(curl));
1422 if (res != CURLE_OK) {
1423 LOGP(alarm, "CURL_perform() failed: {}", curl_easy_strerror(res));
1424 }
1425 curl_easy_cleanup(curl);
1426 curl_slist_free_all(list);
1427 }
1428 }
1429}
1430
1431size_t write_data(void*, size_t size, size_t nmemb, void*)
1432{
1433 return size * nmemb;
1434}
1435
1437{
1438 CurlHandle* curl;
1439 CURLcode res = CURL_LAST;
1440 bool result = false;
1441
1442 curl = curl_easy_init();
1443 curl_easy_setopt(curl, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1444 if (curl) {
1445 // NOTE: mUrl, not getHostUrl(hostIndex), even though hostIndex is unused.
1446 // For a failover setup mUrl is the whole comma-separated list, which curl
1447 // rejects as malformed, so every multi-host instance reports itself
1448 // unreachable however healthy its hosts are -- and testCcdbApiMultipleUrls,
1449 // whose cases are gated on this, is skipped rather than run.
1450 //
1451 // Fixing it is a separate change: the suite then runs for the first time
1452 // and its storeAndRetrieve fails, so the multi-host store/retrieve path
1453 // needs looking at before this can be corrected. Callers outside the tests
1454 // are affected too -- HMPID/PedestalsCalculationSpec sets mWriteToDB from
1455 // this, and TPC workflows branch on it.
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);
1459 curlSetSSLOptions(curl);
1460 res = static_cast<CURLcode>(CURL_perform(curl));
1461 result = (res == CURLE_OK);
1462 }
1463
1464 /* always cleanup */
1465 curl_easy_cleanup(curl);
1466 }
1467 return result;
1468}
1469
1470std::vector<std::string> CcdbApi::parseSubFolders(std::string const& reply) const
1471{
1472 // this needs some text filtering
1473 // go through reply line by line until we see "SubFolders:"
1474 std::stringstream ss(reply.c_str());
1475 std::string line;
1476 std::vector<std::string> folders;
1477
1478 size_t numberoflines = std::count(reply.begin(), reply.end(), '\n');
1479 bool inSubFolderSection = false;
1480
1481 for (size_t linenumber = 0; linenumber < numberoflines; ++linenumber) {
1482 std::getline(ss, line);
1483 if (inSubFolderSection && line.size() > 0) {
1484 // remove all white space
1485 folders.push_back(sanitizeObjectName(line));
1486 }
1487
1488 if (line.compare("Subfolders:") == 0) {
1489 inSubFolderSection = true;
1490 }
1491 }
1492 return folders;
1493}
1494
1495namespace
1496{
1497size_t header_callback(char* buffer, size_t size, size_t nitems, void* userdata)
1498{
1499 auto* headers = static_cast<std::vector<std::string>*>(userdata);
1500 auto header = std::string(buffer, size * nitems);
1501 headers->emplace_back(std::string(header.data()));
1502 return size * nitems;
1503}
1504} // namespace
1505
1506bool stdmap_to_jsonfile(std::map<std::string, std::string> const& meta, std::string const& filename)
1507{
1508
1509 // create directory structure if necessary
1510 auto p = std::filesystem::path(filename).parent_path();
1511 if (!std::filesystem::exists(p)) {
1512 std::filesystem::create_directories(p);
1513 }
1514
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());
1521 }
1522 writer.EndObject();
1523
1524 // Write JSON to file
1525 std::ofstream file(filename);
1526 if (file.is_open()) {
1527 file << buffer.GetString();
1528 file.close();
1529 } else {
1530 return false;
1531 }
1532 return true;
1533}
1534
1535bool jsonfile_to_stdmap(std::map<std::string, std::string>& meta, std::string const& filename)
1536{
1537 // Read JSON from file
1538 std::ifstream file(filename);
1539 if (!file.is_open()) {
1540 std::cerr << "Failed to open file for reading." << std::endl;
1541 return false;
1542 }
1543
1544 std::string jsonStr((std::istreambuf_iterator<char>(file)), std::istreambuf_iterator<char>());
1545
1546 // Parse JSON
1547 rapidjson::Document document;
1548 document.Parse(jsonStr.c_str());
1549
1550 if (document.HasParseError()) {
1551 std::cerr << "Error parsing JSON" << std::endl;
1552 return false;
1553 }
1554
1555 // Convert JSON to std::map
1556 for (auto itr = document.MemberBegin(); itr != document.MemberEnd(); ++itr) {
1557 meta[itr->name.GetString()] = itr->value.GetString();
1558 }
1559 return true;
1560}
1561
1562std::map<std::string, std::string> CcdbApi::retrieveHeaders(std::string const& path, std::map<std::string, std::string> const& metadata, long timestamp) const
1563{
1564 // lambda that actually does the call to the CCDB server
1565 auto do_remote_header_call = [this, &path, &metadata, timestamp]() -> std::map<std::string, std::string> {
1566 CurlHandle* curl = curl_easy_init();
1567 CURLcode res = CURL_LAST;
1568 std::string fullUrl = getFullUrlForRetrieval(curl, path, metadata, timestamp);
1569 std::map<std::string, std::string> headers;
1570
1571 if (curl != nullptr) {
1572 struct curl_slist* list = nullptr;
1573 list = curl_slist_append(list, ("If-None-Match: " + std::to_string(timestamp)).c_str());
1574 list = appendGateToken(list, fullUrl);
1575
1576 curl_easy_setopt(curl, CURLOPT_HTTPHEADER, list);
1577
1578 /* get us the resource without a body! */
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());
1584
1585 curlSetSSLOptions(curl);
1586
1587 // Perform the request, res will get the return code
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) {
1594 // We take out the unsupported protocol error because we are only querying
1595 // header info which is returned in any case. Unsupported protocol error
1596 // occurs sometimes because of redirection to alien for blobs.
1597 LOG(error) << "CURL_perform() failed: " << curl_easy_strerror(res);
1598 }
1599 getCodeRes = curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &httpCode);
1600 }
1601 if (httpCode == 404) {
1602 headers.clear();
1603 }
1604 curl_easy_cleanup(curl);
1605 }
1606 return headers;
1607 };
1608
1609 if (!mSnapshotCachePath.empty()) {
1610 // protect this sensitive section by a multi-process named semaphore
1611 auto semaphore_barrier = std::make_unique<CCDBSemaphore>(mSnapshotCachePath + std::string("_headers"), path);
1612
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";
1617 }
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";
1621
1622 // if file not already here and valid --> snapshot it
1623 auto meta = do_remote_header_call();
1624
1625 // cache the result
1626 if (!stdmap_to_jsonfile(meta, snapshotfile)) {
1627 LOG(warn) << "Failed to cache the header information to disc";
1628 }
1629 return meta;
1630 } else {
1631 out << "CCDB-header-access[" << getpid() << "] ... " << mUniqueAgentID << "serving from local snapshot " << snapshotfile << "\n";
1632 std::map<std::string, std::string> meta;
1633 if (!jsonfile_to_stdmap(meta, snapshotfile)) {
1634 LOG(warn) << "Failed to read cached information from disc";
1635 return do_remote_header_call();
1636 }
1637 return meta;
1638 }
1639 }
1640 return do_remote_header_call();
1641}
1642
1643bool CcdbApi::getCCDBEntryHeaders(std::string const& url, std::string const& etag, std::vector<std::string>& headers, const std::string& agentID)
1644{
1645 auto curl = curl_easy_init();
1646 headers.clear();
1647 if (!curl) {
1648 return true;
1649 }
1650
1651 struct curl_slist* list = nullptr;
1652 list = curl_slist_append(list, ("If-None-Match: " + etag).c_str());
1653 list = appendGateToken(list, url);
1654
1655 curl_easy_setopt(curl, CURLOPT_HTTPHEADER, list);
1656
1657 curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
1658 /* get us the resource without a body! */
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());
1665 }
1666
1667 curlSetSSLOptions(curl);
1668
1669 /* Perform the request */
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) {
1674 return false;
1675 }
1676 return true;
1677}
1678
1679void CcdbApi::parseCCDBHeaders(std::vector<std::string> const& headers, std::vector<std::string>& pfns, std::string& etag)
1680{
1681 static std::string etagHeader = "ETag: ";
1682 static std::string locationHeader = "Content-Location: ";
1683 // Trimmed: `headers` holds raw header lines, CRLF and all, and the etag goes
1684 // straight back out as an If-None-Match request header.
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())));
1690 }
1691 }
1692}
1693
1695{
1696 auto object = file.GetObjectChecked(CCDBQUERY_ENTRY, TClass::GetClass(typeid(o2::ccdb::CCDBQuery)));
1697 if (object) {
1698 return static_cast<CCDBQuery*>(object);
1699 }
1700 return nullptr;
1701}
1702
1703std::map<std::string, std::string>* CcdbApi::retrieveMetaInfo(TFile& file)
1704{
1705 auto object = file.GetObjectChecked(CCDBMETA_ENTRY, TClass::GetClass(typeid(std::map<std::string, std::string>)));
1706 if (object) {
1707 return static_cast<std::map<std::string, std::string>*>(object);
1708 }
1709 return nullptr;
1710}
1711
1712namespace
1713{
1714void traverseAndFillFolders(CcdbApi const& api, std::string const& top, std::vector<std::string>& folders)
1715{
1716 // LOG(info) << "Querying " << top;
1717 auto reply = api.list(top);
1718 folders.emplace_back(top);
1719 // LOG(info) << reply;
1720 auto subfolders = api.parseSubFolders(reply);
1721 if (subfolders.size() > 0) {
1722 // LOG(info) << subfolders.size() << " folders in " << top;
1723 for (auto& sub : subfolders) {
1724 traverseAndFillFolders(api, sub, folders);
1725 }
1726 } else {
1727 // LOG(info) << "NO subfolders in " << top;
1728 }
1729}
1730} // namespace
1731
1732std::vector<std::string> CcdbApi::getAllFolders(std::string const& top) const
1733{
1734 std::vector<std::string> folders;
1735 traverseAndFillFolders(*this, top, folders);
1736 return folders;
1737}
1738
1739TClass* CcdbApi::tinfo2TClass(std::type_info const& tinfo)
1740{
1741 TClass* cl = TClass::GetClass(tinfo);
1742 if (!cl) {
1743 throw std::runtime_error(fmt::format("Could not retrieve ROOT dictionary for type {}, aborting", tinfo.name()));
1744 return nullptr;
1745 }
1746 return cl;
1747}
1748
1749int CcdbApi::updateMetadata(std::string const& path, std::map<std::string, std::string> const& metadata, long timestamp, std::string const& id, long newEOV)
1750{
1751 int ret = -1;
1752 CurlHandle* curl = curl_easy_init();
1753 curl_easy_setopt(curl, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1754 if (curl != nullptr) {
1755 CURLcode res;
1756 for (size_t hostIndex = 0; hostIndex < hostsPool.size(); hostIndex++) {
1757 // Inside the loop: hoisted out, the stream accumulates and the second
1758 // host's URL is the first with the second appended.
1759 stringstream fullUrl;
1760 fullUrl << getHostUrl(hostIndex) << "/" << path << "/" << timestamp;
1761 if (newEOV > 0) {
1762 fullUrl << "/" << newEOV;
1763 }
1764 if (!id.empty()) {
1765 fullUrl << "/" << id;
1766 }
1767 fullUrl << "?";
1768
1769 for (auto& kv : metadata) {
1770 std::string mfirst = kv.first;
1771 std::string msecond = kv.second;
1772 // same trick for the metadata as for the object type
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);
1778 }
1779
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"); // make sure we use PUT
1784 curl_easy_setopt(curl, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1785 curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);
1786 // A PUT is a write, so it needs the gate token as storing does -- per
1787 // host, since the token is per endpoint (see appendGateToken).
1788 struct curl_slist* list = appendGateToken(nullptr, fullUrl.str());
1789 curl_easy_setopt(curl, CURLOPT_HTTPHEADER, list);
1790 curlSetSSLOptions(curl);
1791
1792 // Perform the request, res will get the return code
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));
1796 ret = int(res);
1797 } else {
1798 ret = 0;
1799 }
1800 curl_slist_free_all(list);
1801 }
1802 }
1803 // After the loop, not inside it: cleaning up per host left every later
1804 // iteration using a freed handle.
1805 curl_easy_cleanup(curl);
1806 }
1807 return ret;
1808}
1809
1810void CcdbApi::initHostsPool(std::string hosts)
1811{
1812 hostsPool.clear();
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());
1816 }
1817}
1818
1819std::string CcdbApi::getHostUrl(int hostIndex) const
1820{
1821 return hostsPool.at(hostIndex);
1822}
1823
1824void CcdbApi::scheduleDownload(RequestContext& requestContext, size_t* requestCounter) const
1825{
1826 auto data = new DownloaderRequestData(); // Deleted in transferFinished of CCDBDownloader.cxx
1827 data->hoPair.object = &requestContext.dest;
1828
1829 std::function<bool(std::string)> localContentCallback = [this, &requestContext](std::string url) {
1830 return this->loadLocalContentToMemory(requestContext.dest, url);
1831 };
1832
1833 auto writeCallback = [](void* contents, size_t size, size_t nmemb, void* chunkptr) {
1834 auto& ho = *static_cast<HeaderObjectPair_t*>(chunkptr);
1835 auto& chunk = *ho.object;
1836 size_t realsize = size * nmemb, sz = 0;
1837 ho.counter++;
1838 try {
1839 if (chunk.capacity() < chunk.size() + realsize) {
1840 // estimate headers size when converted to annotated text string
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);
1847 } else {
1848 sz = hsize + std::max(chunk.size() * 2, chunk.size() + realsize);
1849 // LOGP(debug, "SIZE IS NOT IN HEADER, allocate {}", sz);
1850 }
1851 chunk.reserve(sz);
1852 }
1853 char* contC = (char*)contents;
1854 chunk.insert(chunk.end(), contC, contC + realsize);
1855 } catch (std::exception e) {
1856 // LOGP(alarm, "failed to reserve {} bytes in CURL write callback (realsize = {}): {}", sz, realsize, e.what());
1857 realsize = 0;
1858 }
1859 return realsize;
1860 };
1861
1862 CurlHandle* curl_handle = curl_easy_init();
1863 curl_easy_setopt(curl_handle, CURLOPT_USERAGENT, mUniqueAgentID.c_str());
1864 std::string fullUrl = getFullUrlForRetrieval(curl_handle, requestContext.path, requestContext.metadata, requestContext.timestamp);
1865
1866 data->headers = &requestContext.headers;
1867 data->hosts = hostsPool;
1868 data->path = requestContext.path;
1869 data->timestamp = requestContext.timestamp;
1870 data->localContentCallback = localContentCallback;
1871 data->userAgent = mUniqueAgentID;
1872
1873 // One header list per host, built HERE because this is where the gate-token
1874 // table is visible -- the downloader only indexes them. A single shared list
1875 // sent the first host's token to every host it failed over to, which a broker
1876 // answers 401: the failover then retrieved nothing while looking like a
1877 // network failure (testCcdbApi multi_host_test).
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,
1882 requestContext.timestamp, hostIndex);
1883 initCurlHTTPHeaderOptionsForRetrieve(curl_handle, hostOptions, requestContext.timestamp, &requestContext.headers,
1884 requestContext.etag, requestContext.createdNotAfter, requestContext.createdNotBefore,
1885 hostUrl);
1886 data->optionsLists.push_back(hostOptions);
1887 }
1888 // initCurlHTTPHeaderOptionsForRetrieve sets CURLOPT_HTTPHEADER as a side
1889 // effect, so the handle currently points at the LAST host's list. Point it
1890 // back at host 0, which is the one this transfer starts with.
1891 if (!data->optionsLists.empty()) {
1892 curl_easy_setopt(curl_handle, CURLOPT_HTTPHEADER, data->optionsLists.front());
1893 }
1894
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);
1900 curlSetSSLOptions(curl_handle);
1901
1902 asynchPerform(curl_handle, requestCounter);
1903}
1904
1905std::string CcdbApi::determineSemaphoreName(std::string const& basedir, std::string const& ccdbpath)
1906{
1907 std::hash<std::string> hasher;
1908 std::string semhashedstring = "aliceccdb" + std::to_string(hasher(basedir + ccdbpath)).substr(0, 16);
1909 return semhashedstring;
1910}
1911
1912boost::interprocess::named_semaphore* CcdbApi::createNamedSemaphore(std::string const& path) const
1913{
1914 std::string semhashedstring = determineSemaphoreName(mSnapshotCachePath, path);
1915 // LOG(info) << "Creating named semaphore with name " << semhashedstring.c_str();
1916 try {
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";
1920 return nullptr;
1921 }
1922}
1923
1924void CcdbApi::releaseNamedSemaphore(boost::interprocess::named_semaphore* sem, std::string const& path) const
1925{
1926 if (sem) {
1927 sem->post();
1928 if (sem->try_wait()) { // if nobody else is waiting remove the semaphore resource
1929 sem->post();
1930 boost::interprocess::named_semaphore::remove(determineSemaphoreName(mSnapshotCachePath, path).c_str());
1931 }
1932 }
1933}
1934
1935bool CcdbApi::removeSemaphore(std::string const& semaname, bool remove)
1936{
1937 // removes a given named semaphore from the system
1938 try {
1939 boost::interprocess::named_semaphore semaphore(boost::interprocess::open_only, semaname.c_str());
1940 std::cout << "Found CCDB semaphore: " << semaname << "\n";
1941 if (remove) {
1942 auto success = boost::interprocess::named_semaphore::remove(semaname.c_str());
1943 if (success) {
1944 std::cout << "Removed CCDB semaphore: " << semaname << "\n";
1945 }
1946 return success;
1947 }
1948 return true;
1949 } catch (std::exception const& e) {
1950 // no EXISTING under this name semaphore found
1951 // nothing to be done
1952 }
1953 return false;
1954}
1955
1956// helper function checking for leaking semaphores associated to CCDB cache files and removing them
1957// walks a local CCDB snapshot tree and checks
1958void CcdbApi::removeLeakingSemaphores(std::string const& snapshotdir, bool remove)
1959{
1960 namespace fs = std::filesystem;
1961 std::string fileName{"snapshot.root"};
1962 try {
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;
1970
1971 // we need to obtain the path relative to snapshotdir
1972 auto pathtokens = o2::utils::Str::tokenize(currentDir, '/', true);
1973 auto numtokens = pathtokens.size();
1974 if (numtokens < 3) {
1975 // cannot be a CCDB path
1976 continue;
1977 }
1978 // path are last 3 entries
1979 std::string path = pathtokens[numtokens - 3] + "/" + pathtokens[numtokens - 2] + "/" + pathtokens[numtokens - 1];
1980 auto semaname = o2::ccdb::CcdbApi::determineSemaphoreName(absolutesnapshotdir, path);
1981 removeSemaphore(semaname, remove);
1982 }
1983 }
1984 }
1985 } catch (std::exception const& e) {
1986 LOG(info) << "Semaphore search had exception " << e.what();
1987 }
1988}
1989
1990void CcdbApi::getFromSnapshot(bool createSnapshot, std::string const& path,
1991 long timestamp, std::map<std::string, std::string>& headers,
1992 std::string& snapshotpath, o2::pmr::vector<char>& dest, int& fromSnapshot, std::string const& etag) const
1993{
1994 if (createSnapshot) { // create named semaphore
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";
1999 }
2000 }
2001 if (mInSnapshotMode) { // file must be there, otherwise a fatal will be produced;
2002 if (etag.empty()) {
2003 loadFileToMemory(dest, getSnapshotFile(mSnapshotTopPath, path), &headers);
2004 }
2005 fromSnapshot = 1;
2006 } else if (mPreferSnapshotCache && std::filesystem::exists(snapshotpath)) {
2007 // if file is available, use it, otherwise cache it below from the server. Do this only when etag is empty since otherwise the object was already fetched and cached
2008 if (etag.empty()) {
2009 loadFileToMemory(dest, snapshotpath, &headers);
2010 }
2011 fromSnapshot = 2;
2012 }
2013}
2014
2015void CcdbApi::saveSnapshot(RequestContext& requestContext) const
2016{
2017 // Consider saving snapshot
2018 if (!mSnapshotCachePath.empty() && !(mInSnapshotMode && mSnapshotTopPath == mSnapshotCachePath)) { // store in the snapshot only if the object was not read from the snapshot
2019 auto semaphore_barrier = std::make_unique<CCDBSemaphore>(mSnapshotCachePath, requestContext.path);
2020
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";
2027 }
2028 { // dump image to a file
2029 LOGP(debug, "creating snapshot {} -> {}", requestContext.path, snapshotpath);
2030 CCDBQuery querysummary(requestContext.path, requestContext.metadata, requestContext.timestamp);
2031 {
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));
2034 }
2035 // now open the same file as root file and store metadata
2036 updateMetaInformationInLocalFile(snapshotpath, &requestContext.headers, &querysummary);
2037 }
2038 }
2039}
2040
2041void CcdbApi::loadFileToMemory(std::vector<char>& dest, std::string const& path,
2042 std::map<std::string, std::string> const& metadata, long timestamp,
2043 std::map<std::string, std::string>* headers, std::string const& etag,
2044 const std::string& createdNotAfter, const std::string& createdNotBefore, bool considerSnapshot) const
2045{
2047 destP.reserve(dest.size());
2048 loadFileToMemory(destP, path, metadata, timestamp, headers, etag, createdNotAfter, createdNotBefore, considerSnapshot);
2049 dest.clear();
2050 dest.reserve(destP.size());
2051 for (const auto c : destP) {
2052 dest.push_back(c);
2053 }
2054}
2055
2057 std::map<std::string, std::string> const& metadata, long timestamp,
2058 std::map<std::string, std::string>* headers, std::string const& etag,
2059 const std::string& createdNotAfter, const std::string& createdNotBefore, bool considerSnapshot) const
2060{
2061 RequestContext requestContext(dest, metadata, *headers);
2062 requestContext.path = path;
2063 // std::map<std::string, std::string> metadataCopy = metadata; // Create a copy because metadata will be passed as a pointer so it cannot be constant. The const in definition is for backwards compatability.
2064 // requestContext.metadata = metadataCopy;
2065 requestContext.timestamp = timestamp;
2066 requestContext.etag = etag;
2067 requestContext.createdNotAfter = createdNotAfter;
2068 requestContext.createdNotBefore = createdNotBefore;
2069 requestContext.considerSnapshot = considerSnapshot;
2070 std::vector<RequestContext> contexts = {requestContext};
2071 vectoredLoadFileToMemory(contexts);
2072}
2073
2074void CcdbApi::appendFlatHeader(o2::pmr::vector<char>& dest, const std::map<std::string, std::string>& headers)
2075{
2076 size_t hsize = getFlatHeaderSize(headers), cnt = dest.size();
2077 dest.resize(cnt + hsize);
2078 auto addString = [&dest, &cnt](const std::string& s) {
2079 for (char c : s) {
2080 dest[cnt++] = c;
2081 }
2082 dest[cnt++] = 0;
2083 };
2084
2085 for (auto& h : headers) {
2086 addString(h.first);
2087 addString(h.second);
2088 }
2089 *reinterpret_cast<int*>(&dest[cnt]) = hsize; // store size
2090 std::memcpy(&dest[cnt + sizeof(int)], FlatHeaderAnnot, sizeof(FlatHeaderAnnot)); // annotate the flattened headers map
2091}
2092
2093void CcdbApi::navigateSourcesAndLoadFile(RequestContext& requestContext, int& fromSnapshot, size_t* requestCounter) const
2094{
2095 LOGP(debug, "loadFileToMemory {} ETag=[{}]", requestContext.path, requestContext.etag);
2096 bool createSnapshot = requestContext.considerSnapshot && !mSnapshotCachePath.empty(); // create snaphot if absent
2097
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);
2101 // if we are in snapshot mode we can simply open the file, unless the etag is non-empty:
2102 // this would mean that the object was is already fetched and in this mode we don't to validity checks!
2103 getFromSnapshot(createSnapshot, requestContext.path, requestContext.timestamp, requestContext.headers, snapshotpath, requestContext.dest, fromSnapshot, requestContext.etag);
2104 } else { // look on the server
2105 scheduleDownload(requestContext, requestCounter);
2106 }
2107}
2108
2109void CcdbApi::vectoredLoadFileToMemory(std::vector<RequestContext>& requestContexts) const
2110{
2111 std::vector<int> fromSnapshots(requestContexts.size());
2112 size_t requestCounter = 0;
2113
2114 // Get files from snapshots and schedule downloads
2115 for (int i = 0; i < requestContexts.size(); i++) {
2116 // navigateSourcesAndLoadFile either retrieves file from snapshot immediately, or schedules it to be downloaded when mDownloader->runLoop is ran at a later time
2117 auto& requestContext = requestContexts.at(i);
2118 navigateSourcesAndLoadFile(requestContext, fromSnapshots.at(i), &requestCounter);
2119 }
2120
2121 // Download the rest
2122 while (requestCounter > 0) {
2123 mDownloader->runLoop(0);
2124 }
2125
2126 // Save snapshots
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) {
2133 saveSnapshot(requestContext);
2134 }
2135 }
2136 }
2137}
2138
2140{
2141 if (url.find("alien:/", 0) != std::string::npos) {
2142 std::map<std::string, std::string> localHeaders;
2143 loadFileToMemory(dest, url, &localHeaders, false);
2144 auto it = localHeaders.find("Error");
2145 if (it != localHeaders.end() && it->second == "An error occurred during retrieval") {
2146 return false;
2147 } else {
2148 return true;
2149 }
2150 }
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;
2155 loadFileToMemory(dest, url, &localHeaders, o2::utils::Str::endsWith(path, ".root"));
2156 auto it = localHeaders.find("Error");
2157 if (it != localHeaders.end() && it->second == "An error occurred during retrieval") {
2158 return false;
2159 } else {
2160 return true;
2161 }
2162 }
2163 }
2164 return false;
2165}
2166
2167void CcdbApi::loadFileToMemory(o2::pmr::vector<char>& dest, const std::string& path, std::map<std::string, std::string>* localHeaders, bool fetchLocalMetaData) const
2168{
2169 // Read file to memory as vector. For special case of the locally cached file retriev metadata stored directly in the file
2170 constexpr size_t MaxCopySize = 0x1L << 25;
2171 auto signalError = [&dest, localHeaders]() {
2172 dest.clear();
2173 dest.reserve(1);
2174 if (localHeaders) { // indicate that an error occurred ---> used by caching layers (such as CCDBManager)
2175 (*localHeaders)["Error"] = "An error occurred during retrieval";
2176 }
2177 };
2178 if (path.find("alien:/") == 0 && !initTGrid()) {
2179 signalError();
2180 return;
2181 }
2182 std::string fname(path);
2183 if (fname.find("?filetype=raw") == std::string::npos) {
2184 fname += "?filetype=raw";
2185 }
2186 std::unique_ptr<TFile> sfile{TFile::Open(fname.c_str())};
2187 if (!sfile || sfile->IsZombie()) {
2188 LOG(error) << "Failed to open file " << fname;
2189 signalError();
2190 return;
2191 }
2192 size_t totalread = 0, fsize = sfile->GetSize(), b00 = sfile->GetBytesRead();
2193 dest.resize(fsize);
2194 char* dptr = dest.data();
2195 sfile->Seek(0);
2196 long nread = 0;
2197 do {
2198 size_t b0 = sfile->GetBytesRead(), b1 = b0 - b00;
2199 size_t readsize = fsize - b1 > MaxCopySize ? MaxCopySize : fsize - b1;
2200 if (readsize == 0) {
2201 break;
2202 }
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";
2208 signalError();
2209 return;
2210 }
2211 dptr += nread;
2212 totalread += nread;
2213 } while (nread == (long)MaxCopySize);
2214
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);
2218 if (storedmeta) {
2219 *localHeaders = *storedmeta; // do a simple deep copy
2220 delete storedmeta;
2221 }
2222 if ((isSnapshotMode() || mPreferSnapshotCache) && localHeaders->find("ETag") == localHeaders->end()) { // generate dummy ETag to profit from the caching
2223 (*localHeaders)["ETag"] = path;
2224 }
2225 if (localHeaders->find("fileSize") == localHeaders->end()) {
2226 (*localHeaders)["fileSize"] = fmt::format("{}", memFile.GetEND());
2227 }
2228 }
2229 return;
2230}
2231
2232void CcdbApi::checkMetadataKeys(std::map<std::string, std::string> const& metadata) const
2233{
2234
2235 // function to check if any key contains invalid characters
2236 // if so, a fatal will be issued
2237
2238 const std::regex regexPatternSearch(R"([ :;.,\\/'?!\‍(\)\{\}\[\]@<>=+*#$&`|~^%])");
2239 bool isInvalid = false;
2240
2241 for (auto& el : metadata) {
2242 auto keyMd = el.first;
2243 auto tmp = keyMd;
2244 std::smatch searchRes;
2245 while (std::regex_search(keyMd, searchRes, regexPatternSearch)) {
2246 isInvalid = true;
2247 LOG(error) << "Invalid character found in metadata key '" << tmp << "\': '" << searchRes.str() << "\'";
2248 keyMd = searchRes.suffix();
2249 }
2250 }
2251 if (isInvalid) {
2252 LOG(fatal) << "Some metadata keys have invalid characters, please fix!";
2253 }
2254 return;
2255}
2256
2257void CcdbApi::logReading(const std::string& path, long ts, const std::map<std::string, std::string>* headers, const std::string& comment) const
2258{
2259 std::string upath{path};
2260 if (headers) {
2261 auto ent = headers->find("Valid-From");
2262 if (ent != headers->end()) {
2263 upath += "/" + ent->second;
2264 }
2265 ent = headers->find("ETag");
2266 if (ent != headers->end()) {
2267 upath += "/" + ent->second;
2268 }
2269 }
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);
2272}
2273
2274void CcdbApi::asynchPerform(CurlHandle* handle, size_t* requestCounter) const
2275{
2276 mDownloader->asynchSchedule(handle, requestCounter);
2277}
2278
2279int CcdbApi::CURL_perform(CurlHandle* handle) const
2280{
2281 if (mIsCCDBDownloaderPreferred) {
2282 return mDownloader->perform(handle);
2283 }
2284 CURLcode result;
2285 for (int i = 1; i <= mCurlRetries && (result = curl_easy_perform(handle)) != CURLE_OK; i++) {
2286 usleep(mCurlDelayRetries * i);
2287 }
2288 return result;
2289}
2290
2295CCDBSemaphore::CCDBSemaphore(std::string const& snapshotpath, std::string const& path)
2296{
2297 LOG(debug) << "Entering semaphore barrier";
2298 mSemName = CcdbApi::determineSemaphoreName(snapshotpath, path);
2299 try {
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";
2303 mSem = nullptr;
2304 }
2305 // automatically wait
2306 if (mSem) {
2307 gSemaRegistry.add(this);
2308 mSem->wait();
2309 }
2310}
2311
2313{
2314 LOG(debug) << "Ending semaphore barrier";
2315 if (mSem) {
2316 mSem->post();
2317 if (mSem->try_wait()) { // if nobody else is waiting remove the semaphore resource
2318 mSem->post();
2319 boost::interprocess::named_semaphore::remove(mSemName.c_str());
2320 }
2321 gSemaRegistry.remove(this);
2322 }
2323}
2324
2326{
2327 LOG(debug) << "Cleaning up semaphore registry with count " << mStore.size();
2328 for (auto& s : mStore) {
2329 delete s;
2330 mStore.erase(s);
2331 }
2332}
2333
2335{
2336 mStore.insert(ptr);
2337}
2338
2340{
2341 mStore.erase(ptr);
2342}
2343
2344} // namespace o2::ccdb
std::string createdNotBefore
std::string createdNotAfter
size_t maxSize
std::string etag
std::string url
std::ostringstream debug
std::vector< std::string > header
std::vector< long > entries
int32_t i
#define failed(...)
Definition Utils.h:42
uint8_t errorflag
Definition RawData.h:0
uint32_t res
Definition RawData.h:0
uint32_t c
Definition RawData.h:2
TBranch * ptr
StringRef key
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)
CCDBSemaphore(std::string const &cachepath, std::string const &path)
Definition CcdbApi.cxx:2295
static std::string generateFileName(const std::string &inp)
Definition CcdbApi.cxx:902
std::string list(std::string const &path="", bool latestOnly=false, std::string const &returnFormat="text/plain", long createdNotAfter=-1, long createdNotBefore=-1) const
Definition CcdbApi.cxx:1302
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
Definition CcdbApi.cxx:410
static bool checkAlienToken()
Definition CcdbApi.cxx:200
void runDownloaderLoop(bool noWait)
Definition CcdbApi.cxx:342
void releaseNamedSemaphore(boost::interprocess::named_semaphore *sem, std::string const &path) const
Definition CcdbApi.cxx:1924
static std::map< std::string, std::string > * retrieveMetaInfo(TFile &)
Definition CcdbApi.cxx:1703
void scheduleDownload(RequestContext &requestContext, size_t *requestCounter) const
Definition CcdbApi.cxx:1824
TObject * retrieve(std::string const &path, std::map< std::string, std::string > const &metadata, long timestamp) const
Definition CcdbApi.cxx:874
void init(std::string const &hosts)
Definition CcdbApi.cxx:237
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
Definition CcdbApi.cxx:912
std::string const & getURL() const
Definition CcdbApi.h:93
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
Definition CcdbApi.cxx:1990
bool loadLocalContentToMemory(o2::pmr::vector< char > &dest, std::string &url) const
Definition CcdbApi.cxx:2139
static void removeLeakingSemaphores(std::string const &basedir, bool remove=false)
Definition CcdbApi.cxx:1958
void saveSnapshot(RequestContext &requestContext) const
Definition CcdbApi.cxx:2015
static std::unique_ptr< std::vector< char > > createObjectImage(const T *obj, CcdbObjectInfo *info=nullptr)
Definition CcdbApi.h:109
static void * extractFromTFile(TFile &file, TClass const *cl, const char *what=CCDBOBJECT_ENTRY)
Definition CcdbApi.cxx:984
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
Definition CcdbApi.cxx:550
void snapshot(std::string const &ccdbrootpath, std::string const &localDir, long timestamp) const
Definition CcdbApi.cxx:974
bool isSnapshotMode() const
Definition CcdbApi.h:99
bool isHostReachable() const
Definition CcdbApi.cxx:1436
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
Definition CcdbApi.cxx:2041
static std::string determineSemaphoreName(std::string const &basedir, std::string const &objectpath)
Definition CcdbApi.cxx:1905
void deleteObject(std::string const &path, long timestamp=-1) const
Definition CcdbApi.cxx:1356
static void appendFlatHeader(o2::pmr::vector< char > &dest, const std::map< std::string, std::string > &headers)
Definition CcdbApi.cxx:2074
std::vector< std::string > getAllFolders(std::string const &top) const
Definition CcdbApi.cxx:1732
void vectoredLoadFileToMemory(std::vector< RequestContext > &requestContext) const
Definition CcdbApi.cxx:2109
static void curlSetSSLOptions(CurlHandle *curl)
Definition CcdbApi.cxx:696
boost::interprocess::named_semaphore * createNamedSemaphore(std::string const &path) const
Definition CcdbApi.cxx:1912
static bool removeSemaphore(std::string const &name, bool remove=false)
Definition CcdbApi.cxx:1935
std::map< std::string, std::string > retrieveHeaders(std::string const &path, std::map< std::string, std::string > const &metadata, long timestamp=-1) const
Definition CcdbApi.cxx:1562
CcdbApi()
Default constructor.
Definition CcdbApi.cxx:173
static bool getCCDBEntryHeaders(std::string const &url, std::string const &etag, std::vector< std::string > &headers, const std::string &agentID="")
Definition CcdbApi.cxx:1643
static CCDBQuery * retrieveQueryInfo(TFile &)
Definition CcdbApi.cxx:1694
static constexpr const char * CCDBQUERY_ENTRY
Definition CcdbApi.h:341
void truncate(std::string const &path) const
Definition CcdbApi.cxx:1394
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
Definition CcdbApi.cxx:426
static void parseCCDBHeaders(std::vector< std::string > const &headers, std::vector< std::string > &pfns, std::string &etag)
Definition CcdbApi.cxx:1679
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
Definition CcdbApi.cxx:919
int updateMetadata(std::string const &path, std::map< std::string, std::string > const &metadata, long timestamp, std::string const &id="", long newEOV=0)
Definition CcdbApi.cxx:1749
static constexpr const char * CCDBMETA_ENTRY
Definition CcdbApi.h:342
static constexpr const char * CCDBOBJECT_ENTRY
Definition CcdbApi.h:343
void navigateSourcesAndLoadFile(RequestContext &requestContext, int &fromSnapshot, size_t *requestCounter) const
Definition CcdbApi.cxx:2093
std::vector< std::string > parseSubFolders(std::string const &reply) const
Definition CcdbApi.cxx:1470
virtual ~CcdbApi()
Default destructor.
Definition CcdbApi.cxx:189
void add(CCDBSemaphore const *ptr)
Definition CcdbApi.cxx:2334
void remove(CCDBSemaphore const *ptr)
Definition CcdbApi.cxx:2339
const GLfloat * m
Definition glcorearb.h:4066
GLuint64EXT * result
Definition glcorearb.h:5662
GLint void * img
Definition glcorearb.h:550
GLuint buffer
Definition glcorearb.h:655
GLuint entry
Definition glcorearb.h:5735
GLsizeiptr size
Definition glcorearb.h:659
GLuint GLuint end
Definition glcorearb.h:469
GLuint index
Definition glcorearb.h:781
GLdouble GLdouble GLdouble GLdouble top
Definition glcorearb.h:4077
GLdouble f
Definition glcorearb.h:310
GLboolean GLboolean GLboolean b
Definition glcorearb.h:1233
GLenum GLint * range
Definition glcorearb.h:1899
GLsizei const GLfloat * value
Definition glcorearb.h:819
GLboolean * data
Definition glcorearb.h:298
GLsizei const GLchar *const * path
Definition glcorearb.h:3591
GLuint object
Definition glcorearb.h:4041
GLuint start
Definition glcorearb.h:469
GLboolean GLboolean GLboolean GLboolean a
Definition glcorearb.h:1233
GLuint GLuint stream
Definition glcorearb.h:1806
GLbitfield GLuint64 timeout
Definition glcorearb.h:1573
GLuint id
Definition glcorearb.h:650
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)
Definition CcdbApi.cxx:1506
size_t write_data(void *, size_t size, size_t nmemb, void *)
Definition CcdbApi.cxx:1431
long getCurrentTimestamp()
returns the timestamp in long corresponding to "now"
size_t(*)(void *, size_t, size_t, void *) CurlWriteCallback
Definition CcdbApi.cxx:722
std::string sanitizeObjectName(const std::string &objectName)
Definition CcdbApi.cxx:373
std::mutex gIOMutex
Definition CcdbApi.cxx:64
bool jsonfile_to_stdmap(std::map< std::string, std::string > &meta, std::string const &filename)
Definition CcdbApi.cxx:1535
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>
Definition CcdbApi.h:63
size_t CurlWrite_CallbackFunc_StdString2(void *contents, size_t size, size_t nmemb, std::string *s)
Definition CcdbApi.cxx:1287
std::string timestamp() noexcept
Definition Clock.h:84
Defining ITS Vertex explicitly as messageable.
Definition Cartesian.h:288
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)
Definition common.h:52
std::string filename()
void empty(int)
std::map< std::string, std::string > const & metadata
Definition CcdbApi.h:366
std::map< std::string, std::string > & headers
Definition CcdbApi.h:368
o2::pmr::vector< char > & dest
Definition CcdbApi.h:364
o2::pmr::vector< char > * object
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)
Definition StringUtils.h:70
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()))
const std::string str
uint64_t const void const *restrict const msg
Definition x9.h:153