Merge pull request #6925 from cylc/8.5.x-sync

🤖 Merge 8.5.x-sync into master

Author: MetRonnie

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/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/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/*'),
 }
 
 

cylc/flow/task_events_mgr.py

@@ -797,11 +797,9 @@ def process_message(
             # ... but either way update the job ID in the job proxy (it only
             # comes in via the submission message).
             if itask.run_mode != RunMode.SIMULATION:
-                job_tokens = itask.tokens.duplicate(
-                    job=str(itask.submit_num)
-                )
                 self.data_store_mgr.delta_job_attr(
-                    job_tokens, 'job_id', itask.summary['submit_method_id'])
+                    itask, 'job_id', itask.summary['submit_method_id']
+                )
             else:
                 # In simulation mode submitted implies started:
                 self.spawn_children(itask, TASK_OUTPUT_STARTED, forced)
@@ -1354,9 +1352,8 @@ def _process_message_failed(
         if event_time is None:
             event_time = get_current_time_string()
         itask.set_summary_time('finished', event_time)
-        job_tokens = itask.tokens.duplicate(job=str(itask.submit_num))
-        self.data_store_mgr.delta_job_time(job_tokens, 'finished', event_time)
-        self.data_store_mgr.delta_job_state(job_tokens, TASK_STATUS_FAILED)
+        self.data_store_mgr.delta_job_time(itask, 'finished', event_time)
+        self.data_store_mgr.delta_job_state(itask, TASK_STATUS_FAILED)
         self.workflow_db_mgr.put_update_task_jobs(itask, {
             "run_status": 1,
             "time_run_exit": event_time,
@@ -1394,12 +1391,11 @@ def _process_message_failed(
 
     def _process_message_started(self, itask, event_time, forced):
         """Helper for process_message, handle a started message."""
+        self.data_store_mgr.delta_job_time(itask, 'started', event_time)
+        self.data_store_mgr.delta_job_state(itask, TASK_STATUS_RUNNING)
         if itask.job_vacated:
             itask.job_vacated = False
             LOG.info(f"[{itask}] Vacated job restarted")
-        job_tokens = itask.tokens.duplicate(job=str(itask.submit_num))
-        self.data_store_mgr.delta_job_time(job_tokens, 'started', event_time)
-        self.data_store_mgr.delta_job_state(job_tokens, TASK_STATUS_RUNNING)
         itask.set_summary_time('started', event_time)
         self.workflow_db_mgr.put_update_task_jobs(itask, {
             "time_run": itask.summary['started_time_string']})
@@ -1430,9 +1426,8 @@ def _process_message_succeeded(self, itask, event_time, forced):
         Ignore forced.
         """
 
-        job_tokens = itask.tokens.duplicate(job=str(itask.submit_num))
-        self.data_store_mgr.delta_job_time(job_tokens, 'finished', event_time)
-        self.data_store_mgr.delta_job_state(job_tokens, TASK_STATUS_SUCCEEDED)
+        self.data_store_mgr.delta_job_time(itask, 'finished', event_time)
+        self.data_store_mgr.delta_job_state(itask, TASK_STATUS_SUCCEEDED)
         itask.set_summary_time('finished', event_time)
         self.workflow_db_mgr.put_update_task_jobs(itask, {
             "run_status": 0,
@@ -1505,13 +1500,9 @@ def _process_message_submit_failed(
             self.setup_event_handlers(itask, self.EVENT_SUBMIT_RETRY, msg)
 
         # Register newly submit-failed job with the database and datastore.
-        job_tokens = itask.tokens.duplicate(job=str(itask.submit_num))
         self._insert_task_job(
             itask, event_time, self.JOB_SUBMIT_FAIL_FLAG, forced=forced)
-        self.data_store_mgr.delta_job_state(
-            job_tokens,
-            TASK_STATUS_SUBMIT_FAILED
-        )
+        self.data_store_mgr.delta_job_state(itask, TASK_STATUS_SUBMIT_FAILED)
         self._reset_job_timers(itask)
 
         return no_retries
@@ -1560,24 +1551,12 @@ def _process_message_submitted(
         # Do after itask has changed state
         self._insert_task_job(
             itask, event_time, self.JOB_SUBMIT_SUCCESS_FLAG, forced=forced)
-        job_tokens = itask.tokens.duplicate(job=str(itask.submit_num))
-        self.data_store_mgr.delta_job_time(
-            job_tokens,
-            'submitted',
-            event_time,
-        )
+        self.data_store_mgr.delta_job_time(itask, 'submitted', event_time)
         if itask.run_mode == RunMode.SIMULATION:
             # Simulate job started as well.
-            self.data_store_mgr.delta_job_time(
-                job_tokens,
-                'started',
-                event_time,
-            )
+            self.data_store_mgr.delta_job_time(itask, 'started', event_time)
         else:
-            self.data_store_mgr.delta_job_state(
-                job_tokens,
-                TASK_STATUS_SUBMITTED,
-            )
+            self.data_store_mgr.delta_job_state(itask, TASK_STATUS_SUBMITTED)
 
     def _insert_task_job(
         self,

cylc/flow/task_job_mgr.py

@@ -1220,10 +1220,10 @@ def _prep_submit_task_job(
                 f"\"{itask.identity}\" the following are not compatible:\n"
             )
 
-        host_n, platform_name = None, None
+        host_name, platform_name = None, None
         try:
             if rtconfig['remote']['host'] is not None:
-                host_n = self.task_remote_mgr.eval_host(
+                host_name = self.task_remote_mgr.eval_host(
                     rtconfig['remote']['host']
                 )
             else:
@@ -1242,27 +1242,26 @@ def _prep_submit_task_job(
             return False
         else:
             # host/platform select not ready
-            if host_n is None and platform_name is None:
+            if host_name is None and platform_name is None:
                 return None
             elif (
-                host_n is None
+                host_name is None
                 and rtconfig['platform']
                 and rtconfig['platform'] != platform_name
             ):
                 LOG.debug(
                     f"for task {itask.identity}: platform = "
                     f"{rtconfig['platform']} evaluated as {platform_name}"
                 )
-                rtconfig['platform'] = platform_name
+
             elif (
                 platform_name is None
-                and rtconfig['remote']['host'] != host_n
+                and rtconfig['remote']['host'] != host_name
             ):
                 LOG.debug(
                     f"[{itask}] host = "
-                    f"{rtconfig['remote']['host']} evaluated as {host_n}"
+                    f"{rtconfig['remote']['host']} evaluated as {host_name}"
                 )
-                rtconfig['remote']['host'] = host_n
 
             try:
                 platform = cast(
@@ -1271,7 +1270,10 @@ def _prep_submit_task_job(
                     # return early if the subshell is still evaluating.
                     'dict',
                     get_platform(
-                        rtconfig, itask.tdef.name, bad_hosts=self.bad_hosts
+                        platform_name or rtconfig,
+                        itask.tdef.name,
+                        bad_hosts=self.bad_hosts,
+                        evaluated_host=host_name,
                     ),
                 )
             except PlatformLookupError as exc:

cylc/flow/task_proxy.py

@@ -315,6 +315,11 @@ def __init__(
                 )
             )
 
+    @property
+    def job_tokens(self) -> 'Tokens':
+        """Return the job tokens for this task proxy."""
+        return self.tokens.duplicate(job=str(self.submit_num))
+
     def __repr__(self) -> str:
         return f"<{type(self).__name__} {self.identity} {self.state}>"