changes

Author: wxtim

cylc/flow/review.py

@@ -19,14 +19,15 @@
 
 import sqlite3
 import os
+from pathlib import Path
 import tarfile
 import re
 from glob import glob
 from sqlite3 import OperationalError
 
-from cylc.rundb import CylcSuiteDAO
-from cylc.task_state import TASK_STATUS_GROUPS
-"""Provide data access object to the suite runtime database for Cylc Review."""
+from cylc.flow.rundb import CylcWorkflowDAO
+from cylc.flow.task_state import TASK_STATUS_GROUPS
+from cylc.flow.workflow_files import WorkflowFiles
 
 
 class CylcReviewDAO(object):
@@ -105,29 +106,30 @@ def __init__(self):
         self.daos = {}
 
     def _db_init(self, user_name, suite_name):
-        """Initialise a named CylcSuiteDAO database connection."""
+        """Initialise a named CylcWorkflowDAO database connection."""
         key = (user_name, suite_name)
         if key not in self.daos:
             prefix = "~"
             if user_name:
                 prefix += user_name
             for name in [os.path.join("log", "db"), "cylc-suite.db"]:
-                db_f_name = os.path.expanduser(os.path.join(
-                    prefix, os.path.join("cylc-run", suite_name, name)))
-                self.daos[key] = CylcSuiteDAO(db_f_name, is_public=True)
+                db_f_name = os.path.expanduser(
+                    os.path.join(
+                        prefix, os.path.join("cylc-run", suite_name, name)))
+                self.daos[key] = CylcWorkflowDAO(db_f_name, is_public=True)
                 if os.path.exists(db_f_name):
                     break
         self.is_cylc8 = self.set_is_cylc8(user_name, suite_name)
         return self.daos[key]
 
     def _db_close(self, user_name, suite_name):
-        """Close a named CylcSuiteDAO database connection."""
+        """Close a named CylcWorkflowDAO database connection."""
         key = (user_name, suite_name)
         if self.daos.get(key) is not None:
             self.daos[key].close()
 
     def _db_exec(self, user_name, suite_name, stmt, stmt_args=None):
-        """Execute a query on a named CylcSuiteDAO database connection."""
+        """Execute a query on a named CylcWorkflowDAO database connection."""
         daos = self._db_init(user_name, suite_name)
         if stmt_args is None:
             stmt_args = []
@@ -140,10 +142,12 @@ def _db_exec(self, user_name, suite_name, stmt, stmt_args=None):
             except sqlite3.OperationalError as exc:
                 # At Cylc 8.0.1+ Workflows installed but not run will not yet
                 # have a database.
-                if (os.path.exists(os.path.dirname(
-                    self.daos.values()[0].db_file_name) + '/flow.cylc') or
-                    os.path.exists(os.path.dirname(
-                        self.daos.values()[0].db_file_name) + '/suite.rc')):
+                wf_dir = Path(
+                    list(self.daos.values())[0].db_file_name
+                ).parent
+                if (wf_dir / WorkflowFiles.FLOW_FILE).exists() or (
+                    wf_dir / WorkflowFiles.SUITE_RC
+                ).exists():
                     return []
                 else:
                     raise exc
@@ -152,7 +156,7 @@ def get_suite_broadcast_states(self, user_name, suite_name):
         """Return broadcast states of a suite.
         [[point, name, key, value], ...]
         """
-        stmt = CylcSuiteDAO.pre_select_broadcast_states(
+        stmt = CylcWorkflowDAO.pre_select_broadcast_states(
             self._db_init(user_name, suite_name), order="ASC")[0]
         broadcast_states = []
         for row in self._db_exec(user_name, suite_name, stmt):
@@ -164,7 +168,7 @@ def get_suite_broadcast_events(self, user_name, suite_name):
         """Return broadcast events of a suite.
         [[time, change, point, name, key, value], ...]
         """
-        stmt = CylcSuiteDAO.pre_select_broadcast_events(
+        stmt = CylcWorkflowDAO.pre_select_broadcast_events(
             self._db_init(user_name, suite_name), order="DESC")[0]
         broadcast_events = []
         for row in self._db_exec(user_name, suite_name, stmt):
@@ -175,16 +179,27 @@ def get_suite_broadcast_events(self, user_name, suite_name):
 
     @staticmethod
     def set_is_cylc8(user_name, suite_name):
-        from cylc.review import CylcReviewService
+        from cylc.flow.review import CylcReviewService
+
         suite_dir = os.path.join(
             CylcReviewService._get_user_home(user_name),
             "cylc-run",
             suite_name)
         return CylcReviewService.is_cylc8(suite_dir)
 
     def get_suite_job_entries(
-            self, user_name, suite_name, cycles, tasks, task_status,
-            job_status, order, limit, offset, flow_nums='flow_nums'):
+        self,
+        user_name,
+        suite_name,
+        cycles,
+        tasks,
+        task_status,
+        job_status,
+        order,
+        limit,
+        offset,
+        flow_nums='flow_nums',
+    ):
         """Query suite runtime database to return a listing of task jobs.
         user -- A string containing a valid user ID
         suite -- A string containing a valid suite ID
@@ -226,18 +241,20 @@ def get_suite_job_entries(
 
         # Get number of entries
         of_n_entries = 0
-        stmt = ("SELECT COUNT(*)" +
-                " FROM task_states LEFT JOIN task_jobs USING (name, cycle)" +
-                where_expr)
+        stmt = (
+            "SELECT COUNT(*)"
+            + " FROM task_states LEFT JOIN task_jobs USING (name, cycle)"
+            + where_expr
+        )
         try:
             for row in self._db_exec(user_name, suite_name, stmt, where_args):
                 of_n_entries = row[0]
                 break
             else:
                 self._db_close(user_name, suite_name)
-                return ([], 0)
+                return ([], 0, self.is_cylc8)
         except sqlite3.Error:
-            return ([], 0)
+            return ([], 0, self.is_cylc8)
         if self.is_cylc8:
             stmt = (
                 "SELECT" +
@@ -250,7 +267,9 @@ def get_suite_job_entries(
                 " time_run, time_run_exit, run_signal, run_status," +
                 " platform_name, job_runner_name, job_id" +
                 " FROM task_states LEFT JOIN task_jobs USING " +
-                "(cycle, name, " + flow_nums + ") " +
+                "(cycle, name, " +
+                flow_nums +
+                ") " +
                 where_expr +
                 " ORDER BY " +
                 self.JOB_ORDERS.get(order, self.JOB_ORDERS["time_desc"])
@@ -293,18 +312,28 @@ def get_suite_job_entries(
             if exc.message == self.CANNOT_JOIN_FLOW_NUMS:
                 stmt = stmt.replace('flow_nums', 'submit_num')
                 db_data = self._db_exec(
-                   user_name, suite_name, stmt, where_args + limit_args
+                    user_name, suite_name, stmt, where_args + limit_args
                 )
                 eight_zero_warning = True
             else:
                 raise exc
 
         for row in db_data:
             (
-                cycle, name, submit_num, submit_num_max, task_status,
-                time_submit, submit_status,
-                time_run, time_run_exit, run_signal, run_status,
-                user_at_host, batch_sys_name, batch_sys_job_id
+                cycle,
+                name,
+                submit_num,
+                submit_num_max,
+                task_status,
+                time_submit,
+                submit_status,
+                time_run,
+                time_run_exit,
+                run_signal,
+                run_status,
+                user_at_host,
+                batch_sys_name,
+                batch_sys_job_id,
             ) = row[1:]
             entry = {
                 "cycle": cycle,
@@ -469,7 +498,7 @@ def _get_job_logs(self, user_name, suite_name, entries, entry_of):
                     entry["seq_logs_indexes"][seq_key] = int_indexes
                 except ValueError:
                     pass
-            for filename, log_dict in entry["logs"].items():
+            for _, log_dict in entry["logs"].items():
                 # Unset seq_key for singular items
                 if log_dict["seq_key"] not in entry["seq_logs_indexes"]:
                     log_dict["seq_key"] = None
@@ -499,8 +528,10 @@ def get_suite_cycles_summary(
         """
 
         of_n_entries = 0
-        stmt = ("SELECT COUNT(DISTINCT cycle) FROM task_states WHERE " +
-                "submit_num > 0")
+        stmt = (
+            "SELECT COUNT(DISTINCT cycle) FROM task_states WHERE "
+            + "submit_num > 0"
+        )
         try:
             for row in self._db_exec(user_name, suite_name, stmt):
                 of_n_entries = row[0]
@@ -555,7 +586,7 @@ def get_suite_cycles_summary(
             " cycle," +
             " max(time_updated)," +
             " sum(" + states_stmt["active"] + ") AS n_active," +
-            " sum(" + states_stmt["success"] + ") AS n_success,"
+            " sum(" + states_stmt["success"] + ") AS n_success," +
             " sum(" + states_stmt["fail"] + ") AS n_fail"
             " FROM task_states" +
             " GROUP BY cycle" +
@@ -612,12 +643,13 @@ def get_suite_cycles_summary(
                 " FROM task_jobs GROUP BY cycle")
         else:
             fail_events_stmt = " OR ".join(
-                ["event=='%s'" % (name)
-                 for name in TASK_STATUS_GROUPS["fail"]])
+                ["event=='%s'" % (name) for name in TASK_STATUS_GROUPS["fail"]]
+            )
             stmt = (
                 "SELECT cycle," +
                 " sum(" + fail_events_stmt + ") AS n_job_fail" +
-                " FROM task_events GROUP BY cycle")
+                " FROM task_events GROUP BY cycle"
+            )
         self._db_close(user_name, suite_name)
         for cycle, n_job_active, n_job_success, n_job_fail in self._db_exec(
                 user_name, suite_name, stmt):

cylc/flow/review_dao.py

@@ -625,6 +625,15 @@ def select_broadcast_states(self, callback, sort=None):
         for row_idx, row in enumerate(self.connect().execute(stmt)):
             callback(row_idx, list(row))
 
+    def pre_select_broadcast_events(self, order=None):
+        """Query statement and args formation for select_broadcast_events."""
+        form_stmt = r"SELECT time,change,point,namespace,key,value FROM %s"
+        if order == "DESC":
+            ordering = (" ORDER BY " +
+                        "time DESC, point DESC, namespace DESC, key DESC")
+            form_stmt = form_stmt + ordering
+        return form_stmt % self.TABLE_BROADCAST_EVENTS, []
+
     def select_workflow_params(self) -> Iterable[Tuple[str, Optional[str]]]:
         """Select all from workflow_params.
 

cylc/flow/rundb.py

@@ -293,9 +293,6 @@ def get_version(long=False):
         'cylc set-outputs (cylc 8.0-8.2) has been replaced by cylc set',
     'restart':
         'cylc run & cylc restart have been replaced by cylc play',
-    'review':
-        'cylc review has been removed; the latest Cylc 7 version is forward'
-        ' compatible with Cylc 8.',
     'suite-state':
         'cylc suite-state has been replaced by cylc workflow-state',
     'run':

cylc/flow/scripts/cylc.py

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+# THIS FILE IS PART OF THE CYLC SUITE ENGINE.
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""cylc review [start|stop]
+
+Start/stop ad-hoc Cylc Review web service server for browsing users' suite
+logs via an HTTP interface.
+
+With no arguments, the status of the ad-hoc web service server is printed.
+"""
+
+import os
+import signal
+
+from cylc.flow.option_parsers import CylcOptionParser as COP
+from cylc.flow.review import CylcReviewService
+from cylc.flow.terminal import cli_function
+from cylc.flow.ws import _ws_init, _get_server_status
+
+
+IRRELEVANT_OPTS = [
+    '--host',
+    '--user',
+    '--verbose',
+    '--debug',
+    '--quiet',
+    '--timestamp',
+    '--no-timestamp',
+    '--color',
+]
+START = 'start'
+STOP = 'stop'
+
+
+def get_option_parser():
+    parser = COP(
+        __doc__,
+        argdoc=[
+            COP.optional(("start", "Start ad-hoc web service server.")),
+            COP.optional(("stop", "Stop ad-hoc web service server.")),
+        ],
+    )
+
+    parser.add_option(
+        '--port',
+        '-p',
+        help="Port to use for Cylc Review (for start only).",
+        default=8080,
+        action='store',
+        type=int,
+    )
+    parser.add_option(
+        "--non-interactive",
+        "--yes",
+        "-y",
+        help="Switch off interactive prompting i.e. answer yes to everything"
+        " (for stop only).",
+        action="store_true",
+        default=False,
+        dest="non_interactive",
+    )
+    parser.add_option(
+        "--service-root",
+        "-R",
+        help="Include web service name under root of URL (for start only).",
+        action="store",
+        default='/',
+        dest="service_root",
+    )
+    return parser
+
+
+@cli_function(get_option_parser, remove_opts=IRRELEVANT_OPTS)
+def main(_, opts, *args):
+    """Start/Stop the Cylc Review server."""
+    subcmd = args[0] if args else ''
+
+    # Get current server status:
+    status = _get_server_status(CylcReviewService)
+
+    # User has asked to start the server, and it's not already running:
+    if subcmd == START:
+        _ws_init(
+            service_cls=CylcReviewService,
+            port=opts.port,
+            service_root=opts.service_root,
+        )
+    # User has asked to stop or get info on server, server _not_ running:
+    elif not status:
+        print(f'No {CylcReviewService.TITLE} service server running.')
+    else:
+        # Report on status of server:
+        for key, value in sorted(status.items()):
+            print(f'{key}={value}')
+
+        # User has asked to stop the server:
+        if (
+            subcmd == STOP  # User asked for server stop
+            and status.get('pid')  # Server is running
+            and (  # User really wants to stop the server
+                opts.non_interactive
+                or input("Stop server by termination? y/n(default=n)") == "y"
+            )
+        ):
+            try:
+                os.killpg(int(status["pid"]), signal.SIGTERM)
+            except OSError:
+                print("Termination signal failed.")