Merge pull request #30 from schastel/wfcommons-wfc-1.3_to_dask

Merged/Added wfc-1.3_to_dask

Author: rafaelfsilva

.gitignore

@@ -43,3 +43,9 @@ venv
 
 # Workflow Schema
 workflow-schema.json
+
+# wfc-1.3_to_dask specifics
+/wfcommons/wfc-1.3_to_dask/__pycache__/
+/wfcommons/wfc-1.3_to_dask/.coverage
+/wfcommons/wfc-1.3_to_dask/out
+/wfcommons/wfc-1.3_to_dask/htmlcov

wfcommons/wfc-1.3_to_dask/README.md

@@ -0,0 +1,58 @@
+# Objective
+wfc2dask.py is a tool that attempts to translate workflows specified using the WFCommons workflow format (version 1.3; 
+see https://wfcommons.org/format or https://github.com/wfcommons/wfformat) to a workflow actionnable through dask
+
+# TL;DR
+* You need to install dask (see https://docs.dask.org/en/stable/install.html that will tell you to run `conda install
+dask` or `pip install 'dask[distributed]'`).  //TODO Check if this works if dask is not installed//
+* `python wfc2dask.py -h` to get help
+* `python wfc2dask.py <workflow_filename>` creates a directory (named out by default) where the Python code needed
+  to execute the workflow is stored
+* `cd out` and then `python application.py` to run the workflow in a local dask
+
+# The Gory Details
+## Configuration
+* The dask client is defined in `/out/dask_client.py`. Feel free to modify to your local configuration.
+* The workflow tasks and dask DAG are defined in `/out/run_workflow.py`. Feel free to modify as well but take into 
+  account the comments of the `__doc__` of that file
+* That's it for the configuration. You don't need to change the other files
+
+## Implementation
+`/wfc2dask.py` contains the main. The WFCommons JSON document is digested using `/wfc2dask/wfctask.py` and its tasks 
+are grouped in a `WFDag` defined in `/wfc2dask/wfdag.py`. Once all tasks have been ingested, the DAG is built and Python
+code is written to the output directory.
+
+## Contents of the 'samples' directory 
+### Contents of the 'samples/unittests' directory
+
+Files necessary for the proper execution of unit tests
+
+* hello-world-sequence.json:
+The first task (hello) creates a file whose contents are used in the second task (world). 
+The DAG is the following:
+```
+(init) -> hello -> world -> (end)
+```
+
+* hello-world-join.json:
+Three tasks in this workflow. The 'hello' and 'world' tasks each create a different file. Those files
+are then used in the third 'join' task
+The DAG is the following:
+```
+      /-> hello -\
+(init)            -> join -> (end)
+      \-> world -/
+```
+
+## Unit tests/Coverage
+### Unit tests
+```commandline
+python -m unittest discover -s tests
+````
+
+## Coverage
+```
+conda install coverage  # or pip install coverage
+coverage run --source=. --omit='code_templates/*' -m unittest discover -s tests/ && coverage html
+```
+Open htmlcov/index.html

wfcommons/wfc-1.3_to_dask/code_templates/__init__.py

@@ -0,0 +1,3 @@
+"""
+Looks like this is required
+"""

wfcommons/wfc-1.3_to_dask/code_templates/application.py

@@ -0,0 +1,32 @@
+"""
+You don't need to modify this unless you know what you're doing
+
+This is what you need to run with Python to execute the dask workflow
+application, main, driver, or custom name, feel free to rename this file.
+"""
+from dask_client import build_dask_client
+from run_workflow import run_workflow
+import json
+
+
+def to_json(obj):
+    return json.dumps(obj, indent=2, default=lambda o: o.__dict__)
+
+
+def process_arguments():
+    import argparse
+    import sys
+    parser = argparse.ArgumentParser(prog=sys.argv[0],
+                                     description='Runs a workflow through dask')  # TODO
+    parser.add_argument("-nosim", "--do-not-simulate",
+                        help="Do not simulate all tasks (default: do simulate all tasks)", action="store_false")
+    parser.add_argument("-s", "--seed", help="Randomizer seed (used when simulating)")
+    return parser.parse_args()
+
+
+if __name__ == '__main__':
+    args = process_arguments()
+    with build_dask_client() as client:
+        tasks = run_workflow(client, args.do_not_simulate, seed=int(args.seed))
+    with open("run.json", "w") as fp:
+        fp.write(to_json(tasks))

wfcommons/wfc-1.3_to_dask/code_templates/dask_client.py

@@ -0,0 +1,14 @@
+"""
+Feel free to modify this to target your local dask configuration
+
+Lots of info there:
+https://docs.dask.org/en/stable/configuration.html
+https://dask.pydata.org/en/latest/scheduling.html
+"""
+from dask.distributed import Client
+
+
+def build_dask_client():
+    cpu_count = 8        # default is 4 for me
+    threads_per_cpu = 4  # default value is 4 for me
+    return Client(n_workers=cpu_count, threads_per_worker=threads_per_cpu)

wfcommons/wfc-1.3_to_dask/code_templates/helpers.py

@@ -0,0 +1,38 @@
+"""
+This is the method that should allow the execution of a task
+
+"""
+# Those imports are required when pretending to run the commands
+import os
+import pathlib
+import time
+import logging
+from workflow_task import WorkflowTask
+
+
+logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+
+def execute_task(task: WorkflowTask, fut_inputs_list) -> WorkflowTask:
+    """
+    :param task: The task to be executed (it holds all relevant information)
+    :param fut_inputs_list: Unused here but necessary for dask to build its own DAG
+    :return:
+    """
+    logger.info("Executing task %s/%s: %s / in=%s / out=%s" % (task.name, task.dag_id, task.command_arguments, task.inputs, task.outputs))
+    start = time.time()
+    if task.simulate or task.command_arguments is None or len(task.command_arguments) == 0:
+        logger.info("Simulating execution of task %s" % task.name)
+        # Pretend we do something/Wait some time
+        task.simulate_execution()
+        for output in task.outputs:
+            logger.debug("Simulating %s => %s" % (task.command_arguments, output))
+            pathlib.Path(output).touch()
+    else:
+        command = " ".join(task.command_arguments)
+        logger.info("Running command for task %s/%s: %s" % (task.name, task.dag_id, command))
+        os.system(command)  # TODO Use subprocess?
+    task.execution_time = time.time()-start
+    logger.info("End of task %s/%s (%f)" % (task.name, task.dag_id, task.execution_time))
+    return task

wfcommons/wfc-1.3_to_dask/code_templates/run_workflow.py

@@ -0,0 +1,17 @@
+"""
+The contents of this file have been largely generated
+
+You may want to modify this to tune to your taste.
+It consists of two parts:
++ The first part is the workflow tasks definition. Feel free to change the values if you need
++ The second part is the dask code to execute the workflow. I wouldn't touch it too much if I were you (it's likely
+better to modify the WFCommons JSON workflow definition in my opinion but you hold the chainsaw eventually).
+"""
+from helpers import execute_task
+import random
+from workflow_task import WorkflowTask
+
+
+def run_workflow(client, simulate: bool, seed: int=42) -> list[WorkflowTask]:
+# Generated code goes here
+    return TASKS

wfcommons/wfc-1.3_to_dask/code_templates/workflow_task.py

@@ -0,0 +1,52 @@
+"""
+
+A class used to define and help execute a task in a dask workflow
+
+You don't need to modify this unless you know what you're doing
+
+TODO The JSON should actually be a WFCTask
+"""
+import random
+import time
+
+
+class WorkflowTask:
+    def __init__(self,
+                 dag_id: str = None,
+                 name: str = None,
+                 command_arguments: list[str] = None,
+                 inputs: list[str] = None,
+                 outputs: list[str] = None,
+                 simulate: bool = False,
+                 randomizer: random.Random = random.Random(),
+                 simulate_minimum_execution_time: float = 0.1,
+                 simulate_maximum_execution_time: float = 1.1,
+                 execution_time: float = None,  # This is an execution output
+                 ):
+        self.dag_id = dag_id
+        self.name = name
+        self.command_arguments = command_arguments
+        self.inputs = inputs
+        self.outputs = outputs
+        self.simulate = simulate
+        self.randomizer = randomizer
+        self.simulate_minimum_execution_time = simulate_minimum_execution_time
+        self.simulate_maximum_execution_time = simulate_maximum_execution_time
+        self.execution_time = execution_time
+
+    def simulate_execution(self):
+        time.sleep(self.randomizer.uniform(self.simulate_minimum_execution_time,
+                                           self.simulate_maximum_execution_time))
+
+    def pythonize(self, randomizer_varname: str = "randomizer") -> list[str]:
+        codelines = ["WorkflowTask(dag_id = '%s'," % self.dag_id,
+                     "             name = '%s'," % self.name,
+                     "             command_arguments = %s," % self.command_arguments,
+                     "             inputs = %s," % self.inputs,
+                     "             outputs = %s," % self.outputs,
+                     "             simulate = simulate,",
+                     "             randomizer = %s," % randomizer_varname,
+                     "             simulate_minimum_execution_time = %s," % self.simulate_minimum_execution_time,
+                     "             simulate_maximum_execution_time = %s," % self.simulate_maximum_execution_time,
+                     "             )"]
+        return codelines

wfcommons/wfc-1.3_to_dask/samples/unittests/hello-world-join.json

@@ -0,0 +1,43 @@
+{
+  "name": "hello-world-sequence",
+  "description": "See README.md",
+  "workflow": {
+    "tasks": [
+      {
+        "name": "hello",
+        "files": [
+            {
+                "link": "output",
+                "name": "hello.dat"
+            }
+        ]
+      },
+      {
+        "name": "world",
+        "files": [
+            {
+                "link": "output",
+                "name": "world.dat"
+            }
+        ]
+      },
+      {
+        "name": "hello-world",
+        "files": [
+            {
+                "link": "input",
+                "name": "hello.dat"
+            },
+            {
+                "link": "input",
+                "name": "world.dat"
+            },
+          {
+            "link": "output",
+                "name": "helloworld.dat"
+          }
+        ]
+      }
+    ]
+  }
+}

wfcommons/wfc-1.3_to_dask/samples/unittests/hello-world-sequence.json

@@ -0,0 +1,31 @@
+{
+  "name": "hello-world-sequence",
+  "description": "See README.md",
+  "workflow": {
+    "tasks": [
+      {
+        "name": "hello",
+        "files": [
+            {
+                "link": "output",
+                "name": "hello.dat"
+            }
+        ]
+      },
+      {
+        "name": "world",
+        "parents": ["hello"],
+        "files": [
+          {
+                "link": "input",
+                "name": "hello.dat"
+            },
+            {
+                "link": "output",
+                "name": "world.dat"
+            }
+        ]
+      }
+    ]
+  }
+}