Ensure that kafka config parameters are propagated to post-processing

Feature introduced by QC-271 could not really work without allowing the user to propagate the kafka broker url and the right topic to the SOR/EOR triggers.
This commit does the necessary plumbering.

Author: knopers8

Framework/include/QualityControl/CommonSpec.h

@@ -44,6 +44,8 @@ struct CommonSpec {
   LogDiscardParameters infologgerDiscardParameters;
   double postprocessingPeriod = 30.0;
   std::string bookkeepingUrl;
+  std::string kafkaBrokersUrl;
+  std::string kafkaTopicAliECSRun = "aliecs.run";
 };
 
 } // namespace o2::quality_control::core

Framework/include/QualityControl/PostProcessingConfig.h

@@ -39,7 +39,7 @@ struct PostProcessingConfig : public o2::quality_control::core::UserCodeConfig {
   std::vector<std::string> updateTriggers = {};
   std::vector<std::string> stopTriggers = {};
   std::string kafkaBrokersUrl;
-  std::string kafkaTopic;
+  std::string kafkaTopicAliECSRun;
   core::Activity activity;
   bool matchAnyRunNumber = false;
   bool critical;

Framework/src/InfrastructureSpecReader.cxx

@@ -81,6 +81,8 @@ CommonSpec InfrastructureSpecReader::readSpecEntry<CommonSpec>(const std::string
   };
   spec.postprocessingPeriod = commonTree.get<double>("postprocessing.periodSeconds", spec.postprocessingPeriod);
   spec.bookkeepingUrl = commonTree.get<std::string>("bookkeeping.url", spec.bookkeepingUrl);
+  spec.kafkaBrokersUrl = commonTree.get<std::string>("kafka.url", spec.kafkaBrokersUrl);
+  spec.kafkaTopicAliECSRun = commonTree.get<std::string>("kafka.topicAliecsRun", spec.kafkaTopicAliECSRun);
 
   return spec;
 }

Framework/src/PostProcessingConfig.cxx

@@ -39,6 +39,9 @@ PostProcessingConfig::PostProcessingConfig(const std::string& id, const boost::p
   detectorName = config.get<std::string>("qc.postprocessing." + id + ".detectorName", "MISC");
   ccdbUrl = config.get<std::string>("qc.config.conditionDB.url", "");
   consulUrl = config.get<std::string>("qc.config.consul.url", "");
+  kafkaBrokersUrl = config.get<std::string>("qc.config.kafka.url", "");
+  kafkaTopicAliECSRun = config.get<std::string>("qc.config.kafka.topicAliecsRun", "aliecs.run");
+
   // if available, use the source repo as defined in the postprocessing task, otherwise the general QCDB
   auto sourceRepo = config.get_child_optional("qc.postprocessing." + id + ".sourceRepo");
   auto databasePath = sourceRepo ? "qc.postprocessing." + id + ".sourceRepo" : "qc.config.database";

Framework/src/TriggerHelpers.cxx

@@ -95,9 +95,9 @@ TriggerFcn triggerFactory(const std::string& trigger, const PostProcessingConfig
   } else if (triggerLowerCase == "always") {
     return triggers::Always(activity);
   } else if (triggerLowerCase == "sor" || triggerLowerCase == "startofrun") {
-    return triggers::StartOfRun(config.kafkaBrokersUrl, config.kafkaTopic, config.detectorName, config.taskName, activity);
+    return triggers::StartOfRun(config.kafkaBrokersUrl, config.kafkaTopicAliECSRun, config.detectorName, config.taskName, activity);
   } else if (triggerLowerCase == "eor" || triggerLowerCase == "endofrun") {
-    return triggers::EndOfRun(config.kafkaBrokersUrl, config.kafkaTopic, config.detectorName, config.taskName, activity);
+    return triggers::EndOfRun(config.kafkaBrokersUrl, config.kafkaTopicAliECSRun, config.detectorName, config.taskName, activity);
   } else if (triggerLowerCase == "sof" || triggerLowerCase == "startoffill") {
     return triggers::StartOfFill(activity);
   } else if (triggerLowerCase == "eof" || triggerLowerCase == "endoffill") {

doc/Advanced.md

@@ -1670,6 +1670,10 @@ should not be present in real configuration files.
       "bookkeeping": {                    "": "Configuration of the bookkeeping (optional)",
         "url": "localhost:4001",          "": "Url of the bookkeeping API (port is usually different from web interface)"
       },
+      "kafka": {
+        "url": "kafka-broker:123",        "": "url of the kafka broker",
+        "topicAliecsRun":"aliecs.run",    "": "the topic where AliECS publishes Run Events, 'aliecs.run' by default"
+      },
       "postprocessing": {                 "": "Configuration parameters for post-processing",
         "periodSeconds": 10.0,            "": "Sets the interval of checking all the triggers. One can put a very small value",
                                           "": "for async processing, but use 10 or more seconds for synchronous operations",