Reworked QC infrastructure generation (#87)

This commit introduces a main qc factory which can be used to generate a full topology with just one function call. As the TaskRunner can be placed either locally on FLP or EPN machine or remotely on dedicated QC servers, a distinction between this two has been introduced. When a task is 'local', the taskRunners will be generated by generateLocalInfrastructure(), but mergers and checkers by generateRemoteInfrastructe(). If a task is 'remote', both taskRunner and checker will be generated as remote infrastructure. The commit removes also some deprecated config files.
AliceO2Group · Nov 12, 2018 · 977119d · 977119d
1 parent bca755f
commit 977119d
Show file tree

Hide file tree

Showing 18 changed files with 456 additions and 265 deletions.
diff --git a/Framework/CMakeLists.txt b/Framework/CMakeLists.txt
@@ -17,6 +17,7 @@ set(
   src/TaskInterface.cxx
   src/RepositoryBenchmark.cxx
   src/HistoMerger.cxx
+  src/InfrastructureGenerator.cxx
 )
 
 set(
@@ -31,6 +32,7 @@ set(
   include/QualityControl/TaskRunner.h
   include/QualityControl/TaskRunnerFactory.h
   include/QualityControl/HistoMerger.h
+  include/QualityControl/InfrastructureGenerator.h
 )
 
 set(
@@ -202,6 +204,7 @@ set(
   test/testMonitorObject.cxx
   test/testPublisher.cxx
   test/testQcInfoLogger.cxx
+  test/testInfrastructureGenerator.cxx
   test/testQCTask.cxx
   test/testQuality.cxx
 )
@@ -216,6 +219,12 @@ foreach(test ${TEST_SRCS})
   set_tests_properties(${test_name} PROPERTIES TIMEOUT 60)
 endforeach()
 
+install(
+  FILES
+  test/testQCFactory.json
+  DESTINATION test
+)
+
 # ---- Install ----
 
 # Build targets with install rpath on Mac to dramatically speed up installation
@@ -280,15 +289,13 @@ install(
 install(PROGRAMS script/qcDatabaseSetup.sh DESTINATION bin)
 install(
   FILES
-    example-default.ini
     example-default.json
     alfa.json
     dataDump.json
   DESTINATION etc
 )
 install(
   FILES
-    qcTaskDplConfig.ini
     basic.json
     readout.json
     readoutForDataDump.json

diff --git a/Framework/basic.json b/Framework/basic.json
@@ -13,18 +13,18 @@
         "type": "2"
       }
     },
-    "tasks_config": {
+    "tasks": {
       "QcTask": {
-        "taskDefinition": "QcTaskDefinition"
-      },
-      "QcTaskDefinition": {
+        "active": "true",
         "className": "o2::quality_control_modules::skeleton::SkeletonTask",
         "moduleName": "QcSkeleton",
-        "outputDataOrigin": "ITS",
         "cycleDurationSeconds": "10",
         "maxNumberCycles": "-1",
-        "outputDataDescription": "HIST_SKLT_TASK",
-        "dataSamplingPolicy": "its-raw"
+        "dataSamplingPolicy": "its-raw",
+        "taskParameters": {
+          "nothing": "rien"
+        },
+        "location": "remote"
       }
     }
   },

diff --git a/Framework/example-default.ini b/Framework/example-default.ini
diff --git a/Framework/example-default.json b/Framework/example-default.json
@@ -1,9 +1,6 @@
 {
   "qc": {
     "config": {
-      "DataSampling": {
-        "implementation": "MockSampler"
-      },
       "database": {
         "username": "qc_user",
         "password": "qc_user",
@@ -16,53 +13,71 @@
         "type": "2"
       }
     },
-    "tasks_config": {
+    "tasks": {
       "myTask_1": {
-        "taskDefinition": "taskDefinition_1"
-      },
-      "myTask_2": {
-        "taskDefinition": "taskDefinition_1"
-      },
-      "taskDefinition_1": {
         "className": "o2::quality_control_modules::example::ExampleTask",
         "moduleName": "QcExample",
         "cycleDurationSeconds": "10",
-        "maxNumberCycles": "-1"
+        "maxNumberCycles": "-1",
+        "dataSamplingPolicy": "ex1",
+        "location": "remote"
       },
       "daqTask": {
-        "taskDefinition": "daqTaskDefinition"
-      },
-      "daqTaskDefinition": {
         "className": "o2::quality_control_modules::daq::DaqTask",
         "moduleName": "QcDaq",
         "maxNumberCycles": "-1",
-        "cycleDurationSeconds": "10"
+        "cycleDurationSeconds": "10",
+        "dataSamplingPolicy": "mftclusters",
+        "location": "remote"
       },
       "benchmarkTask_0": {
-        "taskDefinition": "benchmark"
-      },
-      "benchmark": {
         "className": "o2::quality_control_modules::example::BenchmarkTask",
         "moduleName": "QcExample",
-        "numberHistos": "1",
-        "numberChecks": "1",
-        "typeOfChecks": "o2::quality_control_modules::example::FakeCheck",
-        "moduleOfChecks": "QcExample",
-        "cycleDurationSeconds": "1"
+        "cycleDurationSeconds": "1",
+        "dataSamplingPolicy": "ex1",
+        "location": "local",
+        "machines": [
+          "o2flp1",
+          "o2flp2"
+        ]
       }
+    }
+  },
+  "dataSamplingPoliciesFile_comment": "In case that policies are stored in different file, specify its path below. When both dataSamplingPolicies and dataSamplingPoliciesFile are specified, the latter has higher priority",
+  "dataSamplingPoliciesFile": "json:///home/genghiskhan/alice/QualityControl/dataSamplingConfig.json",
+  "dataSamplingPolicies_comment": "this is ignored when dataSamplingPoliciesFile is specified",
+  "dataSamplingPolicies": [
+    {
+      "id": "ex1",
+      "active": "true",
+      "dataHeaders": [
+        {
+          "binding": "data",
+          "dataOrigin": "TST",
+          "dataDescription": "DATA"
+        }
+      ],
+      "subSpec": "0",
+      "samplingConditions": []
     },
-    "checkers_config": {
-      "numberCheckers": "1",
-      "numberTasks": "1",
-      "tasksAddresses": "tcp://localhost:5556,tcp://localhost:5557,tcp://localhost:5558,tcp://localhost:5559",
-      "checker_0": {
-        "broadcast": "0",
-        "broadcastAddress": "tcp://*:5600",
-        "id": "0"
-      },
-      "Checks": {
-        "checkMeanIsAbove/threshold": "1"
-      }
+    {
+      "id": "mftclusters",
+      "active": "true",
+      "dataHeaders": [
+        {
+          "binding": "mft-clusters",
+          "dataOrigin": "MFT",
+          "dataDescription": "CLUSTERS"
+        }
+      ],
+      "subSpec": "0",
+      "samplingConditions": [
+        {
+          "condition": "payloadSize",
+          "upperLimit": "5000",
+          "lowerLimit": "1000"
+        }
+      ]
     }
-  }
+  ]
 }
diff --git a/Framework/include/QualityControl/InfrastructureGenerator.h b/Framework/include/QualityControl/InfrastructureGenerator.h
@@ -0,0 +1,57 @@
+///
+/// \file   QualityControlFactory.h
+/// \author Piotr Konopka
+///
+
+#ifndef QC_CORE_QUALITYCONTROLFACTORY_H
+#define QC_CORE_QUALITYCONTROLFACTORY_H
+
+#include <string>
+#include <Framework/WorkflowSpec.h>
+
+namespace o2
+{
+namespace quality_control
+{
+namespace core
+{
+
+/// \brief A factory class which can generate QC topologies given a configuration file.
+///
+/// A factory class which can generate QC topologies given a configuration file (example in Framework/basic.json and
+/// Framework/example-default.json). As QC topologies will be spread on both processing chain machines and dedicated
+/// QC servers, a _local_ vs. _remote_ distinction was introduced. Tasks which are _local_ should have taskRunners
+/// placed on FLP or EPN machines and their results should be merged and checked on QC servers. The 'remote' option
+/// means, that full QC chain should be located on remote (QC) machines. For the laptop development, use 'remote' tasks
+/// and generateRemoteInfrastructure() to obtain the full topology in one go.
+///
+/// \author Piotr Konopka
+class InfrastructureGenerator
+{
+ public:
+  InfrastructureGenerator() = delete;
+
+  /// \brief Generates the local part of the QC infrastructure for a specified host.
+  ///
+  /// Generates the local part of the QC infrastructure for a specified host - taskRunners which are declared in the
+  /// configuration to be 'local'.
+  ///
+  /// \param configurationSource - full path to configuration file, preceded with the backend (f.e. "json://")
+  /// \param host - name of the machine
+  /// \return generated local QC workflow
+  static o2::framework::WorkflowSpec generateLocalInfrastructure(std::string configurationSource, std::string host);
+
+  /// \brief Generates the remote part of the QC infrastructure
+  ///
+  /// Generates the remote part of the QC infrastructure - mergers and checkers for 'local' tasks and full QC chain for
+  /// 'remote' tasks.
+  ///
+  /// \param configurationSource - full path to configuration file, preceded with the backend (f.e. "json://")
+  /// \return generated remote QC workflow
+  static o2::framework::WorkflowSpec generateRemoteInfrastructure(std::string configurationSource);
+};
+}
+}
+}
+
+#endif //QC_CORE_QUALITYCONTROLFACTORY_H
diff --git a/Framework/include/QualityControl/TaskFactory.h b/Framework/include/QualityControl/TaskFactory.h
@@ -55,7 +55,7 @@ class TaskFactory
     std::string library = "lib" + taskConfig.moduleName;
     logger << "Loading library " << library << AliceO2::InfoLogger::InfoLogger::endm;
     int libLoaded = gSystem->Load(library.c_str(), "", true);
-    if (libLoaded) {
+    if (libLoaded < 0) {
       BOOST_THROW_EXCEPTION(FatalException() << errinfo_details("Failed to load Detector Publisher Library"));
     }
 

diff --git a/Framework/include/QualityControl/TaskRunner.h b/Framework/include/QualityControl/TaskRunner.h
@@ -65,7 +65,12 @@ using namespace std::chrono;
 class TaskRunner
 {
  public:
-  TaskRunner(std::string taskName, std::string configurationSource);
+  /// \brief Constructor
+  ///
+  /// \param taskName - name of the task, which exists in tasks list in the configuration file
+  /// \param configurationSource - absolute path to configuration file, preceded with backend (f.e. "json://")
+  /// \param id - subSpecification for taskRunner's OutputSpec, useful to avoid outputs collisions one more complex topologies
+  TaskRunner(std::string taskName, std::string configurationSource, size_t id = 0);
   ~TaskRunner();
 
   /// \brief To be invoked during initialization of Data Processor
@@ -80,8 +85,10 @@ class TaskRunner
 
   void setResetAfterPublish(bool);
 
+  /// \brief Unified DataOrigin for Quality Control tasks
+  static header::DataOrigin createTaskDataOrigin();
   /// \brief Unified DataDescription naming scheme for all tasks
-  static o2::header::DataDescription createTaskDataDescription(const std::string taskName);
+  static header::DataDescription createTaskDataDescription(const std::string& taskName);
 
  private:
   void populateConfig(std::string taskName);
@@ -94,8 +101,8 @@ class TaskRunner
  private:
   std::string mTaskName;
   TaskConfig mTaskConfig;
-  std::shared_ptr<o2::configuration::ConfigurationInterface> mConfigFile; // used in init only
-  std::shared_ptr<o2::monitoring::Monitoring> mCollector;
+  std::shared_ptr<configuration::ConfigurationInterface> mConfigFile; // used in init only
+  std::shared_ptr<monitoring::Monitoring> mCollector;
   TaskInterface* mTask;
   bool mResetAfterPublish;
   std::shared_ptr<ObjectsManager> mObjectsManager;

diff --git a/Framework/include/QualityControl/TaskRunnerFactory.h b/Framework/include/QualityControl/TaskRunnerFactory.h
@@ -22,7 +22,7 @@ class TaskRunnerFactory
   TaskRunnerFactory();
   virtual ~TaskRunnerFactory();
 
-  o2::framework::DataProcessorSpec create(std::string taskName, std::string configurationSource);
+  o2::framework::DataProcessorSpec create(std::string taskName, std::string configurationSource, size_t id = 0);
 };
 
 } // namespace core