Merge branch '8.5.x' into group-trigger-followup

Author: hjoliver

changes.d/6836.fix.md

@@ -0,0 +1 @@
+Fix a bug causing the results of `platform = $(subshell commands)` to be cached, and preventing re-evaluation for each task with the same config.

changes.d/6911.fix.md

@@ -0,0 +1 @@
+Fix an issue where trigger could fail to run a task, removing it instead.

changes.d/6914.fix.md

@@ -0,0 +1 @@
+Fixed bug where jobs in the UI could regress to an earlier state.

changes.d/6924.feat.md

@@ -0,0 +1 @@
+Include `log/version/` files in `cylc cat-log` and the UI log view.

cylc/flow/cfgspec/workflow.py

@@ -1158,14 +1158,60 @@ def get_script_common_text(this: str, example: Optional[str] = None):
                 .. versionadded:: 8.3.0
             ''')
             Conf('platform', VDR.V_STRING, desc='''
-                The name of a compute resource defined in
-                :cylc:conf:`global.cylc[platforms]` or
-                :cylc:conf:`global.cylc[platform groups]`.
+                The platform to submit jobs to.
 
-                The platform specifies the host(s) that the tasks' jobs
+                The
+                :cylc:conf:`platform <global.cylc[platforms][<platform name>]>`
+                specifies the host(s) that the tasks' jobs
                 will run on and where (if necessary) files need to be
                 installed, and what job runner will be used.
 
+                This can be:
+
+                * A :cylc:conf:`platform <global.cylc[platforms]>`,
+                * a :cylc:conf:`platform group <global.cylc[platform groups]>`,
+                * or a command which returns a platform or platform group.
+
+                To see what platforms have been configured at your site, run
+                ``cylc config --platform-names``.
+
+                .. rubric:: Commands:
+
+                The ``platform`` can be set to a command which returns the name
+                of the platform to submit jobs to using the ``$()`` syntax,
+                i.e:
+
+                .. code-block:: cylc
+
+                   platform = $(command)
+
+                The configured command will be evaluated for each job
+                submission (i.e, different submissions of the same task may
+                submit on different platforms).
+
+                Cylc batches job submissions, so when multiple jobs are
+                submitted at the same time, using a platform defined by the
+                same command, the command will be run once, and all jobs in the
+                batch will submit to the same platform.
+
+                Note: do not use a command to configure a list of login nodes.
+                Instead, define a platform and configure the login nodes it
+                can use; see
+                :ref:`config.platforms.cluster_with_multiple_login_nodes`.
+
+                .. rubric:: Examples:
+
+                .. code-block:: cylc
+
+                   # run the job on the same host the Cylc scheduler runs on
+                   platform = localhost
+
+                   # run the job on a platform (or platform group) called hpc
+                   platform = hpc
+
+                   # run a command to select the platform (or platform group):
+                   platform = $(select-platform)
+
                 .. versionadded:: 8.0.0
             ''')
             Conf('inherit', VDR.V_STRING_LIST, desc='''

cylc/flow/commands.py

@@ -92,7 +92,6 @@
 from cylc.flow.task_id import TaskID
 from cylc.flow.task_state import (
     TASK_STATUS_PREPARING,
-    TASK_STATUS_WAITING,
     TASK_STATUSES_ACTIVE,
 )
 from cylc.flow.taskdef import generate_graph_children
@@ -751,7 +750,7 @@ def _force_trigger_tasks(
         # Remove non group start and final-status group start tasks, and
         # trigger them from scratch (so only the TaskDef matters).
 
-        # Waiting group start tasks are not removed, but a reload would
+        # Group start tasks are not removed, but a reload would
         # replace them, so using the TaskDef is fine.
 
         if not any(
@@ -774,12 +773,13 @@ def _force_trigger_tasks(
                             (str(itask.point), itask.tdef.name)] = (label, msg)
 
             if itask.state(TASK_STATUS_PREPARING, *TASK_STATUSES_ACTIVE):
+                # This is a live active group start task
                 warnings_has_job.append(str(itask))
                 # Just merge the flows.
                 schd.pool.merge_flows(itask, flow_nums)
 
-            elif itask.state(TASK_STATUS_WAITING):
-                # This is a waiting active group start task...
+            else:
+                # This is a non-live active group start task...
                 # ... satisfy off-group (i.e. all) prerequisites
                 itask.state.set_all_task_prerequisites_satisfied()
                 # ... and satisfy all xtrigger prerequisites.
@@ -791,8 +791,7 @@ def _force_trigger_tasks(
 
                 # Trigger group start task.
                 schd.pool.queue_or_trigger(itask, on_resume)
-            else:
-                active_to_remove.append(itask)
+
         else:
             active_to_remove.append(itask)
 

cylc/flow/data_store_mgr.py

@@ -67,6 +67,7 @@
     Set,
     TYPE_CHECKING,
     Tuple,
+    Union,
 )
 import zlib
 
@@ -1609,13 +1610,20 @@ def _apply_broadcasts_to_runtime(self, tokens, rtconfig):
             poverride(rtconfig, overrides, prepend=True)
         return rtconfig
 
-    def insert_job(self, name, cycle_point, status, job_conf):
+    def insert_job(
+        self,
+        name: str,
+        cycle_point: Union['PointBase', str],
+        status: str,
+        job_conf: dict,
+    ):
         """Insert job into data-store.
 
         Args:
-            name (str): Corresponding task name.
-            cycle_point (str|PointBase): Cycle point string
-            job_conf (dic):
+            name: Corresponding task name.
+            cycle_point: Cycle point string
+            status: The task's state.
+            job_conf:
                 Dictionary of job configuration used to generate
                 the job script.
                 (see TaskJobManager._prep_submit_task_job_impl)
@@ -1625,6 +1633,11 @@ def insert_job(self, name, cycle_point, status, job_conf):
             None
 
         """
+        if status not in JOB_STATUS_SET:
+            # Ignore task-only states e.g. preparing
+            # https://github.com/cylc/cylc-flow/issues/4994
+            return
+
         sub_num = job_conf['submit_num']
         tp_tokens = self.id_.duplicate(
             cycle=str(cycle_point),
@@ -1641,9 +1654,6 @@ def insert_job(self, name, cycle_point, status, job_conf):
             # Job already exists (i.e. post-submission submit failure)
             return
 
-        if status not in JOB_STATUS_SET:
-            return
-
         j_buf = PbJob(
             stamp=f'{j_id}@{update_time}',
             id=j_id,
@@ -2761,12 +2771,12 @@ def delta_job_msg(self, tokens: Tokens, msg: str) -> None:
 
     def delta_job_attr(
         self,
-        tokens: Tokens,
+        itask: 'TaskProxy',
         attr_key: str,
         attr_val: Any,
     ) -> None:
         """Set job attribute."""
-        j_id, job = self.store_node_fetcher(tokens)
+        j_id, job = self.store_node_fetcher(itask.job_tokens)
         if not job:
             return
         j_delta = PbJob(stamp=f'{j_id}@{time()}')
@@ -2779,12 +2789,19 @@ def delta_job_attr(
 
     def delta_job_state(
         self,
-        tokens: Tokens,
+        itask: 'TaskProxy',
         status: str,
     ) -> None:
         """Set job state."""
-        j_id, job = self.store_node_fetcher(tokens)
-        if not job or status not in JOB_STATUS_SET:
+        if status not in JOB_STATUS_SET:
+            # Ignore task-only states e.g. preparing
+            return
+        j_id, job = self.store_node_fetcher(itask.job_tokens)
+        if not job or (
+            # Don't cause backwards state change:
+            JOB_STATUSES_ALL.index(status) <= JOB_STATUSES_ALL.index(job.state)
+            and not itask.job_vacated
+        ):
             return
         j_delta = PbJob(
             stamp=f'{j_id}@{time()}',
@@ -2798,15 +2815,15 @@ def delta_job_state(
 
     def delta_job_time(
         self,
-        tokens: Tokens,
+        itask: 'TaskProxy',
         event_key: str,
         time_str: Optional[str] = None,
     ) -> None:
         """Set an event time in job pool object.
 
         Set values of both event_key + '_time' and event_key + '_time_string'.
         """
-        j_id, job = self.store_node_fetcher(tokens)
+        j_id, job = self.store_node_fetcher(itask.job_tokens)
         if not job:
             return
         j_delta = PbJob(stamp=f'{j_id}@{time()}')

cylc/flow/install_plugins/log_vc_info.py

@@ -285,8 +285,8 @@ def write_vc_info(
     )
     info_file.parent.mkdir(exist_ok=True, parents=True)
     with open(info_file, 'w') as f:
-        f.write(
-            json.dumps(info, indent=JSON_INDENT)
+        print(
+            json.dumps(info, indent=JSON_INDENT), file=f
         )
 
 
@@ -356,7 +356,7 @@ def write_diff(
         try:
             _run_cmd(vcs, args, repo_path, stdout=f)
         except VCSMissingBaseError as exc:
-            f.write(f"# No diff - {exc}")
+            print(f"# No diff - {exc}", file=f)
     return diff_file
 
 

cylc/flow/platforms.py

@@ -83,7 +83,8 @@ def log_platform_event(
 def get_platform(
     task_conf: Optional[str] = None,
     task_name: str = UNKNOWN_TASK,
-    bad_hosts: Optional[Set[str]] = None
+    bad_hosts: Optional[Set[str]] = None,
+    evaluated_host: Optional[str] = None,
 ) -> Dict[str, Any]:
     ...
 
@@ -92,7 +93,8 @@ def get_platform(
 def get_platform(
     task_conf: Union[dict, 'OrderedDictWithDefaults'],
     task_name: str = UNKNOWN_TASK,
-    bad_hosts: Optional[Set[str]] = None
+    bad_hosts: Optional[Set[str]] = None,
+    evaluated_host: Optional[str] = None,
 ) -> Optional[Dict[str, Any]]:
     ...
 
@@ -108,7 +110,8 @@ def get_platform(
 def get_platform(
     task_conf: Union[str, dict, 'OrderedDictWithDefaults', None] = None,
     task_name: str = UNKNOWN_TASK,
-    bad_hosts: Optional[Set[str]] = None
+    bad_hosts: Optional[Set[str]] = None,
+    evaluated_host: Optional[str] = None,
 ) -> Optional[Dict[str, Any]]:
     """Get a platform.
 
@@ -121,6 +124,7 @@ def get_platform(
         task_name: Help produce more helpful error messages.
         bad_hosts: A set of hosts known to be unreachable (had an ssh 255
             error)
+        evaluated_host: Host name evaluated from platform subshell.
 
     Returns:
         platform: A platform definition dictionary. Uses either
@@ -169,6 +173,7 @@ def get_platform(
             glbl_cfg().get(['platforms']),
             task_job_section,
             task_remote_section,
+            evaluated_host,
         ),
         bad_hosts=bad_hosts,
     )
@@ -330,7 +335,8 @@ def get_platform_from_group(
 def platform_name_from_job_info(
     platforms: Union[dict, 'OrderedDictWithDefaults'],
     job: Dict[str, Any],
-    remote: Dict[str, Any]
+    remote: Dict[str, Any],
+    evaluated_host: Optional[str] = None,
 ) -> str:
     """
     Find out which job platform to use given a list of possible platforms
@@ -385,6 +391,7 @@ def platform_name_from_job_info(
         job: Workflow config [runtime][TASK][job] section.
         remote: Workflow config [runtime][TASK][remote] section.
         platforms: Dictionary containing platform definitions.
+        evaluated_host: Host is the result of evaluating a subshell.
 
     Returns:
         platform: string representing a platform from the global config.
@@ -422,7 +429,9 @@ def platform_name_from_job_info(
 
     # NOTE: Do NOT use .get() on OrderedDictWithDefaults -
     # https://github.com/cylc/cylc-flow/pull/4975
-    if 'host' in remote and remote['host']:
+    if evaluated_host:
+        task_host = evaluated_host
+    elif 'host' in remote and remote['host']:
         task_host = remote['host']
     else:
         task_host = 'localhost'

cylc/flow/scripts/cat_log.py

@@ -106,6 +106,7 @@
     ),
     'i': ('install log', r'install/*-*install.log'),
     's': ('scheduler log', r'scheduler/*-*start*.log'),
+    'v': ('version control info (JSON)', r'version/*'),
 }