From 0fcad278fb148725ca989178f3948c5e134ff9f0 Mon Sep 17 00:00:00 2001 From: Andrey Cheptsov Date: Wed, 12 Aug 2026 13:47:07 +0200 Subject: [PATCH 1/4] Keep `now` on the metrics axis while a job is reporting The axis labelled its right edge `now` only when the newest sample was under ten seconds old, borrowing `pretty_date`'s threshold. Collection runs every ten seconds, so samples routinely arrive older than that: polling a live job, three renders in ten showed an absolute time instead, and under `--watch` the edge alternated between the two. Thresholds on three collection intervals instead. A job whose instance goes unreachable still shows its last timestamp -- that gap is minutes, not seconds. Draws `now` bold, in the grey `no data` already uses, so a live run is distinguishable at a glance from one that stopped reporting. Co-Authored-By: Claude Opus 5 (1M context) --- src/dstack/_internal/cli/utils/metrics.py | 14 +++++++++----- src/tests/_internal/cli/utils/test_metrics.py | 15 +++++++++++++++ 2 files changed, 24 insertions(+), 5 deletions(-) diff --git a/src/dstack/_internal/cli/utils/metrics.py b/src/dstack/_internal/cli/utils/metrics.py index 22ce839a0..4db0536f4 100644 --- a/src/dstack/_internal/cli/utils/metrics.py +++ b/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 @@ -230,12 +232,14 @@ def _axis(width: int, first: datetime, last: datetime) -> Text: 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" + 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}" diff --git a/src/tests/_internal/cli/utils/test_metrics.py b/src/tests/_internal/cli/utils/test_metrics.py index 17a81244b..b9ceee70c 100644 --- a/src/tests/_internal/cli/utils/test_metrics.py +++ b/src/tests/_internal/cli/utils/test_metrics.py @@ -8,6 +8,9 @@ from rich.theme import Theme from dstack._internal.cli.utils.metrics import ( + _axis, + _stamp, + _window, 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 (_stamp(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 = _axis(60, *_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): From 7d0ef163f795d3ebb22a0378302e9ff425256bbd Mon Sep 17 00:00:00 2001 From: Andrey Cheptsov Date: Wed, 12 Aug 2026 14:12:40 +0200 Subject: [PATCH 2/4] Drop docstrings that summarise the code they sit on Five of the six in this module restated what the function below already said, or argued for the option taken over one that was not. The remaining one records that the server sends samples newest-first, which is a fact from outside this file and draws every chart backwards if missed. Co-Authored-By: Claude Opus 5 (1M context) --- src/dstack/_internal/cli/utils/metrics.py | 28 ----------------------- 1 file changed, 28 deletions(-) diff --git a/src/dstack/_internal/cli/utils/metrics.py b/src/dstack/_internal/cli/utils/metrics.py index 4db0536f4..363b7e2a9 100644 --- a/src/dstack/_internal/cli/utils/metrics.py +++ b/src/dstack/_internal/cli/utils/metrics.py @@ -73,12 +73,6 @@ def get_metrics_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. - """ 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 @@ -125,13 +119,6 @@ def _add_job( 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] if not windows: return None @@ -140,7 +127,6 @@ def _span(metrics: Sequence[JobMetrics]) -> Optional[tuple[datetime, datetime]]: 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: return 0 @@ -214,18 +200,6 @@ def _cell(spark: Text, label: str) -> Text: def _axis(width: int, first: datetime, last: datetime) -> Text: - """` ┄┄┄ `, never wider than the sparkline above it. - - 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) if len(left) + len(right) + 3 > width: left, right = _stamp(first, clock_only=True), _stamp(last, clock_only=True) @@ -250,8 +224,6 @@ def _window(job_metrics: JobMetrics) -> Optional[tuple[datetime, datetime]]: 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.""" return max((len(metric.timestamps) for metric in job_metrics.metrics), default=0) From 838b933905c88a902f97d6e16989d944f760240c Mon Sep 17 00:00:00 2001 From: Andrey Cheptsov Date: Wed, 12 Aug 2026 14:18:44 +0200 Subject: [PATCH 3/4] Name the rendering helpers after what they do `_axis` and `_stamp` said nothing about building a timeline row or formatting a timestamp for it, and `_span` sat next to `_window` meaning two different windows. Names that carry their meaning at the call site beat a docstring that only carries it at the definition: _axis -> _time_axis builds the timeline row _stamp -> _time_label formats one timestamp for it _span -> _shared_window the window all jobs are drawn against _window -> _job_window one job's own first and last sample _lead -> _blank_cells cells to blank before a job started _drawn -> _cells_drawn cells a job actually fills _cell -> _chart_cell a sparkline joined to its number _level_cell -> _capacity_cell memory as a fraction of capacity _latest -> _latest_value Co-Authored-By: Claude Opus 5 (1M context) --- src/dstack/_internal/cli/utils/metrics.py | 79 ++++++++++--------- src/tests/_internal/cli/utils/test_metrics.py | 10 +-- 2 files changed, 47 insertions(+), 42 deletions(-) diff --git a/src/dstack/_internal/cli/utils/metrics.py b/src/dstack/_internal/cli/utils/metrics.py index 363b7e2a9..d3fd3dc24 100644 --- a/src/dstack/_internal/cli/utils/metrics.py +++ b/src/dstack/_internal/cli/utils/metrics.py @@ -49,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 @@ -61,13 +61,13 @@ 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 @@ -98,51 +98,56 @@ 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]]: - 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]]: + 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: - window = _window(metrics) - if window is None or span is None: +def _blank_cells( + metrics: JobMetrics, window: Optional[tuple[datetime, datetime]], width: int +) -> int: + 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: + 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: + return cell if blanks <= 0 else Text.assemble(Text(" " * blanks), cell) def _cpu_cell(job_metrics: JobMetrics, resources: Optional[Resources], width: int) -> Text: @@ -154,7 +159,7 @@ 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 _chart_cell(sparkline(values, width, HOST_RAMP), f"{values[-1]:.0f}%") def _memory_cell(job_metrics: JobMetrics, resources: Optional[Resources], width: int) -> Text: @@ -162,7 +167,7 @@ def _memory_cell(job_metrics: JobMetrics, resources: Optional[Resources], width: 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( @@ -177,32 +182,32 @@ 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 _chart_cell(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: 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) + return _chart_cell(sparkline(percents, width, ramp), label) -def _cell(spark: Text, label: str) -> Text: +def _chart_cell(spark: Text, label: str) -> Text: return Text.assemble(spark, " ", label) -def _axis(width: int, first: datetime, last: datetime) -> Text: - left, right = _stamp(first), _stamp(last) +def _time_axis(width: int, first: datetime, last: datetime) -> Text: + 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 @@ -211,14 +216,14 @@ def _axis(width: int, first: datetime, last: datetime) -> Text: return axis -def _stamp(moment: datetime, clock_only: bool = False) -> str: +def _time_label(moment: datetime, clock_only: bool = False) -> str: 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]]: stamps = [t for metric in job_metrics.metrics for t in metric.timestamps] return (min(stamps), max(stamps)) if stamps else None @@ -236,7 +241,7 @@ 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 @@ -244,7 +249,7 @@ def _latest(job_metrics: JobMetrics, name: str) -> Optional[Any]: 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 diff --git a/src/tests/_internal/cli/utils/test_metrics.py b/src/tests/_internal/cli/utils/test_metrics.py index b9ceee70c..7fdb6900e 100644 --- a/src/tests/_internal/cli/utils/test_metrics.py +++ b/src/tests/_internal/cli/utils/test_metrics.py @@ -8,9 +8,9 @@ from rich.theme import Theme from dstack._internal.cli.utils.metrics import ( - _axis, - _stamp, - _window, + _job_window, + _time_axis, + _time_label, format_memory, get_metrics_table, job_labels, @@ -192,12 +192,12 @@ def test_a_finished_run_cannot_look_live(self, state: str, live: bool): @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 (_stamp(moment) == "now") == live + 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 = _axis(60, *_window(metrics)) + axis = _time_axis(60, *_job_window(metrics)) styles = {str(span.style) for span in axis.spans} assert ("bold grey58" in styles) == emphasised From e0055d98142e6434d5a16ae625e787adb4c45977 Mon Sep 17 00:00:00 2001 From: Andrey Cheptsov Date: Wed, 12 Aug 2026 14:24:49 +0200 Subject: [PATCH 4/4] Say what each rendering helper does Renaming alone did not carry it: `_time_axis` still did not say it builds a row, `_time_label` hid that it returns `now` for a live job, and `_samples_num` did not say it takes the longest series. Each now has one line stating what it does. `_chart_cell` is gone -- it wrapped a single `Text.assemble` at three call sites and only added a name to look up. Co-Authored-By: Claude Opus 5 (1M context) --- src/dstack/_internal/cli/utils/metrics.py | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/src/dstack/_internal/cli/utils/metrics.py b/src/dstack/_internal/cli/utils/metrics.py index d3fd3dc24..f921d18f0 100644 --- a/src/dstack/_internal/cli/utils/metrics.py +++ b/src/dstack/_internal/cli/utils/metrics.py @@ -73,6 +73,7 @@ def get_metrics_table( def job_labels(jobs: Sequence[Job]) -> List[str]: + """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 @@ -119,6 +120,7 @@ def _add_job( 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 @@ -129,6 +131,7 @@ def _shared_window(metrics: Sequence[JobMetrics]) -> Optional[tuple[datetime, da 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 @@ -142,11 +145,13 @@ def _blank_cells( 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, 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) @@ -159,7 +164,7 @@ 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 _chart_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: @@ -189,22 +194,20 @@ 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 _chart_cell(sparkline(values, width, GPU_RAMP), f"{values[-1]:.0f}%") + return Text.assemble(sparkline(values, width, GPU_RAMP), " ", f"{values[-1]:.0f}%") 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 _chart_cell(sparkline(percents, width, ramp), label) - - -def _chart_cell(spark: Text, label: str) -> Text: - return Text.assemble(spark, " ", label) + return Text.assemble(sparkline(percents, width, ramp), " ", label) 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 = _time_label(first, clock_only=True), _time_label(last, clock_only=True) @@ -217,6 +220,7 @@ def _time_axis(width: int, first: datetime, last: datetime) -> Text: 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() @@ -224,11 +228,13 @@ def _time_label(moment: datetime, clock_only: bool = False) -> str: 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: + """How many samples the longest series holds.""" return max((len(metric.timestamps) for metric in job_metrics.metrics), default=0)