Merge branch 'develop' into issue700/remove_archive_workflows

.github/workflows/integration.yml

@@ -26,13 +26,14 @@ jobs:
     runs-on: ubuntu-22.04
     steps:
       - uses: actions/checkout@v3
-      - name: Dependency Install
-        run: ./ci/deps_install.sh
-      - name: Batch Scheduler Install and Start
-        run: ./ci/batch_scheduler.sh
-      - name: BEE Install
-        run: ./ci/bee_install.sh
-      - name: BEE Config
-        run: ./ci/bee_config.sh
+      - name: Install and Configure
+        run: |
+          . ./ci/env.sh
+          ./ci/deps_install.sh
+          ./ci/batch_scheduler.sh
+          ./ci/bee_install.sh
+          ./ci/bee_config.sh
       - name: Integration Test
-        run: ./ci/integration_test.sh
+        run: |
+          . ./ci/env.sh
+          ./ci/integration_test.sh

.github/workflows/unit-tests.yml

@@ -24,13 +24,14 @@ jobs:
     runs-on: ubuntu-22.04
     steps:
       - uses: actions/checkout@v3
-      - name: Dependency Install
-        run: ./ci/deps_install.sh
-      - name: Slurm Setup and Install
-        run: ./ci/slurm_start.sh
-      - name: BEE Install
-        run: ./ci/bee_install.sh
-      - name: BEE Config
-        run: ./ci/bee_config.sh
+      - name: Install and Configure
+        run: |
+          . ./ci/env.sh
+          ./ci/deps_install.sh
+          ./ci/batch_scheduler.sh
+          ./ci/bee_install.sh
+          ./ci/bee_config.sh
       - name: Unit tests
-        run: ./ci/unit_tests.sh
+        run: |
+          . ./ci/env.sh
+          ./ci/unit_tests.sh

README.rst

@@ -42,6 +42,7 @@ Contributors:
 * Paul Bryant - `paulbry <https://github.com/paulbry>`_
 * Rusty Davis - `rstyd <https://github.com/rstyd>`_
 * Jieyang Chen - `JieyangChen7 <https://github.com/JieyangChen7>`_
+* Krishna Chilleri - `Krishna Chilleri <https://github.com/kchilleri>`_
 * Patricia Grubel - `pagrubel <https://github.com/pagrubel>`_
 * Qiang Guan - `guanxyz <https://github.com/guanxyz>`_
 * Ragini Gupta - `raginigupta6 <https://github.com/raginigupta6>`_
@@ -85,9 +86,8 @@ License can be found `here <https://github.com/lanl/BEE/blob/master/LICENSE>`_
 Publications
 ==========================
 
+- An HPC-Container Based Continuous Integration Tool for Detecting Scaling and Performance Issues in HPC Applications, IEEE Transactions on Services Computing, 2024, `DOI: 10.1109/TSC.2023.3337662 <https://doi.ieeecomputersociety.org/10.1109/TSC.2023.3337662>`_
 - BEE Orchestrator: Running Complex Scientific Workflows on Multiple Systems, HiPC, 2021, `DOI: 10.1109/HiPC53243.2021.00052 <https://doi.org/10.1109/HiPC53243.2021.00052>`_
 - "BeeSwarm: Enabling Parallel Scaling Performance Measurement in Continuous Integration for HPC Applications", ASE, 2021, `DOI: 10.1109/ASE51524.2021.9678805 <https://www.computer.org/csdl/proceedings-article/ase/2021/033700b136/1AjTjgnW2pa#:~:text=10.1109/ASE51524.2021.9678805>`_
 - "BeeFlow: A Workflow Management System for In Situ Processing across HPC and Cloud Systems", ICDCS, 2018, `DOI: 10.1109/ICDCS.2018.00103 <https://ieeexplore.ieee.org/abstract/document/8416366>`_
 - "Build and execution environment (BEE): an encapsulated environment enabling HPC applications running everywhere", IEEE BigData, 2018, `DOI: 10.1109/BigData.2018.8622572 <https://ieeexplore.ieee.org/document/8622572>`_
-
-

beeflow/client/bee_client.py

@@ -26,6 +26,7 @@
 from beeflow.common.parser import CwlParser
 from beeflow.common.wf_data import generate_workflow_id
 from beeflow.client import core
+from beeflow.wf_manager.resources import wf_utils
 
 # Length of a shortened workflow ID
 short_id_len = 6 #noqa: Not a constant
@@ -295,6 +296,19 @@ def is_parent(parent, path):
     if tarball_path:
         os.remove(tarball_path)
 
+    # Store provided arguments in text file for future reference
+    wf_dir = wf_utils.get_workflow_dir(wf_id)
+    sub_wf_dir = wf_dir + "/submit_command_args.txt"
+
+    f_name = open(sub_wf_dir, "w", encoding="utf-8")
+    f_name.write(f"wf_name: {wf_name}\n")
+    f_name.write(f"wf_path: {wf_path}\n")
+    f_name.write(f"main_cwl: {main_cwl}\n")
+    f_name.write(f"yaml: {yaml}\n")
+    f_name.write(f"workdir: {workdir}\n")
+    f_name.write(f"wf_id: {wf_id}")
+    f_name.close()
+
     return wf_id
 
 

beeflow/client/core.py

@@ -171,8 +171,8 @@ def start_wfm():
     def start_task_manager():
         """Start the TM."""
         fp = open_log('task_manager')
-        return launch_with_gunicorn('beeflow.task_manager:flask_app', paths.tm_socket(),
-                                    stdout=fp, stderr=fp)
+        return launch_with_gunicorn('beeflow.task_manager.task_manager:create_app()',
+                                    paths.tm_socket(), stdout=fp, stderr=fp)
 
     @mgr.component('scheduler', ())
     def start_scheduler():

beeflow/common/config_driver.py

@@ -35,6 +35,10 @@
 else:
     raise RuntimeError(f'System "{_SYSTEM}" is not supported')
 
+BEE_CONFIG = os.getenv('BEE_CONFIG')
+if BEE_CONFIG is not None:
+    USERCONFIG_FILE = BEE_CONFIG
+
 
 class BeeConfig:
     r"""Class to manage and store all BEE configuration.

beeflow/common/parser/parser.py

@@ -141,6 +141,11 @@ def parse_step(self, step, workflow_id):
         else:
             step_cwl = step.run
             step_id = _shortname(step.id)
+            # step_input.id needs to have its step.id prefix stripped
+            for step_input in step_cwl.inputs:
+                step_shortname = _shortname(step_input.id)
+                step_input.id = step_input.id.replace(step_shortname,
+                                                      step_shortname.split("/")[-1])
 
         if step_cwl.class_ != "CommandLineTool":
             raise CwlParseError(f"Step {step.id} class must be CommandLineTool")
@@ -302,7 +307,13 @@ def parse_requirements(self, requirements, as_hints=False):
             return reqs
         if as_hints:
             for req in requirements:
-                items = {k: str(v) for k, v in req.items() if k != "class"}
+                items = {}
+                for k, v in req.items():
+                    if k != 'class':
+                        if isinstance(v, (int, float)):
+                            items[k] = v
+                        else:
+                            items[k] = str(v)
                 # Load in the dockerfile at parse time
                 if 'dockerFile' in items:
                     self._read_requirement_file('dockerFile', items)

beeflow/common/wf_data.py

@@ -197,6 +197,36 @@ def get_requirement(self, req_type, req_param, default=None):
                 requirement = default
         return requirement
 
+    def get_full_requirement(self, req_type):
+        """Get the full requirement (or hint) for this task, if it has one.
+
+        :param req_type: the type of requirement (e.g. 'DockerRequirement')
+        :type req_type: str
+
+        This prefers requirements over hints. Returns None if no hint or
+        requirement found.
+        """
+        result = None
+        hints = dict(self.hints)
+        try:
+            # Try to get Hints
+            hint = hints[req_type]
+        except (KeyError, TypeError):
+            # Task Hints are not mandatory. No task hint specified.
+            hint = None
+        try:
+            # Try to get Requirements
+            req = self.requirements[req_type]
+        except (KeyError, TypeError):
+            # Task Requirements are not mandatory. No task requirement specified.
+            req = None
+        # Prefer requirements over hints
+        if req:
+            result = req
+        elif hint:
+            result = hint
+        return result
+
     def __eq__(self, other):
         """Test the equality of two tasks.
 

beeflow/common/wf_interface.py

@@ -94,6 +94,7 @@ def restart_task(self, task, checkpoint_file):
 
         :param task: the task to restart
         :type task: Task
+        :param checkpoint_file: the task checkpoint file
         :rtype: Task or None
         """
         for hint in task.hints:
@@ -102,8 +103,8 @@ def restart_task(self, task, checkpoint_file):
                     hint.params["num_tries"] -= 1
                     hint.params["bee_checkpoint_file__"] = checkpoint_file
                     break
-                state = self.get_task_state(task)
-                self.set_task_state(task, f"FAILED RESTART: {state}")
+                self.set_task_state(task, "FAILED")
+                self.set_workflow_state("FAILED")
                 return None
         else:
             raise ValueError("invalid task for checkpoint restart")