Keep now on the metrics axis while a job is reporting

src/dstack/_internal/cli/utils/metrics.py

@@ -1,4 +1,4 @@
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Any, List, Optional, Sequence
 
 from rich.console import RenderableType
@@ -10,7 +10,6 @@
 from dstack._internal.core.models.instances import Resources
 from dstack._internal.core.models.metrics import JobMetrics
 from dstack._internal.core.models.runs import Job
-from dstack._internal.utils.common import pretty_date
 
 MAX_SAMPLES = 1000
 """A sample count, not a window: outruns the hour a running job retains, so a young run is
@@ -26,6 +25,9 @@
 """What the server keeps for a running job, and so the widest window there can be."""
 
 AXIS_RULE = "┄"
+NOW = "now"
+
+LIVE_THRESHOLD = timedelta(seconds=3 * WATCH_INTERVAL_SECONDS)
 _FIXED_COLUMNS = 30
 """Everything but the sparklines and the job label: the `gpu=N` column, both numbers, and
 the table's padding. Hand-measured against a `589GB/1480GB`-sized number; a wider one
@@ -47,7 +49,7 @@ def get_metrics_table(
     labels = job_labels(jobs)
     label_width = max((len(label) for label in labels), default=0)
     width = _spark_width(console_width or console.width, label_width)
-    span = _span(metrics)
+    window = _shared_window(metrics)
 
     table = Table(box=None)
     # no headers: the cells read `replica=0` and `gpu=1`, which need no naming
@@ -59,24 +61,19 @@ def get_metrics_table(
     for index, (job, job_metrics) in enumerate(zip(jobs, metrics)):
         if index:
             table.add_row("", "", "", "")
-        _add_job(table, job, job_metrics, width, labels[index], span)
+        _add_job(table, job, job_metrics, width, labels[index], window)
 
-    if span is not None:
+    if window is not None:
         table.add_row("", "", "", "")
         # the axis spans the widest chart drawn: a job with fewer samples than cells draws
         # one cell per sample and cannot fill its share
-        axis = _axis(max(_drawn(m, span, width) for m in metrics), *span)
+        axis = _time_axis(max(_cells_drawn(m, window, width) for m in metrics), *window)
         table.add_row("", "", axis, axis)
     return table
 
 
 def job_labels(jobs: Sequence[Job]) -> List[str]:
-    """`replica=`/`group=` only where they distinguish something, as `dstack ps` does --
-    one replica across four nodes is `job=0..3`, not `replica=0 job=0..3`.
-
-    Unlike `ps`, `job=` is always printed. This table is keyed by job, so every row names
-    one; `replica=` joins it only where there is more than one replica to tell apart.
-    """
+    """A label per job, naming only what tells them apart, as `dstack ps` does."""
     groups = {job.job_spec.replica_group for job in jobs}
     show_group = len(groups) > 1
     show_replica = len({job.job_spec.replica_num for job in jobs}) > 1
@@ -102,59 +99,60 @@ def _add_job(
     metrics: JobMetrics,
     width: int,
     label: str,
-    span: Optional[tuple[datetime, datetime]],
+    window: Optional[tuple[datetime, datetime]],
 ) -> None:
     resources = _get_resources(job)
-    lead = _lead(metrics, span, width)
-    cells = width - lead
+    blanks = _blank_cells(metrics, window, width)
+    cells = width - blanks
     table.add_row(
         label,
         "cpu",
-        _pad(_cpu_cell(metrics, resources, cells), lead),
-        _pad(_memory_cell(metrics, resources, cells), lead),
+        _pad(_cpu_cell(metrics, resources, cells), blanks),
+        _pad(_memory_cell(metrics, resources, cells), blanks),
     )
     for index in range(_gpus_num(metrics, resources)):
         table.add_row(
             "",
             f"gpu={index}",
-            _pad(_gpu_util_cell(metrics, index, cells), lead),
-            _pad(_gpu_memory_cell(metrics, resources, index, cells), lead),
+            _pad(_gpu_util_cell(metrics, index, cells), blanks),
+            _pad(_gpu_memory_cell(metrics, resources, index, cells), blanks),
         )
 
 
-def _span(metrics: Sequence[JobMetrics]) -> Optional[tuple[datetime, datetime]]:
-    """The window every job is drawn against: always the full retention hour.
-
-    Fixed rather than fitted to the data, so a row means the same thing in every
-    invocation and across every job. A job younger than the hour fills only its share of
-    the row and the rest is blank -- which is the fact worth seeing about a replica that
-    started two minutes ago.
-    """
-    windows = [w for w in (_window(m) for m in metrics) if w is not None]
+def _shared_window(metrics: Sequence[JobMetrics]) -> Optional[tuple[datetime, datetime]]:
+    """The window every job is charted against: the newest sample back one retention hour."""
+    windows = [w for w in (_job_window(m) for m in metrics) if w is not None]
     if not windows:
         return None
     latest, earliest = max(w[1] for w in windows), min(w[0] for w in windows)
     return min(earliest, latest - RETENTION), latest
 
 
-def _lead(metrics: JobMetrics, span: Optional[tuple[datetime, datetime]], width: int) -> int:
-    """Cells before this job's first sample -- time it was not running for."""
-    window = _window(metrics)
-    if window is None or span is None:
+def _blank_cells(
+    metrics: JobMetrics, window: Optional[tuple[datetime, datetime]], width: int
+) -> int:
+    """How many cells to leave empty before a job's chart, so it starts where it started."""
+    job_window = _job_window(metrics)
+    if job_window is None or window is None:
         return 0
-    total = (span[1] - span[0]).total_seconds()
+    total = (window[1] - window[0]).total_seconds()
     if total <= 0:
         return 0
-    return min(width - 1, max(0, round((window[0] - span[0]).total_seconds() / total * width)))
+    started = (job_window[0] - window[0]).total_seconds()
+    return min(width - 1, max(0, round(started / total * width)))
 
 
-def _drawn(metrics: JobMetrics, span: Optional[tuple[datetime, datetime]], width: int) -> int:
-    lead = _lead(metrics, span, width)
-    return lead + min(width - lead, _samples_num(metrics))
+def _cells_drawn(
+    metrics: JobMetrics, window: Optional[tuple[datetime, datetime]], width: int
+) -> int:
+    """How many cells a job's chart occupies: its empty lead plus one per sample."""
+    blanks = _blank_cells(metrics, window, width)
+    return blanks + min(width - blanks, _samples_num(metrics))
 
 
-def _pad(cell: Text, lead: int) -> Text:
-    return cell if lead <= 0 else Text.assemble(Text(" " * lead), cell)
+def _pad(cell: Text, blanks: int) -> Text:
+    """Prefix a chart cell with empty cells, for time before the job started."""
+    return cell if blanks <= 0 else Text.assemble(Text(" " * blanks), cell)
 
 
 def _cpu_cell(job_metrics: JobMetrics, resources: Optional[Resources], width: int) -> Text:
@@ -166,15 +164,15 @@ def _cpu_cell(job_metrics: JobMetrics, resources: Optional[Resources], width: in
         values = [v / cpus for v in values]
     # no core count: the value is already normalised to it, and unlike memory there is no
     # total to give the number meaning
-    return _cell(sparkline(values, width, HOST_RAMP), f"{values[-1]:.0f}%")
+    return Text.assemble(sparkline(values, width, HOST_RAMP), " ", f"{values[-1]:.0f}%")
 
 
 def _memory_cell(job_metrics: JobMetrics, resources: Optional[Resources], width: int) -> Text:
     values = _metric_values(job_metrics, "memory_working_set_bytes")
     if not values:
         return no_data()
     total = resources.memory_mib * 1024 * 1024 if resources else None
-    return _level_cell(values, total, width, HOST_RAMP)
+    return _capacity_cell(values, total, width, HOST_RAMP)
 
 
 def _gpu_memory_cell(
@@ -189,65 +187,54 @@ def _gpu_memory_cell(
     total = None
     if resources and index < len(resources.gpus):
         total = resources.gpus[index].memory_mib * 1024 * 1024
-    return _level_cell(values, total, width, GPU_RAMP)
+    return _capacity_cell(values, total, width, GPU_RAMP)
 
 
 def _gpu_util_cell(job_metrics: JobMetrics, index: int, width: int) -> Text:
     values = _metric_values(job_metrics, f"gpu_util_percent_gpu{index}")
     if not values:
         return no_data()
-    return _cell(sparkline(values, width, GPU_RAMP), f"{values[-1]:.0f}%")
+    return Text.assemble(sparkline(values, width, GPU_RAMP), " ", f"{values[-1]:.0f}%")
 
 
-def _level_cell(values: List[float], total: Optional[float], width: int, ramp: Ramp) -> Text:
+def _capacity_cell(values: List[float], total: Optional[float], width: int, ramp: Ramp) -> Text:
+    """A memory chart drawn against capacity, labelled `used/total`."""
     percents = [v / total * 100 for v in values] if total else values
     label = format_memory(values[-1], 0)
     if total:
         label += f"/{format_memory(total, 0)}"
-    return _cell(sparkline(percents, width, ramp), label)
-
-
-def _cell(spark: Text, label: str) -> Text:
-    return Text.assemble(spark, " ", label)
-
-
-def _axis(width: int, first: datetime, last: datetime) -> Text:
-    """`<oldest> ┄┄┄ <newest>`, never wider than the sparkline above it.
+    return Text.assemble(sparkline(percents, width, ramp), " ", label)
 
-    The rule is what pairs the two stamps. UTILIZATION and MEMORY each print one, so the
-    row ends up holding four times, and with the rule left blank the only cue is spacing --
-    which points the wrong way above 88 columns: at 200 there are 66 blanks between a
-    column's own two stamps but only 13 between the columns, so each column's newest time
-    reads as belonging to the next column's oldest.
 
-    A run draws one cell per sample, so for its first few minutes there are fewer cells
-    than two dates need. Dropping the date keeps the axis inside its cell; overflowing
-    instead widens the column and pulls MEMORY out of line with the charts.
-    """
-    left, right = _stamp(first), _stamp(last)
+def _time_axis(width: int, first: datetime, last: datetime) -> Text:
+    """The timeline row printed under the charts, exactly `width` columns wide."""
+    left, right = _time_label(first), _time_label(last)
     if len(left) + len(right) + 3 > width:
-        left, right = _stamp(first, clock_only=True), _stamp(last, clock_only=True)
+        left, right = _time_label(first, clock_only=True), _time_label(last, clock_only=True)
     if len(left) + len(right) + 2 > width:
         return Text("")
     fill = width - len(left) - len(right) - 2
-    return Text(f"{left} " + AXIS_RULE * fill + f" {right}", style="grey42")
+    axis = Text(f"{left} " + AXIS_RULE * fill + " ", style="grey42")
+    axis.append(right, style="bold grey58" if right == NOW else "grey42")
+    return axis
 
 
-def _stamp(moment: datetime, clock_only: bool = False) -> str:
-    if pretty_date(moment) == "now":
-        return "now"
+def _time_label(moment: datetime, clock_only: bool = False) -> str:
+    """One timestamp for the axis: `now` while a job is still reporting, a date otherwise."""
+    if datetime.now(timezone.utc) - moment < LIVE_THRESHOLD:
+        return NOW
     local = moment.astimezone()
     return f"{local:%H:%M}" if clock_only else f"{local.day} {local:%b %H:%M}"
 
 
-def _window(job_metrics: JobMetrics) -> Optional[tuple[datetime, datetime]]:
+def _job_window(job_metrics: JobMetrics) -> Optional[tuple[datetime, datetime]]:
+    """The oldest and newest sample timestamps of one job, or None if it has none."""
     stamps = [t for metric in job_metrics.metrics for t in metric.timestamps]
     return (min(stamps), max(stamps)) if stamps else None
 
 
 def _samples_num(job_metrics: JobMetrics) -> int:
-    """`slices` never draws more cells than it has samples, so the axis must stop there
-    too -- else it claims a span nothing was measured over, and Rich widens the column."""
+    """How many samples the longest series holds."""
     return max((len(metric.timestamps) for metric in job_metrics.metrics), default=0)
 
 
@@ -260,15 +247,15 @@ def _metric_values(job_metrics: JobMetrics, name: str) -> List[Any]:
     return []
 
 
-def _latest(job_metrics: JobMetrics, name: str) -> Optional[Any]:
+def _latest_value(job_metrics: JobMetrics, name: str) -> Optional[Any]:
     values = _metric_values(job_metrics, name)
     return values[-1] if values else None
 
 
 def _gpus_num(job_metrics: JobMetrics, resources: Optional[Resources]) -> int:
     if resources is not None and resources.gpus:
         return len(resources.gpus)
-    detected = _latest(job_metrics, "gpus_detected_num")
+    detected = _latest_value(job_metrics, "gpus_detected_num")
     return int(detected) if detected else 0
 
 

src/tests/_internal/cli/utils/test_metrics.py

@@ -8,6 +8,9 @@
 from rich.theme import Theme
 
 from dstack._internal.cli.utils.metrics import (
+    _job_window,
+    _time_axis,
+    _time_label,
     format_memory,
     get_metrics_table,
     job_labels,
@@ -186,6 +189,18 @@ def test_a_finished_run_cannot_look_live(self, state: str, live: bool):
         if not live:
             assert ":" in axis  # a real clock time, not an age
 
+    @pytest.mark.parametrize("age_seconds,live", [(13, True), (45, False)])
+    def test_a_sample_may_lag_a_few_intervals_and_still_read_as_live(self, age_seconds, live):
+        moment = datetime.now(timezone.utc) - timedelta(seconds=age_seconds)
+        assert (_time_label(moment) == "now") == live
+
+    @pytest.mark.parametrize("state,emphasised", [("running", True), ("terminated", False)])
+    def test_only_a_live_edge_is_emphasised(self, state: str, emphasised: bool):
+        job, metrics = make_run("saturated", state=state)
+        axis = _time_axis(60, *_job_window(metrics))
+        styles = {str(span.style) for span in axis.spans}
+        assert ("bold grey58" in styles) == emphasised
+
 
 class TestJobs:
     def test_every_job_is_shown_and_keyed(self):