Cleanup job for GEFS (#2919)

As forecast ensemble jobs are added to the global workflow, this PR ensures the output is being cleaned up properly once it is no longer needed. Resolves #833
NOAA-EMC · Sep 14, 2024 · b7f71e0 · b7f71e0
1 parent a13bbef
commit b7f71e0
Show file tree

Hide file tree

Showing 4 changed files with 38 additions and 2 deletions.
diff --git a/parm/config/gefs/config.cleanup b/parm/config/gefs/config.cleanup
@@ -0,0 +1 @@
+../gfs/config.cleanup
diff --git a/parm/config/gefs/config.resources b/parm/config/gefs/config.resources
@@ -295,6 +295,14 @@ case ${step} in
     export threads_per_task=1
     export memory="4096M"
     ;;
+
+  "cleanup")
+    export walltime="00:15:00"
+    export ntasks=1
+    export tasks_per_node=1
+    export threads_per_task=1
+    export memory="4096M"
+    ;;
   *)
     echo "FATAL ERROR: Invalid job ${step} passed to ${BASH_SOURCE[0]}"
     exit 1

diff --git a/workflow/applications/gefs.py b/workflow/applications/gefs.py
@@ -17,7 +17,7 @@ def _get_app_configs(self):
         """
         Returns the config_files that are involved in gefs
         """
-        configs = ['stage_ic', 'fcst', 'atmos_products', 'arch']
+        configs = ['stage_ic', 'fcst', 'atmos_products', 'arch', 'cleanup']
 
         if self.nens > 0:
             configs += ['efcs', 'atmos_ensstat']
@@ -82,6 +82,6 @@ def get_task_names(self):
         if self.do_extractvars:
             tasks += ['extractvars']
 
-        tasks += ['arch']
+        tasks += ['arch', 'cleanup']
 
         return {f"{self.run}": tasks}
diff --git a/workflow/rocoto/gefs_tasks.py b/workflow/rocoto/gefs_tasks.py
@@ -546,6 +546,9 @@ def arch(self):
                 deps.append(rocoto.add_dependency(dep_dict))
                 dep_dict = {'type': 'metatask', 'name': 'wave_post_bndpnt_bull'}
                 deps.append(rocoto.add_dependency(dep_dict))
+        if self.app_config.do_extractvars:
+            dep_dict = {'type': 'metatask', 'name': 'extractvars'}
+            deps.append(rocoto.add_dependency(dep_dict))
         dependencies = rocoto.create_dependency(dep=deps, dep_condition='and')
 
         resources = self.get_resource('arch')
@@ -564,3 +567,27 @@ def arch(self):
         task = rocoto.create_task(task_dict)
 
         return task
+
+    def cleanup(self):
+        deps = []
+        dep_dict = {'type': 'task', 'name': 'arch'}
+        deps.append(rocoto.add_dependency(dep_dict))
+
+        dependencies = rocoto.create_dependency(dep=deps)
+
+        resources = self.get_resource('cleanup')
+        task_name = 'cleanup'
+        task_dict = {'task_name': task_name,
+                     'resources': resources,
+                     'envars': self.envars,
+                     'cycledef': 'gefs',
+                     'dependency': dependencies,
+                     'command': f'{self.HOMEgfs}/jobs/rocoto/cleanup.sh',
+                     'job_name': f'{self.pslot}_{task_name}_@H',
+                     'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log',
+                     'maxtries': '&MAXTRIES;'
+                     }
+
+        task = rocoto.create_task(task_dict)
+
+        return task