Airflow translator testing

Dockerfile updates
setup.py fix

Author: henricasanova

setup.py

@@ -9,6 +9,11 @@
 # (at your option) any later version.
 
 import sys
+import os
+import stat
+import shutil
+import site
+import sysconfig
 import subprocess
 
 from setuptools import setup, find_packages
@@ -25,16 +30,28 @@ def run(self):
             sys.exit(-1) 
         super().run()
 
+        # Do a by-hand copy of cpu-benchmark to the bin directory, as this is so 
+        # hard to do using data_file in the setup() declaration
+        scripts_dir = os.path.join(site.USER_BASE, "bin")
+        source_path = os.path.join("bin", "cpu-benchmark")
+        target_path = os.path.join(scripts_dir, "cpu-benchmark")
+        # Ensure it's executable (just in case)
+        st = os.stat(source_path)
+        os.chmod(source_path, st.st_mode | stat.S_IEXEC)
+        # Copy to scripts directory
+        shutil.copy2(source_path, target_path)
+
 setup(
     packages=find_packages(),
     include_package_data=True,
     has_ext_modules=lambda: True,
     cmdclass={
         'build_ext': Build,
     },
-    data_files=[
-        ('bin', ['bin/cpu-benchmark', 'bin/wfbench'])
-    ],
+    #data_files=[
+    #    (user_bin_dir, ['bin/cpu-benchmark'])
+    #],
+    scripts=['bin/wfbench'],
     entry_points={
         'console_scripts': [
             'wfchef=wfcommons.wfchef.chef:main'

tests/translators/Dockerfile.airflow

@@ -34,7 +34,7 @@ RUN update-alternatives --install /usr/bin/python python /usr/bin/python3 1
 RUN apt-get -y install stress-ng
 
 # WfCommons
-RUN python3 -m pip install --break-system-packages wfcommons
+#RUN python3 -m pip install --break-system-packages wfcommons
 
 # Install Airflow
 RUN python3 -m pip install --break-system-packages apache-airflow==2.10.2 --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.10.2/constraints-3.12.txt"
@@ -48,8 +48,11 @@ RUN python3 -m pip install --break-system-packages mysqlclient
 # Setup directory
 RUN mkdir /home/wfcommons
 
-# Create an entrypoint script to start mysqld in the background
-# and setup the Airflow DB
+# Create an entrypoint script to start mysqld in the background and setup the Airflow DB
+# This script is then used as the entry points of this Container
+# Create ALSO a script to run a benchmark (only the last line differ), which is used for testing,
+# because using docker-py, things get complicated with different bash sessions!
+
 RUN echo '#!/bin/bash' > /entrypoint.sh && \
     echo 'mysqld --explicit-defaults-for-timestamp &' >> /entrypoint.sh && \
     echo 'until mysqladmin ping -h localhost --silent; do' >> /entrypoint.sh && \
@@ -66,10 +69,16 @@ RUN echo '#!/bin/bash' > /entrypoint.sh && \
     echo "sed -i ./airflow/airflow.cfg -e 's/sqlite:.*/mysql+mysqldb:\/\/airflow_user:airflow_pass@localhost:3306\/airflow_db/'" >> /entrypoint.sh && \
     echo 'airflow db migrate' >> /entrypoint.sh && \
     echo 'echo "Airflow database setup!"' >> /entrypoint.sh && \
-    echo 'mkdir ./airflow/dags' >> /entrypoint.sh && \
-    echo 'exec bash' >> /entrypoint.sh && \
+    echo 'mkdir ./airflow/dags' >> /entrypoint.sh
+
+RUN  cp /entrypoint.sh /run_a_workflow.sh
+
+RUN echo 'exec bash' >> /entrypoint.sh && \
     chmod +x /entrypoint.sh
 
+RUN echo 'airflow dags test $1' >> /run_a_workflow.sh && \
+    chmod +x /run_a_workflow.sh
+
 WORKDIR /home/wfcommons
 
 ENTRYPOINT ["/entrypoint.sh"]

tests/translators/README.airflow

@@ -58,7 +58,7 @@ A much simpler alternative is to use Docker.
 2. Run the docker container in the directory to contains the translated
    workflow (see last section below)
 
-   docker run -it --rm -v .:/home/wfcommons/mount wfcommons-dev /bin/bash
+   docker run -it --rm -v .:/home/wfcommons/translated_workflow wfcommons-dev /bin/bash
 
 
 Running a translated workflow with Airflow
@@ -83,11 +83,12 @@ translator = AirflowTranslator(benchmark.workflow)
 translator.translate(output_folder=pathlib.Path("/tmp/translated_workflow/"))
 ```
 
-The above will create a JSON worfklow file in /tmp/blast-benchmark-45.json.
+The above will create a JSON workflow file in /tmp/blast-benchmark-45.json.
 In that file, the workflow name (this is used below) is set to "Blast-Benchmark".
 
 The above will also create the translated workflow the
-/tmp/translated_workflow/ directory. Some directories and files need to be copied/moved as follows:
+/tmp/translated_workflow/ directory. Some directories and files need to be copied/moved as follows.
+As
 
   cp -r /tmp/translated_workflow/ $AIRFLOW_HOME/dags/
   mv $AIRFLOW_HOME/dags/translated_workflow/workflow.py $AIRFLOW_HOME/dags/
@@ -97,4 +98,6 @@ Finally, run the workflow as:
   airflow dags test Blast-Benchmark    (note the "Blast-Benchmark" workflow name from above)
 
 
+Note that if you're using docker, the above two commands need to be done on the container,
+with the right paths (i.e., /home/wfcommons/translated_workflow instead of /tmp/translated_workflow/)
 

tests/translators/test_translators.py

@@ -26,7 +26,7 @@
 from wfcommons.wfbench import AirflowTranslator
 
 
-def start_docker_container(backend, working_dir):
+def start_docker_container(backend, mounted_dir, working_dir, command = ["sleep", "infinity"]):
     # Pulling the Docker image
     client = docker.from_env()
     sys.stderr.write("Pulling Docker image...\n")
@@ -43,8 +43,8 @@ def start_docker_container(backend, working_dir):
     sys.stderr.write("Starting Docker container...\n")
     container = client.containers.run(
         image_name,
-        "sleep infinity",
-        volumes={working_dir: {'bind': working_dir, 'mode': 'rw'}},
+        command=command,
+        volumes={mounted_dir: {'bind': mounted_dir, 'mode': 'rw'}},
         working_dir=working_dir,
         tty=True,
         detach=True
@@ -91,7 +91,7 @@ def create_workflow_benchmark():
 class TestTranslators:
 
     @pytest.mark.unit
-    @pytest.mark.skip(reason="tmp")
+    # @pytest.mark.skip(reason="tmp")
     def test_dask_translator(self) -> None:
 
         # Create workflow benchmark
@@ -109,12 +109,12 @@ def test_dask_translator(self) -> None:
         translator.translate(output_folder=dirpath)
 
         # Pulling the Docker image
-        container = start_docker_container("dask", str_dirpath)
+        container = start_docker_container("dask", str_dirpath, str_dirpath)
 
         # Installing WfCommons on container
         install_WfCommons_on_container(container)
 
-        # Copy over the wfbench and cpu-benchmark executables to where they should go
+        # Copy over the wfbench and cpu-benchmark executables to where they should go on the container
         exit_code, output = container.exec_run("sudo cp -f /tmp/WfCommons/bin/wfbench " + str_dirpath + "bin/", stdout=True, stderr=True)
         exit_code, output = container.exec_run("sudo cp -f /tmp/WfCommons/bin/cpu-benchmark " + str_dirpath + "bin/", stdout=True, stderr=True)
 
@@ -134,7 +134,7 @@ def test_dask_translator(self) -> None:
 
 
     @pytest.mark.unit
-    @pytest.mark.skip(reason="tmp")
+    # @pytest.mark.skip(reason="tmp")
     def test_parsl_translator(self) -> None:
 
         # Create workflow benchmark
@@ -151,8 +151,8 @@ def test_parsl_translator(self) -> None:
         translator = ParslTranslator(benchmark.workflow)
         translator.translate(output_folder=dirpath)
 
-        # Pulling the Docker image
-        container = start_docker_container("parsl", str_dirpath)
+        # Starting the Docker container
+        container = start_docker_container("parsl", str_dirpath, str_dirpath)
 
         # Installing WfCommons on container
         install_WfCommons_on_container(container)
@@ -178,7 +178,7 @@ def test_parsl_translator(self) -> None:
         assert(num_completed_tasks == num_tasks)
 
     @pytest.mark.unit
-    @pytest.mark.skip(reason="tmp")
+    # @pytest.mark.skip(reason="tmp")
     def test_nextflow_translator(self) -> None:
 
         # Create workflow benchmark
@@ -195,8 +195,8 @@ def test_nextflow_translator(self) -> None:
         translator = NextflowTranslator(benchmark.workflow)
         translator.translate(output_folder=dirpath)
 
-        # Pulling the Docker image
-        container = start_docker_container("nextflow", str_dirpath)
+        # Starting the Docker container
+        container = start_docker_container("nextflow", str_dirpath, str_dirpath)
 
         # Installing WfCommons on container
         install_WfCommons_on_container(container)
@@ -222,7 +222,7 @@ def test_nextflow_translator(self) -> None:
 
 
     @pytest.mark.unit
-    @pytest.mark.skip(reason="tmp")
+    # @pytest.mark.skip(reason="tmp")
     def test_airflow_translator(self) -> None:
 
         # Create workflow benchmark
@@ -239,27 +239,37 @@ def test_airflow_translator(self) -> None:
         translator = AirflowTranslator(benchmark.workflow)
         translator.translate(output_folder=dirpath)
 
-        # Pulling the Docker image
-        container = start_docker_container("airflow", str_dirpath)
+        # Starting the Docker container
+        container = start_docker_container("airflow", str_dirpath, "/home/wfcommons/", command=None)
+        # container = start_docker_container("airflow", str_dirpath, "/home/wfcommons/")
 
         # Installing WfCommons on container
         install_WfCommons_on_container(container)
 
+
         # Copy over the wfbench and cpu-benchmark executables to where they should go
-        # exit_code, output = container.exec_run("sudo cp -f /tmp/WfCommons/bin/wfbench " + str_dirpath + "bin/",
-        #                                        stdout=True, stderr=True)
-        # exit_code, output = container.exec_run("sudo cp -f /tmp/WfCommons/bin/cpu-benchmark " + str_dirpath + "bin/",
-        #                                        stdout=True, stderr=True)
-        #
-        # # Run the workflow!
-        # sys.stderr.write("Running the Airflow workflow on the container...\n")
-        # 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)
-        #
-        # # Kill the container
-        # container.remove(force=True)
-        #
-        # # Do sanity checks
-        # sys.stderr.write("Checking sanity...\n")
-        # assert (exit_code == 0)
-        # assert (task_exit_codes.decode() == num_tasks * "0")
+        exit_code, output = container.exec_run("sudo cp -f /tmp/WfCommons/bin/wfbench /usr/local/bin/",
+                                               stdout=True, stderr=True)
+        exit_code, output = container.exec_run("sudo cp -f /tmp/WfCommons/bin/cpu-benchmark /usr/local/bin/",
+                                               stdout=True, stderr=True)
+
+        # Do the necessary copies (some ugly hardcoded stuff here)
+        airflow_home = "/home/wfcommons/airflow/"  # per the Dockerfile
+        exit_code, output = container.exec_run(f"cp -r /tmp/airflow_translated_workflow {airflow_home}/dags/",stdout=True, stderr=True)
+        exit_code, output = container.exec_run(f"mv {airflow_home}/dags/airflow_translated_workflow/workflow.py {airflow_home}/dags/",stdout=True, stderr=True)
+
+        # # Run the entry point script by hand (such a hack!)
+        # exit_code, output = container.exec_run(f"/bin/bash /entrypoint.sh",stdout=True, stderr=True)
+        # print(output)
+
+        # Run the workflow!
+        sys.stderr.write("Running the Airflow workflow on the container...\n")
+        exit_code, output = container.exec_run(cmd="/bin/bash /run_a_workflow.sh Blast-Benchmark", stdout=True, stderr=True)
+
+        # Kill the container
+        container.remove(force=True)
+
+        # Do sanity checks
+        sys.stderr.write("Checking sanity...\n")
+        assert (exit_code == 0)
+        assert (output.decode().count("completed") == num_tasks * 2)

wfcommons/wfbench/translator/abstract_translator.py

@@ -11,6 +11,7 @@
 import logging
 import os
 import sys
+import site
 import pathlib
 import shutil
 import textwrap
@@ -92,7 +93,8 @@ def _copy_binary_files(self, output_folder: pathlib.Path) -> None:
         bin_folder = output_folder.joinpath("bin")
         bin_folder.mkdir(exist_ok=True)
         # Get the python executable dir
-        python_executable_dir = os.path.dirname(sys.executable)
+        # python_executable_dir = os.path.dirname(sys.executable)
+        python_executable_dir = os.path.join(site.getuserbase(), 'bin')
         # shutil.copy(shutil.which("wfbench"), bin_folder)
         shutil.copy(python_executable_dir + "/wfbench", bin_folder)
         # shutil.copy(shutil.which("cpu-benchmark"), bin_folder)