Merge branch 'main' into stress-ng_cpu_benchmark

Author: henricasanova

.github/workflows/build.yml

@@ -32,6 +32,7 @@ jobs:
           pip install docker
           pip install pygraphviz
           pip install pydot
+          pip install igraph
 
       - name: Check package install
         run: |

tests/test_helpers.py

@@ -143,7 +143,24 @@ def _compare_workflows(workflow_1: Workflow, workflow_2: Workflow):
     # Test the number of tasks
     assert (len(workflow_1.tasks) == len(workflow_2.tasks))
     # Test the task graph topology
-    assert (networkx.is_isomorphic(workflow_1, workflow_2))
+
+    # Slow (too slow for some DAGs) isomorphic check with networkx
+    # assert (networkx.is_isomorphic(workflow_1, workflow_2))
+
+    # Fast isomorphic check using igraph
+    import igraph as ig
+
+    def nx_to_igraph(G):
+        g = ig.Graph(directed=True)
+        g.add_vertices(list(G.nodes()))
+        g.add_edges(list(G.edges()))
+        return g
+
+    g1 = nx_to_igraph(workflow_1)
+    g2 = nx_to_igraph(workflow_2)
+
+    assert(g1.isomorphic(g2))
+
     # Test the total file size sum
     workflow1_input_bytes, workflow2_input_bytes = 0, 0
     workflow1_output_bytes, workflow2_output_bytes = 0, 0

tests/translators_loggers/Dockerfile.snakemake

@@ -0,0 +1,61 @@
+# docker build --platform amd64 -t wfcommons-dev-snakemake -f Dockerfile.snakemake .
+# docker run -it --rm -v `pwd`:/home/wfcommons wfcommons-dev-snakemake /bin/bash
+
+FROM amd64/ubuntu:noble
+
+LABEL org.containers.image.authors="henric@hawaii.edu"
+
+# update repositories
+RUN apt-get update
+
+# set timezone
+RUN echo "America/Los_Angeles" > /etc/timezone && export DEBIAN_FRONTEND=noninteractive && apt-get install -y tzdata
+
+# install useful stuff
+RUN apt-get -y install pkg-config 
+RUN apt-get -y install git
+RUN apt-get -y install wget 
+RUN apt-get -y install curl 
+RUN apt-get -y install make 
+RUN apt-get -y install cmake 
+RUN apt-get -y install cmake-data 
+RUN apt-get -y install sudo 
+RUN apt-get -y install vim --fix-missing
+RUN apt-get -y install gcc 
+RUN apt-get -y install gcc-multilib
+RUN apt-get -y install graphviz libgraphviz-dev
+
+
+# Python stuff
+RUN apt-get -y install python3 python3-pip
+RUN update-alternatives --install /usr/bin/python python /usr/bin/python3 1
+RUN python3 -m pip install --break-system-packages pathos pandas filelock
+RUN python3 -m pip install --break-system-packages networkx scipy matplotlib pygraphviz
+RUN python3 -m pip install --break-system-packages pyyaml jsonschema requests
+RUN python3 -m pip install --break-system-packages --upgrade setuptools
+
+# Stress-ng
+RUN apt-get -y install stress-ng
+
+# Add wfcommons user
+RUN useradd -ms /bin/bash wfcommons
+RUN adduser wfcommons sudo
+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 Pixi
+RUN wget -qO- https://pixi.sh/install.sh | sh
+ENV PATH="$PATH:/home/wfcommons/.pixi/bin"
+
+# Install snakemake
+RUN pixi global install snakemake conda -c conda-forge -c bioconda
+RUN pixi global install snakedeploy -c conda-forge -c bioconda
+RUN ~/.pixi/envs/snakemake/bin/python -m ensurepip && \
+    ~/.pixi/envs/snakemake/bin/python -m pip install snakemake-logger-plugin-snkmt
+

tests/translators_loggers/test_translators_loggers.py

@@ -24,7 +24,8 @@
 from tests.test_helpers import _shutdown_docker_container_and_remove_image
 from tests.test_helpers import _compare_workflows
 
-from wfcommons import BlastRecipe
+from wfcommons import BlastRecipe, EpigenomicsRecipe, BwaRecipe, CyclesRecipe, GenomeRecipe, MontageRecipe, \
+    RnaseqRecipe, SeismologyRecipe, SoykbRecipe, SrasearchRecipe
 from wfcommons.common import Workflow, Task
 from wfcommons.wfbench import WorkflowBenchmark
 from wfcommons.wfbench import DaskTranslator
@@ -34,6 +35,7 @@
 from wfcommons.wfbench import BashTranslator
 from wfcommons.wfbench import TaskVineTranslator
 from wfcommons.wfbench import MakeflowTranslator
+from wfcommons.wfbench import SnakemakeTranslator
 from wfcommons.wfbench import CWLTranslator
 from wfcommons.wfbench import StreamflowTranslator
 from wfcommons.wfbench import PegasusTranslator
@@ -43,16 +45,35 @@
 from wfcommons.wfinstances.logs import TaskVineLogsParser
 from wfcommons.wfinstances.logs import MakeflowLogsParser
 from wfcommons.wfinstances.logs import ROCrateLogsParser
+from wfcommons.wfinstances.logs import SnakemakeLogsParser
 
 
 def _create_workflow_benchmark() -> (WorkflowBenchmark, int):
     # Create a workflow benchmark object to generate specifications based on a recipe (in /tmp/, whatever)
     desired_num_tasks = 45
-    benchmark_full_path = "/tmp/blast-benchmark-{desired_num_tasks}.json"
+    benchmark_full_path = f"/tmp/blast-benchmark-{desired_num_tasks}.json"
+    # benchmark_full_path = f"/tmp/epigenomics-benchmark-{desired_num_tasks}.json"
+    # benchmark_full_path = f"/tmp/bwa-benchmark-{desired_num_tasks}.json"
+    # benchmark_full_path = f"/tmp/cycles-benchmark-{desired_num_tasks}.json"
+    # benchmark_full_path = f"/tmp/genome-benchmark-{desired_num_tasks}.json"
+    # benchmark_full_path = f"/tmp/montage-benchmark-{desired_num_tasks}.json"
+    # benchmark_full_path = f"/tmp/rnaseq-benchmark-{desired_num_tasks}.json"
+    # benchmark_full_path = f"/tmp/seismology-benchmark-{desired_num_tasks}.json"
+    # benchmark_full_path = f"/tmp/soykb-benchmark-{desired_num_tasks}.json"
+    # benchmark_full_path = f"/tmp/srasearch-benchmark-{desired_num_tasks}.json"
     shutil.rmtree(benchmark_full_path, ignore_errors=True)
     benchmark = WorkflowBenchmark(recipe=BlastRecipe, num_tasks=desired_num_tasks)
+    # benchmark = WorkflowBenchmark(recipe=EpigenomicsRecipe, num_tasks=desired_num_tasks)
+    # benchmark = WorkflowBenchmark(recipe=BwaRecipe, num_tasks=desired_num_tasks)
+    # benchmark = WorkflowBenchmark(recipe=CyclesRecipe, num_tasks=desired_num_tasks)
+    # benchmark = WorkflowBenchmark(recipe=GenomeRecipe, num_tasks=desired_num_tasks)
+    # benchmark = WorkflowBenchmark(recipe=MontageRecipe, num_tasks=desired_num_tasks)
+    # benchmark = WorkflowBenchmark(recipe=RnaseqRecipe, num_tasks=desired_num_tasks)
+    # benchmark = WorkflowBenchmark(recipe=SeismologyRecipe, num_tasks=desired_num_tasks)
+    # benchmark = WorkflowBenchmark(recipe=SoykbRecipe, num_tasks=desired_num_tasks)
+    # benchmark = WorkflowBenchmark(recipe=SrasearchRecipe, num_tasks=desired_num_tasks)
     benchmark.create_benchmark(pathlib.Path("/tmp/"), cpu_work=10, data=10, percent_cpu=0.6)
-    with open(f"/tmp/blast-benchmark-{desired_num_tasks}.json", "r") as f:
+    with open(benchmark_full_path, "r") as f:
         generated_json = json.load(f)
         num_tasks = len(generated_json["workflow"]["specification"]["tasks"])
     return benchmark, num_tasks
@@ -119,6 +140,7 @@ def _additional_setup_swiftt(container):
     "bash": noop,
     "taskvine": _additional_setup_taskvine,
     "makeflow": noop,
+    "snakemake": noop,
     "cwl": noop,
     "streamflow": noop,
     "pegasus": _additional_setup_pegasus,
@@ -199,6 +221,15 @@ def run_workflow_makeflow(container, num_tasks, str_dirpath):
     num_completed_jobs = len(re.findall(r'job \d+ completed', output.decode()))
     assert (num_completed_jobs == num_tasks)
 
+def run_workflow_snakemake(container, num_tasks, str_dirpath):
+    # Run the workflow (with full logging)
+    exit_code, output = container.exec_run(cmd=["bash", "-c", "snakemake -s ./workflow.smk --cores 1 --logger snkmt --logger-snkmt-db ./snkmt.sqlite"],
+                                           user="wfcommons", stdout=True, stderr=True)
+    # Check sanity
+    assert (exit_code == 0)
+    num_completed_jobs = len(re.findall(r'Finished jobid: \d+', output.decode()))
+    assert (num_completed_jobs - 1 == num_tasks) # Discounting the "all_tasks" rule
+
 def run_workflow_cwl(container, num_tasks, str_dirpath):
     # Run the workflow!
     # Note that the input file is hardcoded and Blast-specific
@@ -214,7 +245,6 @@ def run_workflow_cwl(container, num_tasks, str_dirpath):
 
 def run_workflow_streamflow(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="streamflow run ./streamflow.yml",
                                            user="wfcommons", stdout=True, stderr=True)
     # Check sanity
@@ -264,6 +294,7 @@ def run_workflow_swiftt(container, num_tasks, str_dirpath):
     "bash": run_workflow_bash,
     "taskvine": run_workflow_taskvine,
     "makeflow": run_workflow_makeflow,
+    "snakemake": run_workflow_snakemake,
     "cwl": run_workflow_cwl,
     "streamflow": run_workflow_streamflow,
     "pegasus": run_workflow_pegasus,
@@ -279,6 +310,7 @@ def run_workflow_swiftt(container, num_tasks, str_dirpath):
     "bash": BashTranslator,
     "taskvine": TaskVineTranslator,
     "makeflow": MakeflowTranslator,
+    "snakemake": SnakemakeTranslator,
     "cwl": CWLTranslator,
     "streamflow": StreamflowTranslator,
     "pegasus": PegasusTranslator,
@@ -291,18 +323,19 @@ class TestTranslators:
     @pytest.mark.parametrize(
         "backend",
         [
-          "swiftt",
-          "dask",
-          "parsl",
-          "nextflow",
-          "nextflow_subworkflow",
-          "airflow",
-          "bash",
-          "taskvine",
-          "makeflow",
-          "cwl",
-          "streamflow",
-          "pegasus",
+           "swiftt",
+           "dask",
+           "parsl",
+           "nextflow",
+           "nextflow_subworkflow",
+           "airflow",
+           "bash",
+           "taskvine",
+           "makeflow",
+           "snakemake",
+           "cwl",
+           "streamflow",
+           "pegasus",
         ])
     @pytest.mark.unit
     # @pytest.mark.skip(reason="tmp")
@@ -359,12 +392,13 @@ def test_translator(self, backend) -> None:
                                        steps_to_ignore=["main.cwl#compile_output_files", "main.cwl#compile_log_files"],
                                        file_extensions_to_ignore=[".out", ".err"],
                                        instruments_to_ignore=["shell.cwl"])
+        elif backend == "snakemake":
+            parser = SnakemakeLogsParser(dirpath, snkmt_db=dirpath / "snkmt.sqlite", rules_to_ignore=["all_wfbench_tasks"])
 
         if parser is not None:
             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"))
-
             original_workflow : Workflow = benchmark.workflow
 
             _compare_workflows(original_workflow, reconstructed_workflow)

wfcommons/wfbench/__init__.py

@@ -18,6 +18,7 @@
                          SwiftTTranslator,
                          TaskVineTranslator,
                          MakeflowTranslator,
+                         SnakemakeTranslator,
                          CWLTranslator,
                          StreamflowTranslator,
                          PyCompssTranslator)

wfcommons/wfbench/translator/__init__.py

@@ -20,3 +20,4 @@
 from .swift_t import SwiftTTranslator
 from .taskvine import TaskVineTranslator
 from .makeflow import MakeflowTranslator
+from .snakemake import SnakemakeTranslator

wfcommons/wfbench/translator/airflow.py

@@ -41,8 +41,6 @@ def __init__(self,
         self.script = f"""
 from __future__ import annotations
 
-import os
-from datetime import datetime
 from airflow.models.dag import DAG
 from airflow.operators.bash import BashOperator
 
@@ -64,9 +62,6 @@ def translate(self, output_folder: pathlib.Path, name: Optional[str] = None) ->
 with DAG(
     "{name}",
     description="airflow translation of a wfcommons instance",
-    schedule="0 0 * * *",
-    start_date=datetime(2021, 1, 1),
-    catchup=False,
     tags=["wfcommons"],
 ) as dag:
 """
@@ -77,12 +72,10 @@ def translate(self, output_folder: pathlib.Path, name: Optional[str] = None) ->
             self.script += f"""
     {self._sanitize_varname(task.task_id)} = BashOperator(
         task_id="{task.task_id}",
-        depends_on_past=False,
         bash_command='{self.task_commands[task.task_id]}',
-        env={{"AIRFLOW_HOME": os.environ["AIRFLOW_HOME"]}},
-        retries=3,
         )
 """
+
         for task in self.tasks.values():
             # Comma-separated list of the task's parents
             parents = ", ".join(map(self._sanitize_varname, self.task_parents[task.task_id]))