Add estimated finish time to Job schema

Author: MetRonnie

changes.d/6987.feat.md

@@ -0,0 +1 @@
+Enable estimated finish times for jobs in the UI.

cylc/flow/data_messages.proto

@@ -158,6 +158,7 @@ message PbJob {
     optional string cycle_point = 31; /* filter item */
     repeated string messages = 32;
     optional PbRuntime runtime = 33;
+    optional string estimated_finish_time = 34;
 }
 
 message PbTask {

cylc/flow/data_messages_pb2.py

@@ -184,7 +184,7 @@ class PbRuntime(_message.Message):
     def __init__(self, platform: _Optional[str] = ..., script: _Optional[str] = ..., init_script: _Optional[str] = ..., env_script: _Optional[str] = ..., err_script: _Optional[str] = ..., exit_script: _Optional[str] = ..., pre_script: _Optional[str] = ..., post_script: _Optional[str] = ..., work_sub_dir: _Optional[str] = ..., execution_polling_intervals: _Optional[str] = ..., execution_retry_delays: _Optional[str] = ..., execution_time_limit: _Optional[str] = ..., submission_polling_intervals: _Optional[str] = ..., submission_retry_delays: _Optional[str] = ..., directives: _Optional[str] = ..., environment: _Optional[str] = ..., outputs: _Optional[str] = ..., completion: _Optional[str] = ..., run_mode: _Optional[str] = ...) -> None: ...
 
 class PbJob(_message.Message):
-    __slots__ = ("stamp", "id", "submit_num", "state", "task_proxy", "submitted_time", "started_time", "finished_time", "job_id", "job_runner_name", "execution_time_limit", "platform", "job_log_dir", "name", "cycle_point", "messages", "runtime")
+    __slots__ = ("stamp", "id", "submit_num", "state", "task_proxy", "submitted_time", "started_time", "finished_time", "job_id", "job_runner_name", "execution_time_limit", "platform", "job_log_dir", "name", "cycle_point", "messages", "runtime", "estimated_finish_time")
     STAMP_FIELD_NUMBER: _ClassVar[int]
     ID_FIELD_NUMBER: _ClassVar[int]
     SUBMIT_NUM_FIELD_NUMBER: _ClassVar[int]
@@ -202,6 +202,7 @@ class PbJob(_message.Message):
     CYCLE_POINT_FIELD_NUMBER: _ClassVar[int]
     MESSAGES_FIELD_NUMBER: _ClassVar[int]
     RUNTIME_FIELD_NUMBER: _ClassVar[int]
+    ESTIMATED_FINISH_TIME_FIELD_NUMBER: _ClassVar[int]
     stamp: str
     id: str
     submit_num: int
@@ -219,7 +220,8 @@ class PbJob(_message.Message):
     cycle_point: str
     messages: _containers.RepeatedScalarFieldContainer[str]
     runtime: PbRuntime
-    def __init__(self, stamp: _Optional[str] = ..., id: _Optional[str] = ..., submit_num: _Optional[int] = ..., state: _Optional[str] = ..., task_proxy: _Optional[str] = ..., submitted_time: _Optional[str] = ..., started_time: _Optional[str] = ..., finished_time: _Optional[str] = ..., job_id: _Optional[str] = ..., job_runner_name: _Optional[str] = ..., execution_time_limit: _Optional[float] = ..., platform: _Optional[str] = ..., job_log_dir: _Optional[str] = ..., name: _Optional[str] = ..., cycle_point: _Optional[str] = ..., messages: _Optional[_Iterable[str]] = ..., runtime: _Optional[_Union[PbRuntime, _Mapping]] = ...) -> None: ...
+    estimated_finish_time: str
+    def __init__(self, stamp: _Optional[str] = ..., id: _Optional[str] = ..., submit_num: _Optional[int] = ..., state: _Optional[str] = ..., task_proxy: _Optional[str] = ..., submitted_time: _Optional[str] = ..., started_time: _Optional[str] = ..., finished_time: _Optional[str] = ..., job_id: _Optional[str] = ..., job_runner_name: _Optional[str] = ..., execution_time_limit: _Optional[float] = ..., platform: _Optional[str] = ..., job_log_dir: _Optional[str] = ..., name: _Optional[str] = ..., cycle_point: _Optional[str] = ..., messages: _Optional[_Iterable[str]] = ..., runtime: _Optional[_Union[PbRuntime, _Mapping]] = ..., estimated_finish_time: _Optional[str] = ...) -> None: ...
 
 class PbTask(_message.Message):
     __slots__ = ("stamp", "id", "name", "meta", "mean_elapsed_time", "depth", "proxies", "namespace", "parents", "first_parent", "runtime")

cylc/flow/data_messages_pb2.pyi

@@ -54,23 +54,30 @@
 
 """
 
+from collections import (
+    Counter,
+    deque,
+)
 from contextlib import suppress
-from collections import Counter, deque
 from copy import deepcopy
 import json
 from time import time
 from typing import (
+    TYPE_CHECKING,
     Any,
     Dict,
-    Optional,
     List,
+    Literal,
+    Optional,
     Set,
-    TYPE_CHECKING,
     Tuple,
 )
 import zlib
 
-from cylc.flow import __version__ as CYLC_VERSION, LOG
+from cylc.flow import (
+    LOG,
+    __version__ as CYLC_VERSION,
+)
 from cylc.flow.cycling.loader import get_point
 from cylc.flow.data_messages_pb2 import (
     AllDeltas,
@@ -98,37 +105,46 @@
 from cylc.flow.parsec.util import (
     listjoin,
     pdeepcopy,
-    poverride
+    poverride,
 )
 from cylc.flow.run_modes import RunMode
-from cylc.flow.workflow_status import (
-    get_workflow_status,
-    get_workflow_status_msg,
+from cylc.flow.task_job_logs import (
+    JOB_LOG_OPTS,
+    get_task_job_log,
 )
-from cylc.flow.task_job_logs import JOB_LOG_OPTS, get_task_job_log
 from cylc.flow.task_proxy import TaskProxy
 from cylc.flow.task_state import (
-    TASK_STATUS_WAITING,
-    TASK_STATUS_SUBMITTED,
-    TASK_STATUS_SUBMIT_FAILED,
+    TASK_STATUS_FAILED,
     TASK_STATUS_RUNNING,
+    TASK_STATUS_SUBMIT_FAILED,
+    TASK_STATUS_SUBMITTED,
     TASK_STATUS_SUCCEEDED,
-    TASK_STATUS_FAILED,
-    TASK_STATUSES_ORDERED
+    TASK_STATUS_WAITING,
+    TASK_STATUSES_FINAL,
+    TASK_STATUSES_ORDERED,
 )
 from cylc.flow.task_state_prop import extract_group_state
-from cylc.flow.taskdef import generate_graph_parents, generate_graph_children
-from cylc.flow.task_state import TASK_STATUSES_FINAL
+from cylc.flow.taskdef import (
+    generate_graph_children,
+    generate_graph_parents,
+)
 from cylc.flow.util import (
+    deserialise_set,
     serialise_set,
-    deserialise_set
 )
 from cylc.flow.wallclock import (
     TIME_ZONE_LOCAL_INFO,
     TIME_ZONE_UTC_INFO,
-    get_utc_mode
+    get_time_string_from_unix_time as time2str,
+    get_unix_time_from_time_string as str2time,
+    get_utc_mode,
+)
+from cylc.flow.workflow_status import (
+    get_workflow_status,
+    get_workflow_status_msg,
 )
 
+
 if TYPE_CHECKING:
     from cylc.flow.cycling import PointBase
     from cylc.flow.flow_mgr import FlowNums
@@ -2809,19 +2825,23 @@ def delta_job_state(
     def delta_job_time(
         self,
         itask: 'TaskProxy',
-        event_key: str,
-        time_str: Optional[str] = None,
+        event_key: Literal['submitted', 'started', 'finished'],
+        time_str: str,
     ) -> None:
-        """Set an event time in job pool object.
-
-        Set values of both event_key + '_time' and event_key + '_time_string'.
-        """
+        """Set an event time in job pool object."""
         j_id, job = self.store_node_fetcher(itask.job_tokens)
         if not job:
             return
         j_delta = PbJob(stamp=f'{j_id}@{time()}')
         time_attr = f'{event_key}_time'
         setbuff(j_delta, time_attr, time_str)
+        if (
+            event_key == 'started'
+            and (run_time := task_mean_elapsed_time(itask.tdef)) is not None
+        ):
+            j_delta.estimated_finish_time = (
+                time2str(str2time(time_str) + run_time) if time_str else ''
+            )
         self.updated[JOBS].setdefault(
             j_id,
             PbJob(id=j_id)

cylc/flow/data_store_mgr.py

@@ -936,6 +936,13 @@ class Meta:
     finished_time = String(
         description='The time this job finished running (if it has yet).',
     )
+    estimated_finish_time = String(
+        description=(
+            "The estimated time this job will finish, if applicable. "
+            "This is based on the task's mean run time, or if not available, "
+            "the execution time limit."
+        ),
+    )
     job_id = ID(
         description='The ID of this job in the job runner it was submitted to.'
     )

cylc/flow/network/schema.py

@@ -14,8 +14,8 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from logging import INFO
 import logging
+from logging import INFO
 from typing import (
     Iterable,
     List,
@@ -25,12 +25,13 @@
 
 import pytest
 
+from cylc.flow import LOG
 from cylc.flow.commands import (
+    force_trigger_tasks,
     run_cmd,
-    force_trigger_tasks
 )
-from cylc.flow import LOG
 from cylc.flow.data_messages_pb2 import (
+    PbJob,
     PbPrerequisite,
     PbTaskProxy,
 )
@@ -42,7 +43,10 @@
     TASKS,
     WORKFLOW,
 )
-from cylc.flow.id import TaskTokens, Tokens
+from cylc.flow.id import (
+    TaskTokens,
+    Tokens,
+)
 from cylc.flow.network.log_stream_handler import ProtobufStreamHandler
 from cylc.flow.scheduler import Scheduler
 from cylc.flow.task_events_mgr import TaskEventsManager
@@ -51,6 +55,7 @@
     TASK_OUTPUT_SUBMITTED,
     TASK_OUTPUT_SUCCEEDED,
 )
+from cylc.flow.task_proxy import TaskProxy
 from cylc.flow.task_state import (
     TASK_STATUS_FAILED,
     TASK_STATUS_PREPARING,
@@ -125,6 +130,11 @@ def get_pb_prereqs(schd: 'Scheduler') -> 'List[PbPrerequisite]':
     ]
 
 
+def get_pb_job(schd: Scheduler, itask: TaskProxy) -> PbJob:
+    """Get the protobuf job for a given task from the data store."""
+    return schd.data_store_mgr.data[schd.id][JOBS][itask.job_tokens.id]
+
+
 @pytest.fixture(scope='module')
 async def mod_harness(mod_flow, mod_scheduler, mod_start):
     flow_def = {
@@ -832,9 +842,6 @@ async def test_log_events(one: Scheduler, start):
 
 async def test_no_backwards_job_state_change(one: Scheduler, start):
     """It should not allow backwards job state changes."""
-    def get_job_state(itask):
-        return one.data_store_mgr.data[one.id][JOBS][itask.job_tokens.id].state
-
     async with start(one):
         itask = one.pool.get_tasks()[0]
         itask.state_reset(TASK_STATUS_PREPARING)
@@ -843,9 +850,59 @@ def get_job_state(itask):
 
         one.task_events_mgr.process_message(itask, INFO, TASK_OUTPUT_STARTED)
         await one.update_data_structure()
-        assert get_job_state(itask) == TASK_STATUS_RUNNING
+        assert get_pb_job(one, itask).state == TASK_STATUS_RUNNING
 
         # Simulate late arrival of "submitted" message
         one.task_events_mgr.process_message(itask, INFO, TASK_OUTPUT_SUBMITTED)
         await one.update_data_structure()
-        assert get_job_state(itask) == TASK_STATUS_RUNNING
+        assert get_pb_job(one, itask).state == TASK_STATUS_RUNNING
+
+
+async def test_job_estimated_finish_time(one_conf, flow, scheduler, start):
+    """It should set estimated_finish_time on job elements along with
+    started_time."""
+    wid = flow({
+        **one_conf,
+        'scheduler': {'UTC mode': True},
+        'runtime': {
+            'one': {'execution time limit': 'PT2M'},
+        },
+    })
+    schd: Scheduler = scheduler(wid)
+    date = '2081-07-02T'
+
+    async def start_job(itask: TaskProxy, start_time: str):
+        if not schd.pool.get_task(itask.point, itask.tdef.name):
+            schd.pool.add_to_pool(itask)
+            await schd.update_data_structure()
+        itask.state_reset(TASK_STATUS_PREPARING)
+        itask.submit_num += 1
+        itask.jobs = []
+        schd.task_events_mgr.process_message(
+            itask, INFO, TASK_OUTPUT_SUBMITTED  # submit time irrelevant
+        )
+        await schd.update_data_structure()
+        schd.task_events_mgr.process_message(
+            itask, INFO, TASK_OUTPUT_STARTED, f'{date}{start_time}'
+        )
+        await schd.update_data_structure()
+
+    async with start(schd):
+        itask = schd.pool.get_tasks()[0]
+        await start_job(itask, '06:00:00Z')
+        # 1st job: estimate based on execution time limit:
+        assert (
+            get_pb_job(schd, itask).estimated_finish_time
+            == f'{date}06:02:00Z'
+        )
+
+        # Finish this job and start a new one:
+        schd.task_events_mgr.process_message(
+            itask, INFO, TASK_OUTPUT_SUCCEEDED, f'{date}06:00:40Z'
+        )
+        await start_job(itask, '06:01:00Z')
+        # >=2nd job: estimate based on mean of previous jobs:
+        assert (
+            get_pb_job(schd, itask).estimated_finish_time
+            == f'{date}06:01:40Z'
+        )