[QC-869] Moving windows of selected plots in sync processing (#1895)

Also brings "--full-chain" flag to o2-qc, which allows to run a QC chain with Mergers between a single instance of a QC task and a CheckRunner. Useful for testing that an object is mergeable and for testing moving windows generated by Mergers.

Author: knopers8

Framework/CMakeLists.txt

@@ -94,9 +94,9 @@ add_library(O2QualityControl
   src/Timekeeper.cxx
   src/TimekeeperSynchronous.cxx
   src/TimekeeperAsynchronous.cxx
+  src/WorkflowType.cxx
   src/TimekeeperFactory.cxx)
 
-
 target_include_directories(
   O2QualityControl
   PUBLIC $<INSTALL_INTERFACE:include>
@@ -241,6 +241,7 @@ add_executable(o2-qc-test-core
                test/testTimekeeper.cxx
                test/testTriggerHelpers.cxx
                test/testVersion.cxx
+               test/testMonitorObjectCollection.cxx
                )
 set_property(TARGET o2-qc-test-core
              PROPERTY RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/tests)

Framework/include/QualityControl/DataSourceSpec.h

@@ -28,6 +28,7 @@ enum class DataSourceType {
   DataSamplingPolicy,
   Direct,
   Task,
+  TaskMovingWindow,
   Check,
   Aggregator,
   PostProcessingTask,

Framework/include/QualityControl/InfrastructureGenerator.h

@@ -68,6 +68,26 @@ class InfrastructureGenerator
   /// \param configurationTree - full QC config ptree
   static void generateStandaloneInfrastructure(framework::WorkflowSpec& workflow, const boost::property_tree::ptree& configurationTree);
 
+  /// \brief Generates a full QC chain infrastructure.
+  ///
+  /// 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,
+  /// as well as Mergers between local QC tasks and Checks.
+  ///
+  /// \param configurationTree - full path to configuration file, preceded with the backend (e.g. "json://")
+  /// \return generated standalone QC workflow
+  static framework::WorkflowSpec generateFullChainInfrastructure(const boost::property_tree::ptree& configurationTree);
+
+  /// \brief Generates a full QC chain infrastructure.
+  ///
+  /// 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,
+  /// as well as Mergers between local QC tasks and Checks.
+  ///
+  /// \param workflow - existing workflow where QC infrastructure should be placed
+  /// \param configurationTree - full QC config ptree
+  static void generateFullChainInfrastructure(framework::WorkflowSpec& workflow, const boost::property_tree::ptree& configurationTree);
+
   /// \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
@@ -190,15 +210,12 @@ class InfrastructureGenerator
   static void generateLocalTaskRemoteProxy(framework::WorkflowSpec& workflow,
                                            const TaskSpec& taskSpec,
                                            size_t numberOfLocalMachines);
-  static void generateMergers(framework::WorkflowSpec& workflow,
-                              const std::string& taskName,
+  static void generateMergers(framework::WorkflowSpec& workflow, const std::string& taskName,
                               size_t numberOfLocalMachines,
                               std::vector<std::pair<size_t, size_t>> cycleDurationSeconds,
-                              const std::string& mergingMode,
-                              size_t resetAfterCycles,
-                              std::string monitoringUrl,
-                              const std::string& detectorName,
-                              std::vector<size_t> mergersPerLayer);
+                              const std::string& mergingMode, size_t resetAfterCycles,
+                              std::string monitoringUrl, const std::string& detectorName,
+                              std::vector<size_t> mergersPerLayer, bool enableMovingWindows);
   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);
@@ -218,6 +235,16 @@ inline void generateStandaloneInfrastructure(framework::WorkflowSpec& workflow,
   core::InfrastructureGenerator::generateStandaloneInfrastructure(workflow, configurationTree);
 }
 
+inline framework::WorkflowSpec generateFullChainInfrastructure(const boost::property_tree::ptree& configurationTree)
+{
+  return core::InfrastructureGenerator::generateFullChainInfrastructure(configurationTree);
+}
+
+inline void generateFullChainInfrastructure(framework::WorkflowSpec& workflow, const boost::property_tree::ptree& configurationTree)
+{
+  core::InfrastructureGenerator::generateFullChainInfrastructure(workflow, configurationTree);
+}
+
 inline framework::WorkflowSpec generateLocalInfrastructure(const boost::property_tree::ptree& configurationTree, std::string host)
 {
   return core::InfrastructureGenerator::generateLocalInfrastructure(configurationTree, host);

Framework/include/QualityControl/InfrastructureSpec.h

@@ -16,6 +16,7 @@
 /// \author Piotr Konopka
 ///
 
+#include "QualityControl/WorkflowType.h"
 #include "QualityControl/CommonSpec.h"
 #include "QualityControl/TaskSpec.h"
 #include "QualityControl/CheckSpec.h"
@@ -29,6 +30,7 @@ namespace o2::quality_control::core
 {
 
 struct InfrastructureSpec {
+  WorkflowType workflowType = WorkflowType::Standalone;
   CommonSpec common;
   std::vector<TaskSpec> tasks;
   std::vector<checker::CheckSpec> checks;

Framework/include/QualityControl/InfrastructureSpecReader.h

@@ -35,7 +35,7 @@ namespace o2::quality_control::core
 namespace InfrastructureSpecReader
 {
 /// \brief Reads the full QC configuration structure.
-InfrastructureSpec readInfrastructureSpec(const boost::property_tree::ptree& wholeTree);
+InfrastructureSpec readInfrastructureSpec(const boost::property_tree::ptree& wholeTree, WorkflowType workflowType);
 
 template <typename T>
 T readSpecEntry(const std::string& entryID, const boost::property_tree::ptree& entryTree, const boost::property_tree::ptree& wholeTree);

Framework/include/QualityControl/MonitorObject.h

@@ -95,6 +95,9 @@ class MonitorObject : public TObject
   void updateValidity(validity_time_t value);
   ValidityInterval getValidity() const;
 
+  void setCreateMovingWindow(bool);
+  bool getCreateMovingWindow() const;
+
   /// \brief Add key value pair that will end up in the database as metadata of the object
   /// Add a metadata (key value pair) to the MonitorObject. It will be stored in the database as metadata.
   /// If the key already exists the value will NOT be updated.
@@ -135,8 +138,10 @@ class MonitorObject : public TObject
   // object.
   // TODO : maybe we should always be the owner ?
   bool mIsOwner;
+  // tells Merger to create an object with data from the last cycle only on the side of the complete object
+  bool mCreateMovingWindow = false;
 
-  ClassDefOverride(MonitorObject, 11);
+  ClassDefOverride(MonitorObject, 12);
 };
 
 } // namespace o2::quality_control::core

Framework/include/QualityControl/MonitorObjectCollection.h

@@ -37,6 +37,8 @@ class MonitorObjectCollection : public TObjArray, public mergers::MergeInterface
   void setDetector(const std::string&);
   const std::string& getDetector() const;
 
+  MergeInterface* cloneMovingWindow() const override;
+
  private:
   std::string mDetector = "TST";
 

Framework/include/QualityControl/ObjectsManager.h

@@ -188,6 +188,9 @@ class ObjectsManager
   const Activity& getActivity() const;
   void setActivity(const Activity& activity);
 
+  void setMovingWindowsList(const std::vector<std::string>&);
+  const std::vector<std::string>& getMovingWindowsList() const;
+
  private:
   std::unique_ptr<MonitorObjectCollection> mMonitorObjects;
   std::string mTaskName;
@@ -196,6 +199,7 @@ class ObjectsManager
   std::unique_ptr<ServiceDiscovery> mServiceDiscovery;
   bool mUpdateServiceDiscovery;
   Activity mActivity;
+  std::vector<std::string> mMovingWindowsList;
 };
 
 } // namespace o2::quality_control::core

Framework/include/QualityControl/PostProcessingRunner.h

@@ -27,6 +27,7 @@
 #include "QualityControl/Triggers.h"
 #include "QualityControl/Activity.h"
 #include "QualityControl/DatabaseInterface.h"
+#include "WorkflowType.h"
 
 namespace o2::framework
 {
@@ -65,7 +66,7 @@ class PostProcessingRunner
   ~PostProcessingRunner() = default;
 
   /// \brief Initialization. Creates configuration structures out of the ptree. Throws on errors.
-  void init(const boost::property_tree::ptree& config);
+  void init(const boost::property_tree::ptree& config, o2::quality_control::core::WorkflowType workflowType);
   /// \brief Initialization. Throws on errors.
   void init(const PostProcessingRunnerConfig& runnerConfig, const PostProcessingConfig& taskConfig);
   /// \brief One iteration over the event loop. Throws on errors. Returns false when it can gracefully exit.

Framework/include/QualityControl/TaskRunner.h

@@ -105,7 +105,7 @@ class TaskRunner : public framework::Task
   /// \brief ID string for all TaskRunner devices
   static std::string createTaskRunnerIdString();
   /// \brief Unified DataOrigin for Quality Control tasks
-  static header::DataOrigin createTaskDataOrigin(const std::string& detectorCode);
+  static header::DataOrigin createTaskDataOrigin(const std::string& detectorCode, bool movingWindows = false);
   /// \brief Unified DataDescription naming scheme for all tasks
   static header::DataDescription createTaskDataDescription(const std::string& taskName);
   /// \brief Unified DataDescription naming scheme for all timers