166 std::unordered_map<std::string, bool> accountedSpecs;
167 auto defHost = options.
get<std::string>(
"condition-backend");
168 auto checkRate = options.
get<
int>(
"condition-tf-per-query");
169 auto checkMult = options.
get<
int>(
"condition-tf-per-query-multiplier");
170 helper.useTFSlice = options.
get<
int>(
"condition-use-slice-for-prescaling");
171 helper.timeToleranceMS = options.
get<int64_t>(
"condition-time-tolerance");
172 helper.queryPeriodGlo = checkRate > 0 ? checkRate : std::numeric_limits<int>::max();
173 helper.queryPeriodFactor = checkMult == 0 ? 1 : checkMult;
174 std::string extraCond{};
175 if (helper.useTFSlice) {
176 extraCond =
". Use TFSlice";
177 if (helper.useTFSlice > 0) {
178 extraCond += fmt::format(
" + max TFcounter jump <= {}", helper.useTFSlice);
181 LOGP(info,
"CCDB Backend at: {}, validity check for every {} TF{}{}", defHost, helper.queryPeriodGlo,
182 helper.queryPeriodFactor == 1 ? std::string{} : (helper.queryPeriodFactor > 0 ? fmt::format(
", (query for high-rate objects downscaled by {})", helper.queryPeriodFactor) : fmt::format(
", (query downscaled as TFcounter%{})", -helper.queryPeriodFactor)),
184 LOGP(info,
"Hook to enable signposts for CCDB messages at {}", (
void*)&private_o2_log_ccdb->stacktrace);
185 auto remapString = options.
get<std::string>(
"condition-remap");
187 if (!
result.error.empty()) {
190 helper.remappings =
result.remappings;
191 helper.apis[
""].init(defHost);
192 LOGP(info,
"Initialised default CCDB host {}", defHost);
194 for (
auto&
entry : helper.remappings) {
195 if (helper.apis.find(
entry.second) == helper.apis.end()) {
197 LOGP(info,
"Initialised custom CCDB host {}",
entry.second);
199 LOGP(info,
"{} is remapped to {}",
entry.first,
entry.second);
201 helper.createdNotBefore =
std::to_string(options.
get<int64_t>(
"condition-not-before"));
202 helper.createdNotAfter =
std::to_string(options.
get<int64_t>(
"condition-not-after"));
204 for (
auto& route : outputRoutes) {
205 if (route.matcher.lifetime != Lifetime::Condition) {
209 if (accountedSpecs.find(specStr) != accountedSpecs.end()) {
212 accountedSpecs[specStr] =
true;
213 helper.routes.push_back(route);
215 for (
auto& metadata : route.matcher.metadata) {
217 LOGP(info,
"- {}: {}", metadata.name, metadata.defaultValue.asString());
253 std::string ccdbMetadataPrefix =
"ccdb-metadata-";
260 O2_SIGNPOST_START(ccdb, sid,
"populateCacheWith",
"Starting to populate cache with CCDB objects");
261 for (
auto& route : helper->routes) {
262 int64_t timestampToUse = timestamp;
267 auto&&
v = allocator.makeVector<
char>(
output);
268 std::map<std::string, std::string> metadata;
269 std::map<std::string, std::string> headers;
270 std::string
path =
"";
271 std::string
etag =
"";
272 int chRate = helper->queryPeriodGlo;
273 bool checkValidity =
false;
274 for (
auto& meta : route.matcher.metadata) {
275 if (meta.name ==
"ccdb-path") {
276 path = meta.defaultValue.get<std::string>();
277 }
else if (meta.name ==
"ccdb-run-dependent" && meta.defaultValue.get<
int>() > 0) {
278 if (meta.defaultValue.get<
int>() == 1) {
279 metadata[
"runNumber"] = dtc.runNumber;
280 }
else if (meta.defaultValue.get<
int>() == 2) {
281 timestampToUse = std::stoi(dtc.runNumber);
283 LOGP(fatal,
"Undefined ccdb-run-dependent option {} for spec {}/{}/{}", meta.defaultValue.get<
int>(), concrete.origin.as<std::string>(), concrete.description.as<std::string>(),
int(concrete.subSpec));
285 }
else if (
isPrefix(ccdbMetadataPrefix, meta.name)) {
286 std::string
key = meta.name.substr(ccdbMetadataPrefix.size());
287 auto value = meta.defaultValue.get<std::string>();
290 }
else if (meta.name ==
"ccdb-query-rate") {
291 chRate = std::max(1, meta.defaultValue.get<
int>()) * helper->queryPeriodFactor;
294 const auto url2uuid = helper->mapURL2UUID.find(
path);
295 if (url2uuid != helper->mapURL2UUID.end()) {
296 etag = url2uuid->second.etag;
298 uint64_t validUntil = url2uuid->second.cacheValidUntil;
303 bool cacheExpired = (validUntil <= timestampToUse) || (timestamp <
cachePopulatedAt);
304 if (isOnline || cacheExpired) {
305 if (!helper->useTFSlice) {
306 checkValidity = chRate > 0 ? (std::abs(
int(timingInfo.tfCounter - url2uuid->second.lastCheckedTF)) >= chRate) : (timingInfo.tfCounter % -chRate) == 0;
308 checkValidity = chRate > 0 ? (std::abs(
int(timingInfo.timeslice - url2uuid->second.lastCheckedSlice)) >= chRate) : (timingInfo.timeslice % -chRate) == 0;
309 if (!checkValidity && helper->useTFSlice > std::abs(chRate)) {
310 checkValidity = std::abs(
int(timingInfo.tfCounter) - url2uuid->second.lastCheckedTF) > helper->useTFSlice;
315 checkValidity =
true;
318 O2_SIGNPOST_EVENT_EMIT(ccdb, sid,
"populateCacheWith",
"checkValidity is %{public}s for tf%{public}s %zu of %{public}s", checkValidity ?
"true" :
"false", helper->useTFSlice ?
"ID" :
"Slice", helper->useTFSlice ? timingInfo.timeslice : timingInfo.tfCounter,
path.data());
320 const auto& api = helper->getAPI(
path);
321 if (checkValidity && (!api.isSnapshotMode() ||
etag.empty())) {
322 LOGP(detail,
"Loading {} for timestamp {}",
path, timestampToUse);
323 api.loadFileToMemory(
v,
path, metadata, timestampToUse, &headers,
etag, helper->createdNotAfter, helper->createdNotBefore);
324 if ((headers.count(
"Error") != 0) || (
etag.empty() &&
v.empty())) {
325 LOGP(fatal,
"Unable to find CCDB object {}/{}",
path, timestampToUse);
330 if (headers.find(
"default") != headers.end()) {
331 LOGP(detail,
"******** Default entry used for {} ********",
path);
333 helper->mapURL2UUID[
path].lastCheckedTF = timingInfo.tfCounter;
334 helper->mapURL2UUID[
path].lastCheckedSlice = timingInfo.timeslice;
336 helper->mapURL2UUID[
path].etag = headers[
"ETag"];
337 helper->mapURL2UUID[
path].cachePopulatedAt = timestampToUse;
338 helper->mapURL2UUID[
path].cacheMiss++;
339 helper->mapURL2UUID[
path].minSize = std::min(
v.size(), helper->mapURL2UUID[
path].minSize);
340 helper->mapURL2UUID[
path].maxSize = std::max(
v.size(), helper->mapURL2UUID[
path].maxSize);
341 api.appendFlatHeader(
v, headers);
343 helper->mapURL2DPLCache[
path] = cacheId;
344 O2_SIGNPOST_EVENT_EMIT(ccdb, sid,
"populateCacheWith",
"Caching %{public}s for %{public}s (DPL id %" PRIu64
")",
path.data(), headers[
"ETag"].data(), cacheId.value);
349 helper->mapURL2UUID[
path].etag = headers[
"ETag"];
350 helper->mapURL2UUID[
path].cachePopulatedAt = timestampToUse;
351 helper->mapURL2UUID[
path].cacheValidUntil = headers[
"Cache-Valid-Until"].empty() ? 0 : std::stoul(headers[
"Cache-Valid-Until"]);
352 helper->mapURL2UUID[
path].cacheMiss++;
353 helper->mapURL2UUID[
path].minSize = std::min(
v.size(), helper->mapURL2UUID[
path].minSize);
354 helper->mapURL2UUID[
path].maxSize = std::max(
v.size(), helper->mapURL2UUID[
path].maxSize);
355 api.appendFlatHeader(
v, headers);
357 helper->mapURL2DPLCache[
path] = cacheId;
358 O2_SIGNPOST_EVENT_EMIT(ccdb, sid,
"populateCacheWith",
"Caching %{public}s for %{public}s (DPL id %" PRIu64
")",
path.data(), headers[
"ETag"].data(), cacheId.value);
364 helper->mapURL2UUID[
path].cacheValidUntil = headers[
"Cache-Valid-Until"].empty() ? 0 : std::stoul(headers[
"Cache-Valid-Until"]);
368 auto cacheId = helper->mapURL2DPLCache[
path];
369 O2_SIGNPOST_EVENT_EMIT(ccdb, sid,
"populateCacheWith",
"Reusing %{public}s for %{public}s (DPL id %" PRIu64
")",
path.data(), headers[
"ETag"].data(), cacheId.value);
370 helper->mapURL2UUID[
path].cacheHit++;
374 O2_SIGNPOST_END(ccdb, sid,
"populateCacheWith",
"Finished populating cache with CCDB objects");
380 std::shared_ptr<CCDBFetcherHelper> helper = std::make_shared<CCDBFetcherHelper>();
385 LOGP(info,
"CCDB cache miss/hit ratio:");
386 for (
auto&
entry : helper->mapURL2UUID) {
387 LOGP(info,
" {}: {}/{} ({}-{} bytes)",
entry.first,
entry.second.cacheMiss,
entry.second.cacheHit,
entry.second.minSize,
entry.second.maxSize);
393 O2_SIGNPOST_START(ccdb, sid,
"fetchFromCCDB",
"Fetching CCDB objects for timeslice %" PRIu64, (uint64_t)timingInfo.
timeslice);
394 static Long64_t orbitResetTime = -1;
395 static size_t lastTimeUsed = -1;
397 LOGP(info,
"Dummy creation time is not supported for CCDB objects. Setting creation to last one used {}.", lastTimeUsed);
403 const std::string
path =
"CTP/Calib/OrbitReset";
404 std::map<std::string, std::string> metadata;
405 std::map<std::string, std::string> headers;
408 bool checkValidity = std::abs(
int(
counter - helper->lastCheckedTFCounterOrbReset)) >= helper->queryPeriodGlo;
409 const auto url2uuid = helper->mapURL2UUID.find(
path);
410 if (url2uuid != helper->mapURL2UUID.end()) {
411 etag = url2uuid->second.etag;
413 checkValidity =
true;
415 O2_SIGNPOST_EVENT_EMIT(ccdb, sid,
"fetchFromCCDB",
"checkValidity is %{public}s for tf%{public}s %d of %{public}s",
416 checkValidity ?
"true" :
"false", helper->useTFSlice ?
"ID" :
"Slice",
counter,
path.data());
418 Long64_t newOrbitResetTime = orbitResetTime;
420 const auto& api = helper->getAPI(
path);
421 if (checkValidity && (!api.isSnapshotMode() ||
etag.empty())) {
422 helper->lastCheckedTFCounterOrbReset =
counter;
423 api.loadFileToMemory(
v,
path, metadata, timingInfo.
creation, &headers,
etag, helper->createdNotAfter, helper->createdNotBefore);
424 if ((headers.count(
"Error") != 0) || (
etag.empty() &&
v.empty())) {
425 LOGP(fatal,
"Unable to find CCDB object {}/{}",
path, timingInfo.
creation);
430 helper->mapURL2UUID[
path].etag = headers[
"ETag"];
431 helper->mapURL2UUID[
path].cacheMiss++;
432 helper->mapURL2UUID[
path].minSize = std::min(
v.size(), helper->mapURL2UUID[
path].minSize);
433 helper->mapURL2UUID[
path].maxSize = std::max(
v.size(), helper->mapURL2UUID[
path].maxSize);
435 api.appendFlatHeader(
v, headers);
437 helper->mapURL2DPLCache[
path] = cacheId;
438 O2_SIGNPOST_EVENT_EMIT(ccdb, sid,
"fetchFromCCDB",
"Caching %{public}s for %{public}s (DPL id %" PRIu64
")",
path.data(), headers[
"ETag"].data(), cacheId.value);
439 }
else if (
v.size()) {
441 helper->mapURL2UUID[
path].etag = headers[
"ETag"];
442 helper->mapURL2UUID[
path].cacheMiss++;
443 helper->mapURL2UUID[
path].minSize = std::min(
v.size(), helper->mapURL2UUID[
path].minSize);
444 helper->mapURL2UUID[
path].maxSize = std::max(
v.size(), helper->mapURL2UUID[
path].maxSize);
446 api.appendFlatHeader(
v, headers);
448 helper->mapURL2DPLCache[
path] = cacheId;
449 O2_SIGNPOST_EVENT_EMIT(ccdb, sid,
"fetchFromCCDB",
"Caching %{public}s for %{public}s (DPL id %" PRIu64
")",
path.data(), headers[
"ETag"].data(), cacheId.value);
455 auto cacheId = helper->mapURL2DPLCache[
path];
456 O2_SIGNPOST_EVENT_EMIT(ccdb, sid,
"fetchFromCCDB",
"Reusing %{public}s for %{public}s (DPL id %" PRIu64
")",
path.data(), headers[
"ETag"].data(), cacheId.value);
457 helper->mapURL2UUID[
path].cacheHit++;
460 if (newOrbitResetTime != orbitResetTime) {
461 O2_SIGNPOST_EVENT_EMIT(ccdb, sid,
"fetchFromCCDB",
"Orbit reset time changed from %lld to %lld", orbitResetTime, newOrbitResetTime);
462 orbitResetTime = newOrbitResetTime;
468 if (std::abs(int64_t(timingInfo.
creation) - timestamp) > helper->timeToleranceMS) {
469 static bool notWarnedYet =
true;
471 LOGP(warn,
"timestamp {} for orbit {} and orbit reset time {} differs by >{} from the TF creation time {}, use the latter", timestamp, timingInfo.
firstTForbit, orbitResetTime / 1000, helper->timeToleranceMS, timingInfo.
creation);
472 notWarnedYet =
false;
479 O2_SIGNPOST_EVENT_EMIT(ccdb, sid,
"fetchFromCCDB",
"Fetching objects. Run %{public}s. OrbitResetTime %lld. Creation %lld. Timestamp %lld. firstTForbit %" PRIu32,
#define O2_DECLARE_DYNAMIC_LOG(name)
#define O2_SIGNPOST_END(log, id, name, format,...)
#define O2_SIGNPOST_EVENT_EMIT(log, id, name, format,...)
#define O2_SIGNPOST_START(log, id, name, format,...)
long orbitResetTimeMUS
The start time of the first orbit in microseconds(!)
DeploymentMode deploymentMode
Where we thing this is running.