diff --git a/src/dstack/_internal/cli/utils/metrics.py b/src/dstack/_internal/cli/utils/metrics.py index 22ce839a0..f921d18f0 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 @@ -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,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 _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: @@ -174,7 +172,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( @@ -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: - """` ┄┄┄ `, 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,7 +247,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 @@ -268,7 +255,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 17a81244b..7fdb6900e 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 ( + _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):