diff --git a/Framework/basic.json b/Framework/basic.json index 942014ceff..9bf3ca6c98 100644 --- a/Framework/basic.json +++ b/Framework/basic.json @@ -39,6 +39,7 @@ "tasks": { "QcTask": { "active": "true", + "critical": "false", "": "if false the task is allowed to die without stopping the workflow, default: true", "className": "o2::quality_control_modules::skeleton::SkeletonTask", "moduleName": "QcSkeleton", "detectorName": "TST", @@ -105,8 +106,7 @@ "fraction": "0.1", "seed": "1234" } - ], - "blocking": "false" + ] } ] } diff --git a/Framework/include/QualityControl/InfrastructureGenerator.h b/Framework/include/QualityControl/InfrastructureGenerator.h index cc102fd150..09484af311 100644 --- a/Framework/include/QualityControl/InfrastructureGenerator.h +++ b/Framework/include/QualityControl/InfrastructureGenerator.h @@ -219,9 +219,13 @@ class InfrastructureGenerator static void generateMergers(framework::WorkflowSpec& workflow, const std::string& taskName, size_t numberOfLocalMachines, std::vector> cycleDurationSeconds, - const std::string& mergingMode, size_t resetAfterCycles, - std::string monitoringUrl, const std::string& detectorName, - std::vector mergersPerLayer, bool enableMovingWindows); + const std::string& mergingMode, + size_t resetAfterCycles, + std::string monitoringUrl, + const std::string& detectorName, + std::vector mergersPerLayer, + bool enableMovingWindows, + bool critical); static void generateCheckRunners(framework::WorkflowSpec& workflow, const InfrastructureSpec& infrastructureSpec); static void generateAggregator(framework::WorkflowSpec& workflow, const InfrastructureSpec& infrastructureSpec); static void generatePostProcessing(framework::WorkflowSpec& workflow, const InfrastructureSpec& infrastructureSpec); diff --git a/Framework/include/QualityControl/PostProcessingConfig.h b/Framework/include/QualityControl/PostProcessingConfig.h index 712652a9b1..eb128ba549 100644 --- a/Framework/include/QualityControl/PostProcessingConfig.h +++ b/Framework/include/QualityControl/PostProcessingConfig.h @@ -47,6 +47,7 @@ struct PostProcessingConfig { std::string consulUrl; core::Activity activity; bool matchAnyRunNumber = false; + bool critical; core::CustomParameters customParameters; }; diff --git a/Framework/include/QualityControl/PostProcessingTaskSpec.h b/Framework/include/QualityControl/PostProcessingTaskSpec.h index 6bdee3d446..4847416eab 100644 --- a/Framework/include/QualityControl/PostProcessingTaskSpec.h +++ b/Framework/include/QualityControl/PostProcessingTaskSpec.h @@ -41,6 +41,7 @@ struct PostProcessingTaskSpec { std::string id = "Invalid"; std::string taskName = "Invalid"; bool active = true; + bool critical = true; std::string detectorName = "Invalid"; boost::property_tree::ptree tree = {}; core::CustomParameters customParameters; diff --git a/Framework/include/QualityControl/TaskRunnerConfig.h b/Framework/include/QualityControl/TaskRunnerConfig.h index 48646c922d..e39a9aac2f 100644 --- a/Framework/include/QualityControl/TaskRunnerConfig.h +++ b/Framework/include/QualityControl/TaskRunnerConfig.h @@ -47,6 +47,7 @@ struct TaskRunnerConfig { std::string className; std::vector> cycleDurations = {}; int maxNumberCycles; + bool critical; std::string consulUrl{}; std::string conditionUrl{}; std::string monitoringUrl{}; diff --git a/Framework/include/QualityControl/TaskSpec.h b/Framework/include/QualityControl/TaskSpec.h index 9fb668b08b..5afe0af726 100644 --- a/Framework/include/QualityControl/TaskSpec.h +++ b/Framework/include/QualityControl/TaskSpec.h @@ -59,6 +59,7 @@ struct TaskSpec { DataSourceSpec dataSource; // advanced bool active = true; + bool critical = true; int maxNumberCycles = -1; size_t resetAfterCycles = 0; std::string saveObjectsToFile; diff --git a/Framework/postprocessing.json b/Framework/postprocessing.json index 7c2e8a29db..5362227b9d 100644 --- a/Framework/postprocessing.json +++ b/Framework/postprocessing.json @@ -42,6 +42,7 @@ "postprocessing": { "ExamplePostprocessing": { "active": "true", + "critical": "false", "": "if false the task is allowed to die without stopping the workflow, default: true", "className": "o2::quality_control_modules::skeleton::SkeletonPostProcessing", "moduleName": "QcSkeleton", "detectorName": "TST", diff --git a/Framework/src/AggregatorRunnerFactory.cxx b/Framework/src/AggregatorRunnerFactory.cxx index b9b5aa6a2a..a2d44f1a56 100644 --- a/Framework/src/AggregatorRunnerFactory.cxx +++ b/Framework/src/AggregatorRunnerFactory.cxx @@ -49,6 +49,8 @@ DataProcessorSpec AggregatorRunnerFactory::create(const core::CommonSpec& common }; newAggregatorRunner.labels.emplace_back(o2::framework::ecs::qcReconfigurable); newAggregatorRunner.labels.emplace_back(AggregatorRunner::getLabel()); + framework::DataProcessorLabel resilientLabel = { "resilient" }; + newAggregatorRunner.labels.emplace_back(resilientLabel); newAggregatorRunner.algorithm = adaptFromTask(std::move(aggregatorRunner)); return newAggregatorRunner; } diff --git a/Framework/src/CheckRunner.cxx b/Framework/src/CheckRunner.cxx index 83a05d8325..2142c695a8 100644 --- a/Framework/src/CheckRunner.cxx +++ b/Framework/src/CheckRunner.cxx @@ -357,6 +357,7 @@ QualityObjectsType CheckRunner::check() QualityObjectsType allQOs; for (auto& [checkName, check] : mChecks) { if (updatePolicyManager.isReady(check.getName())) { + ILOG(Debug, Support) << "Monitor Objects for the check '" << checkName << "' are ready --> check()" << ENDM; auto newQOs = check.check(mMonitorObjects); mTotalNumberCheckExecuted += newQOs.size(); @@ -366,7 +367,7 @@ QualityObjectsType CheckRunner::check() // Was checked, update latest revision updatePolicyManager.updateActorRevision(checkName); } else { - ILOG(Info, Support) << "Monitor Objects for the check '" << checkName << "' are not ready, ignoring" << ENDM; + ILOG(Debug, Support) << "Monitor Objects for the check '" << checkName << "' are not ready, ignoring" << ENDM; } } return allQOs; diff --git a/Framework/src/CheckRunnerFactory.cxx b/Framework/src/CheckRunnerFactory.cxx index 7d9e3f6276..fe6f2a41a2 100644 --- a/Framework/src/CheckRunnerFactory.cxx +++ b/Framework/src/CheckRunnerFactory.cxx @@ -45,6 +45,7 @@ DataProcessorSpec CheckRunnerFactory::create(CheckRunnerConfig checkRunnerConfig options }; newCheckRunner.labels.emplace_back(o2::framework::ecs::qcReconfigurable); newCheckRunner.labels.emplace_back(CheckRunner::getCheckRunnerLabel()); + newCheckRunner.labels.emplace_back(framework::DataProcessorLabel{ "resilient" }); newCheckRunner.algorithm = adaptFromTask(std::move(qcCheckRunner)); return newCheckRunner; } @@ -61,7 +62,7 @@ DataProcessorSpec CheckRunnerFactory::createSinkDevice(const CheckRunnerConfig& checkRunnerConfig.options, {}, { o2::framework::ecs::qcReconfigurable } }; - + newCheckRunner.labels.emplace_back(framework::DataProcessorLabel{ "resilient" }); return newCheckRunner; } diff --git a/Framework/src/InfrastructureGenerator.cxx b/Framework/src/InfrastructureGenerator.cxx index f10b4c2636..dbc70961ff 100644 --- a/Framework/src/InfrastructureGenerator.cxx +++ b/Framework/src/InfrastructureGenerator.cxx @@ -144,7 +144,7 @@ framework::WorkflowSpec InfrastructureGenerator::generateFullChainInfrastructure bool enableMovingWindows = !taskSpec.movingWindows.empty(); generateMergers(workflow, taskSpec.taskName, 1, cycleDurationsMultiplied, taskSpec.mergingMode, resetAfterCycles, infrastructureSpec.common.monitoringUrl, - taskSpec.detectorName, taskSpec.mergersPerLayer, enableMovingWindows); + taskSpec.detectorName, taskSpec.mergersPerLayer, enableMovingWindows, taskSpec.critical); } else { // TaskLocationSpec::Remote auto taskConfig = TaskRunnerFactory::extractConfig(infrastructureSpec.common, taskSpec, 0, taskSpec.resetAfterCycles); workflow.emplace_back(TaskRunnerFactory::create(taskConfig)); @@ -277,9 +277,8 @@ o2::framework::WorkflowSpec InfrastructureGenerator::generateRemoteInfrastructur std::for_each(cycleDurationsMultiplied.begin(), cycleDurationsMultiplied.end(), [taskSpec](std::pair& p) { p.first *= taskSpec.mergerCycleMultiplier; }); bool enableMovingWindows = !taskSpec.movingWindows.empty(); - generateMergers(workflow, taskSpec.taskName, numberOfLocalMachines, cycleDurationsMultiplied, - taskSpec.mergingMode, resetAfterCycles, infrastructureSpec.common.monitoringUrl, - taskSpec.detectorName, taskSpec.mergersPerLayer, enableMovingWindows); + generateMergers(workflow, taskSpec.taskName, numberOfLocalMachines, cycleDurationsMultiplied, taskSpec.mergingMode, + resetAfterCycles, infrastructureSpec.common.monitoringUrl, taskSpec.detectorName, taskSpec.mergersPerLayer, enableMovingWindows, taskSpec.critical); } else if (taskSpec.location == TaskLocationSpec::Remote) { @@ -550,6 +549,9 @@ void InfrastructureGenerator::generateLocalTaskLocalProxy(framework::WorkflowSpe { proxyInput }, channelConfig.c_str())); workflow.back().labels.emplace_back(taskSpec.localControl == "odc" ? ecs::preserveRawChannelsLabel : ecs::uniqueProxyLabel); + if (!taskSpec.critical) { + workflow.back().labels.emplace_back(framework::DataProcessorLabel{ "expendable" }); + } if (getenv("O2_QC_KILL_PROXIES") != nullptr) { workflow.back().metadata.push_back(DataProcessorMetadata{ ecs::privateMemoryKillThresholdMB, proxyMemoryKillThresholdMB }); } @@ -577,6 +579,9 @@ void InfrastructureGenerator::generateLocalTaskRemoteProxy(framework::WorkflowSp channelConfig.c_str(), dplModelAdaptor()); proxy.labels.emplace_back(taskSpec.localControl == "odc" ? ecs::preserveRawChannelsLabel : ecs::uniqueProxyLabel); + if (!taskSpec.critical) { + workflow.back().labels.emplace_back(framework::DataProcessorLabel{ "expendable" }); + } // if not in RUNNING, we should drop all the incoming messages, we set the corresponding proxy option. enableDraining(proxy.options); if (getenv("O2_QC_KILL_PROXIES") != nullptr) { @@ -585,11 +590,9 @@ void InfrastructureGenerator::generateLocalTaskRemoteProxy(framework::WorkflowSp workflow.emplace_back(std::move(proxy)); } void InfrastructureGenerator::generateMergers(framework::WorkflowSpec& workflow, const std::string& taskName, - size_t numberOfLocalMachines, - std::vector> cycleDurations, - const std::string& mergingMode, size_t resetAfterCycles, - std::string monitoringUrl, const std::string& detectorName, - std::vector mergersPerLayer, bool enableMovingWindows) + size_t numberOfLocalMachines, std::vector> cycleDurations, + const std::string& mergingMode, size_t resetAfterCycles, std::string monitoringUrl, + const std::string& detectorName, std::vector mergersPerLayer, bool enableMovingWindows, bool critical) { Inputs mergerInputs; for (size_t id = 1; id <= numberOfLocalMachines; id++) { @@ -617,8 +620,9 @@ void InfrastructureGenerator::generateMergers(framework::WorkflowSpec& workflow, mergerConfig.topologySize = { TopologySize::MergersPerLayer, mergersPerLayer }; mergerConfig.monitoringUrl = std::move(monitoringUrl); mergerConfig.detectorName = detectorName; - mergerConfig.parallelismType = { (mergerConfig.inputObjectTimespan.value == InputObjectsTimespan::LastDifference) ? ParallelismType::RoundRobin : ParallelismType::SplitInputs }; + mergerConfig.labels.push_back({ "resilient" }); mergerConfig.publishMovingWindow = { enableMovingWindows ? PublishMovingWindow::Yes : PublishMovingWindow::No }; + mergerConfig.parallelismType = { (mergerConfig.inputObjectTimespan.value == InputObjectsTimespan::LastDifference) ? ParallelismType::RoundRobin : ParallelismType::SplitInputs }; mergersBuilder.setConfig(mergerConfig); mergersBuilder.generateInfrastructure(workflow); @@ -786,6 +790,10 @@ void InfrastructureGenerator::generatePostProcessing(WorkflowSpec& workflow, con ppTask.getOptions() }; dataProcessorSpec.labels.emplace_back(PostProcessingDevice::getLabel()); + if (!ppTaskSpec.critical) { + framework::DataProcessorLabel expendableLabel = { "expendable" }; + dataProcessorSpec.labels.emplace_back(expendableLabel); + } dataProcessorSpec.algorithm = adaptFromTask(std::move(ppTask)); workflow.emplace_back(std::move(dataProcessorSpec)); diff --git a/Framework/src/InfrastructureSpecReader.cxx b/Framework/src/InfrastructureSpecReader.cxx index 38f6640c05..cffa04de04 100644 --- a/Framework/src/InfrastructureSpecReader.cxx +++ b/Framework/src/InfrastructureSpecReader.cxx @@ -110,6 +110,7 @@ TaskSpec InfrastructureSpecReader::readSpecEntry(const std::string& ta } ts.dataSource = readSpecEntry(taskID, taskTree.get_child("dataSource"), wholeTree); ts.active = taskTree.get("active", ts.active); + ts.critical = taskTree.get("critical", ts.critical); ts.maxNumberCycles = taskTree.get("maxNumberCycles", ts.maxNumberCycles); ts.resetAfterCycles = taskTree.get("resetAfterCycles", ts.resetAfterCycles); ts.saveObjectsToFile = taskTree.get("saveObjectsToFile", ts.saveObjectsToFile); @@ -363,6 +364,7 @@ PostProcessingTaskSpec ppts.id = ppTaskId; ppts.taskName = ppTaskTree.get("taskName", ppts.id); ppts.active = ppTaskTree.get("active", ppts.active); + ppts.critical = ppTaskTree.get("critical", ppts.critical); ppts.detectorName = ppTaskTree.get("detectorName", ppts.detectorName); ppts.tree = wholeTree; diff --git a/Framework/src/PostProcessingConfig.cxx b/Framework/src/PostProcessingConfig.cxx index 088374b199..21ac362ae7 100644 --- a/Framework/src/PostProcessingConfig.cxx +++ b/Framework/src/PostProcessingConfig.cxx @@ -37,7 +37,8 @@ PostProcessingConfig::PostProcessingConfig(const std::string& id, const boost::p config.get("qc.config.Activity.provenance", "qc"), { config.get("qc.config.Activity.start", 0), config.get("qc.config.Activity.end", -1) }), - matchAnyRunNumber(config.get("qc.config.postprocessing.matchAnyRunNumber", false)) + matchAnyRunNumber(config.get("qc.config.postprocessing.matchAnyRunNumber", false)), + critical(true) { for (const auto& initTrigger : config.get_child("qc.postprocessing." + id + ".initTrigger")) { initTriggers.push_back(initTrigger.second.get_value()); diff --git a/Framework/src/TaskRunner.cxx b/Framework/src/TaskRunner.cxx index 7024ba7031..e28d119de6 100644 --- a/Framework/src/TaskRunner.cxx +++ b/Framework/src/TaskRunner.cxx @@ -413,6 +413,7 @@ void TaskRunner::printTaskConfig() const << " / Module name : " << mTaskConfig.moduleName // << " / Detector name : " << mTaskConfig.detectorName // << " / Max number cycles : " << mTaskConfig.maxNumberCycles // + << " / critical : " << mTaskConfig.critical // << " / Save to file : " << mTaskConfig.saveToFile << " / Cycle duration seconds : "; for (auto& [cycleDuration, period] : mTaskConfig.cycleDurations) { diff --git a/Framework/src/TaskRunnerFactory.cxx b/Framework/src/TaskRunnerFactory.cxx index 4a692dfb92..3bd046660e 100644 --- a/Framework/src/TaskRunnerFactory.cxx +++ b/Framework/src/TaskRunnerFactory.cxx @@ -53,6 +53,10 @@ o2::framework::DataProcessorSpec TaskRunnerFactory::create(const TaskRunnerConfi }; newTask.labels.emplace_back(o2::framework::ecs::qcReconfigurable); newTask.labels.emplace_back(TaskRunner::getTaskRunnerLabel()); + if (!taskConfig.critical) { + framework::DataProcessorLabel expendableLabel = { "expendable" }; + newTask.labels.emplace_back(expendableLabel); + } return newTask; } @@ -144,6 +148,7 @@ TaskRunnerConfig TaskRunnerFactory::extractConfig(const CommonSpec& globalConfig taskSpec.className, multipleCycleDurations, taskSpec.maxNumberCycles, + taskSpec.critical, globalConfig.consulUrl, globalConfig.conditionDBUrl, globalConfig.monitoringUrl, diff --git a/Framework/test/testTaskInterface.cxx b/Framework/test/testTaskInterface.cxx index a8e48bac05..77cbc1de19 100644 --- a/Framework/test/testTaskInterface.cxx +++ b/Framework/test/testTaskInterface.cxx @@ -173,6 +173,7 @@ TEST_CASE("test_task_factory") "o2::quality_control_modules::skeleton::SkeletonTask", { { 10, 1 } }, -1, + true, "" }; diff --git a/doc/Advanced.md b/doc/Advanced.md index 663053ea28..e7df8973bc 100644 --- a/doc/Advanced.md +++ b/doc/Advanced.md @@ -5,6 +5,7 @@ Advanced topics + * [Advanced topics](#advanced-topics) * [Framework](#framework) * [Plugging the QC to an existing DPL workflow](#plugging-the-qc-to-an-existing-dpl-workflow) * [Production of QC objects outside this framework](#production-of-qc-objects-outside-this-framework) @@ -18,6 +19,7 @@ Advanced topics * [Monitor cycles](#monitor-cycles) * [Writing a DPL data producer](#writing-a-dpl-data-producer) * [Custom merging](#custom-merging) + * [Critical and non-critical tasks](#critical-and-non-critical-tasks) * [QC with DPL Analysis](#qc-with-dpl-analysis) * [Uploading objects to QCDB](#uploading-objects-to-qcdb) * [Getting AODs in QC Tasks](#getting-aods-in-qc-tasks) @@ -502,6 +504,54 @@ Feel free to consult the existing usage examples among other modules in the QC r Once a custom class is implemented, one should let QCG know how to display it correctly, which is explained in the subsection [Display a non-standard ROOT object in QCG](#display-a-non-standard-root-object-in-qcg). +## Critical, resilient and non-critical tasks + +DPL devices can be marked as expendable, resilient or critical. Expendable tasks can die without affecting the run. +Resilient tasks can survive having one or all their inputs coming from an expendable task but they will stop the system if they themselves die. +Critical tasks (default) will stop the system if they die and will not accept input from expendable tasks. + +In QC we use these `labels`. + +### QC tasks + +In QC, one can mark a task as critical or non-critical: +```json + "tasks": { + "QcTask": { + "active": "true", + "critical": "false", "": "if false the task is allowed to die without stopping the workflow, default: true", +``` +By default they are `critical` meaning that their failure will stop the run. +If they are not critical, they will be `expendable` and will not stop the run if they die. + +### Auto-generated proxies + +They adopt the criticality of the task they are proxying. + +### QC mergers + +Mergers are `resilient`. + +### QC check runners + +CheckRunners are `resilient`. + +### QC aggregators + +Aggregators are `resilient`. + +### QC post-processing tasks + +Post-processing tasks can be marked as critical or non-critical: +```json + "postprocessing": { + "ExamplePostprocessing": { + "active": "true", + "critical": "false", "": "if false the task is allowed to die without stopping the workflow, default: true", +``` +By default, they are critical meaning that their failure will stop the run. +If they are not critical, they will be `expendable` and will not stop the run if they die. + ## QC with DPL Analysis QC offers several ways to interact with the DPL Analysis framework. @@ -1308,6 +1358,7 @@ the "tasks" path. "className": "namespace::of::Task", "": "Class name of the QC Task with full namespace.", "moduleName": "QcSkeleton", "": "Library name. It can be found in CMakeLists of the detector module.", "detectorName": "TST", "": "3-letter code of the detector.", + "critical": "true", "": "if false the task is allowed to die without stopping the workflow, default: true", "cycleDurationSeconds": "10", "": "Cycle duration (how often objects are published), 10 seconds minimum.", "": "The first cycle will be randomly shorter. ", "": "Alternatively, one can specify different cycle durations for different periods. The last item in cycleDurations will be used for the rest of the duration whatever the period. The first cycle will be randomly shorter.",