Merge pull request #135 from wfcommons/makeflow_logger

henricasanova · web-flow · commit 71b8389ff985 · 2026-02-02T11:28:28.000-10:00
Improved Dockerized-testing
diff --git a/tests/test_helpers.py b/tests/test_helpers.py
@@ -50,20 +50,22 @@ def _install_WfCommons_on_container(container):
     tar_data = _make_tarfile_of_wfcommons()
     container.put_archive(target_path, tar_data)
     # Cleanup files that came from the host
-    exit_code, output = container.exec_run("sudo /bin/rm -rf /tmp/WfCommons/build/", stdout=True, stderr=True)
-    exit_code, output = container.exec_run("sudo /bin/rm -rf /tmp/WfCommons/*.egg-info/", stdout=True, stderr=True)
+    exit_code, output = container.exec_run("sudo /bin/rm -rf /tmp/WfCommons/build/", user="wfcommons", stdout=True, stderr=True)
+    exit_code, output = container.exec_run("sudo /bin/rm -rf /tmp/WfCommons/*.egg-info/", user="wfcommons", stdout=True, stderr=True)
     # Clean up and force a rebuild of cpu-benchmark (because it may be compiled for the wrong architecture)
-    exit_code, output = container.exec_run("sudo /bin/rm -rf /tmp/WfCommons/bin/cpu-benchmark.o", stdout=True,
+    exit_code, output = container.exec_run("sudo /bin/rm -rf /tmp/WfCommons/bin/cpu-benchmark.o", user="wfcommons", stdout=True,
                                            stderr=True)
-    exit_code, output = container.exec_run("sudo /bin/rm -rf /tmp/WfCommons/bin/cpu-benchmark", stdout=True,
+    exit_code, output = container.exec_run("sudo /bin/rm -rf /tmp/WfCommons/bin/cpu-benchmark", user="wfcommons", stdout=True,
                                            stderr=True)
 
     # Install WfCommons on the container (to install wfbench and cpu-benchmark really)
     exit_code, output = container.exec_run("sudo python3 -m pip install . --break-system-packages",
+                                           user="wfcommons",
                                            workdir="/tmp/WfCommons", stdout=True, stderr=True)
-    # print(output.decode())
     if exit_code != 0:
-        raise RuntimeError("Failed to install WfCommons on the container")
+        print(output.decode())
+        raise RuntimeError("Failed to pip install WfCommons on the container")
+
 
 def _start_docker_container(backend, mounted_dir, working_dir, bin_dir, command=None):
     if command is None:
@@ -86,6 +88,7 @@ def _start_docker_container(backend, mounted_dir, working_dir, bin_dir, command=
         command=command,
         volumes={mounted_dir: {'bind': mounted_dir, 'mode': 'rw'}},
         working_dir=working_dir,
+        user="wfcommons",
         tty=True,
         detach=True
     )
@@ -97,22 +100,19 @@ def _start_docker_container(backend, mounted_dir, working_dir, bin_dir, command=
     if bin_dir:
         sys.stderr.write(f"[{backend}] Copying wfbench and cpu-benchmark...\n")
         exit_code, output = container.exec_run(["sh", "-c", "sudo cp -f `which wfbench` " + bin_dir],
+                                               user="wfcommons",
                                                stdout=True, stderr=True)
         if exit_code != 0:
             raise RuntimeError("Failed to copy wfbench script to the bin directory")
 
         exit_code, output = container.exec_run(["sh", "-c", "sudo cp -f `which cpu-benchmark` " + bin_dir],
+                                               user="wfcommons",
                                                stdout=True, stderr=True)
         if exit_code != 0:
             raise RuntimeError("Failed to copy cpu-benchmark executable to the bin directory")
     else:
         sys.stderr.write(f"[{backend}] Not Copying wfbench and cpu-benchmark...\n")
 
-    # Change file permissions
-    exit_code, output = container.exec_run(["sh", "-c", "sudo chown -R wfcommons:wfcommons "],
-                                           stdout=True, stderr=True)
-
-
     container.backend = backend
     return container
 
diff --git a/tests/translators_loggers/Dockerfile.airflow b/tests/translators_loggers/Dockerfile.airflow
@@ -77,4 +77,7 @@ RUN echo '%sudo ALL=(ALL) NOPASSWD:ALL' >> /etc/sudoers
 ENV PATH="$PATH:/home/wfcommons/.local/bin/"
 
 USER wfcommons
-WORKDIR /home/wfcommons
+WORKDIR /home/wfcommons
+# Making this directory world rwx to facilitate testing
+RUN chmod -R 777 /home/wfcommons
+
diff --git a/tests/translators_loggers/Dockerfile.bash b/tests/translators_loggers/Dockerfile.bash
@@ -42,4 +42,6 @@ ENV PATH="$PATH:/home/wfcommons/.local/bin/"
 
 USER wfcommons
 WORKDIR /home/wfcommons
+# Making this directory world rwx to facilitate testing
+RUN chmod -R 777 /home/wfcommons
 
diff --git a/tests/translators_loggers/Dockerfile.cwl b/tests/translators_loggers/Dockerfile.cwl
@@ -46,4 +46,6 @@ ENV PATH="$PATH:/home/wfcommons/.local/bin/"
 
 USER wfcommons
 WORKDIR /home/wfcommons
+# Making this directory world rwx to facilitate testing
+RUN chmod -R 777 /home/wfcommons
 
diff --git a/tests/translators_loggers/Dockerfile.dask b/tests/translators_loggers/Dockerfile.dask
@@ -46,4 +46,6 @@ ENV PATH="$PATH:/home/wfcommons/.local/bin/"
 
 USER wfcommons
 WORKDIR /home/wfcommons
+# Making this directory world rwx to facilitate testing
+RUN chmod -R 777 /home/wfcommons
 
diff --git a/tests/translators_loggers/Dockerfile.makeflow b/tests/translators_loggers/Dockerfile.makeflow
@@ -43,6 +43,9 @@ ENV PATH="$PATH:/home/wfcommons/.local/bin/"
 
 USER wfcommons
 WORKDIR /home/wfcommons
+# Making this directory world rwx to facilitate testing
+RUN chmod -R 777 /home/wfcommons
+
 
 # Install Miniforge
 RUN wget -O Miniforge3.sh "https://github.com/conda-forge/miniforge/releases/latest/download/Miniforge3-$(uname)-$(uname -m).sh" && \
diff --git a/tests/translators_loggers/Dockerfile.nextflow b/tests/translators_loggers/Dockerfile.nextflow
@@ -45,6 +45,9 @@ RUN echo '%sudo ALL=(ALL) NOPASSWD:ALL' >> /etc/sudoers
 ENV PATH="$PATH:/home/wfcommons/.local/bin/"
 USER wfcommons
 WORKDIR /home/wfcommons
+# Making this directory world rwx to facilitate testing
+RUN chmod -R 777 /home/wfcommons
+
 
 # Install Nextflow and dependencies
 RUN curl -s https://get.nextflow.io | bash  && sudo mv ./nextflow /usr/local/bin/
diff --git a/tests/translators_loggers/Dockerfile.parsl b/tests/translators_loggers/Dockerfile.parsl
@@ -45,4 +45,6 @@ ENV PATH="$PATH:/home/wfcommons/.local/bin/"
 
 USER wfcommons
 WORKDIR /home/wfcommons
+# Making this directory world rwx to facilitate testing
+RUN chmod -R 777 /home/wfcommons
 
diff --git a/tests/translators_loggers/Dockerfile.pegasus b/tests/translators_loggers/Dockerfile.pegasus
@@ -110,6 +110,9 @@ ENV PATH="$PATH:/home/wfcommons/.local/bin/"
 
 USER wfcommons
 WORKDIR /home/wfcommons
+# Making this directory world rwx to facilitate testing
+RUN chmod -R 777 /home/wfcommons
+
 
 # Create script to start HTCondor
 RUN echo "#!/bin/bash" >> ./start_condor.sh && \
diff --git a/tests/translators_loggers/Dockerfile.pycompss b/tests/translators_loggers/Dockerfile.pycompss
@@ -51,4 +51,6 @@ ENV PATH="$PATH:/home/wfcommons/.local/bin/"
 
 USER wfcommons
 WORKDIR /home/wfcommons
+# Making this directory world rwx to facilitate testing
+RUN chmod -R 777 /home/wfcommons
 
diff --git a/tests/translators_loggers/Dockerfile.swiftt b/tests/translators_loggers/Dockerfile.swiftt
@@ -65,4 +65,6 @@ ENV PATH="/opt/conda/envs/swiftt-env/bin:$PATH"
 
 USER wfcommons
 WORKDIR /home/wfcommons
+# Making this directory world rwx to facilitate testing
+RUN chmod -R 777 /home/wfcommons
 
diff --git a/tests/translators_loggers/Dockerfile.taskvine b/tests/translators_loggers/Dockerfile.taskvine
@@ -43,6 +43,8 @@ ENV PATH="$PATH:/home/wfcommons/.local/bin/"
 
 USER wfcommons
 WORKDIR /home/wfcommons
+# Making this directory world rwx to facilitate testing
+RUN chmod -R 777 /home/wfcommons
 
 # Install Miniforge
 RUN wget -O Miniforge3.sh "https://github.com/conda-forge/miniforge/releases/latest/download/Miniforge3-$(uname)-$(uname -m).sh" && \
diff --git a/tests/translators_loggers/test_translators_loggers.py b/tests/translators_loggers/test_translators_loggers.py
@@ -65,6 +65,7 @@ def _additional_setup_taskvine(container):
     # Create the poncho package
     exit_code, output = container.exec_run(cmd=["bash", "-c",
                                                 "source ~/conda/etc/profile.d/conda.sh && conda activate && poncho_package_create taskvine_poncho.json taskvine_poncho.tar.gz"],
+                                           user="wfcommons",
                                            stdout=True, stderr=True)
     if exit_code != 0:
         raise Exception("Failed to setup TaskVine: cannot create poncho package")
@@ -79,25 +80,27 @@ def _additional_setup_pegasus(container):
     # Start Condor
     exit_code, output = container.exec_run(cmd=["bash", "-c",
                                                 "bash /home/wfcommons/start_condor.sh"],
+                                           user="wfcommons",
                                            stdout=True, stderr=True)
     if exit_code != 0:
         raise Exception("Failed to setup Pegasus: cannot start HTCondor")
     # Run pegasus script
     exit_code, output = container.exec_run(cmd=["bash", "-c",
                                                 "python3 ./pegasus_workflow.py"],
+                                           user="wfcommons",
                                            stdout=True, stderr=True)
     if exit_code != 0:
         raise Exception("Failed to setup Pegasus: error while running the pegasus_workflow.py script")
 
 def _additional_setup_swiftt(container):
     # Start a redis server in the background
     exit_code, output = container.exec_run(
-        cmd=["bash", "-c", "redis-server"], detach=True, stdout=True, stderr=True)
+        cmd=["bash", "-c", "redis-server"], user="wfcommons", detach=True, stdout=True, stderr=True)
     # Note that exit_code will always be None because of detach=True.
 
     # Check that the redis-server is up
     exit_code, output = container.exec_run(
-        cmd=["bash", "-c", "redis-cli ping"], stdout=True, stderr=True)
+        cmd=["bash", "-c", "redis-cli ping"], user="wfcommons", stdout=True, stderr=True)
     if output.decode().strip() != 'PONG':
         raise Exception("Failed to start redis-server...")
 
@@ -120,24 +123,24 @@ def _additional_setup_swiftt(container):
 #############################################################################
 
 def run_workflow_dask(container, num_tasks, str_dirpath):
-    exit_code, output = container.exec_run("python ./dask_workflow.py", stdout=True, stderr=True)
+    exit_code, output = container.exec_run("python ./dask_workflow.py", user="wfcommons", stdout=True, stderr=True)
     # Check sanity
     assert (exit_code == 0)
     assert (output.decode().count("completed!")  == num_tasks)
     # TODO: Look at the (I think) generated run.json file on the container?
 
 def run_workflow_parsl(container, num_tasks, str_dirpath):
-    exit_code, output = container.exec_run("python ./parsl_workflow.py", stdout=True, stderr=True)
-    ignored, output = container.exec_run(f"cat {str_dirpath}/runinfo/000/parsl.log", stdout=True, stderr=True)
+    exit_code, output = container.exec_run("python ./parsl_workflow.py", user="wfcommons", stdout=True, stderr=True)
+    ignored, output = container.exec_run(f"cat {str_dirpath}/runinfo/000/parsl.log", user="wfcommons", stdout=True, stderr=True)
     # Check sanity
     assert (exit_code == 0)
     assert ("completed" in output.decode())
     assert (output.decode().count("_complete_task") == num_tasks)
 
 def run_workflow_nextflow(container, num_tasks, str_dirpath):
     # Run the workflow!
-    exit_code, output = container.exec_run(f"nextflow run ./workflow.nf --pwd .", stdout=True, stderr=True)
-    ignored, task_exit_codes = container.exec_run("find . -name .exitcode -exec cat {} \;", stdout=True, stderr=True)
+    exit_code, output = container.exec_run(f"nextflow run ./workflow.nf --pwd .", user="wfcommons", stdout=True, stderr=True)
+    ignored, task_exit_codes = container.exec_run("find . -name .exitcode -exec cat {} \;", user="wfcommons", stdout=True, stderr=True)
     # Check sanity
     assert (exit_code == 0)
     assert (task_exit_codes.decode() == num_tasks * "0")
@@ -146,29 +149,30 @@ def run_workflow_airflow(container, num_tasks, str_dirpath):
     # Run the workflow! (use a specific working directory)
     # TODO: Remove the hardcoded Blast-Benchmark as it's ugly
     exit_code, output = container.exec_run(cmd=["sh", "-c", "cd /home/wfcommons/ && sudo /bin/bash /run_a_workflow.sh Blast-Benchmark"],
-                                           stdout=True,
-                                           stderr=True)
+                                           user="wfcommons", stdout=True, stderr=True)
     # Check sanity
     assert (exit_code == 0)
     assert (output.decode().count("completed") == num_tasks * 2)
 
 def run_workflow_bash(container, num_tasks, str_dirpath):
     # Run the workflow!
-    exit_code, output = container.exec_run(cmd="/bin/bash ./run_workflow.sh", stdout=True, stderr=True)
+    exit_code, output = container.exec_run(cmd="/bin/bash ./run_workflow.sh", user="wfcommons", stdout=True, stderr=True)
     # Check sanity
     assert (exit_code == 0)
     assert (output.decode().count("completed") == num_tasks)
 
 def run_workflow_taskvine(container, num_tasks, str_dirpath):
     # Run the workflow!
-    exit_code, output = container.exec_run(cmd=["bash", "-c", "source ~/conda/etc/profile.d/conda.sh && conda activate && python3 ./taskvine_workflow.py"], stdout=True, stderr=True)
+    exit_code, output = container.exec_run(cmd=["bash", "-c", "source ~/conda/etc/profile.d/conda.sh && conda activate && python3 ./taskvine_workflow.py"],
+                                           user="wfcommons", stdout=True, stderr=True)
     # Check sanity
     assert (exit_code == 0)
     assert (output.decode().count("completed") == num_tasks)
 
 def run_workflow_makeflow(container, num_tasks, str_dirpath):
     # Run the workflow (with full logging)
-    exit_code, output = container.exec_run(cmd=["bash", "-c", "source ~/conda/etc/profile.d/conda.sh && conda activate && makeflow --log-verbose  --monitor=./monitor_data/ ./workflow.makeflow"], stdout=True, stderr=True)
+    exit_code, output = container.exec_run(cmd=["bash", "-c", "source ~/conda/etc/profile.d/conda.sh && conda activate && makeflow --log-verbose  --monitor=./monitor_data/ ./workflow.makeflow"],
+                                           user="wfcommons", stdout=True, stderr=True)
     # Check sanity
     assert (exit_code == 0)
     num_completed_jobs = len(re.findall(r'job \d+ completed', output.decode()))
@@ -177,7 +181,8 @@ def run_workflow_makeflow(container, num_tasks, str_dirpath):
 def run_workflow_cwl(container, num_tasks, str_dirpath):
     # Run the workflow!
     # Note that the input file is hardcoded and Blast-specific
-    exit_code, output = container.exec_run(cmd="cwltool ./main.cwl --split_fasta_00000001_input ./data/workflow_infile_0001 ", stdout=True, stderr=True)
+    exit_code, output = container.exec_run(cmd="cwltool ./main.cwl --split_fasta_00000001_input ./data/workflow_infile_0001 ",
+                                           user="wfcommons", stdout=True, stderr=True)
     # Check sanity
     assert (exit_code == 0)
     # this below is ugly (the 3 is for "workflow", "compile_output_files" and "compile_log_files",
@@ -186,14 +191,16 @@ def run_workflow_cwl(container, num_tasks, str_dirpath):
 
 def run_workflow_pegasus(container, num_tasks, str_dirpath):
     # Run the workflow!
-    exit_code, output = container.exec_run(cmd="bash /home/wfcommons/run_workflow.sh", stdout=True, stderr=True)
+    exit_code, output = container.exec_run(cmd="bash /home/wfcommons/run_workflow.sh",
+                                           user="wfcommons", stdout=True, stderr=True)
     # Check sanity
     assert(exit_code == 0)
     assert("success" in output.decode())
 
 def run_workflow_swiftt(container, num_tasks, str_dirpath):
     # Run the workflow!
-    exit_code, output = container.exec_run(cmd="swift-t workflow.swift", stdout=True, stderr=True)
+    exit_code, output = container.exec_run(cmd="swift-t workflow.swift",
+                                           user="wfcommons", stdout=True, stderr=True)
     # sys.stderr.write(output.decode())
     # Check sanity
     assert(exit_code == 0)
@@ -266,11 +273,12 @@ def test_translator(self, backend) -> None:
             translator = translator_classes[backend](benchmark.workflow)
         translator.translate(output_folder=dirpath)
 
-        # # Make the directory that holds the translation world-writable,
-        # # so that docker commands won't fail
-        # TODO: Explore whether this below makes tests runnable on Linux due to
-        #       different Docker permission schemes, etc.
-        # os.chmod(dirpath, 0o777)
+        # Make the directory that holds the translation world-writable,
+        # so that we don't have any permission shenanigans
+        for directory, directory_name, filenames in os.walk(dirpath):
+            os.chmod(directory, 0o777)
+            for filename in filenames:
+                os.chmod(os.path.join(directory, filename), 0o777)
 
         # Start the Docker container
         container = _start_docker_container(backend if backend != "nextflow_subworkflow" else "nextflow", str_dirpath, str_dirpath, str_dirpath + "bin/")
@@ -284,6 +292,12 @@ def test_translator(self, backend) -> None:
         run_workflow_methods[backend](container, num_tasks, str_dirpath)
         sys.stderr.write(f"[{backend}] Workflow ran in %.2f seconds\n" % (time.time() - start_time))
 
+        # Make the whole content world-writable/readable
+        # so that we don't have any permission shenanigans for loggers below
+        exit_code, output = container.exec_run(cmd=["bash", "-c", "chmod -R 0777 ."],
+                                               user="wfcommons",
+                                               stdout=True, stderr=True)
+
         # Run the log parser if any
         if backend == "pegasus":
             parser = PegasusLogsParser(dirpath / "work/wfcommons/pegasus/Blast-Benchmark/run0001/")
@@ -295,7 +309,7 @@ def test_translator(self, backend) -> None:
             parser = None
 
         if parser:
-            sys.stderr.write(f"\n[{backend}] Parsing the logs...\n")
+            sys.stderr.write(f"[{backend}] Parsing the logs...\n")
             reconstructed_workflow : Workflow = parser.build_workflow(f"reconstructed_workflow_{backend}")
             reconstructed_workflow.write_json(pathlib.Path("/tmp/reconstructed_workflow.json"))
 
diff --git a/wfcommons/wfbench/translator/parsl.py b/wfcommons/wfbench/translator/parsl.py
@@ -96,6 +96,9 @@ def translate(self, output_folder: pathlib.Path) -> None:
         self._copy_binary_files(output_folder)
         self._generate_input_files(output_folder)
 
+        # README file
+        self._write_readme_file(output_folder)
+
     def _parsl_wftasks_codelines(self) -> list[str]:
         codelines = ["task_arr = []\n"]
 
@@ -161,3 +164,14 @@ def _parsl_wftasks_codelines(self) -> list[str]:
         codelines.extend(cleanup_code)
 
         return codelines
+
+    def _write_readme_file(self, output_folder: pathlib.Path) -> None:
+        """
+        Write the README  file.
+
+        :param output_folder: The path of the output folder.
+        :type output_folder: pathlib.Path
+        """
+        readme_file_path = output_folder.joinpath("README")
+        with open(readme_file_path, "w") as out:
+            out.write(f"Simply run: python3 ./parsl_workflow.py\n")
diff --git a/wfcommons/wfbench/translator/templates/parsl_template.py b/wfcommons/wfbench/translator/templates/parsl_template.py
@@ -16,25 +16,20 @@
 from parsl.app.app import bash_app
 from parsl.config import Config
 from parsl.executors import HighThroughputExecutor
+from parsl.executors.threads import ThreadPoolExecutor
 from parsl.providers import LocalProvider
 from parsl.data_provider.files import File
+
 # NOTE: Uncomment the following lines to enable Parsl workflow monitoring
 # from parsl.monitoring.monitoring import MonitoringHub
 # from parsl.addresses import address_by_hostname
 
 # NOTE: update the configuration below for your desired system (https://parsl.readthedocs.io/en/stable/userguide/configuring.html)
 config = Config(
     executors=[
-        HighThroughputExecutor(
-            label="local_htex",
-            worker_debug=True,
-            cores_per_worker=1,
-            max_workers_per_node=1,
-            provider=LocalProvider(
-                init_blocks=1,
-                max_blocks=1,
-            ),
-        )
+        ThreadPoolExecutor(
+            max_threads=1,
+            label='local_threads'),
     ],
     strategy=None,
 
