[QC-589] Reduce the amount of configuration needed for mutlinode setups (#715)

knopers8 · Barthelemy · web-flow · commit 165a40ce5f39 · 2021-06-09T15:21:39.000+02:00
* Do not require remote machine and port when running with AliECS

* One less indentation in infrastucture generator

* Allow for optional return type for DataSampling::PortForPolicy

* Expand and update the multinode setups documentation

* trigger rebuild

* Update InfrastructureGenerator.cxx

* Wrong if condition

Co-authored-by: Barthélémy von Haller &lt;barthelemy.von.haller@gmail.com&gt;
Co-authored-by: Barthélémy von Haller &lt;barthelemy.von.haller@cern.ch&gt;
diff --git a/Framework/src/InfrastructureGenerator.cxx b/Framework/src/InfrastructureGenerator.cxx
@@ -48,6 +48,9 @@ using SubSpec = o2::header::DataHeader::SubSpecificationType;
 namespace o2::quality_control::core
 {
 
+const char* defaultRemotePort = "36543";
+uint16_t defaultPolicyPort = 42349;
+
 framework::WorkflowSpec InfrastructureGenerator::generateStandaloneInfrastructure(std::string configurationSource)
 {
   WorkflowSpec workflow;
@@ -88,46 +91,63 @@ WorkflowSpec InfrastructureGenerator::generateLocalInfrastructure(std::string co
   }
 
   for (const auto& [taskName, taskConfig] : config->getRecursive("qc.tasks")) {
-    if (taskConfig.get<bool>("active")) {
-      if (taskConfig.get<std::string>("location") == "local") {
+    if (!taskConfig.get<bool>("active")) {
+      ILOG(Info, Devel) << "Task " << taskName << " is disabled, ignoring." << ENDM;
+      continue;
+    }
 
-        if (taskConfig.get_child("localMachines").empty()) {
-          throw std::runtime_error("No local machines specified for task " + taskName + " in its configuration");
-        }
+    if (taskConfig.get<std::string>("location") == "local") {
+      if (taskConfig.get_child("localMachines").empty()) {
+        throw std::runtime_error("No local machines specified for task " + taskName + " in its configuration");
+      }
 
-        size_t id = 1;
-        for (const auto& machine : taskConfig.get_child("localMachines")) {
-          // We spawn a task and proxy only if we are on the right machine.
-          if (machine.second.get<std::string>("") == host) {
-            // Generate QC Task Runner
-            bool needsResetAfterCycle = taskConfig.get<std::string>("mergingMode", "delta") == "delta";
-            workflow.emplace_back(taskRunnerFactory.create(taskName, configurationSource, id, needsResetAfterCycle));
-            // Generate an output proxy
-            // These should be removed when we are able to declare dangling output in normal DPL devices
-            generateLocalTaskLocalProxy(workflow, id, taskName, taskConfig.get<std::string>("remoteMachine"), taskConfig.get<std::string>("remotePort"));
-            break;
+      size_t id = 1;
+      for (const auto& machine : taskConfig.get_child("localMachines")) {
+        // We spawn a task and proxy only if we are on the right machine.
+        if (machine.second.get<std::string>("") == host) {
+          // Generate QC Task Runner
+          bool needsResetAfterCycle = taskConfig.get<std::string>("mergingMode", "delta") == "delta";
+          workflow.emplace_back(taskRunnerFactory.create(taskName, configurationSource, id, needsResetAfterCycle));
+          // Generate an output proxy
+          // These should be removed when we are able to declare dangling output in normal DPL devices
+          auto remoteMachine = taskConfig.get_optional<std::string>("remoteMachine");
+          if (!remoteMachine.has_value()) {
+            ILOG(Warning, Devel)
+              << "No remote machine was specified for a multinode QC setup."
+                 " This is fine if running with AliECS, but it will fail in standalone mode."
+              << ENDM;
           }
-          id++;
-        }
-      } else // if (taskConfig.get<std::string>("location") == "remote")
-      {
-        // Collecting Data Sampling Policies
-        auto dataSourceTree = taskConfig.get_child("dataSource");
-        std::string type = dataSourceTree.get<std::string>("type");
-        if (type == "dataSamplingPolicy") {
-          samplingPoliciesUsed.insert(dataSourceTree.get<std::string>("name"));
-        } else if (type == "direct") {
-          throw std::runtime_error("Configuration error: Remote QC tasks such as " + taskName + " cannot use direct data sources");
-        } else {
-          throw std::runtime_error("Configuration error: dataSource type unknown : " + type);
+          auto remotePort = taskConfig.get_optional<std::string>("remotePort");
+          if (!remotePort.has_value()) {
+            ILOG(Warning, Devel)
+              << "No remote port was specified for a multinode QC setup."
+                 " This is fine if running with AliECS, but it might fail in standalone mode."
+              << ENDM;
+          }
+          generateLocalTaskLocalProxy(workflow, id, taskName, remoteMachine.value_or("any"), remotePort.value_or(defaultRemotePort));
+          break;
         }
+        id++;
+      }
+    } else // if (taskConfig.get<std::string>("location") == "remote")
+    {
+      // Collecting Data Sampling Policies
+      auto dataSourceTree = taskConfig.get_child("dataSource");
+      std::string type = dataSourceTree.get<std::string>("type");
+      if (type == "dataSamplingPolicy") {
+        samplingPoliciesUsed.insert(dataSourceTree.get<std::string>("name"));
+      } else if (type == "direct") {
+        throw std::runtime_error("Configuration error: Remote QC tasks such as " + taskName + " cannot use direct data sources");
+      } else {
+        throw std::runtime_error("Configuration error: dataSource type unknown : " + type);
       }
     }
   }
 
   // Creating Data Sampling Policies proxies
   for (const auto& policyName : samplingPoliciesUsed) {
-    std::string port = std::to_string(DataSampling::PortForPolicy(config.get(), policyName));
+    // todo: leave only the new way once the return type is changed
+    std::string port = std::to_string(std::optional<uint16_t>(DataSampling::PortForPolicy(config.get(), policyName)).value_or(defaultPolicyPort));
     Inputs inputSpecs = DataSampling::InputSpecsForPolicy(config.get(), policyName);
 
     std::vector<std::string> machines = DataSampling::MachinesForPolicy(config.get(), policyName);
@@ -157,43 +177,51 @@ o2::framework::WorkflowSpec InfrastructureGenerator::generateRemoteInfrastructur
   if (config->getRecursive("qc").count("tasks")) {
     TaskRunnerFactory taskRunnerFactory;
     for (const auto& [taskName, taskConfig] : config->getRecursive("qc.tasks")) {
-      if (taskConfig.get<bool>("active", true)) {
-
-        if (taskConfig.get<std::string>("location") == "local") {
-          // if tasks are LOCAL, generate input proxies + mergers + checkers
+      if (!taskConfig.get<bool>("active", true)) {
+        ILOG(Info, Devel) << "Task " << taskName << " is disabled, ignoring." << ENDM;
+        continue;
+      }
 
-          size_t numberOfLocalMachines = taskConfig.get_child("localMachines").size() > 1 ? taskConfig.get_child("localMachines").size() : 1;
-          // Generate an input proxy
-          // These should be removed when we are able to declare dangling inputs in normal DPL devices
-          generateLocalTaskRemoteProxy(workflow, taskName, numberOfLocalMachines, taskConfig.get<std::string>("remotePort"));
+      if (taskConfig.get<std::string>("location") == "local") {
+        // if tasks are LOCAL, generate input proxies + mergers + checkers
+
+        size_t numberOfLocalMachines = taskConfig.get_child("localMachines").size() > 1 ? taskConfig.get_child("localMachines").size() : 1;
+        // Generate an input proxy
+        // These should be removed when we are able to declare dangling inputs in normal DPL devices
+        auto remotePort = taskConfig.get_optional<std::string>("remotePort");
+        if (!remotePort.has_value()) {
+          ILOG(Warning, Devel) << "No remote port was specified for a multinode QC setup."
+                                  " This is fine if running with AliECS, but it might fail in standalone mode."
+                               << ENDM;
+        }
+        generateLocalTaskRemoteProxy(workflow, taskName, numberOfLocalMachines, remotePort.value_or(defaultRemotePort));
 
-          generateMergers(workflow, taskName, numberOfLocalMachines,
-                          taskConfig.get<double>("cycleDurationSeconds"),
-                          taskConfig.get<std::string>("mergingMode", "delta"));
+        generateMergers(workflow, taskName, numberOfLocalMachines,
+                        taskConfig.get<double>("cycleDurationSeconds"),
+                        taskConfig.get<std::string>("mergingMode", "delta"));
 
-        } else if (taskConfig.get<std::string>("location") == "remote") {
+      } else if (taskConfig.get<std::string>("location") == "remote") {
 
-          // -- if tasks are REMOTE, generate dispatcher proxies + tasks + checkers
-          // (for the time being we don't foresee parallel tasks on QC servers, so no mergers here)
+        // -- if tasks are REMOTE, generate dispatcher proxies + tasks + checkers
+        // (for the time being we don't foresee parallel tasks on QC servers, so no mergers here)
 
-          // fixme: ideally we should check if we are on the right remote machine, but now we support only n -> 1 setups,
-          //  so there is no point. Also, I expect that we should be able to generate one big topology or its parts
-          //  and we would place it among QC servers using AliECS, not by configuration files.
+        // fixme: ideally we should check if we are on the right remote machine, but now we support only n -> 1 setups,
+        //  so there is no point. Also, I expect that we should be able to generate one big topology or its parts
+        //  and we would place it among QC servers using AliECS, not by configuration files.
 
-          // Collecting Data Sampling Policies
-          auto dataSourceTree = taskConfig.get_child("dataSource");
-          std::string type = dataSourceTree.get<std::string>("type");
-          if (type == "dataSamplingPolicy") {
-            samplingPoliciesUsed.insert(dataSourceTree.get<std::string>("name"));
-          } else if (type == "direct") {
-            throw std::runtime_error("Configuration error: Remote QC tasks such as " + taskName + " cannot use direct data sources");
-          } else {
-            throw std::runtime_error("Configuration error: dataSource type unknown : " + type);
-          }
-
-          // Creating the remote task
-          workflow.emplace_back(taskRunnerFactory.create(taskName, configurationSource, 0));
+        // Collecting Data Sampling Policies
+        auto dataSourceTree = taskConfig.get_child("dataSource");
+        std::string type = dataSourceTree.get<std::string>("type");
+        if (type == "dataSamplingPolicy") {
+          samplingPoliciesUsed.insert(dataSourceTree.get<std::string>("name"));
+        } else if (type == "direct") {
+          throw std::runtime_error("Configuration error: Remote QC tasks such as " + taskName + " cannot use direct data sources");
+        } else {
+          throw std::runtime_error("Configuration error: dataSource type unknown : " + type);
         }
+
+        // Creating the remote task
+        workflow.emplace_back(taskRunnerFactory.create(taskName, configurationSource, 0));
       }
     }
   }
@@ -202,7 +230,9 @@ o2::framework::WorkflowSpec InfrastructureGenerator::generateRemoteInfrastructur
   for (const auto& policyName : samplingPoliciesUsed) {
     // todo now we have to generate one proxy per local machine and policy, because of the proxy limitations.
     //  Use one proxy per policy when it is possible.
-    std::string port = std::to_string(DataSampling::PortForPolicy(config.get(), policyName));
+
+    // todo: leave only the new way once the return type is changed
+    std::string port = std::to_string(std::optional<uint16_t>(DataSampling::PortForPolicy(config.get(), policyName)).value_or(defaultPolicyPort));
     Outputs outputSpecs = DataSampling::OutputSpecsForPolicy(config.get(), policyName);
     std::vector<std::string> machines = DataSampling::MachinesForPolicy(config.get(), policyName);
     for (const auto& machine : machines) {
@@ -428,7 +458,7 @@ vector<OutputSpec> InfrastructureGenerator::generateCheckRunners(framework::Work
         break;
       }
     }
-    if (!isStored) {
+    if (!isStored) { // fixme: statement is always true
       // If there is no Check for a given input, create a candidate for a sink device
       InputNames singleEntry{ label };
       // Init empty Check vector to appear in the next step
diff --git a/doc/Advanced.md b/doc/Advanced.md
@@ -132,10 +132,9 @@ locally produced Monitor Objects should be merged on QC servers and then have Ch
 By **remote QC tasks** we mean those which run on QC servers (**remote machines**), while **local QC Tasks**
 run on FLPs and EPNs (**local machines**).
 
-While it is responsibility of the run operators to run all the processing topologies during the
-data taking, here we show how to achieve such multinode workflows on development setups, running
-them just with DPL driver. Note that for now we support cases with one or more local machines,
-but just only one remote machine.
+Setting up a multinode setup to run standalone or with AliECS requires different amount of parameters,
+as some of them are overwritten by AliECS anyway. Such parameters are marked accordingly. Please note
+that for now we support cases with one or more local machines, but just only one remote machine.
 
 In our example, we assume having two local processing nodes (`localnode1`, `localnode2`) and one
 QC node (`qcnode`). There are two types of QC Tasks declared:
@@ -145,7 +144,9 @@ QC node (`qcnode`). There are two types of QC Tasks declared:
   `localnode2` only. Mergers are not needed in this case, but there is a process running Checks against
   Monitor Objects generated by this Task.
 
-We use the `SkeletonTask` class for both, but any Task can be used of course. Should a Task be local, all its `MonitorObject`s need to be mergeable - they should be one of the mergeable ROOT types (histograms, TTrees) or inherit [MergeInterface](https://github.com/AliceO2Group/AliceO2/blob/dev/Utilities/Mergers/include/Mergers/MergeInterface.h).
+We use the `SkeletonTask` class for both, but any Task can be used of course. Should a Task be local,
+all its `MonitorObject`s need to be mergeable - they should be one of the mergeable ROOT types (histograms, TTrees)
+or inherit [MergeInterface](https://github.com/AliceO2Group/AliceO2/blob/dev/Utilities/Mergers/include/Mergers/MergeInterface.h).
 
 These are the steps to follow to get a multinode setup:
 
@@ -165,18 +166,21 @@ added:
           "localnode1",
           "localnode2"
         ],
-        "remoteMachine": "qcnode",
-        "remotePort": "30132",
-        "mergingMode": "delta"
+        "remoteMachine": "qcnode",   "":"not needed with AliECS",
+        "remotePort": "30132",       "":"not needed with AliECS",
+        "mergingMode": "delta",      "":"if absent, delta is default"
       }
     },
 ```
 List the local processing machines in the `localMachines` array. `remoteMachine` should contain the host name which
-will serve as a QC server and `remotePort` should be a port number on which Mergers will wait for upcoming MOs. Make
-sure it is not used by other service. If different QC Tasks are run in parallel, use separate ports for each. One
-also may choose the merging mode - `delta` is the default and recommended (tasks are reset after each cycle, so they
-send only updates), but if it is not feasible, Mergers may expect `entire` objects - tasks are not reset, they
-always send entire objects and the latest versions are combined in Mergers.
+ will serve as a QC server and `remotePort` should be a port number on which Mergers will wait for upcoming MOs. Make
+ sure it is not used by other service. If different QC Tasks are run in parallel, use separate ports for each. One
+ also may choose the merging mode - `delta` is the default and recommended (tasks are reset after each cycle, so they
+ send only updates), but if it is not feasible, Mergers may expect `entire` objects - tasks are not reset, they
+ always send entire objects and the latest versions are combined in Mergers.
+ 
+With the `delta` mode, one can cheat by specifying just one local machine name and referencing only that one later.
+This is not possible with `entire` mode, because then Mergers need identifiable data sources to merge objects correctly.
 
 In case of a remote task, choosing `"remote"` option for the `"location"` parameter is enough.
 
@@ -196,29 +200,28 @@ In case of a remote task, choosing `"remote"` option for the `"location"` parame
     }
 ```
 
-However in both cases, one has to specify the machines where data should be sampled, as below. If data should be
-published to external machines (with remote tasks), one has to add a local port number. Use separate ports for each
-Data Sampling Policy.
+In case the task is running remotely, one has to specify the machines where data should be published to external 
+machines (with remote tasks) and a local port number. Use separate ports for each Data Sampling Policy.
 ```json
 {
   "dataSamplingPolicies": [
     ...
     {
       "id": "rnd-little",
       "active": "true",
-      "machines": [
+      "machines": [        "","needed only for remote QC tasks",
         "localnode2"
       ],
-      "port": "30333"
+      "port": "30333",     "":"not needed with AliECS",
       ...
     }
   ]
 }
 ```
 /
 2. Make sure that the firewalls are properly configured. If your machines block incoming/outgoing connections by
-   default, you can add these rules to the firewall (run as sudo). Consider enabling only concrete ports or a small
-   range of those.
+ default, you can add these rules to the firewall (run as sudo). Consider enabling only concrete ports or a small
+ range of those.
 
 ```
 # localnode1 and localnode2 :
@@ -230,6 +233,11 @@ iptables -I OUTPUT -p tcp -m conntrack --ctstate NEW,ESTABLISHED -d localnode1 -
 iptables -I INPUT -p tcp -m conntrack --ctstate NEW,ESTABLISHED -s localnode2 -j ACCEPT
 iptables -I OUTPUT -p tcp -m conntrack --ctstate NEW,ESTABLISHED -d localnode2 -j ACCEPT
 ```
+If your network is isolated, you might consider disabling the firewall as an alternative. Be wary of the security risks.
+```
+systemctl stop firewalld     # to disable until reboot
+systemctl disable firewalld  # to disable permanently
+```
 
 3. Install the same version of the QC software on each of these nodes. We cannot guarantee that different QC versions will talk to each other without problems. Also, make sure the configuration file that you will use is the same everywhere.
 
@@ -246,7 +254,10 @@ o2-qc --config json:/${QUALITYCONTROL_ROOT}/etc/multiNode.json --remote
 If there are no problems, on QCG you should see the `example` histogram updated under the paths `qc/TST/MO/MultiNodeLocal`
 and `qc/TST/MO/MultiNodeRemote`, and corresponding Checks under the path `qc/TST/QO/`.
 
-## Writing a DPL data producer
+When using AliECS, one has to generate workflow templates and upload them to the corresponding repository. Please
+contact the QC or AliECS developers to receive assistance or instruction on how to do that.
+
+## Writing a DPL data producer 
 
 For your convenience, and although it does not lie within the QC scope, we would like to document how to write a simple data producer in the DPL. The DPL documentation can be found [here](https://github.com/AliceO2Group/AliceO2/blob/dev/Framework/Core/README.md) and for questions please head to the [forum](https://alice-talk.web.cern.ch/).
 
