Repository: fractal-analytics-platform/fractal-client (language: python)

Issue:
Client status monitoring
While general monitoring is a broader topic (see https://github.com/fractal-analytics-platform/fractal/issues/224), is there a simple way we could add some status monitoring to the client? Such that a user could request updates after submitting a job and know things like:
1. Is my job still running? / Have there been any errors?
2. How long has my job been running?
3. What task is currently being run?
Client status monitoring
While general monitoring is a broader topic (see https://github.com/fractal-analytics-platform/fractal/issues/224), is there a simple way we could add some status monitoring to the client? Such that a user could request updates after submitting a job and know things like:
1. Is my job still running? / Have there been any errors?
2. How long has my job been running?
3. What task is currently being run?


Discussion / hints:
And in the same logic: If there were errors, can the user check those error messages using the client?

I'm just imagining the demo scenario. If 10 different users submit something to the server and set some parameters wrong, it's impossible from the server log to know which error was affecting which user
The current status of logging in `fractal-server` is the one described in https://github.com/fractal-analytics-platform/fractal-server/issues/52#issuecomment-1268166445, that is:

> * What did fractal-server do for this workflow (e.g. it defined a set of apps, it started running, it completed running). This is currently in server-dir/logs/workflow_XXXXXX/workflow.log.
> * What did each task do (tasks may include some internal logging (e.g. "I created /some/path/plate.zarr"), which should be accessible. This is currently distributed in the files in server-dir/logs/workflow_XXXXXX/scripts.
> * Task errors (e.g. a task raises a ValueError). This currently goes only to the server log stream, which is not useful to retrieve it on a per-workflow level. For testing, we can always use the dummy_fail task, which simply raises an exception with a message.
> * SLURM errors (e.g. memory error). This currently goes in some of the files in the files in server-dir/logs/workflow_XXXXXX/scripts. For testing, we can submit a labeling task to a cpu-low executor, and we'll get something like "Virtual Memory Error".
> * Monitoring of resources (especially memory), either in real-time or ex-post. This is written in the monitoring.db parsl database, but at the moment we are not able to retrieve it in a simple way.

Let's keep the current issue open to decide what the client should do within this scheme. Should there be some `check_info` client command that points to some of the content mentioned above?
Having a client command to check this would be great! Something like `check_status` and then a way to get the relevant logs :) 
With the current logs, the easiest way to serve logs would be to merge all the `stderr` files in a specific folder and then sort their lines by time. This would offer a comprehensive log stream, without thinking too much about how files are organized (which is related to parsl executor blocks, and at the moment is not totally transparent).

As for the status, this is probably better discussed after https://github.com/fractal-analytics-platform/fractal-server/issues/107.
That's certainly a good start. A user would then request the log for a given workflow, which makes sense :)
Once https://github.com/fractal-analytics-platform/fractal-server/pull/327 is merged, we should add the corresponding client command that queries the status of a worfklow-execution job
And in the same logic: If there were errors, can the user check those error messages using the client?

I'm just imagining the demo scenario. If 10 different users submit something to the server and set some parameters wrong, it's impossible from the server log to know which error was affecting which user
The current status of logging in `fractal-server` is the one described in https://github.com/fractal-analytics-platform/fractal-server/issues/52#issuecomment-1268166445, that is:

> * What did fractal-server do for this workflow (e.g. it defined a set of apps, it started running, it completed running). This is currently in server-dir/logs/workflow_XXXXXX/workflow.log.
> * What did each task do (tasks may include some internal logging (e.g. "I created /some/path/plate.zarr"), which should be accessible. This is currently distributed in the files in server-dir/logs/workflow_XXXXXX/scripts.
> * Task errors (e.g. a task raises a ValueError). This currently goes only to the server log stream, which is not useful to retrieve it on a per-workflow level. For testing, we can always use the dummy_fail task, which simply raises an exception with a message.
> * SLURM errors (e.g. memory error). This currently goes in some of the files in the files in server-dir/logs/workflow_XXXXXX/scripts. For testing, we can submit a labeling task to a cpu-low executor, and we'll get something like "Virtual Memory Error".
> * Monitoring of resources (especially memory), either in real-time or ex-post. This is written in the monitoring.db parsl database, but at the moment we are not able to retrieve it in a simple way.

Let's keep the current issue open to decide what the client should do within this scheme. Should there be some `check_info` client command that points to some of the content mentioned above?
Having a client command to check this would be great! Something like `check_status` and then a way to get the relevant logs :) 
With the current logs, the easiest way to serve logs would be to merge all the `stderr` files in a specific folder and then sort their lines by time. This would offer a comprehensive log stream, without thinking too much about how files are organized (which is related to parsl executor blocks, and at the moment is not totally transparent).

As for the status, this is probably better discussed after https://github.com/fractal-analytics-platform/fractal-server/issues/107.
That's certainly a good start. A user would then request the log for a given workflow, which makes sense :)
Once https://github.com/fractal-analytics-platform/fractal-server/pull/327 is merged, we should add the corresponding client command that queries the status of a worfklow-execution job

Fix patch:
diff --git a/fractal/cmd/__init__.py b/fractal/cmd/__init__.py
index c8f2a9e3b..74d9e2693 100644
--- a/fractal/cmd/__init__.py
+++ b/fractal/cmd/__init__.py
@@ -26,6 +26,7 @@
 from ._workflow import workflow_edit_task
 from ._workflow import workflow_list
 from ._workflow import workflow_new
+from ._workflow import workflow_query_job_status
 from ._workflow import workflow_remove_task
 from ._workflow import workflow_show
 
@@ -165,6 +166,8 @@ async def workflow(
         iface = await workflow_remove_task(client, **kwargs)
     elif subcmd == "apply":
         iface = await workflow_apply(client, **kwargs)
+    elif subcmd == "job-status":
+        iface = await workflow_query_job_status(client, batch=batch, **kwargs)
     return iface
 
 
diff --git a/fractal/cmd/_workflow.py b/fractal/cmd/_workflow.py
index 758c4731a..bd3482ca7 100644
--- a/fractal/cmd/_workflow.py
+++ b/fractal/cmd/_workflow.py
@@ -20,6 +20,26 @@
 from .utils import get_cached_task_by_name
 
 
+async def workflow_query_job_status(
+    client: AuthClient,
+    job_id: int,
+    batch: bool = False,
+    **kwargs,
+) -> BaseInterface:
+    """
+    Query the status of a workflow job
+    """
+
+    res = await client.get(f"{settings.BASE_URL}/job/{job_id}")
+    job = check_response(
+        res, expected_status_code=200, coerce=ApplyWorkflowRead
+    )
+    if batch:
+        return PrintInterface(retcode=0, data=job.status)
+    else:
+        return RichJsonInterface(retcode=0, data=job.dict())
+
+
 async def workflow_new(
     client: AuthClient,
     name: str,
diff --git a/fractal/parser.py b/fractal/parser.py
index 5b4d40bc9..4b43a85f4 100644
--- a/fractal/parser.py
+++ b/fractal/parser.py
@@ -416,6 +416,7 @@
     help="Change the project associated with the current workflow",
 )
 
+# workflow apply
 workflow_apply_parser = workflow_subparsers.add_parser(
     "apply", help="Apply workflow to dataset", argument_default=ap.SUPPRESS
 )
@@ -441,6 +442,18 @@
     help="Command to be run before starting a worker",
 )
 
+# workflow job-status
+workflow_job_status_parser = workflow_subparsers.add_parser(
+    "job-status",
+    help="Query workflow-execution status",
+    argument_default=ap.SUPPRESS,
+)
+workflow_job_status_parser.add_argument(
+    "--job-id",
+    help="Id of the job",
+)
+
+
 # VERSION GROUP
 version_parser = subparsers_main.add_parser(
     "version", help="Print verison and exit"


Test patch:
diff --git a/tests/test_workflow.py b/tests/test_workflow.py
index 98bde1d1a..92fd061c1 100644
--- a/tests/test_workflow.py
+++ b/tests/test_workflow.py
@@ -1,11 +1,14 @@
 import asyncio
 import json
 import logging
+import time
 from pathlib import Path
 
 import pytest  # noqa F401
 from devtools import debug
 
+TIMEOUT = 15.0
+
 
 async def test_workflow_new(register_user, invoke):
     PROJECT_NAME = "project_name"
@@ -244,7 +247,9 @@ async def test_edit_workflow_task(
         res = await invoke(cmd)
 
 
-async def test_workflow_apply(register_user, invoke, testdata_path):
+async def test_workflow_apply(
+    register_user, invoke, testdata_path: Path, tmp_path: Path
+):
     """
     GIVEN a project and a nontrivial workflow
     WHEN the client requests to apply the workflow to the project
@@ -261,12 +266,14 @@ async def test_workflow_apply(register_user, invoke, testdata_path):
     debug(venv_path)
     state_id = res0.data["id"]
 
+    starting_time = time.perf_counter()
     while True:
         res1 = await invoke(f"task check-collection {state_id}")
         if res1.data["data"]["status"] == "OK":
             debug(res1.data)
             break
         await asyncio.sleep(1)
+        assert time.perf_counter() - starting_time < TIMEOUT
 
     res = await invoke("project new testproject prjpath")
     debug(res)
@@ -293,6 +300,9 @@ async def test_workflow_apply(register_user, invoke, testdata_path):
     TASK_ID = 1
     res = await invoke(f"workflow add-task {workflow_id} {TASK_ID}")
     assert res.retcode == 0
+    debug(res.data)
+    TASK_NAME = res.data["task_list"][0]["task"]["name"]
+    debug(TASK_NAME)
 
     cmd = (
         f"workflow apply {workflow_id} {input_dataset_id} "
@@ -301,6 +311,58 @@ async def test_workflow_apply(register_user, invoke, testdata_path):
     debug(cmd)
     res = await invoke(cmd)
     debug(res.data)
+    job_id = res.data["id"]
+    assert res.retcode == 0
+    assert res.data["status"] == "submitted"
+
+    # Check that job completed successfully
+    cmd = f"workflow job-status --job-id {job_id}"
+    starting_time = time.perf_counter()
+    debug(cmd)
+    while True:
+        res = await invoke(cmd)
+        assert res.retcode == 0
+        if res.data["status"] == "done":
+            break
+        await asyncio.sleep(1)
+        assert time.perf_counter() - starting_time < TIMEOUT
+    debug(res.data)
+    assert res.data["history"][0] == TASK_NAME
+
+    # Test output with --batch
+    cmd = f"--batch workflow job-status --job-id {job_id}"
+    res = await invoke(cmd)
+    assert res.retcode == 0
+    assert res.data == "done"
+
+    # Prepare and run a workflow with a failing task
+    args_file = str(tmp_path / "args.json")
+    with open(args_file, "w") as f:
+        json.dump({"raise_error": True}, f)
+    res = await invoke(
+        f"workflow add-task {workflow_id} {TASK_ID}"
+        f" --args-file {args_file}"
+    )
+    assert res.retcode == 0
+    cmd = (
+        f"workflow apply {workflow_id} {input_dataset_id} "
+        f"-o {output_dataset_id} -p {prj['id']}"
+    )
+    debug(cmd)
+    res = await invoke(cmd)
     assert res.retcode == 0
+    job_id = res.data["id"]
+
+    # Note: we may want to use a while loop to wait for the failure of the
+    # workflow execution
 
-    # TODO: verify outout
+    # Verify that status is failed, and that there is a log
+    cmd = f"workflow job-status --job-id {job_id}"
+    res = await invoke(cmd)
+    assert res.retcode == 0
+    debug(res.data)
+    assert res.data["status"] == "failed"
+    assert res.data["log"] is not None
+    # Note: the failing task is not added to the history
+    assert len(res.data["history"]) > 0
+    print(res.data["log"])