[QC-602] Batch processing in QC

Framework/CMakeLists.txt

@@ -75,7 +75,9 @@ add_library(O2QualityControl
   src/AdvancedWorkflow.cxx
   src/QualitiesToTRFCollectionConverter.cxx
   src/Calculators.cxx
-  src/DataSourceSpec.cxx)
+  src/DataSourceSpec.cxx
+  src/RootFileSink.cxx
+  src/RootFileSource.cxx)
 
 target_include_directories(
   O2QualityControl
@@ -313,6 +315,14 @@ set_tests_properties(multinode_test
 set_property(TEST multinode_test PROPERTY LABELS slow)
 set_property(TEST multinode_test PROPERTY TIMEOUT 75)
 
+# Batch processing test
+string(RANDOM UNIQUE_ID)
+configure_file(batch-test.json.in ${CMAKE_BINARY_DIR}/tests/batch-test.json) # substitute the unique id in the task name
+add_test(NAME batch_test COMMAND o2-qc-batch-test.sh)
+set_tests_properties(batch_test PROPERTIES ENVIRONMENT "JSON_DIR=${CMAKE_BINARY_DIR}/tests;UNIQUE_ID=${UNIQUE_ID}")
+set_property(TEST batch_test PROPERTY LABELS slow)
+set_property(TEST batch_test PROPERTY TIMEOUT 45)
+
 # disable some tests in the CI to avoid un-expected failures.
 # to be removed when it is fixed / understood
 #set_property(TEST functional_test PROPERTY LABELS manual)
@@ -395,6 +405,7 @@ install(FILES example-default.json
 install(PROGRAMS script/RepoCleaner/o2-qc-repo-cleaner
                  script/o2-qc-functional-test.sh
                  script/o2-qc-multinode-test.sh
+                 script/o2-qc-batch-test.sh
         DESTINATION bin)
 
 install(FILES script/RepoCleaner/rules/1_per_hour.py

Framework/batch-test.json.in

@@ -0,0 +1,64 @@
+{
+  "qc": {
+    "config": {
+      "database": {
+        "implementation": "CCDB",
+        "host": "ccdb-test.cern.ch:8080",
+        "username": "not_applicable",
+        "password": "not_applicable",
+        "name": "not_applicable"
+      },
+      "Activity": {
+        "number": "42",
+        "type": "2"
+      },
+      "monitoring": {
+        "url": "infologger:///debug?qc"
+      },
+      "consul": {
+        "url": ""
+      },
+      "conditionDB": {
+        "url": "ccdb-test.cern.ch:8080"
+      }
+    },
+    "tasks": {
+      "BatchTestTask@UNIQUE_ID@": {
+        "active": "true",
+        "className": "o2::quality_control_modules::skeleton::SkeletonTask",
+        "moduleName": "QcSkeleton",
+        "detectorName": "TST",
+        "cycleDurationSeconds": "5",
+        "maxNumberCycles": "-1",
+        "dataSource": {
+          "type": "dataSamplingPolicy",
+          "name": "tst-raw"
+        }
+      }
+    },
+    "checks": {
+      "BatchTestCheck@UNIQUE_ID@": {
+        "active": "true",
+        "className": "o2::quality_control_modules::skeleton::SkeletonCheck",
+        "moduleName": "QcSkeleton",
+        "policy": "OnAny",
+        "detectorName": "TST",
+        "dataSource": [{
+          "type": "Task",
+          "name": "BatchTestTask@UNIQUE_ID@",
+          "MOs": ["example"]
+        }]
+      }
+    }
+  },
+  "dataSamplingPolicies": [
+    {
+      "id": "tst-raw",
+      "active": "true",
+      "machines": [],
+      "query": "random:TST/RAWDATA/0",
+      "samplingConditions": [],
+      "blocking": "false"
+    }
+  ]
+}

Framework/include/QualityControl/InfrastructureGenerator.h

@@ -52,7 +52,7 @@ class InfrastructureGenerator
   /// Generates a full QC infrastructure from a configuration file. This function is aimed to use for standalone setups
   /// and local development. It will create both local and remote QC tasks, and CheckRunners running associated Checks.
   ///
-  /// \param configurationSource - full path to configuration file, preceded with the backend (f.e. "json://")
+  /// \param configurationSource - full path to configuration file, preceded with the backend (e.g. "json://")
   /// \return generated standalone QC workflow
   static framework::WorkflowSpec generateStandaloneInfrastructure(std::string configurationSource);
 
@@ -62,15 +62,15 @@ class InfrastructureGenerator
   /// and local development. It will create both local and remote QC tasks, and CheckRunners running associated Checks.
   ///
   /// \param workflow - existing workflow where QC infrastructure should be placed
-  /// \param configurationSource - full path to configuration file, preceded with the backend (f.e. "json://")
+  /// \param configurationSource - full path to configuration file, preceded with the backend (e.g. "json://")
   static void generateStandaloneInfrastructure(framework::WorkflowSpec& workflow, std::string configurationSource);
 
   /// \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 configurationSource - full path to configuration file, preceded with the backend (e.g. "json://")
   /// \param targetHost - name of the machine
   /// \return generated local QC workflow
   static framework::WorkflowSpec generateLocalInfrastructure(std::string configurationSource, std::string targetHost);
@@ -81,7 +81,7 @@ class InfrastructureGenerator
   /// configuration to be 'local'.
   ///
   /// \param workflow - existing workflow where QC infrastructure should be placed
-  /// \param configurationSource - full path to configuration file, preceded with the backend (f.e. "json://")
+  /// \param configurationSource - full path to configuration file, preceded with the backend (e.g. "json://")
   /// \param host - name of the machine
   /// \return generated local QC workflow
   static void generateLocalInfrastructure(framework::WorkflowSpec& workflow, std::string configurationSource, std::string host);
@@ -91,7 +91,7 @@ class InfrastructureGenerator
   /// 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://")
+  /// \param configurationSource - full path to configuration file, preceded with the backend (e.g. "json://")
   /// \return generated remote QC workflow
   static o2::framework::WorkflowSpec generateRemoteInfrastructure(std::string configurationSource);
 
@@ -101,10 +101,45 @@ class InfrastructureGenerator
   /// 'remote' tasks.
   ///
   /// \param workflow - existing workflow where QC infrastructure should be placed
-  /// \param configurationSource - full path to configuration file, preceded with the backend (f.e. "json://")
-  /// \return generated remote QC workflow
+  /// \param configurationSource - full path to configuration file, preceded with the backend (e.g. "json://")
   static void generateRemoteInfrastructure(framework::WorkflowSpec& workflow, std::string configurationSource);
 
+  /// \brief Generates the local batch part of the QC infrastructure.
+  ///
+  /// Generates the local batch part of the QC infrastructure - tasks and a file sink/merger.
+  ///
+  /// \param workflow - existing workflow where QC infrastructure should be placed
+  /// \param configurationSource - full path to configuration file, preceded with the backend (e.g. "json://")
+  /// \param sinkFilePath - path to the output file
+  static void generateLocalBatchInfrastructure(framework::WorkflowSpec& workflow, std::string configurationSource, std::string sinkFilePath);
+
+  /// \brief Generates the local batch part of the QC infrastructure.
+  ///
+  /// Generates the local batch part of the QC infrastructure - tasks and a file sink/merger.
+  ///
+  /// \param configurationSource - full path to configuration file, preceded with the backend (f.e. "json://")
+  /// \param sinkFilePath - path to the output file
+  /// \return generated local QC workflow
+  static framework::WorkflowSpec generateLocalBatchInfrastructure(std::string configurationSource, std::string sinkFilePath);
+
+  /// \brief Generates the remote batch part of the QC infrastructure.
+  ///
+  /// Generates the remote batch part of the QC infrastructure - file reader, check runners, aggregator runners.
+  ///
+  /// \param workflow - existing workflow where QC infrastructure should be placed
+  /// \param configurationSource - full path to configuration file, preceded with the backend (f.e. "json://")
+  /// \param sourceFilePath - path to the input file
+  static void generateRemoteBatchInfrastructure(framework::WorkflowSpec& workflow, std::string configurationSource, std::string sourceFilePath);
+
+  /// \brief Generates the remote batch part of the QC infrastructure.
+  ///
+  /// Generates the remote batch part of the QC infrastructure - file reader, check runners, aggregator runners.
+  ///
+  /// \param configurationSource - full path to configuration file, preceded with the backend (f.e. "json://")
+  /// \param sourceFilePath - path to the input file
+  /// \return generated remote batch QC workflow
+  static framework::WorkflowSpec generateRemoteBatchInfrastructure(std::string configurationSource, std::string sourceFilePath);
+
   /// \brief Provides necessary customization of the QC infrastructure.
   ///
   /// Provides necessary customization of the Completion Policies of the QC infrastructure. This is necessary to make
@@ -187,6 +222,26 @@ inline framework::WorkflowSpec generateRemoteInfrastructure(std::string configur
   return core::InfrastructureGenerator::generateRemoteInfrastructure(configurationSource);
 }
 
+inline framework::WorkflowSpec generateLocalBatchInfrastructure(std::string configurationSource, std::string sinkFilePath)
+{
+  return core::InfrastructureGenerator::generateLocalBatchInfrastructure(configurationSource, std::move(sinkFilePath));
+}
+
+inline void generateLocalBatchInfrastructure(framework::WorkflowSpec& workflow, std::string configurationSource, std::string sinkFilePath)
+{
+  core::InfrastructureGenerator::generateLocalBatchInfrastructure(workflow, configurationSource, std::move(sinkFilePath));
+}
+
+inline framework::WorkflowSpec generateRemoteBatchInfrastructure(std::string configurationSource, std::string sourceFilePath)
+{
+  return core::InfrastructureGenerator::generateRemoteBatchInfrastructure(configurationSource, std::move(sourceFilePath));
+}
+
+inline void generateRemoteBatchInfrastructure(framework::WorkflowSpec& workflow, std::string configurationSource, std::string sourceFilePath)
+{
+  core::InfrastructureGenerator::generateRemoteBatchInfrastructure(workflow, configurationSource, std::move(sourceFilePath));
+}
+
 inline void generateRemoteInfrastructure(framework::WorkflowSpec& workflow, std::string configurationSource)
 {
   core::InfrastructureGenerator::generateRemoteInfrastructure(workflow, configurationSource);

Framework/include/QualityControl/RootFileSink.h

@@ -0,0 +1,56 @@
+// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+///
+/// \file   RootFileSink.h
+/// \author Piotr Konopka
+///
+
+#ifndef QUALITYCONTROL_ROOTFILESINK_H
+#define QUALITYCONTROL_ROOTFILESINK_H
+
+#include <Framework/Task.h>
+#include <Framework/CompletionPolicy.h>
+#include <Framework/DataProcessorLabel.h>
+
+class TFile;
+
+namespace o2::quality_control::core
+{
+
+/// \brief A Data Processor which stores MonitorObjectCollections in a specified file
+class RootFileSink : public framework::Task
+{
+ public:
+  explicit RootFileSink(std::string filePath);
+  ~RootFileSink() override;
+
+  void init(framework::InitContext& ictx) override;
+  void run(framework::ProcessingContext& pctx) override;
+
+  static framework::DataProcessorLabel getLabel()
+  {
+    return { "QC-ROOT-FILE-SINK" };
+  }
+
+  static void customizeInfrastructure(std::vector<framework::CompletionPolicy>& policies);
+
+ private:
+  void reset();
+
+ private:
+  std::string mFilePath;
+  TFile* mFile = nullptr;
+};
+
+} // namespace o2::quality_control::core
+
+#endif //QUALITYCONTROL_ROOTFILESINK_H

Framework/include/QualityControl/RootFileSource.h

@@ -0,0 +1,42 @@
+// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+///
+/// \file   RootFileSource.h
+/// \author Piotr Konopka
+///
+
+#ifndef QUALITYCONTROL_ROOTFILESOURCE_H
+#define QUALITYCONTROL_ROOTFILESOURCE_H
+
+#include <Framework/Task.h>
+#include <string>
+
+namespace o2::quality_control::core
+{
+
+/// \brief A Data Processor which reads MonitorObjectCollections from a specified file
+class RootFileSource : public framework::Task
+{
+ public:
+  RootFileSource(std::string filePath);
+  ~RootFileSource() override = default;
+
+  void init(framework::InitContext& ictx) override;
+  void run(framework::ProcessingContext& pctx) override;
+
+ private:
+  std::string mFilePath;
+};
+
+} // namespace o2::quality_control::core
+
+#endif //QUALITYCONTROL_ROOTFILESOURCE_H