Issue735/clear workflow

README.rst

@@ -60,6 +60,7 @@ Contributors:
 * Quincy Wofford - `qwofford <https://github.com/qwofford>`_
 * Tim Randles - `trandles-lanl <https://github.com/trandles-lanl>`_
 * Jacob Tronge - `jtronge <https://github.com/jtronge>`_
+* Kabir Vats - `kabir-vats <https://github.com/kabir-vats>`_
 
 Concept and Design Contributors
 

beeflow/common/config_driver.py

@@ -235,6 +235,8 @@ def filepath_completion_input(*pargs, **kwargs):
                  info='backend workload scheduler to interact with ')
 VALIDATOR.option('DEFAULT', 'use_archive', validator=validation.bool_, default=True,
                  info='use the BEE archiving functinality')
+VALIDATOR.option('DEFAULT', 'delete_completed_workflow_dirs', validator=validation.bool_,
+                 default=True, info='delete workflow directory for completed jobs')
 VALIDATOR.option('DEFAULT', 'neo4j_image', validator=validation.file_,
                  info='neo4j container image',
                  input_fn=filepath_completion_input)

beeflow/wf_manager/common/dep_manager.py

@@ -225,3 +225,5 @@ def kill_gdb(pid):
         os.kill(pid, signal.SIGTERM)
     except OSError:
         dep_log.info('Process already killed')
+        return False
+    return True

beeflow/wf_manager/resources/wf_update.py

@@ -15,7 +15,7 @@
 
 from beeflow.common.db import wfm_db
 from beeflow.common.db.bdb import connect_db
-
+from beeflow.common.config_driver import BeeConfig as bc
 
 log = bee_logging.setup(__name__)
 db_path = wf_utils.get_db_path()
@@ -39,13 +39,21 @@ def archive_workflow(db, wf_id, final_state=None):
     # We use tar directly since tarfile is apparently very slow
     workflows_dir = wf_utils.get_workflows_dir()
     subprocess.call(['tar', '-czf', archive_path, wf_id], cwd=workflows_dir)
+    pid = db.workflows.get_gdb_pid(wf_id)
+    # Wait for Graph database to be down (max 10 seconds)
+    for _ in range(10):
+        if not dep_manager.kill_gdb(pid):
+            break
+        time.sleep(1)
+    remove_wf_dir = bc.get('DEFAULT', 'delete_completed_workflow_dirs')
+    if remove_wf_dir:
+        log.info('Removing Workflow Directory')
+        wf_utils.remove_wf_dir(wf_id)
 
 
 def archive_fail_workflow(db, wf_id):
     """Archive and fail a workflow."""
     archive_workflow(db, wf_id, final_state='Failed')
-    pid = db.workflows.get_gdb_pid(wf_id)
-    dep_manager.kill_gdb(pid)
 
 
 def set_dependent_tasks_dep_fail(db, wfi, wf_id, task):
@@ -132,11 +140,9 @@ def handle_state_change(self, state_update, task, wfi, db):
                 wf_utils.schedule_submit_tasks(state_update.wf_id, tasks)
 
             if wfi.workflow_completed():
-                log.info("Workflow Completed")
                 wf_id = wfi.workflow_id
                 archive_workflow(db, state_update.wf_id)
-                pid = db.workflows.get_gdb_pid(state_update.wf_id)
-                dep_manager.kill_gdb(pid)
+                log.info('Workflow Completed')
 
         # If the job failed and it doesn't include a checkpoint-restart hint,
         # then fail the entire workflow