[QC-489] Make sure aggregators are run in the right order (#560)

Author: Barthelemy

Framework/CMakeLists.txt

@@ -304,7 +304,7 @@ endforeach()
 
 foreach(t testTaskInterface testWorkflow testTaskRunner testCheckWorkflow
         testInfrastructureGenerator testPostProcessingConfig testPostProcessingInterface
-        testPostProcessingRunner testCheck testCheckRunner testTrendingTask)
+        testPostProcessingRunner testCheck testCheckRunner testTrendingTask testAggregatorRunner)
   target_sources(${t} PRIVATE
                  ${CMAKE_BINARY_DIR}/getTestDataDirectory.cxx)
   target_include_directories(${t} PRIVATE ${CMAKE_SOURCE_DIR})

Framework/include/QualityControl/Aggregator.h

@@ -33,6 +33,13 @@ namespace o2::quality_control::checker
 {
 
 class AggregatorInterface;
+enum AggregatorSourceType { check,
+                            aggregator };
+struct AggregatorSource {
+  AggregatorSource(const std::string& t, const std::string& n);
+  AggregatorSourceType type;
+  std::string name;
+};
 
 /// \brief An aggregator as found in the configuration.
 ///
@@ -64,10 +71,13 @@ class Aggregator
   std::string getPolicyName() const;
   std::vector<std::string> getObjectsNames() const;
   bool getAllObjectsOption() const;
+  std::vector<AggregatorSource> getSources();
+  std::vector<AggregatorSource> getSources(AggregatorSourceType type);
 
  private:
   CheckConfig mAggregatorConfig; // we reuse checkConfig, just consider that Check = Aggregator
   AggregatorInterface* mAggregatorInterface = nullptr;
+  std::vector<AggregatorSource> mSources;
 };
 
 } // namespace o2::quality_control::checker

Framework/include/QualityControl/AggregatorRunner.h

@@ -66,6 +66,7 @@ class TClass;
 
 namespace o2::quality_control::checker
 {
+struct AggregatorSource;
 
 /// \brief The class in charge of running the aggregators on the QualityObjects.
 ///
@@ -99,8 +100,9 @@ class AggregatorRunner : public framework::Task
   /// \brief AggregatorRunner process callback
   void run(framework::ProcessingContext& ctx) override;
 
-  framework::Inputs getInputs() { return mInputs; };
-  std::string getDeviceName() { return mDeviceName; };
+  framework::Inputs getInputs() { return mInputs; }
+  std::string getDeviceName() { return mDeviceName; }
+  const std::vector<std::shared_ptr<Aggregator>>& getAggregators() const { return mAggregators; }
 
   static std::string createAggregatorRunnerIdString() { return "QC-AGGREGATOR-RUNNER"; };
   static std::string createAggregatorRunnerName();
@@ -128,14 +130,29 @@ class AggregatorRunner : public framework::Task
   inline void initServiceDiscovery();
   inline void initAggregators();
 
+  /**
+   * Reorder the aggregators stored in mAggregators.
+   */
+  void reorderAggregators();
+
+  /**
+   * Checks whether all sources provided are already in the aggregators vector.
+   * The match is done by name.
+   * @param sources
+   * @param aggregators
+   * @return true if all sources are found, by name, in the vector of aggregators.
+   */
+  bool areSourcesIn(const std::vector<AggregatorSource>& sources,
+                    const std::vector<std::shared_ptr<Aggregator>>& aggregators);
+
   /**
    * Send metrics to the monitoring system if the time has come.
    */
   void sendPeriodicMonitoring();
 
   // General state
   std::string mDeviceName;
-  std::map<std::string, std::shared_ptr<Aggregator>> mAggregatorsMap;
+  std::vector<std::shared_ptr<Aggregator>> mAggregators;
   std::shared_ptr<o2::quality_control::repository::DatabaseInterface> mDatabase;
   std::shared_ptr<o2::configuration::ConfigurationInterface> mConfigFile;
   core::QualityObjectsMapType mQualityObjects; // where we cache the incoming quality objects and the output of the aggregators

Framework/src/Aggregator.cxx

@@ -17,11 +17,15 @@
 #include "QualityControl/Aggregator.h"
 #include "QualityControl/RootClassFactory.h"
 #include "QualityControl/AggregatorInterface.h"
+#include <Common/Exceptions.h>
 
 using namespace o2::quality_control::checker;
 using namespace o2::quality_control::core;
 using namespace std;
 
+namespace o2::quality_control::checker
+{
+
 Aggregator::Aggregator(const std::string& aggregatorName, const boost::property_tree::ptree& configuration)
 {
   mAggregatorConfig.name = aggregatorName;
@@ -44,10 +48,7 @@ Aggregator::Aggregator(const std::string& aggregatorName, const boost::property_
     if (auto sourceType = dataSource.get<std::string>("type"); sourceType == "Aggregator" || sourceType == "Check") {
       auto sourceName = dataSource.get<std::string>("name");
       ILOG(Info, Devel) << "   Found a source : " << sourceName << ENDM;
-
-      cout << "dataSource.count(QOs) : " << dataSource.count("QOs") << endl;
-      if (dataSource.count("QOs") != 0)
-        cout << "dataSource.get<std::string>(\"QOs\") : " << dataSource.get<std::string>("QOs") << endl;
+      mSources.emplace_back(sourceType, sourceName);
 
       if (dataSource.count("QOs") == 0) {
         ILOG(Info, Devel) << "      (no QOs specified, we take all)" << ENDM;
@@ -123,4 +124,32 @@ std::vector<std::string> Aggregator::getObjectsNames() const
 bool Aggregator::getAllObjectsOption() const
 {
   return mAggregatorConfig.allObjects;
+}
+
+std::vector<AggregatorSource> Aggregator::getSources()
+{
+  return mSources;
+}
+
+std::vector<AggregatorSource> Aggregator::getSources(AggregatorSourceType type)
+{
+  std::vector<AggregatorSource> matches;
+  std::copy_if(mSources.begin(), mSources.end(), std::back_inserter(matches), [&](const AggregatorSource& source) {
+    return source.type == type;
+  });
+  return matches;
+}
+
+AggregatorSource::AggregatorSource(const std::string& t, const std::string& n)
+{
+  if (t == "Aggregator") {
+    type = aggregator;
+  } else if (t == "Check") {
+    type = check;
+  } else {
+    BOOST_THROW_EXCEPTION(AliceO2::Common::Exception() << AliceO2::Common::errinfo_details("Unknown type of Aggregator: " + t));
+  }
+  name = n;
+}
+
 }

Framework/src/AggregatorRunner.cxx

@@ -127,7 +127,8 @@ QualityObjectsType AggregatorRunner::aggregate()
   ILOG(Debug, Trace) << "Aggregate called in AggregatorRunner, QOs in cache: " << mQualityObjects.size() << ENDM;
 
   QualityObjectsType allQOs;
-  for (auto const& [aggregatorName, aggregator] : mAggregatorsMap) {
+  for (auto const& aggregator : mAggregators) {
+    string aggregatorName = aggregator->getName();
     ILOG(Info, Devel) << "Processing aggregator: " << aggregatorName << ENDM;
 
     if (updatePolicyManager.isReady(aggregatorName)) {
@@ -200,17 +201,93 @@ void AggregatorRunner::initAggregators()
 
     // For every aggregator definition, create an Aggregator
     for (const auto& [aggregatorName, aggregatorConfig] : mConfigFile->getRecursive("qc.aggregators")) {
-
       ILOG(Info, Devel) << ">> Aggregator name : " << aggregatorName << ENDM;
+
       if (aggregatorConfig.get<bool>("active", true)) {
-        // create Aggregator and store it.
-        auto aggregator = make_shared<Aggregator>(aggregatorName, aggregatorConfig);
-        aggregator->init();
-        updatePolicyManager.addPolicy(aggregator->getName(), aggregator->getPolicyName(), aggregator->getObjectsNames(), aggregator->getAllObjectsOption(), false);
-        mAggregatorsMap[aggregatorName] = aggregator;
+        try {
+          auto aggregator = make_shared<Aggregator>(aggregatorName, aggregatorConfig);
+          aggregator->init();
+          updatePolicyManager.addPolicy(aggregator->getName(),
+                                        aggregator->getPolicyName(),
+                                        aggregator->getObjectsNames(),
+                                        aggregator->getAllObjectsOption(),
+                                        false);
+          mAggregators.push_back(aggregator);
+        } catch (...) {
+          // catch the configuration exception and print it to avoid losing it
+          ILOG(Error, Ops) << "Error creating aggregator '" << aggregatorName << "'"
+                           << current_diagnostic(true) << ENDM;
+          continue; // skip this aggregator, it might still fail fatally later if another aggregator depended on it
+        }
       }
     }
   }
+
+  reorderAggregators();
+}
+
+bool AggregatorRunner::areSourcesIn(const std::vector<AggregatorSource>& sources,
+                                    const std::vector<std::shared_ptr<Aggregator>>& aggregators)
+{
+  for (auto source : sources) {
+    auto it = find_if(aggregators.begin(), aggregators.end(),
+                      [&](const std::shared_ptr<Aggregator>& agg) { return (agg->getName() == source.name); });
+    if (it == aggregators.end()) {
+      return false;
+    }
+  }
+
+  return true;
+}
+
+void AggregatorRunner::reorderAggregators()
+{
+  // Implementation
+  // This is a simple, light-weight, but sub-optimal implementation.
+  // One could build a proper tree (e.g. with boost Graph) and then apply a complex algorithm to order
+  // the nodes and find cycles.
+  // Instead this implementation goes through the aggregators and for each checks whether
+  // there are no dependencies or if they are all fulfilled. If it is the case the aggregator
+  // is moved at the end of the resulting vector.
+  // In case we have looped through all the remaining aggregators and nothing has been done,
+  // ie. no aggregator has its dependencies fulfilled, we stop and raise an error.
+  // This means that there is a cycle or that one of the dependencies does not exist.
+  // Note that by "fulfilled" we mean that all the sources of an aggregator are already
+  // in the result vector.
+
+  std::vector<std::shared_ptr<Aggregator>> originals = mAggregators;
+  std::vector<std::shared_ptr<Aggregator>> results;
+  bool modificationLastIteration = true;
+  // As long as there are items in original and we did some modifications in the last iteration
+  while (!originals.empty() && modificationLastIteration) {
+    modificationLastIteration = false;
+    std::vector<std::shared_ptr<Aggregator>> toBeMoved; // we need it because we cannot modify the vectors while iterating over them
+    // Loop over remaining items in the original list
+    for (const auto& orig : originals) {
+      // if no Aggregator dependencies or Aggregator sources are all already in result
+      auto sources = orig->getSources(aggregator);
+      if (sources.empty() || areSourcesIn(sources, results)) {
+        // move from original to result
+        toBeMoved.push_back(orig);
+        modificationLastIteration = true;
+      }
+    }
+    // move the items from one vector to the other
+    for (const auto& item : toBeMoved) {
+      results.push_back(item);
+      originals.erase(std::remove(originals.begin(), originals.end(), item), originals.end());
+    }
+  }
+
+  if (!originals.empty()) {
+    string msg =
+      "Error in the aggregators definition : either there is a cycle "
+      "or an aggregator depends on an aggregator that does not exist.";
+    ILOG(Error, Ops) << msg << ENDM;
+    BOOST_THROW_EXCEPTION(FatalException() << errinfo_details(msg));
+  }
+  assert(results.size() != mAggregators.size());
+  mAggregators = results;
 }
 
 void AggregatorRunner::sendPeriodicMonitoring()

Framework/test/testAggregatorRunner.cxx

@@ -13,8 +13,13 @@
 /// \author  Barthelemy von Haller
 ///
 
+#include "getTestDataDirectory.h"
 #include "QualityControl/AggregatorRunnerFactory.h"
 #include "QualityControl/AggregatorRunner.h"
+#include "QualityControl/Aggregator.h"
+#include <Framework/InitContext.h>
+#include <Framework/ConfigParamRegistry.h>
+#include <Framework/ConfigParamStore.h>
 
 #define BOOST_TEST_MODULE CheckRunner test
 #define BOOST_TEST_MAIN
@@ -27,9 +32,30 @@ using namespace std;
 using namespace o2::framework;
 using namespace o2::header;
 
-BOOST_AUTO_TEST_CASE(test_check_runner_static)
+BOOST_AUTO_TEST_CASE(test_aggregator_runner_static)
 {
   BOOST_CHECK(AggregatorRunner::createAggregatorRunnerDataDescription("qwertyuiop") == DataDescription("qwertyuiop"));
   BOOST_CHECK(AggregatorRunner::createAggregatorRunnerDataDescription("012345678901234567890") == DataDescription("0123456789012345"));
   BOOST_CHECK_THROW(AggregatorRunner::createAggregatorRunnerDataDescription(""), AliceO2::Common::FatalException);
 }
+
+BOOST_AUTO_TEST_CASE(test_aggregator_runner)
+{
+  std::string configFilePath = std::string("json://") + getTestDataDirectory() + "testSharedConfig.json";
+  AggregatorRunner aggregatorRunner{ configFilePath, { OutputSpec{ { "mo" }, "QC", "abcTask-mo", 123 } } };
+
+  std::unique_ptr<ConfigParamStore> store;
+  ConfigParamRegistry cfReg(std::move(store));
+  ServiceRegistry sReg;
+  InitContext initContext{ cfReg, sReg };
+  aggregatorRunner.init(initContext);
+
+  BOOST_CHECK_EQUAL(aggregatorRunner.getDeviceName(), "QC-AGGREGATOR-RUNNER");
+
+  // check the reordering
+  const std::vector<std::shared_ptr<Aggregator>> aggregators = aggregatorRunner.getAggregators();
+  BOOST_CHECK(aggregators.at(0)->getName() == "MyAggregatorC" || aggregators.at(0)->getName() == "MyAggregatorB");
+  BOOST_CHECK(aggregators.at(1)->getName() == "MyAggregatorC" || aggregators.at(1)->getName() == "MyAggregatorB");
+  BOOST_CHECK(aggregators.at(2)->getName() == "MyAggregatorA");
+  BOOST_CHECK(aggregators.at(3)->getName() == "MyAggregatorD");
+}

Framework/test/testSharedConfig.json

@@ -181,6 +181,69 @@
           "once"
         ]
       }
+    },
+    "aggregators": {
+      "MyAggregatorA": {
+        "active": "true",
+        "className": "o2::quality_control_modules::skeleton::SkeletonAggregator",
+        "moduleName": "QcSkeleton",
+        "policy": "OnAll",
+        "detectorName": "TST",
+        "dataSource": [{
+          "type": "Aggregator",
+          "name": "MyAggregatorB", "": "no QOs parameter -> all QOs of this aggregator",
+          "QOs": ["newQuality"], "": "list all objects we are interested in"
+        }, {
+          "type": "Aggregator",
+          "name": "MyAggregatorC",
+          "QOs": ["newQuality", "another"], "": "if we omitted the QOs for a data source we would default to OnAny"
+        }]
+      },
+      "MyAggregatorD": {
+        "active": "true",
+        "className": "o2::quality_control_modules::skeleton::SkeletonAggregator",
+        "moduleName": "QcSkeleton",
+        "policy": "OnAll",
+        "detectorName": "TST",
+        "dataSource": [{
+          "type": "Aggregator",
+          "name": "MyAggregatorA", "": "no QOs parameter -> all QOs of this aggregator",
+          "QOs": ["newQuality"], "": "list all objects we are interested in"
+        }, {
+          "type": "Aggregator",
+          "name": "MyAggregatorC",
+          "QOs": ["newQuality", "another"], "": "if we omitted the QOs for a data source we would default to OnAny"
+        }]
+      },
+      "MyAggregatorC": {
+        "active": "true",
+        "className": "o2::quality_control_modules::skeleton::SkeletonAggregator",
+        "moduleName": "QcSkeleton",
+        "policy": "OnAll",
+        "detectorName": "TST",
+        "dataSource": [{
+          "type": "Check",
+          "name": "dataSizeCheck"
+        }, {
+          "type": "Check",
+          "name": "someNumbersCheck"
+        }]
+      },
+      "MyAggregatorB": {
+        "active": "true",
+        "className": "o2::quality_control_modules::skeleton::SkeletonAggregator",
+        "moduleName": "QcSkeleton",
+        "policy": "OnAll",
+        "detectorName": "TST",
+        "dataSource": [{
+          "type": "Check",
+          "name": "dataSizeCheck"
+        }, {
+          "type": "Check", "": "this one is using onEachSeparately and thus it sends under a full path",
+          "name": "someNumbersCheck",
+          "QOs": ["someNumbersTask/example"], "": "also possible to ignore it altogether, meaning we take all objects"
+        }]
+      }
     }
   },
   "dataSamplingPolicies": [