[QC-421] Allow to check PP tasks results in CheckRunners (#506)

knopers8 · web-flow · commit d8091df3b9ef · 2020-10-14T15:15:39.000+02:00
* [QC-421] Allow to check PP tasks results in CheckRunners

* fix infrastructuregenerator test
diff --git a/Framework/postprocessing.json b/Framework/postprocessing.json
@@ -22,6 +22,20 @@
         "url": "ccdb-test.cern.ch:8080"
       }
     },
+    "checks": {
+      "ExamplePPCheck": {
+        "active": "true",
+        "className": "o2::quality_control_modules::skeleton::SkeletonCheck",
+        "moduleName": "QcSkeleton",
+        "policy": "OnAny",
+        "detectorName": "TST",
+        "dataSource": [{
+          "type": "PostProcessing",
+          "name": "ExampleTrend",
+          "MOs": ["mean_of_histogram"]
+        }]
+      }
+    },
     "postprocessing": {
       "ExamplePostprocessing": {
         "active": "true",
diff --git a/Framework/src/Check.cxx b/Framework/src/Check.cxx
@@ -25,6 +25,7 @@
 #include "QualityControl/TaskRunner.h"
 #include "QualityControl/InputUtils.h"
 #include "QualityControl/RootClassFactory.h"
+#include "QualityControl/PostProcessingDevice.h"
 // Fairlogger
 #include <fairlogger/Logger.h>
 
@@ -34,6 +35,7 @@ using namespace o2::configuration;
 
 using namespace o2::quality_control::checker;
 using namespace o2::quality_control::core;
+using namespace o2::quality_control::postprocessing;
 using namespace std;
 
 /// Static functions
@@ -88,12 +90,15 @@ void Check::initConfig(std::string checkName)
   mNumberOfTaskSources = 0;
   for (const auto& [_key, dataSource] : checkConfig.get_child("dataSource")) {
     (void)_key;
-    if (dataSource.get<std::string>("type") == "Task" || dataSource.get<std::string>("type") == "ExternalTask") {
+    if (auto sourceType = dataSource.get<std::string>("type");
+        sourceType == "Task" || sourceType == "ExternalTask" || sourceType == "PostProcessing") {
       auto taskName = dataSource.get<std::string>("name");
       mNumberOfTaskSources++;
 
       if (dataSource.get<std::string>("type") == "Task") {
         mInputs.push_back({ taskName, TaskRunner::createTaskDataOrigin(), TaskRunner::createTaskDataDescription(taskName) });
+      } else if (dataSource.get<std::string>("type") == "PostProcessing") {
+        mInputs.push_back({ taskName, PostProcessingDevice::createPostProcessingDataOrigin(), PostProcessingDevice::createPostProcessingDataDescription(taskName) });
       } else if (dataSource.get<std::string>("type") == "ExternalTask") {
         auto query = config->getString("qc.externalTasks." + taskName + ".query").get();
         framework::Inputs input = o2::framework::DataDescriptorQueryBuilder::parse(query.c_str());
diff --git a/Framework/src/CheckRunner.cxx b/Framework/src/CheckRunner.cxx
@@ -274,7 +274,7 @@ void CheckRunner::prepareCacheData(framework::InputRecord& inputRecord)
 void CheckRunner::sendPeriodicMonitoring()
 {
   if (mTimer.isTimeout()) {
-    mTimer.reset(1000000); // 10 s.
+    mTimer.reset(10000000); // 10 s.
     mCollector->send({ mTotalNumberObjectsReceived, "qc_objects_received" }, DerivedMetricMode::RATE);
     mCollector->send({ mTotalNumberCheckExecuted, "qc_checks_executed" }, DerivedMetricMode::RATE);
     mCollector->send({ mTotalNumberQOStored, "qc_qo_stored" }, DerivedMetricMode::RATE);
@@ -397,7 +397,7 @@ void CheckRunner::initMonitoring()
   mCollector->enableProcessMonitoring();
   mCollector->addGlobalTag(tags::Key::Subsystem, tags::Value::QC);
   mCollector->addGlobalTag("CheckRunnerName", mDeviceName);
-  mTimer.reset(1000000); // 10 s.
+  mTimer.reset(10000000); // 10 s.
 }
 
 void CheckRunner::initServiceDiscovery()
diff --git a/Framework/src/InfrastructureGenerator.cxx b/Framework/src/InfrastructureGenerator.cxx
@@ -367,6 +367,15 @@ void InfrastructureGenerator::generateCheckRunners(framework::WorkflowSpec& work
     }
   }
 
+  if (config->getRecursive("qc").count("postprocessing")) {
+    for (const auto& [ppTaskName, ppTaskConfig] : config->getRecursive("qc.postprocessing")) {
+      if (ppTaskConfig.get<bool>("active", true)) {
+        InputSpec taskOutput{ ppTaskName, PostProcessingDevice::createPostProcessingDataOrigin(), PostProcessingDevice::createPostProcessingDataDescription(ppTaskName) };
+        tasksOutputMap.insert({ DataSpecUtils::label(taskOutput), taskOutput });
+      }
+    }
+  }
+
   // For each external task prepare the InputSpec to be stored in tasksoutputMap
   if (config->getRecursive("qc").count("externalTasks")) {
     for (const auto& [taskName, taskConfig] : config->getRecursive("qc.externalTasks")) {
@@ -467,7 +476,7 @@ void InfrastructureGenerator::generatePostProcessing(WorkflowSpec& workflow, std
         ppTask.getOptions()
       };
 
-      ppTaskSpec.algorithm = std::move(adaptFromTask<PostProcessingDevice>(std::move(ppTask)));
+      ppTaskSpec.algorithm = adaptFromTask<PostProcessingDevice>(std::move(ppTask));
 
       workflow.emplace_back(std::move(ppTaskSpec));
     }
diff --git a/Framework/src/PostProcessingDevice.cxx b/Framework/src/PostProcessingDevice.cxx
@@ -52,11 +52,9 @@ void PostProcessingDevice::init(framework::InitContext& ctx)
 
 void PostProcessingDevice::run(framework::ProcessingContext& ctx)
 {
-  // todo: enable once check runners can store the results.
-  //  Now we still choose the default option - storing data directly to the repo.
   // we set the publication callback each time, because we cannot be sure that
   // the reference to DataAllocator does not change
-  // mRunner.setPublicationCallback(publishToDPL(ctx.outputs(), outputBinding));
+  mRunner->setPublicationCallback(publishToDPL(ctx.outputs(), outputBinding));
 
   // When run returns false, it has done its processing.
   if (!mRunner->run()) {
diff --git a/Framework/test/testCheck.cxx b/Framework/test/testCheck.cxx
@@ -116,3 +116,22 @@ BOOST_AUTO_TEST_CASE(test_check_dont_invoke_beautify)
   // Beautify should not run - more than one MO declared
   BOOST_CHECK(!testCheck.mBeautify);
 }
+
+BOOST_AUTO_TEST_CASE(test_check_postprocessing)
+{
+  std::string configFilePath = std::string("json://") + getTestDataDirectory() + "testSharedConfig.json";
+
+  Check check("checkAnyPP", configFilePath);
+  check.init();
+
+  TestCheck testCheck;
+  check.setCheckInterface(dynamic_cast<CheckInterface*>(&testCheck));
+
+  std::map<std::string, std::shared_ptr<MonitorObject>> moMap = { { "SkeletonPostProcessing/example", std::shared_ptr<MonitorObject>(new MonitorObject()) } };
+
+  check.check(moMap);
+  // Check should run
+  BOOST_CHECK(testCheck.mCheck);
+  // Beautify should run - single MO declared
+  BOOST_CHECK(testCheck.mBeautify);
+}
diff --git a/Framework/test/testInfrastructureGenerator.cxx b/Framework/test/testInfrastructureGenerator.cxx
@@ -81,7 +81,8 @@ BOOST_AUTO_TEST_CASE(qc_factory_remote_test)
 
   // the infrastructure should consist of a proxy, merger and checker for the 'skeletonTask' (its taskRunner is declared to be
   // local) and also taskRunner and checker for the 'abcTask' and 'xyzTask'.
-  BOOST_REQUIRE_EQUAL(workflow.size(), 9);
+  // Post processing adds one process for the task and one for checks.
+  BOOST_REQUIRE_EQUAL(workflow.size(), 10);
 
   auto tcpclustProxy = std::find_if(
     workflow.begin(), workflow.end(),
@@ -142,7 +143,7 @@ BOOST_AUTO_TEST_CASE(qc_factory_remote_test)
       return d.name.find("QC-CHECK-RUNNER") != std::string::npos &&
              d.inputs.size() == 1;
     });
-  BOOST_REQUIRE_EQUAL(checkRunnerCount, 3);
+  BOOST_REQUIRE_EQUAL(checkRunnerCount, 4);
 
   auto postprocessingTask = std::find_if(
     workflow.begin(), workflow.end(),
@@ -159,8 +160,8 @@ BOOST_AUTO_TEST_CASE(qc_factory_standalone_test)
   std::string configFilePath = std::string("json://") + getTestDataDirectory() + "testSharedConfig.json";
   auto workflow = InfrastructureGenerator::generateStandaloneInfrastructure(configFilePath);
 
-  // the infrastructure should consist of 3 TaskRunners, 3 CheckRunners, 1 PostProcessingRunner
-  BOOST_REQUIRE_EQUAL(workflow.size(), 7);
+  // the infrastructure should consist of 3 TaskRunners, 1 PostProcessingRunner, 4 CheckRunners (including one for PP)
+  BOOST_REQUIRE_EQUAL(workflow.size(), 8);
 
   auto taskRunnerSkeleton = std::find_if(
     workflow.begin(), workflow.end(),
@@ -195,7 +196,7 @@ BOOST_AUTO_TEST_CASE(qc_factory_standalone_test)
       return d.name.find("QC-CHECK-RUNNER") != std::string::npos &&
              d.inputs.size() == 1;
     });
-  BOOST_REQUIRE_EQUAL(checkRunnerCount, 3);
+  BOOST_REQUIRE_EQUAL(checkRunnerCount, 4);
 
   auto postprocessingTask = std::find_if(
     workflow.begin(), workflow.end(),
diff --git a/Framework/test/testSharedConfig.json b/Framework/test/testSharedConfig.json
@@ -147,6 +147,19 @@
             "test2"
           ]
         }]
+      },
+      "checkAnyPP": {
+        "active": "true",
+        "className": "o2::quality_control_modules::skeleton::SkeletonCheck",
+        "moduleName": "QcSkeleton",
+        "policy": "OnAny",
+        "dataSource": [{
+          "type": "PostProcessing",
+          "name": "SkeletonPostProcessing",
+          "MOs": [
+            "example"
+          ]
+        }]
       }
     },
     "postprocessing": {
diff --git a/Modules/Skeleton/src/SkeletonCheck.cxx b/Modules/Skeleton/src/SkeletonCheck.cxx
@@ -62,10 +62,10 @@ void SkeletonCheck::beautify(std::shared_ptr<MonitorObject> mo, Quality checkRes
     if (checkResult == Quality::Good) {
       h->SetFillColor(kGreen);
     } else if (checkResult == Quality::Bad) {
-      ILOG(Info, Support) << "Quality::Bad, setting to red";
+      ILOG(Info, Support) << "Quality::Bad, setting to red" << ENDM;
       h->SetFillColor(kRed);
     } else if (checkResult == Quality::Medium) {
-      ILOG(Info, Support) << "Quality::medium, setting to orange";
+      ILOG(Info, Support) << "Quality::medium, setting to orange" << ENDM;
       h->SetFillColor(kOrange);
     }
     h->SetLineColor(kBlack);
diff --git a/doc/Advanced.md b/doc/Advanced.md
@@ -600,7 +600,7 @@ Below the full QC Checks configuration structure is described. Note that more th
         "policy": "OnAny",            "": ["Policy which determines when MOs should be checked. See the documentation",
                                            "of Checks for the list of available policies and their behaviour."],
         "dataSource": [{              "": "List of data source of the Check.",
-          "type": "Task",             "": "Type of the data source, only \"Task\" up to this date", 
+          "type": "Task",             "": "Type of the data source, \"Task\", \"ExternalTask\" or \"PostProcessing\"", 
           "name": "myTask_1",         "": "Name of the Task",
           "MOs": [ "example" ],       "": ["List of MOs to be checked. Use \"all\" (not as a list) to check each MO ",
                                            "which is produced by the Task"]
diff --git a/doc/PostProcessing.md b/doc/PostProcessing.md
@@ -88,6 +88,28 @@ This is a snippet of a JSON structure which configures a post-processing task:
 ```
 Each task is identified by its name (`MyPostProcessingTask`). One can activate it by setting the `"active"` field to `"true"`. The task is loaded given its full `"className"` and a `"moduleName"` where it is located. The `"detectorName"` might be used by tasks to store generated data in correct paths in QCDB. The `"initTrigger"`, `"updateTrigger"` and `"stopTrigger"` lists contain triggers which should invoke corresponding interface methods.
 
+Checks can be applied to the results of Post-processing Tasks just as for normal QC Tasks. However, one should use
+ data source type of `"PostProcessing"` instead of `"Task"`:
+ 
+```
+...
+    "checks": {
+      "ExamplePPCheck": {
+        "active": "true",
+        "className": "o2::quality_control_modules::skeleton::SkeletonCheck",
+        "moduleName": "QcSkeleton",
+        "policy": "OnAny",
+        "detectorName": "TST",
+        "dataSource": [{
+          "type": "PostProcessing",
+          "name": "ExampleTrend",
+          "MOs": ["mean_of_histogram"]
+        }]
+      }
+    },
+...
+```
+
 ### Triggers configuration
 
 Each of the three methods can be invoked by one or more triggers. Below are listed the possible options (case insensitive).
