Repository: cylc/cylc-uiserver (language: python)

Issue:
cat-log: support workflow files and log listing
> **Built Atop:** https://github.com/cylc/cylc-uiserver/tree/update-log-work

> **Sibling:** https://github.com/cylc/cylc-flow/pull/5453

* Support subscribing to other workflow log files.
* Add interface for listing log files.
* Kill the full cat-log process tree recursively when stopping subscriptions (I seemed to hit some situation where that wasn't happening).
* Unify the `workflow` and `task` arguments in the log subscription interface into a single `id` argument.

**Check List**

- [x] I have read `CONTRIBUTING.md` and added my name as a Code Contributor.
- [x] Contains logically grouped changes (else tidy your branch by rebase).
- [x] Does not contain off-topic changes (use other PRs for other changes).
- [x] Applied any dependency changes to both `setup.cfg` (and `conda-environment.yml` if present).
- [x] Tests are included (or explain why tests are not needed).
- [x] `CHANGES.md` entry included if this is a change that can affect users
- [x] [Cylc-Doc](https://github.com/cylc/cylc-doc) pull request opened if required at cylc/cylc-doc/pull/XXXX.
- [x] If this is a bug fix, PR should be raised against the relevant `?.?.x` branch.


Discussion / hints:

Fix patch:
diff --git a/.github/dependabot.yml b/.github/dependabot.yml
index 6718b59e..8ba4f0f2 100644
--- a/.github/dependabot.yml
+++ b/.github/dependabot.yml
@@ -9,3 +9,10 @@ updates:
     directory: "/" # Location of package manifests
     schedule:
       interval: "daily"
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: "weekly"
+    labels:
+      - 'dependencies'
+      - 'infrastructure'
diff --git a/cylc/uiserver/resolvers.py b/cylc/uiserver/resolvers.py
index 4bc596e2..fa4a4416 100644
--- a/cylc/uiserver/resolvers.py
+++ b/cylc/uiserver/resolvers.py
@@ -20,9 +20,19 @@
 from getpass import getuser
 import os
 from copy import deepcopy
+import signal
 from subprocess import Popen, PIPE, DEVNULL
 from typing import (
-    TYPE_CHECKING, Any, Callable, Dict, Iterable, List, Tuple, Union
+    Any,
+    AsyncGenerator,
+    Callable,
+    Dict,
+    Iterable,
+    List,
+    Optional,
+    TYPE_CHECKING,
+    Tuple,
+    Union,
 )
 
 from graphql.language.base import print_ast
@@ -332,19 +342,21 @@ async def enqueue(stream, queue):
             await queue.put(line.decode())
 
     @classmethod
-    async def cat_log(cls, workflow: Tokens, log, info, task=None, file=None):
+    async def cat_log(cls, id_: Tokens, log, info, file=None):
         """Calls `cat log`.
 
         Used for log subscriptions.
         """
-        full_workflow = workflow
-        if file and task:
-            full_workflow = Tokens(f"{workflow.workflow_id}//{task}")
-        cmd = ['cylc', 'cat-log']
+        cmd: List[str] = [
+            'cylc',
+            'cat-log',
+            '--mode=tail',
+            '--force-remote',
+            '--prepend-path',
+            id_.id,
+        ]
         if file:
             cmd += ['-f', file]
-        cmd += ['-m', 't']
-        cmd.append(full_workflow.id)
         log.info(f'$ {" ".join(cmd)}')
 
         # For info, below subprocess is safe (uses shell=false by default)
@@ -365,42 +377,76 @@ async def cat_log(cls, workflow: Tokens, log, info, task=None, file=None):
             while info.context['sub_statuses'].get(op_id) != 'stop':
                 if queue.empty():
                     if buffer:
-                        yield list(buffer)
+                        yield {'lines': list(buffer)}
                         buffer.clear()
                     if proc.returncode not in [None, 0]:
                         (_, stderr) = await proc.communicate()
                         # pass any error onto ui
-                        yield stderr.decode().splitlines()
+                        yield {'error': stderr.decode()}
                         break
                     # sleep set at 1, which matches the `tail` default interval
                     await asyncio.sleep(1)
                 else:
                     if line_count > MAX_LINES:
-                        yield buffer
-                        yield [
-                            '\n\n' + ('-' * 80) + '\n',
-                            (
+                        yield {'lines': buffer}
+                        yield {
+                            'error': (
                                 'This file has been truncated because'
-                                f' it is over {MAX_LINES} lines long.\n'
-                            ),
-                        ]
+                                f' it is over {MAX_LINES} lines long.'
+                            )
+                        }
                         break
+                    elif line_count == 0:
+                        line_count += 1
+                        yield {
+                            'connected': True,
+                            'path': (await queue.get())[2:].strip(),
+                        }
+                        continue
                     line = await queue.get()
                     line_count += 1
                     buffer.append(line)
                     if len(buffer) >= 75:
-                        yield list(buffer)
+                        yield {'lines': list(buffer)}
                         buffer.clear()
                         await asyncio.sleep(0)
         finally:
-            with suppress(psutil.Error):
-                cat_log_proc = psutil.Process(proc.pid)
-                # cat-log process will auto-terminate on tail termination
-                for tail_proc in cat_log_proc.children():
-                    tail_proc.terminate()
+            kill_process_tree(proc.pid)
             enqueue_task.cancel()
             with suppress(asyncio.CancelledError):
                 await enqueue_task
+            yield {'connected': False}
+
+    @classmethod
+    async def cat_log_files(cls, id_: Tokens):
+        """Calls cat log to get list of available log files.
+
+        Note kept separate from the cat_log method above as this is a one off
+        query rather than a process held open for subscription.
+        This uses the Cylc cat-log interface, list dir mode, forcing remote
+        file checking.
+        """
+        cmd: List[str] = ['cylc', 'cat-log', '-m', 'l', '-o', id_.id]
+        proc_job = await asyncio.subprocess.create_subprocess_exec(
+            *cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        # wait for proc to finish
+        await proc_job.wait()
+
+        # MOTD returned in stderr, no use in returning
+        out_job, _ = await proc_job.communicate()
+        if out_job:
+            return sorted(
+                # return the log files in reverse sort order
+                # this means that the most recent log file rotations
+                # will be at the top of the list
+                out_job.decode().splitlines(),
+                reverse=True,
+            )
+        else:
+            return []
 
 
 class Resolvers(BaseResolvers):
@@ -492,15 +538,85 @@ async def subscription_service(
         self,
         info: 'ResolveInfo',
         _command: str,
-        workflows: List[Tokens],
-        task=None,
+        ids: List[Tokens],
         file=None
     ):
         async for ret in Services.cat_log(
-            workflows[0],
+            ids[0],
             self.log,
             info,
-            task,
             file
         ):
             yield ret
+
+    async def query_service(
+        self,
+        id_: Tokens,
+    ):
+        return await Services.cat_log_files(id_)
+
+
+def kill_process_tree(
+    pid,
+    sig=signal.SIGTERM,
+    include_parent=True,
+):
+    """Kill an entire process tree.
+
+    Args:
+        pid: The parent process ID to kill.
+        sig: The signal to send to the processes in this tree.
+        include_parent: Also kill the parent process (pid).
+
+    """
+    try:
+        parent = psutil.Process(pid)
+    except psutil.NoSuchProcess:
+        return
+    children = parent.children(recursive=True)
+    if include_parent:
+        children.append(parent)
+    for p in children:
+        with suppress(psutil.NoSuchProcess):
+            p.send_signal(sig)
+
+
+async def list_log_files(
+    root: Optional[Any],
+    info: 'ResolveInfo',
+    id: str,  # noqa: required to match schema arg name
+):
+    tokens = Tokens(id)
+    resolvers: 'Resolvers' = (
+        info.context.get('resolvers')  # type: ignore[union-attr]
+    )
+    files = await resolvers.query_service(tokens)
+    return {'files': files}
+
+
+async def stream_log(
+    root: Optional[Any],
+    info: 'ResolveInfo',
+    *,
+    command='cat_log',
+    id: str,  # noqa: required to match schema arg name
+    file=None,
+    **kwargs: Any,
+) -> AsyncGenerator[Any, None]:
+    """Cat Log Resolver
+    Expands workflow provided subscription query.
+    """
+    tokens = Tokens(id)
+    if kwargs.get('args', False):
+        kwargs.update(kwargs.get('args', {}))
+        kwargs.pop('args')
+    resolvers: 'Resolvers' = (
+        info.context.get('resolvers')  # type: ignore[union-attr]
+    )
+    async for item in resolvers.subscription_service(
+        info,
+        command,
+        [tokens],
+        file
+    ):
+        yield item
diff --git a/cylc/uiserver/schema.py b/cylc/uiserver/schema.py
index 01cdcee6..c49eb42b 100644
--- a/cylc/uiserver/schema.py
+++ b/cylc/uiserver/schema.py
@@ -20,7 +20,7 @@
 """
 
 from functools import partial
-from typing import TYPE_CHECKING, Any, List, Optional, AsyncGenerator
+from typing import TYPE_CHECKING, Any, List, Optional
 
 import graphene
 from graphene.types.generic import GenericScalar
@@ -42,7 +42,11 @@
     sstrip,
     get_nodes_all
 )
-from cylc.uiserver.resolvers import Resolvers
+from cylc.uiserver.resolvers import (
+    Resolvers,
+    list_log_files,
+    stream_log,
+)
 
 if TYPE_CHECKING:
     from graphql import ResolveInfo
@@ -464,6 +468,26 @@ class UISTask(Task):
 
 class UISQueries(Queries):
 
+    class LogFiles(graphene.ObjectType):
+        # Example GraphiQL query:
+        # {
+        #    logFiles(workflowID: "<workflow_id>", task: "<task_id>") {
+        #      files
+        #    }
+        # }
+        files = graphene.List(graphene.String)
+
+    log_files = graphene.Field(
+        LogFiles,
+        description='List available job logs',
+        id=graphene.Argument(
+            graphene.ID,
+            description='workflow//[cycle/task]',
+            required=True,
+        ),
+        resolver=list_log_files
+    )
+
     tasks = graphene.List(
         UISTask,
         description=Task._meta.description,
@@ -488,55 +512,26 @@ class UISSubscriptions(Subscriptions):
     #   }
     # }
 
-    async def resolve_logs(
-        root: Optional[Any],
-        info: 'ResolveInfo',
-        *,
-        command='cat_log',
-        workflow: str,
-        task=None,
-        file=None,
-        **kwargs: Any,
-    ) -> AsyncGenerator[Any, None]:
-        """Cat Log Resolver
-        Expands workflow provided subscription query.
-        """
-        parsed_workflows = [Tokens(workflow)]
-        if kwargs.get('args', False):
-            kwargs.update(kwargs.get('args', {}))
-            kwargs.pop('args')
-        resolvers: 'Resolvers' = (
-            info.context.get('resolvers')  # type: ignore[union-attr]
-        )
-        async for item in resolvers.subscription_service(
-            info,
-            command,
-            parsed_workflows,
-            task,
-            file
-        ):
-            yield {'lines': item}
-
     class Logs(graphene.ObjectType):
         lines = graphene.List(graphene.String)
+        connected = graphene.Boolean()
+        path = graphene.String()
+        error = graphene.String()
 
     logs = graphene.Field(
         Logs,
         description='Workflow & job logs',
-        workflow=graphene.Argument(
-            ID
-        ),
-        task=graphene.Argument(
-            graphene.String,
-            required=False,
-            description='cylc/task/job ID'
+        id=graphene.Argument(
+            graphene.ID,
+            description='workflow//[cycle/task[/job]]',
+            required=True,
         ),
         file=graphene.Argument(
             graphene.String,
             required=False,
             description='File name of job log to fetch, e.g. job.out'
         ),
-        resolver=resolve_logs
+        resolver=stream_log
     )
 
 


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 1fadb74a..00d728a2 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -48,14 +48,14 @@ jobs:
         with:
           python-version: ${{ matrix.python }}
 
-      # NOTE: Install cylc-uiserver in editable mode for correct coverage results
-      - name: Install cylc-uiserver & dependencies
-        run: |
-          pip install git+https://github.com/cylc/cylc-flow@master
-          pip install -e .[all]
+      - name: install cylc-flow
+        uses: cylc/release-actions/install-cylc-components@v1
+        with:
+          cylc_flow: true
+          cylc_flow_opts: ''
 
-      - name: (Debug - pip list)
-        run: pip list --format=freeze
+      - name: install cylc-uiserver
+        run: pip install -e .[all]
 
       - name: Style test
         run: flake8
diff --git a/cylc/uiserver/tests/test_resolvers.py b/cylc/uiserver/tests/test_resolvers.py
index a2c5fb4b..6962e7cb 100644
--- a/cylc/uiserver/tests/test_resolvers.py
+++ b/cylc/uiserver/tests/test_resolvers.py
@@ -4,6 +4,7 @@
 import pytest
 from unittest import mock
 
+from cylc.flow import CYLC_LOG
 from cylc.flow.id import Tokens
 from cylc.flow.scripts.clean import CleanOptions
 from cylc.uiserver.resolvers import (
@@ -52,7 +53,7 @@ async def test_cat_log(workflow_run_dir):
     returns all the logs. Note the log content should be over 20 lines to check
     the buffer logic.
     """
-    (flow_name, log_dir) = workflow_run_dir
+    (id_, log_dir) = workflow_run_dir
     log_file_content = """2022-11-08T11:10:05Z DEBUG - Starting
     2022-11-08T11:10:05Z DEBUG -
     2022-11-08T11:10:05Z DEBUG -
@@ -75,23 +76,39 @@ async def test_cat_log(workflow_run_dir):
     2022-11-08T11:14:09Z DEBUG -
     2022-11-08T11:14:11Z INFO - DONE
     """
-    log_file = log_dir / 'log'
+    log_file = log_dir / '01-start-01.log'
     log_file.write_text(log_file_content)
     expected = log_file.read_text()
     info = mock.MagicMock()
     info.root_value = 2
     # mock the context
     info.context = {'sub_statuses': {2: "start"}}
-    workflow = Tokens(flow_name)
-    log = logging.getLogger('cylc')
+    workflow = Tokens(id_)
+    log = logging.getLogger(CYLC_LOG)
     # note - timeout tests that the cat-log process is being stopped correctly
+
+    first_response = None
     async with timeout(10):
         ret = services.cat_log(workflow, log, info)
-        actual = str()
-        async for buffered_return in ret:
-            for line in buffered_return:
+        actual = ''
+        is_first = True
+        async for response in ret:
+            if is_first:
+                first_response = response
+                is_first = False
+            for line in response.get('lines', []):
                 actual += line
                 if "DONE" in line:
                     info.context['sub_statuses'][2] = 'stop'
             await asyncio.sleep(0)
+
+    # the first response should include the log file path and
+    # connection status
+    assert first_response['path'].endswith('01-start-01.log')
+    assert first_response['connected'] is True
+
+    # the last response should change the connected status
+    assert response['connected'] is False
+
+    # the other responses should contain the log file lines
     assert actual.rstrip() == expected.rstrip()