workflows4: PanDA native workflow engine first merge

PandaPkgInfo.py

@@ -1 +1 @@
-release_version = "0.7.2"
+release_version = "0.8.0"

pandajedi/jedicore/JediDBSchemaInfo.py

@@ -12,6 +12,6 @@ class JediDBSchemaInfo:
     schema_version = None
 
     def method(self):
-        schema_version = "0.0.30"
+        schema_version = "0.1.1"
         _logger.debug(f"PanDA schema version required for JEDI is : {schema_version}")
         return schema_version

pandajedi/jedidog/AtlasWorkflowManagerWatchDog.py

@@ -0,0 +1,63 @@
+import sys
+import traceback
+
+# logger
+from pandacommon.pandalogger.PandaLogger import PandaLogger
+
+from pandajedi.jedicore.MsgWrapper import MsgWrapper
+from pandaserver.workflow.workflow_core import WorkflowInterface
+
+from .WatchDogBase import WatchDogBase
+
+logger = PandaLogger().getLogger(__name__.split(".")[-1])
+
+
+class AtlasWorkflowManagerWatchDog(WatchDogBase):
+    """
+    Workflow manager watchdog for ATLAS
+    """
+
+    # constructor
+    def __init__(self, taskBufferIF, ddmIF):
+        WatchDogBase.__init__(self, taskBufferIF, ddmIF)
+        self.vo = "atlas"
+        self.workflow_interface = WorkflowInterface(taskBufferIF)
+
+    def doProcessWorkflows(self):
+        """
+        Action to process active workflows
+        """
+        tmpLog = MsgWrapper(logger, " #ATM #KV doProcessWorkflows")
+        tmpLog.debug("start")
+        try:
+            # watchdog lock
+            got_lock = self.get_process_lock("AtlasWFManagerDog.doProcessWorkflows", timeLimit=5)
+            if not got_lock:
+                tmpLog.debug("locked by another watchdog process. Skipped")
+                return
+            tmpLog.debug("got watchdog lock")
+            # process active workflows
+            stats = self.workflow_interface.process_active_workflows()
+            tmpLog.info(f"processed workflows: {stats}")
+            # done
+            tmpLog.debug("done")
+        except Exception:
+            errtype, errvalue = sys.exc_info()[:2]
+            tmpLog.error(f"failed with {errtype} {errvalue} {traceback.format_exc()}")
+
+    # main
+    def doAction(self):
+        try:
+            # get logger
+            origTmpLog = MsgWrapper(logger)
+            origTmpLog.debug("start")
+            # clean up
+            # check
+            # process workflows
+            self.doProcessWorkflows()
+        except Exception:
+            errtype, errvalue = sys.exc_info()[:2]
+            origTmpLog.error(f"failed with {errtype} {errvalue}")
+        # return
+        origTmpLog.debug("done")
+        return self.SC_SUCCEEDED

pandajedi/jedimsgprocessor/workflow_manager_msg_processor.py

@@ -0,0 +1,89 @@
+import json
+
+from pandacommon.pandalogger import logger_utils
+
+from pandajedi.jedimsgprocessor.base_msg_processor import BaseMsgProcPlugin
+from pandaserver.workflow.workflow_core import WorkflowInterface
+
+base_logger = logger_utils.setup_logger(__name__.split(".")[-1])
+
+
+# Workflow manager message processor plugin
+class WorkflowManagerMsgProcPlugin(BaseMsgProcPlugin):
+    """
+    Message-driven workflow manager
+    """
+
+    def initialize(self):
+        """
+        Initialize the plugin
+        """
+        BaseMsgProcPlugin.initialize(self)
+        self.workflow_interface = WorkflowInterface(self.tbIF)
+
+    def process(self, msg_obj):
+        """
+        Process the message
+        Typical message data looks like:
+            {"msg_type":"workflow", "workflow_id": 123, "timestamp": 987654321}
+            {"msg_type":"wfstep", "step_id": 456, "timestamp": 987654321}
+            {"msg_type":"wfdata", "data_id": 789, "timestamp": 987654321}
+
+        Args:
+            msg_obj: message object
+        """
+        tmp_log = logger_utils.make_logger(base_logger, token=self.get_pid(), method_name="process")
+        # start
+        tmp_log.info("start")
+        tmp_log.debug(f"sub_id={msg_obj.sub_id} ; msg_id={msg_obj.msg_id}")
+        # parse json
+        try:
+            msg_dict = json.loads(msg_obj.data)
+        except Exception as e:
+            err_str = f"failed to parse message json {msg_obj.data} , skipped. {e.__class__.__name__} : {e}"
+            tmp_log.error(err_str)
+            raise
+        # sanity check
+        try:
+            msg_type = msg_dict["msg_type"]
+        except Exception as e:
+            err_str = f"failed to parse message object dict {msg_dict} , skipped. {e.__class__.__name__} : {e}"
+            tmp_log.error(err_str)
+            raise
+        if msg_type not in ("workflow", "wfstep", "wfdata"):
+            err_str = f"got unknown msg_type {msg_type} , skipped "
+            tmp_log.error(err_str)
+            raise
+        # run
+        try:
+            tmp_log.info(f"got message {msg_dict}")
+            if msg_type == "workflow":
+                workflow_id = msg_dict["workflow_id"]
+                workflow_spec = self.tbIF.get_workflow(workflow_id)
+                if workflow_spec is None:
+                    tmp_log.warning(f"workflow_id={workflow_id} not found; skipped")
+                    return
+                stats, workflow_spec = self.workflow_interface.process_workflow(workflow_spec, by="msg")
+                tmp_log.info(f"processed workflow_id={workflow_id}")
+            elif msg_type == "wfstep":
+                step_id = msg_dict["step_id"]
+                step_spec = self.tbIF.get_workflow_step(step_id)
+                if step_spec is None:
+                    tmp_log.warning(f"step_id={step_id} not found; skipped")
+                    return
+                stats, step_spec = self.workflow_interface.process_step(step_spec, by="msg")
+                tmp_log.info(f"processed step_id={step_id}")
+            elif msg_type == "wfdata":
+                data_id = msg_dict["data_id"]
+                data_spec = self.tbIF.get_workflow_data(data_id)
+                if data_spec is None:
+                    tmp_log.warning(f"data_id={data_id} not found; skipped")
+                    return
+                stats, data_spec = self.workflow_interface.process_data(data_spec, by="msg")
+                tmp_log.info(f"processed data_id={data_id}")
+        except Exception as e:
+            err_str = f"failed to run, skipped. {e.__class__.__name__} : {e}"
+            tmp_log.error(err_str)
+            raise
+        # done
+        tmp_log.info("done")

pandajedi/jediorder/ContentsFeeder.py

@@ -185,6 +185,7 @@ def feed_contents_to_tasks(self, task_ds_list, real_run=True):
                                 parentOutDatasets.add(tmpParentOutDataset.containerName + "/")
             # loop over all datasets
             nFilesMaster = 0
+            nFilesMasterReady = 0
             checkedMaster = False
             setFrozenTime = True
             master_offset = None
@@ -240,6 +241,9 @@ def feed_contents_to_tasks(self, task_ds_list, real_run=True):
                         ):
                             # dummy metadata when parent is running
                             tmpMetadata = {"state": "mutable"}
+                        # set mutable when workflow holdup is set
+                        if taskSpec.is_workflow_holdup():
+                            tmpMetadata = {"state": "mutable"}
                     except Exception:
                         errtype, errvalue = sys.exc_info()[:2]
                         tmpLog.error(f"{self.__class__.__name__} failed to get metadata to {errtype.__name__}:{errvalue}")
@@ -529,7 +533,7 @@ def feed_contents_to_tasks(self, task_ds_list, real_run=True):
                                 orderBy = None
                             # feed files to the contents table
                             tmpLog.debug("update contents")
-                            retDB, missingFileList, nFilesUnique, diagMap = self.taskBufferIF.insertFilesForDataset_JEDI(
+                            res_dict = self.taskBufferIF.insertFilesForDataset_JEDI(
                                 datasetSpec,
                                 tmpRet,
                                 tmpMetadata["state"],
@@ -565,6 +569,10 @@ def feed_contents_to_tasks(self, task_ds_list, real_run=True):
                                 maxFileRecords,
                                 skip_short_output,
                             )
+                            retDB = res_dict["ret_val"]
+                            missingFileList = res_dict["missingFileList"]
+                            nFilesUnique = res_dict["numUniqueLfn"]
+                            diagMap = res_dict["diagMap"]
                             if retDB is False:
                                 taskSpec.setErrDiag(f"failed to insert files for {datasetSpec.datasetName}. {diagMap['errMsg']}")
                                 allUpdated = False
@@ -595,6 +603,7 @@ def feed_contents_to_tasks(self, task_ds_list, real_run=True):
                                 if datasetSpec.isMaster():
                                     checkedMaster = True
                                     nFilesMaster += nFilesUnique
+                                    nFilesMasterReady += res_dict.get("nReady", 0)
                                     master_offset = datasetSpec.getOffset()
                             # running task
                             if diagMap["isRunningTask"]:
@@ -618,6 +627,11 @@ def feed_contents_to_tasks(self, task_ds_list, real_run=True):
                                 setFrozenTime = False
                                 skip_secondaries = True
                     tmpLog.debug("end loop")
+            # task holdup by workflow if no master inputs are ready
+            if not taskOnHold and not taskBroken and allUpdated and nFilesMasterReady == 0 and checkedMaster and taskSpec.is_workflow_holdup():
+                # hold up by the workflow
+                taskOnHold = True
+                tmpLog.debug("task to hold up by workflow")
             # no master input
             if not taskOnHold and not taskBroken and allUpdated and nFilesMaster == 0 and checkedMaster:
                 tmpErrStr = "no master input files. input dataset is empty"

pandajedi/jedirefine/TaskRefinerBase.py

@@ -432,6 +432,8 @@ def extractCommon(self, jediTaskID, taskParamMap, workQueueMapper, splitRule):
             self.setSplitRule(None, 1, JediTaskSpec.splitRuleToken["allowIncompleteInDS"])
         if "noAutoPause" in taskParamMap and taskParamMap["noAutoPause"]:
             self.setSplitRule(None, 1, JediTaskSpec.splitRuleToken["noAutoPause"])
+        if "workflowHoldup" in taskParamMap and taskParamMap["workflowHoldup"]:
+            self.setSplitRule(None, 1, JediTaskSpec.splitRuleToken["workflowHoldup"])
         # work queue
         workQueue = None
         if "workQueueName" in taskParamMap:

pandaserver/api/v1/workflow_api.py

@@ -0,0 +1,136 @@
+import datetime
+import json
+from concurrent.futures import ThreadPoolExecutor
+from threading import Lock
+from typing import Any, Dict, List
+
+from pandacommon.pandalogger.LogWrapper import LogWrapper
+from pandacommon.pandalogger.PandaLogger import PandaLogger
+from pandacommon.pandautils.PandaUtils import naive_utcnow
+
+from pandaserver.api.v1.common import (
+    MESSAGE_DATABASE,
+    TIME_OUT,
+    TimedMethod,
+    generate_response,
+    get_dn,
+    has_production_role,
+    request_validation,
+)
+from pandaserver.srvcore.panda_request import PandaRequest
+from pandaserver.taskbuffer.TaskBuffer import TaskBuffer
+from pandaserver.workflow.workflow_core import WorkflowInterface
+
+_logger = PandaLogger().getLogger("api_workflow")
+
+# These global variables are initialized in the init_task_buffer method
+global_task_buffer = None
+global_wfif = None
+
+# These global variables don't depend on DB access and can be initialized here
+# global_proxy_cache = panda_proxy_cache.MyProxyInterface()
+# global_token_cache = token_cache.TokenCache()
+
+
+def init_task_buffer(task_buffer: TaskBuffer) -> None:
+    """
+    Initialize the task buffer and other interfaces. This method needs to be called before any other method in this module.
+    """
+    global global_task_buffer
+    global_task_buffer = task_buffer
+
+    global global_wfif
+    global_wfif = WorkflowInterface(global_task_buffer)
+
+
+@request_validation(_logger, secure=True, production=False, request_method="POST")
+def submit_workflow_raw_request(req: PandaRequest, params: dict | str) -> dict:
+    """
+    Submit raw request of PanDA native workflow.
+
+    API details:
+        HTTP Method: POST
+        Path: /v1/workflow/submit_workflow_raw_request
+
+    Args:
+        req(PandaRequest): internally generated request object containing the env variables
+        params (dict|str): dictionary or JSON of parameters of the raw request
+
+    Returns:
+        dict: dictionary `{'success': True/False, 'message': 'Description of error', 'data': <requested data>}`
+    """
+
+    user_dn = get_dn(req)
+    prodsourcelabel = "user"
+
+    # FIXME: only for analysis temporarily
+    # if has_production_role(req):
+    #     prodsourcelabel = "managed"
+
+    tmp_logger = LogWrapper(_logger, f'submit_workflow_raw_request prodsourcelabel={prodsourcelabel} user_dn="{user_dn}" ')
+    tmp_logger.debug("Start")
+    success, message, data = False, "", None
+    time_start = naive_utcnow()
+
+    if isinstance(params, str):
+        try:
+            params = json.loads(params)
+        except Exception as exc:
+            message = f"Failed to parse params: {params} {str(exc)}"
+            tmp_logger.error(message)
+            return generate_response(success, message, data)
+
+    workflow_id = global_wfif.register_workflow(prodsourcelabel, user_dn, raw_request_params=params)
+
+    if workflow_id is not None:
+        success = True
+        data = {"workflow_id": workflow_id}
+    else:
+        message = "Failed to submit raw workflow request"
+
+    time_delta = naive_utcnow() - time_start
+    tmp_logger.debug(f"Done. Took {time_delta.seconds}.{time_delta.microseconds // 1000:03d} sec")
+
+    return generate_response(success, message, data)
+
+
+@request_validation(_logger, secure=True, production=False, request_method="POST")
+def submit_workflow(req: PandaRequest, workflow_definition: dict) -> dict:
+    """
+    Submit a PanDA native workflow.
+
+    API details:
+        HTTP Method: POST
+        Path: /v1/workflow/submit_workflow
+
+    Args:
+        req(PandaRequest): internally generated request object containing the env variables
+        workflow_definition (dict): dictionary of workflow definition
+
+    Returns:
+        dict: dictionary `{'success': True/False, 'message': 'Description of error', 'data': <requested data>}`
+    """
+
+    user_dn = get_dn(req)
+    prodsourcelabel = "user"
+    if has_production_role(req):
+        prodsourcelabel = "managed"
+    workflow_name = workflow_definition.get("workflow_name", None)
+
+    tmp_logger = LogWrapper(_logger, f'submit_workflow prodsourcelabel={prodsourcelabel} user_dn="{user_dn}" workflow_name={workflow_name}')
+    tmp_logger.debug("Start")
+    success, message, data = False, "", None
+    time_start = naive_utcnow()
+
+    workflow_id = global_wfif.register_workflow(prodsourcelabel, user_dn, workflow_name, workflow_definition)
+
+    if workflow_id is not None:
+        success = True
+        data = {"workflow_id": workflow_id}
+    else:
+        message = "Failed to submit workflow"
+
+    time_delta = naive_utcnow() - time_start
+    tmp_logger.debug(f"Done. Took {time_delta.seconds}.{time_delta.microseconds // 1000:03d} sec")
+
+    return generate_response(success, message, data)