[QC-1010] Make activity available in aggregator (#1953)

* [QC-1010] Aggregator -> add activity

* doc

* Update Advanced.md

* actually use the setActivity and switch to a shared pointer (as it is done in the Checks)

* format

* make mActivity private

Author: Barthelemy

Framework/include/QualityControl/Aggregator.h

@@ -75,6 +75,7 @@ class Aggregator
   std::vector<AggregatorSource> getSources() const;
   std::vector<AggregatorSource> getSources(core::DataSourceType type);
   const std::string& getDetector() const { return mAggregatorConfig.detectorName; };
+  void setActivity(std::shared_ptr<core::Activity> activity);
 
   static AggregatorConfig extractConfig(const core::CommonSpec&, const AggregatorSpec&);
 

Framework/include/QualityControl/AggregatorInterface.h

@@ -44,8 +44,13 @@ class AggregatorInterface : public o2::quality_control::core::UserCodeInterface
   /// @return The new qualities, associated with a name.
   virtual std::map<std::string, o2::quality_control::core::Quality> aggregate(std::map<std::string, std::shared_ptr<const o2::quality_control::core::QualityObject>>& qoMap) = 0;
 
- protected:
-  ClassDef(AggregatorInterface, 2)
+  void setActivity(std::shared_ptr<core::Activity> activity) { mActivity = activity; }
+  std::shared_ptr<const core::Activity> getActivity() const { return mActivity; }
+
+ private:
+  std::shared_ptr<core::Activity> mActivity; // TODO should probably go to UserCodeInterface
+
+  ClassDef(AggregatorInterface, 3)
 };
 
 } // namespace o2::quality_control::checker

Framework/include/QualityControl/AggregatorRunner.h

@@ -29,6 +29,7 @@
 #include "QualityControl/Activity.h"
 #include "QualityControl/AggregatorRunnerConfig.h"
 #include "QualityControl/AggregatorConfig.h"
+#include "QualityControl/Activity.h"
 
 namespace o2::framework
 {
@@ -175,7 +176,7 @@ class AggregatorRunner : public framework::Task
 
   // General state
   std::string mDeviceName;
-  core::Activity mActivity;
+  std::shared_ptr<core::Activity> mActivity; // shareable with the Aggregators
   std::vector<std::shared_ptr<Aggregator>> mAggregators;
   std::shared_ptr<o2::quality_control::repository::DatabaseInterface> mDatabase;
   AggregatorRunnerConfig mRunnerConfig;

Framework/src/Aggregator.cxx

@@ -224,4 +224,13 @@ AggregatorConfig Aggregator::extractConfig(const core::CommonSpec& commonSpec, c
   };
 }
 
+void Aggregator::setActivity(std::shared_ptr<core::Activity> activity)
+{
+  if (mAggregatorInterface) {
+    mAggregatorInterface->setActivity(std::move(activity));
+  } else {
+    throw std::runtime_error("Trying to set Activity on an empty AggregatorInterface '" + mAggregatorConfig.name + "'");
+  }
+}
+
 } // namespace o2::quality_control::checker

Framework/src/AggregatorRunner.cxx

@@ -214,7 +214,7 @@ QualityObjectsType AggregatorRunner::aggregate()
 
     if (updatePolicyManager.isReady(aggregatorName)) {
       ILOG(Info, Devel) << "   Quality Objects for the aggregator '" << aggregatorName << "' are  ready, aggregating" << ENDM;
-      auto newQOs = aggregator->aggregate(mQualityObjects, mActivity); // we give the whole list
+      auto newQOs = aggregator->aggregate(mQualityObjects, *mActivity); // we give the whole list
       mTotalNumberObjectsProduced += newQOs.size();
       mTotalNumberAggregatorExecuted++;
       // we consider the output of the aggregators the same way we do the output of a check
@@ -421,17 +421,20 @@ void AggregatorRunner::sendPeriodicMonitoring()
 
 void AggregatorRunner::start(ServiceRegistryRef services)
 {
-  mActivity = computeActivity(services, mRunnerConfig.fallbackActivity);
+  mActivity = std::make_shared<Activity>(computeActivity(services, mRunnerConfig.fallbackActivity));
   mTimerTotalDurationActivity.reset();
-  QcInfoLogger::setRun(mActivity.mId);
-  QcInfoLogger::setPartition(mActivity.mPartitionName);
-  ILOG(Info, Support) << "Starting run " << mActivity.mId << ENDM;
+  QcInfoLogger::setRun(mActivity->mId);
+  QcInfoLogger::setPartition(mActivity->mPartitionName);
+  ILOG(Info, Support) << "Starting run " << mActivity->mId << ENDM;
+  for (auto& aggregator : mAggregators) {
+    aggregator->setActivity(mActivity);
+  }
 
   // register ourselves to the BK
   if (gSystem->Getenv("O2_QC_REGISTER_IN_BK")) { // until we are sure it works, we have to turn it on
     ILOG(Debug, Devel) << "Registering aggregator to BookKeeping" << ENDM;
     try {
-      Bookkeeping::getInstance().registerProcess(mActivity.mId, mDeviceName, AggregatorRunner::getDetectorName(mAggregators), bookkeeping::DPL_PROCESS_TYPE_QC_AGGREGATOR, "");
+      Bookkeeping::getInstance().registerProcess(mActivity->mId, mDeviceName, AggregatorRunner::getDetectorName(mAggregators), bookkeeping::DPL_PROCESS_TYPE_QC_AGGREGATOR, "");
     } catch (std::runtime_error& error) {
       ILOG(Warning, Devel) << "Failed registration to the BookKeeping: " << error.what() << ENDM;
     }
@@ -440,7 +443,7 @@ void AggregatorRunner::start(ServiceRegistryRef services)
 
 void AggregatorRunner::stop()
 {
-  ILOG(Info, Support) << "Stopping run " << mActivity.mId << ENDM;
+  ILOG(Info, Support) << "Stopping run " << mActivity->mId << ENDM;
 }
 
 void AggregatorRunner::reset()
@@ -449,7 +452,7 @@ void AggregatorRunner::reset()
 
   try {
     mCollector.reset();
-    mActivity = Activity();
+    mActivity = make_shared<Activity>();
   } catch (...) {
     // we catch here because we don't know where it will go in DPL's CallbackService
     ILOG(Error, Support) << "Error caught in reset() : "

doc/Advanced.md

@@ -1129,6 +1129,10 @@ In a task, the `activity` is provided in `startOfActivity`.
 
 In a Check, it is returned by `getActivity()`.
 
+In an Aggregator, it is returned by `getActivity()`.
+
+In a postprocessing task, it is available in the objects manager: `getObjectsManager()->getActivity()`
+
 ## Definition of new arguments
 
 One can also tell the DPL driver to accept new arguments. This is done using the `customize` method at the top of your workflow definition (usually called "runXXX" in the QC).