From d604ea40872261389997678423f91479c2e04cbd Mon Sep 17 00:00:00 2001
From: chausman <chausman@nrao.edu>
Date: Thu, 13 Oct 2022 15:39:53 -0600
Subject: [PATCH] - make workflow setup account for case where previous condor
files and logs exist in processing directory - catch up with 2.5.2 email
change
---
.../workspaces/capability/schema.py | 4 +--
.../workflow/services/workflow_service.py | 27 +++++++++++++++++++
2 files changed, 29 insertions(+), 2 deletions(-)
diff --git a/shared/workspaces/workspaces/capability/schema.py b/shared/workspaces/workspaces/capability/schema.py
index 4622f5ff4..1e9f13402 100644
--- a/shared/workspaces/workspaces/capability/schema.py
+++ b/shared/workspaces/workspaces/capability/schema.py
@@ -197,7 +197,7 @@ class SingleQAStateMachine(StateMachine):
SendMessage(arguments="execution_complete"),
SendNotification(arguments=json.dumps({"template": "workflow-complete", "send_to_qa": True})),
SendNotification(
- arguments=json.dumps({"template": workflow_name + "_complete", "send_to_qa": True})
+ arguments=json.dumps({"template": workflow_name + "_complete", "needs_contacts": True})
),
],
("Error", "ingestion-failed"): [
@@ -272,7 +272,7 @@ class DoubleQAStateMachine(StateMachine):
SendMessage(arguments="execution_complete"),
SendNotification(arguments=json.dumps({"template": "workflow-complete", "send_to_qa": True})),
SendNotification(
- arguments=json.dumps({"template": workflow_name + "_complete", "send_to_qa": True})
+ arguments=json.dumps({"template": workflow_name + "_complete", "needs_contacts": True})
),
],
("Error", "ingestion-failed"): [
diff --git a/shared/workspaces/workspaces/workflow/services/workflow_service.py b/shared/workspaces/workspaces/workflow/services/workflow_service.py
index f4ebaae71..56be6352e 100644
--- a/shared/workspaces/workspaces/workflow/services/workflow_service.py
+++ b/shared/workspaces/workspaces/workflow/services/workflow_service.py
@@ -16,6 +16,8 @@
# You should have received a copy of the GNU General Public License
# along with Workspaces. If not, see <https://www.gnu.org/licenses/>.
""" This is the workflow service. """
+import glob
+import itertools
# pylint: disable=E0401, E0402, W1203
@@ -23,6 +25,7 @@ import json
import logging
import os
import pathlib
+import shutil
import stat
import subprocess
from pathlib import Path
@@ -375,6 +378,7 @@ class WorkflowService(WorkflowServiceIF):
workflow_files = self._determine_usable_files(request, templated_files)
# prepare files for condor execution
+ self._check_for_old_condor_files(temp_folder)
self._prepare_files_for_condor(workflow_files, temp_folder)
self.info.save_request(request)
@@ -544,6 +548,7 @@ class WorkflowService(WorkflowServiceIF):
:param files: a dictionary of filename -> content
:return: a Path
"""
+
# 1. spool each of the temp files to tmp directory
for file in files:
path = temp_folder / file.filename
@@ -555,6 +560,28 @@ class WorkflowService(WorkflowServiceIF):
logger.info("Making %s executable", file)
file.chmod(file.stat().st_mode | stat.S_IEXEC)
+ @staticmethod
+ def _check_for_old_condor_files(dir_path: Path):
+ """
+ It is possible that a follow-on workflow will use an existing, previously used, directory, and this directory
+ might contain previously used condor submit files and their associated output. Move these aside so they don't
+ interfere with the new workflow execution
+ Example: VLASS Image Caching post SECI or Quicklook
+
+ :param dir_path: path of the directory that will be submitted to condor
+ :return:
+ """
+ extensions = ["*.condor", ".dag", ".log"]
+
+ old_files = itertools.chain.from_iterable(dir_path.glob(pattern) for pattern in extensions)
+ if len(list(old_files)) > 0:
+ logger.info("Shifting old condor files aside...")
+ aside = dir_path / "old_condor"
+ aside.mkdir(0o777, parents=False, exist_ok=True)
+ for file in old_files:
+ shutil.move(file, aside)
+ logger.info("Finished shifting old condor files.")
+
def _execute_prepared(self, folder: Path, run_cv: bool) -> [Path, bool]:
"""
Execute HTCondor using the named folder as the source of the files.
--
GitLab