diff --git a/mkdocs/blog/posts/agentic-orchestration.md b/mkdocs/blog/posts/agentic-orchestration.md
index 71ad35e87..15e2c1b8e 100644
--- a/mkdocs/blog/posts/agentic-orchestration.md
+++ b/mkdocs/blog/posts/agentic-orchestration.md
@@ -250,8 +250,12 @@ $ dstack event --within-run train-qwen
```shell
$ dstack metrics train-qwen
- NAME STATUS CPU MEMORY GPU
- train-qwen running 92% 118GB/200GB gpu=0 mem=71GB/80GB util=97%
+ UTILIZATION MEMORY
+ cpu ▅▄▄▆▆▆▆▆▆▆▆▆▆▆▆▆▆▅▆▆▆▆▆▆▆▆▆ 91% of 32 ▃▃▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄ 116GB/200GB
+
+ gpu=0 ▁▂▃▆▆▆▆▆▆▆▆▆▆▆▆▆▆▁▆▆▆▆▆▆▆▆▆ 92% ▄▅▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆ 70GB/80GB
+
+ 4 Aug 14:10 ┄┄┄┄┄┄┄┄┄┄┄ now 4 Aug 14:10 ┄┄┄┄┄┄┄┄┄┄┄ now
```
diff --git a/mkdocs/blog/posts/dstack-metrics.md b/mkdocs/blog/posts/dstack-metrics.md
index d3bf4ffa6..eeccdc45b 100644
--- a/mkdocs/blog/posts/dstack-metrics.md
+++ b/mkdocs/blog/posts/dstack-metrics.md
@@ -16,7 +16,23 @@ While it's possible to use third-party monitoring tools with `dstack`, it is oft
track metrics out of the box. That's why, with the latest release, `dstack` introduced [`dstack stats`](../../docs/reference/cli/dstack/metrics.md), a new CLI (and API)
for monitoring container metrics, including GPU usage for `NVIDIA`, `AMD`, and other accelerators.
-
+
+
+```shell
+$ dstack metrics llama-70b-sft
+
+ UTILIZATION MEMORY
+ cpu ▅▄▄▄▃▃▃▃▃▃▃▃▃▃▃▃▅▅▄▂▃▃▃▃▃▃▃ 39% of 64 ▃▃▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄ 297GB/480GB
+
+ gpu=0 ▁▂▃▆▆▆▆▆▆▆▆▆▆▆▆▆▆▁▆▆▆▆▆▆▆▆▆ 89% ▄▅▅▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆ 67GB/80GB
+ gpu=1 ▁▂▆▆▆▅▆▆▆▆▆▆▆▆▆▆▁▁▅▆▆▅▆▆▆▆▆ 84% ▄▅▅▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆ 67GB/80GB
+ gpu=2 ▁▂▆▆▆▆▆▆▆▆▆▆▆▆▆▆▁▆▆▆▆▆▆▆▆▆▆ 87% ▄▅▅▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆ 67GB/80GB
+ gpu=3 ▂▃▆▅▅▅▅▅▅▆▅▅▆▆▆▆▁▅▅▅▅▅▆▅▅▅▅ 82% ▄▅▅▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆ 67GB/80GB
+
+ 4 Aug 14:10 ┄┄┄┄┄┄┄┄┄┄┄ now 4 Aug 14:10 ┄┄┄┄┄┄┄┄┄┄┄ now
+```
+
+
@@ -31,6 +47,9 @@ difference is that `dstack stats` includes GPU VRAM usage and GPU utilization pe
Similar to `kubectl top`, if a run consists of multiple jobs (such as distributed training or an auto-scalable service),
`dstack stats` will display metrics per job.
+> Note, `dstack metrics` now shows one job at a time, like `dstack logs`. Use `--replica` and `--job` to
+> choose it; both default to `0`.
+
!!! info "HTTP API"
In addition to the `dstack stats` CLI commands, metrics can be obtained via the
[`/api/project/{project_name}/metrics/job/{run_name}`](../../docs/reference/http/metrics.md) HTTP endpoint.
diff --git a/mkdocs/docs/concepts/metrics.md b/mkdocs/docs/concepts/metrics.md
index 889c7bc9b..8e916bd98 100644
--- a/mkdocs/docs/concepts/metrics.md
+++ b/mkdocs/docs/concepts/metrics.md
@@ -18,27 +18,36 @@ This tab displays key CPU, memory, and GPU metrics collected during the last hou
## CLI
-As an alternative to the UI, you can track real-time essential metrics via the CLI.
-The `dstack metrics` command displays the most recently tracked CPU, memory, and GPU metrics.
+As an alternative to the UI, you can track essential metrics via the CLI.
+The `dstack metrics` command charts CPU, memory, and GPU utilization over the last hour of the
+job, with the latest value beside each chart.
```shell
dstack metrics gentle-mayfly-1
- NAME STATUS CPU MEMORY GPU
- gentle-mayfly-1 done 0% 16.27GB/2000GB gpu=0 mem=72.48GB/80GB util=0%
- gpu=1 mem=64.99GB/80GB util=0%
- gpu=2 mem=580MB/80GB util=0%
- gpu=3 mem=4MB/80GB util=0%
- gpu=4 mem=4MB/80GB util=0%
- gpu=5 mem=4MB/80GB util=0%
- gpu=6 mem=4MB/80GB util=0%
- gpu=7 mem=292MB/80GB util=0%
+ UTILIZATION MEMORY
+ cpu ▅▄▄▄▃▃▃▃▃▃▃▃▃▃▃▃▅▅▄▃▃▃▃▃▃▃▃ 41% of 128 ▃▃▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄ 581GB/960GB
+
+ gpu=0 ▁▂▃▆▆▆▆▆▆▆▆▆▆▆▆▆▆▁▆▆▆▆▆▆▆▆▆ 89% ▄▅▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆ 71GB/80GB
+ gpu=1 ▁▂▆▆▆▅▆▆▆▆▆▆▆▆▆▆▁▁▅▆▆▅▆▆▆▆▆ 84% ▄▅▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆ 71GB/80GB
+ gpu=2 ▁▂▆▆▆▆▆▆▆▆▆▆▆▆▆▆▁▆▆▆▆▆▆▆▆▆▆ 87% ▄▅▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆ 71GB/80GB
+ gpu=3 ▂▃▆▅▅▅▅▅▅▆▅▅▆▆▆▆▁▅▅▅▅▅▆▅▅▅▅ 82% ▄▅▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆ 71GB/80GB
+ gpu=4 ▂▆▆▆▆▆▆▆▆▆▆▆▆▆▆▁▁▆▆▆▆▆▆▆▆▆▆ 90% ▄▅▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆ 71GB/80GB
+ gpu=5 ▂▆▆▆▆▆▆▆▆▆▆▆▆▆▆▁▆▆▆▆▆▆▆▆▆▆▆ 85% ▄▅▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆ 71GB/80GB
+ gpu=6 ▃▆▆▅▅▆▆▆▆▆▆▆▆▆▆▁▅▅▅▅▆▆▆▅▅▅▅ 83% ▄▅▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆ 71GB/80GB
+ gpu=7 ▃▆▆▆▆▆▆▆▆▆▆▆▆▆▁▁▆▆▆▆▆▆▆▆▆▆▆ 88% ▄▅▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆ 71GB/80GB
+
+ 4 Aug 14:10 ┄┄┄┄┄┄┄┄┄┄┄ now 4 Aug 14:10 ┄┄┄┄┄┄┄┄┄┄┄ now
```
+Like [`dstack logs`](../reference/cli/dstack/logs.md), the command shows a single job.
+Use `--replica` and `--job` to select one; both default to `0`.
+Pass `-w` to keep the charts updating.
+
## Prometheus
To enable exporting metrics to Prometheus, set the
diff --git a/mkdocs/docs/guides/migration/slurm.md b/mkdocs/docs/guides/migration/slurm.md
index 2791075e8..fbe2dc3c3 100644
--- a/mkdocs/docs/guides/migration/slurm.md
+++ b/mkdocs/docs/guides/migration/slurm.md
@@ -1475,8 +1475,13 @@ Check real-time metrics:
```shell
$ dstack metrics training-job
- NAME STATUS CPU MEMORY GPU
- training-job running 45% 16.27GB/200GB gpu=0 mem=72.48GB/80GB util=95%
+
+ UTILIZATION MEMORY
+ cpu ▅▄▄▄▃▃▃▃▃▃▃▃▃▃▃▃▅▅▄▃▃▃▃▃▃▃▃ 45% of 32 ▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁ 16GB/200GB
+
+ gpu=0 ▁▂▃▆▆▆▆▆▆▆▆▆▆▆▆▆▆▁▆▆▆▆▆▆▆▆▆ 90% ▄▅▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆▆ 71GB/80GB
+
+ 4 Aug 14:10 ┄┄┄┄┄┄┄┄┄┄┄ now 4 Aug 14:10 ┄┄┄┄┄┄┄┄┄┄┄ now
```
diff --git a/src/dstack/_internal/cli/commands/metrics.py b/src/dstack/_internal/cli/commands/metrics.py
index 16092748c..3166640df 100644
--- a/src/dstack/_internal/cli/commands/metrics.py
+++ b/src/dstack/_internal/cli/commands/metrics.py
@@ -1,21 +1,22 @@
import argparse
import time
-from typing import Any, List, Optional
from rich.live import Live
-from rich.table import Table
from dstack._internal.cli.commands import APIBaseCommand
from dstack._internal.cli.services.completion import RunNameCompleter
from dstack._internal.cli.utils.common import (
- LIVE_TABLE_PROVISION_INTERVAL_SECS,
LIVE_TABLE_REFRESH_RATE_PER_SEC,
- add_row_from_dict,
console,
)
+from dstack._internal.cli.utils.metrics import (
+ MAX_SAMPLES,
+ WATCH_INTERVAL_SECONDS,
+ get_metrics_table,
+)
from dstack._internal.core.errors import CLIError
-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.api._public import Client
from dstack.api._public.runs import Run
@@ -33,121 +34,61 @@ def _register(self):
help="Watch run metrics in realtime",
action="store_true",
)
+ self._parser.add_argument(
+ "--replica",
+ help="The replica number. Defaults to 0.",
+ type=int,
+ default=0,
+ )
+ self._parser.add_argument(
+ "--job",
+ help="The job number inside the replica. Defaults to 0.",
+ type=int,
+ default=0,
+ )
def _command(self, args: argparse.Namespace):
super()._command(args)
- run = self.api.runs.get(run_name=args.run_name)
- if run is None:
- raise CLIError(f"Run {args.run_name} not found")
- metrics = _get_run_jobs_metrics(api=self.api, run=run)
+ job, metrics = self._fetch(args)
if not args.watch:
- console.print(_get_metrics_table(run, metrics))
+ console.print(get_metrics_table(job, metrics))
return
try:
with Live(console=console, refresh_per_second=LIVE_TABLE_REFRESH_RATE_PER_SEC) as live:
while True:
- live.update(_get_metrics_table(run, metrics))
- time.sleep(LIVE_TABLE_PROVISION_INTERVAL_SECS)
- run = self.api.runs.get(run_name=args.run_name)
- if run is None:
- raise CLIError(f"Run {args.run_name} not found")
- metrics = _get_run_jobs_metrics(api=self.api, run=run)
+ live.update(get_metrics_table(job, metrics))
+ time.sleep(WATCH_INTERVAL_SECONDS)
+ job, metrics = self._fetch(args)
except KeyboardInterrupt:
pass
+ def _fetch(self, args: argparse.Namespace) -> tuple[Job, JobMetrics]:
+ run = self.api.runs.get(run_name=args.run_name)
+ if run is None:
+ raise CLIError(f"Run {args.run_name} not found")
+ job = _get_job(run, args.replica, args.job)
+ return job, _get_job_metrics(self.api, run, job)
-def _get_run_jobs_metrics(api: Client, run: Run) -> List[JobMetrics]:
- metrics = []
- for job in run._run.jobs:
- job_metrics = api.client.metrics.get_job_metrics(
- project_name=api.project,
- run_name=run.name,
- replica_num=job.job_spec.replica_num,
- job_num=job.job_spec.job_num,
- )
- metrics.append(job_metrics)
- return metrics
-
-
-def _get_metrics_table(run: Run, metrics: List[JobMetrics]) -> Table:
- table = Table(box=None)
- table.add_column("NAME", style="bold", no_wrap=True)
- table.add_column("STATUS")
- table.add_column("CPU")
- table.add_column("MEMORY")
- table.add_column("GPU")
-
- run_row = {"NAME": run.name, "STATUS": run.status.value}
- if len(run._run.jobs) != 1:
- add_row_from_dict(table, run_row)
-
- for job, job_metrics in zip(run._run.jobs, metrics):
- jrd = job.job_submissions[-1].job_runtime_data
- jpd = job.job_submissions[-1].job_provisioning_data
- resources: Optional[Resources] = None
- if jrd is not None and jrd.offer is not None:
- resources = jrd.offer.instance.resources
- elif jpd is not None:
- resources = jpd.instance_type.resources
- cpu_usage = _get_metric_value(job_metrics, "cpu_usage_percent")
- if cpu_usage is not None:
- if resources is not None:
- cpu_usage = cpu_usage / resources.cpus
- cpu_usage = f"{cpu_usage:.0f}%"
- memory_usage = _get_metric_value(job_metrics, "memory_working_set_bytes")
- if memory_usage is not None:
- memory_usage = _format_memory(memory_usage, 2)
- if resources is not None:
- memory_usage += f"/{_format_memory(resources.memory_mib * 1024 * 1024, 2)}"
- gpu_metrics = ""
- gpus_detected_num = _get_metric_value(job_metrics, "gpus_detected_num")
- if gpus_detected_num is not None:
- for i in range(gpus_detected_num):
- gpu_memory_usage = _get_metric_value(job_metrics, f"gpu_memory_usage_bytes_gpu{i}")
- gpu_util_percent = _get_metric_value(job_metrics, f"gpu_util_percent_gpu{i}")
- if gpu_memory_usage is not None:
- if i != 0:
- gpu_metrics += "\n"
- gpu_metrics += f"gpu={i} mem={_format_memory(gpu_memory_usage, 2)}"
- if resources is not None:
- gpu_metrics += (
- f"/{_format_memory(resources.gpus[i].memory_mib * 1024 * 1024, 2)}"
- )
- gpu_metrics += f" util={gpu_util_percent}%"
-
- job_row = {
- "NAME": f" replica={job.job_spec.replica_num} job={job.job_spec.job_num}",
- "STATUS": job.job_submissions[-1].status.value,
- "CPU": cpu_usage or "-",
- "MEMORY": memory_usage or "-",
- "GPU": gpu_metrics or "-",
- }
- if len(run._run.jobs) == 1:
- job_row.update(run_row)
- add_row_from_dict(table, job_row)
-
- return table
-
-
-def _get_metric_value(job_metrics: JobMetrics, name: str) -> Optional[Any]:
- for metric in job_metrics.metrics:
- if metric.name == name:
- return metric.values[-1]
- return None
-
-
-def _format_memory(memory_bytes: int, decimal_places: int) -> str:
- """See test_format_memory in tests/_internal/cli/commands/test_metrics.py for examples."""
- memory_mb = memory_bytes / 1024 / 1024
- if memory_mb >= 1024:
- value = memory_mb / 1024
- unit = "GB"
- else:
- value = memory_mb
- unit = "MB"
- if decimal_places == 0:
- return f"{round(value)}{unit}"
- return f"{value:.{decimal_places}f}".rstrip("0").rstrip(".") + unit
+def _get_job(run: Run, replica_num: int, job_num: int) -> Job:
+ for job in run._run.jobs:
+ if job.job_spec.replica_num == replica_num and job.job_spec.job_num == job_num:
+ return job
+ raise CLIError(
+ f"Run {run.name} has no replica={replica_num} job={job_num}."
+ " Use --replica and --job to select one."
+ )
+
+
+def _get_job_metrics(api: Client, run: Run, job: Job) -> JobMetrics:
+ """`limit` must be sent explicitly: the endpoint declares it `limit: int = 1`, not
+ Optional, so omitting it caps the response at one sample."""
+ return api.client.metrics.get_job_metrics(
+ project_name=api.project,
+ run_name=run.name,
+ replica_num=job.job_spec.replica_num,
+ job_num=job.job_spec.job_num,
+ limit=MAX_SAMPLES,
+ )
diff --git a/src/dstack/_internal/cli/utils/metrics.py b/src/dstack/_internal/cli/utils/metrics.py
new file mode 100644
index 000000000..ccf840a99
--- /dev/null
+++ b/src/dstack/_internal/cli/utils/metrics.py
@@ -0,0 +1,209 @@
+from datetime import datetime
+from typing import Any, List, Optional
+
+from rich.console import RenderableType
+from rich.table import Table
+from rich.text import Text
+
+from dstack._internal.cli.utils.common import console
+from dstack._internal.cli.utils.sparkline import GPU_RAMP, HOST_RAMP, Ramp, no_data, sparkline
+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
+never under-filled. Matches the UI."""
+
+WATCH_INTERVAL_SECONDS = 10
+"""Matched to the server's collection cadence; a new point cannot arrive faster."""
+
+MIN_SPARK_WIDTH = 10
+MAX_SPARK_WIDTH = 80
+
+AXIS_RULE = "┄"
+_FIXED_COLUMNS = 34
+"""Labels, numbers and padding. Hand-measured against a `589GB/1480GB`-sized label; a
+wider one (a 2000GB host prints `1218GB/2000GB`) overflows and Rich ellipsizes the row
+labels rather than shrinking the sparklines."""
+
+_SPARKLINE_COLUMNS = 2
+
+
+def _spark_width(console_width: int) -> int:
+ budget = console_width - _FIXED_COLUMNS
+ return max(MIN_SPARK_WIDTH, min(MAX_SPARK_WIDTH, budget // _SPARKLINE_COLUMNS))
+
+
+def get_metrics_table(
+ job: Job, metrics: JobMetrics, console_width: Optional[int] = None
+) -> RenderableType:
+ resources = _get_resources(job)
+ width = _spark_width(console_width or console.width)
+
+ table = Table(box=None)
+ # no header: every cell in this column already reads `cpu` or `gpu=N`
+ table.add_column("", style="secondary", no_wrap=True)
+ table.add_column("UTILIZATION", no_wrap=True)
+ table.add_column("MEMORY", no_wrap=True)
+
+ table.add_row(
+ "cpu",
+ _cpu_cell(metrics, resources, width),
+ _memory_cell(metrics, resources, width),
+ )
+ table.add_row("", "", "") # host and devices are different things; separate them
+ for index in range(_gpus_num(metrics, resources)):
+ table.add_row(
+ f"gpu={index}",
+ _gpu_util_cell(metrics, index, width),
+ _gpu_memory_cell(metrics, resources, index, width),
+ )
+ window = _window(metrics)
+ if window is not None:
+ axis = _axis(min(width, _samples_num(metrics)), *window)
+ table.add_row("", "", "")
+ table.add_row("", axis, axis)
+ return table
+
+
+def _cpu_cell(job_metrics: JobMetrics, resources: Optional[Resources], width: int) -> Text:
+ values = _metric_values(job_metrics, "cpu_usage_percent")
+ if not values:
+ return no_data()
+ cpus = resources.cpus if resources else None
+ if cpus:
+ values = [v / cpus for v in values]
+ label = f"{values[-1]:.0f}%"
+ if cpus:
+ label += f" of {cpus}"
+ return _cell(sparkline(values, width, HOST_RAMP), label)
+
+
+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)
+
+
+def _gpu_memory_cell(
+ job_metrics: JobMetrics,
+ resources: Optional[Resources],
+ index: int,
+ width: int,
+) -> Text:
+ values = _metric_values(job_metrics, f"gpu_memory_usage_bytes_gpu{index}")
+ if not values:
+ return no_data()
+ 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)
+
+
+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}%")
+
+
+def _level_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)
+
+
+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.
+
+ 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)
+ 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")
+
+
+def _stamp(moment: datetime, clock_only: bool = False) -> str:
+ if pretty_date(moment) == "now":
+ 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]]:
+ 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."""
+ return max((len(metric.timestamps) for metric in job_metrics.metrics), default=0)
+
+
+def _metric_values(job_metrics: JobMetrics, name: str) -> List[Any]:
+ """Values for `name`, oldest first. The server sends latest-first; reversing here, once,
+ is what keeps every sparkline downstream running left-to-right in time."""
+ for metric in job_metrics.metrics:
+ if metric.name == name:
+ return list(reversed(metric.values))
+ return []
+
+
+def _latest(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")
+ return int(detected) if detected else 0
+
+
+def _get_resources(job: Job) -> Optional[Resources]:
+ submission = job.job_submissions[-1]
+ jrd = submission.job_runtime_data
+ if jrd is not None and jrd.offer is not None:
+ return jrd.offer.instance.resources
+ jpd = submission.job_provisioning_data
+ if jpd is not None:
+ return jpd.instance_type.resources
+ return None
+
+
+def format_memory(memory_bytes: float, decimal_places: int) -> str:
+ memory_mb = memory_bytes / 1024 / 1024
+ if memory_mb >= 1024:
+ value = memory_mb / 1024
+ unit = "GB"
+ else:
+ value = memory_mb
+ unit = "MB"
+
+ if decimal_places == 0:
+ return f"{round(value)}{unit}"
+ return f"{value:.{decimal_places}f}".rstrip("0").rstrip(".") + unit
diff --git a/src/dstack/_internal/cli/utils/sparkline.py b/src/dstack/_internal/cli/utils/sparkline.py
new file mode 100644
index 000000000..93beddfc5
--- /dev/null
+++ b/src/dstack/_internal/cli/utils/sparkline.py
@@ -0,0 +1,71 @@
+from typing import List, Optional, Sequence
+
+from rich.text import Text
+
+SPARKS = "▁▂▃▄▅▆▇"
+"""No full block: it fills the cell to the top edge, fusing consecutive rows into one mass."""
+
+NO_DATA = "no data"
+
+Ramp = Sequence[tuple[float, str]]
+
+# Colour encodes scope: device metrics share one ramp, host metrics another, so a job row is
+# never mistaken for a device row.
+GPU_RAMP: Ramp = ((25, "grey42"), (50, "chartreuse4"), (75, "chartreuse3"), (101, "green1"))
+HOST_RAMP: Ramp = (
+ (50, "steel_blue3"),
+ (90, "deep_sky_blue3"),
+ (97, "dark_orange3"),
+ (101, "red3"),
+)
+
+
+def ramp_style(value: float, ramp: Ramp) -> str:
+ for threshold, style in ramp:
+ if value < threshold:
+ return style
+ return ramp[-1][1]
+
+
+def slices(values: Sequence[float], width: int) -> List[tuple[float, float]]:
+ """`(peak, mean)` per cell, oldest first -- height is the peak, colour the mean, so a
+ card that *touched* 100% reads differently from one that *held* it.
+
+ Cells span the whole series, not its tail. The last cell is the latest sample, so it
+ cannot contradict the number printed beside the sparkline.
+ """
+ vals = list(values)
+ if width < 1:
+ return []
+ if width == 1 or len(vals) <= width:
+ return [(v, v) for v in vals[-width:]]
+ history, out = vals[:-1], []
+ for i in range(width - 1):
+ lo = int(i * len(history) / (width - 1))
+ hi = max(lo + 1, int((i + 1) * len(history) / (width - 1)))
+ chunk = history[lo:hi]
+ out.append((max(chunk), sum(chunk) / len(chunk)))
+ out.append((vals[-1], vals[-1]))
+ return out
+
+
+def no_data() -> Text:
+ """Spelled out: next to sparklines a bare `-` reads as a stray glyph."""
+ return Text(NO_DATA, style="grey58")
+
+
+def sparkline(
+ values: Optional[Sequence[float]],
+ width: int,
+ ramp: Optional[Ramp] = None,
+ vmax: float = 100.0,
+) -> Text:
+ """Fixed 0..vmax scale, never autoscaled: height means the same thing on every row."""
+ if not values:
+ return no_data()
+ text = Text()
+ for peak, mean in slices(values, width):
+ index = int(max(0.0, min(vmax, peak)) / vmax * (len(SPARKS) - 1))
+ shade = max(0.0, min(vmax, mean)) / vmax * 100
+ text.append(SPARKS[index], style=ramp_style(shade, ramp) if ramp else "cyan")
+ return text
diff --git a/src/dstack/api/server/_metrics.py b/src/dstack/api/server/_metrics.py
index b9604461d..2a5fafc20 100644
--- a/src/dstack/api/server/_metrics.py
+++ b/src/dstack/api/server/_metrics.py
@@ -1,3 +1,6 @@
+from datetime import datetime
+from typing import Any, Dict, Optional
+
from dstack._internal.core.models.common import validate_extra_ignore
from dstack._internal.core.models.metrics import JobMetrics
from dstack.api.server._group import APIClientGroup
@@ -10,13 +13,28 @@ def get_job_metrics(
run_name: str,
replica_num: int = 0,
job_num: int = 0,
+ after: Optional[datetime] = None,
+ before: Optional[datetime] = None,
+ limit: Optional[int] = None,
) -> JobMetrics:
+ """
+ Returns job metrics ordered from the latest sample to the earliest.
+
+ Without `after`/`before`/`limit`, the server returns one latest sample.
+ """
+ params: Dict[str, Any] = {
+ "replica_num": replica_num,
+ "job_num": job_num,
+ }
+ if after is not None:
+ params["after"] = after.isoformat()
+ if before is not None:
+ params["before"] = before.isoformat()
+ if limit is not None:
+ params["limit"] = limit
resp = self._request(
f"/api/project/{project_name}/metrics/job/{run_name}",
method="GET",
- params={
- "replica_num": replica_num,
- "job_num": job_num,
- },
+ params=params,
)
return validate_extra_ignore(JobMetrics, resp.json())
diff --git a/src/tests/_internal/cli/commands/test_metrics.py b/src/tests/_internal/cli/commands/test_metrics.py
index 5e9492a4a..2309a480f 100644
--- a/src/tests/_internal/cli/commands/test_metrics.py
+++ b/src/tests/_internal/cli/commands/test_metrics.py
@@ -1,34 +1,47 @@
+from unittest.mock import MagicMock
+
import pytest
-from dstack._internal.cli.commands.metrics import _format_memory
-
-
-@pytest.mark.parametrize(
- "bytes_value,decimal_places,expected",
- [
- # Test MB values with different decimal places
- (512 * 1024 * 1024, 0, "512MB"), # exact MB, no decimals
- (512 * 1024 * 1024, 2, "512MB"), # exact MB, with decimals
- (512.5 * 1024 * 1024, 0, "512MB"), # decimal MB, no decimals
- (512.5 * 1024 * 1024, 2, "512.5MB"), # decimal MB, 2 decimals
- (512.5 * 1024 * 1024, 3, "512.5MB"), # decimal MB, 3 decimals
- (999 * 1024 * 1024, 0, "999MB"), # just under 1GB, no decimals
- (999 * 1024 * 1024, 2, "999MB"), # just under 1GB, with decimals
- # Test GB values with different decimal places
- (1.5 * 1024 * 1024 * 1024, 0, "2GB"), # decimal GB, no decimals
- (1.5 * 1024 * 1024 * 1024, 2, "1.5GB"), # decimal GB, 2 decimals
- (1.5 * 1024 * 1024 * 1024, 3, "1.5GB"), # decimal GB, 3 decimals
- (2 * 1024 * 1024 * 1024, 0, "2GB"), # exact GB, no decimals
- (2 * 1024 * 1024 * 1024, 2, "2GB"), # exact GB, with decimals
- # Test edge cases
- (0, 0, "0MB"), # zero bytes, no decimals
- (0, 2, "0MB"), # zero bytes, with decimals
- (1023 * 1024, 0, "1MB"), # just under 1MB, no decimals
- (1023 * 1024, 2, "1MB"), # just under 1MB, with decimals
- (1024 * 1024 * 1024 - 1, 0, "1024MB"), # just under 1GB, no decimals
- (1024 * 1024 * 1024 - 1, 2, "1024MB"), # just under 1GB, with decimals
- ],
-)
-def test_format_memory(bytes_value: int, decimal_places: int, expected: str):
- result = _format_memory(bytes_value, decimal_places)
- assert result == expected
+from dstack._internal.cli.commands.metrics import _get_job, _get_job_metrics
+from dstack._internal.cli.utils.metrics import MAX_SAMPLES
+from dstack._internal.core.errors import CLIError
+from dstack._internal.core.models.metrics import JobMetrics
+
+
+def _run(replicas: int = 1, jobs_per_replica: int = 1):
+ run = MagicMock()
+ run.name = "my-run"
+ run._run.jobs = []
+ for replica in range(replicas):
+ for job_num in range(jobs_per_replica):
+ job = MagicMock()
+ job.job_spec.replica_num = replica
+ job.job_spec.job_num = job_num
+ run._run.jobs.append(job)
+ return run
+
+
+class TestJobSelection:
+ def test_defaults_to_the_first_job_of_the_first_replica(self):
+ job = _get_job(_run(replicas=3), replica_num=0, job_num=0)
+ assert (job.job_spec.replica_num, job.job_spec.job_num) == (0, 0)
+
+ def test_selects_by_replica_and_job(self):
+ job = _get_job(_run(replicas=3, jobs_per_replica=2), replica_num=2, job_num=1)
+ assert (job.job_spec.replica_num, job.job_spec.job_num) == (2, 1)
+
+ def test_unknown_job_is_an_error(self):
+ with pytest.raises(CLIError, match="replica=7"):
+ _get_job(_run(replicas=3), replica_num=7, job_num=0)
+
+
+class TestMetricsRequest:
+ def test_limit_is_sent_explicitly(self):
+ api = MagicMock()
+ api.project = "main"
+ api.client.metrics.get_job_metrics.return_value = JobMetrics(metrics=[])
+ run = _run()
+ _get_job_metrics(api, run, run._run.jobs[0])
+ kwargs = api.client.metrics.get_job_metrics.call_args.kwargs
+ assert kwargs["limit"] == MAX_SAMPLES
+ assert "after" not in kwargs and "before" not in kwargs
diff --git a/src/tests/_internal/cli/utils/test_metrics.py b/src/tests/_internal/cli/utils/test_metrics.py
new file mode 100644
index 000000000..2f9d1e6b4
--- /dev/null
+++ b/src/tests/_internal/cli/utils/test_metrics.py
@@ -0,0 +1,216 @@
+import re
+from datetime import datetime, timedelta, timezone
+from typing import List, Tuple
+from unittest.mock import MagicMock
+
+import pytest
+from rich.console import Console
+from rich.theme import Theme
+
+from dstack._internal.cli.utils.metrics import format_memory, get_metrics_table
+from dstack._internal.cli.utils.sparkline import SPARKS
+from dstack._internal.core.models.metrics import JobMetrics, Metric
+
+GIB = 1024**3
+CAPACITY_GB = 80
+
+# utilization percent, and memory as a fraction of capacity, over `t` in 0..1 oldest to newest
+SHAPES = {
+ "idle": (lambda t: 1.0, lambda t: 0.05),
+ "spike": (lambda t: 100.0 if 0.49 < t < 0.51 else 2.0, lambda t: 0.5),
+ "ramp": (lambda t: t * 100.0, lambda t: t * 0.5),
+ "saturated": (lambda t: 95.0, lambda t: 0.95),
+ "low": (lambda t: 19.0, lambda t: 0.19),
+}
+
+
+def make_run(
+ shape: str = "saturated",
+ samples: int = 360,
+ state: str = "running",
+ gpus: int = 1,
+ cpus: int = 8,
+) -> Tuple[MagicMock, JobMetrics]:
+ """A job and its metrics. `state` decides whether the newest sample reads as `now`."""
+ newest = datetime.now(timezone.utc)
+ if state == "terminated":
+ newest -= timedelta(hours=2)
+ timestamps = [newest - timedelta(seconds=10 * i) for i in range(samples)]
+
+ def series(fn) -> List[float]:
+ oldest_first = [fn(i / max(1, samples - 1)) for i in range(samples)]
+ return list(reversed(oldest_first)) # the server returns points newest first
+
+ util, memory = SHAPES[shape]
+ metrics = [
+ Metric(
+ name="cpu_usage_percent",
+ timestamps=timestamps,
+ values=series(lambda t: util(t) * cpus),
+ ),
+ Metric(
+ name="memory_working_set_bytes",
+ timestamps=timestamps,
+ values=series(lambda t: memory(t) * 32 * GIB),
+ ),
+ ]
+ for index in range(gpus):
+ metrics.append(
+ Metric(name=f"gpu_util_percent_gpu{index}", timestamps=timestamps, values=series(util))
+ )
+ metrics.append(
+ Metric(
+ name=f"gpu_memory_usage_bytes_gpu{index}",
+ timestamps=timestamps,
+ values=series(lambda t: memory(t) * CAPACITY_GB * GIB),
+ )
+ )
+
+ job = MagicMock()
+ submission = MagicMock()
+ resources = MagicMock()
+ resources.cpus, resources.memory_mib = cpus, 32 * 1024
+ resources.gpus = [MagicMock(memory_mib=CAPACITY_GB * 1024) for _ in range(gpus)]
+ submission.job_runtime_data.offer.instance.resources = resources
+ job.job_submissions = [submission]
+ return job, JobMetrics(metrics=metrics)
+
+
+def render(job, metrics: JobMetrics, width: int = 200, color: bool = False) -> str:
+ console = Console(
+ width=width,
+ theme=Theme({"secondary": "grey58"}),
+ no_color=not color,
+ force_terminal=color,
+ color_system="truecolor" if color else None,
+ )
+ with console.capture() as capture:
+ console.print(get_metrics_table(job, metrics, console_width=width))
+ return capture.get()
+
+
+def lines(output: str) -> List[str]:
+ return [line.rstrip() for line in output.splitlines() if line.strip()]
+
+
+def row(output: str, label: str) -> str:
+ return next(line for line in lines(output) if line.strip().startswith(label))
+
+
+def bars(line: str) -> List[int]:
+ """Glyph heights of the first sparkline in `line`, left to right."""
+ return [SPARKS.index(glyph) for glyph in re.findall(rf"[{SPARKS}]+", line)[0]]
+
+
+def colours(output: str, label: str) -> set:
+ """Distinct colours among the glyphs of `label`'s row."""
+ line = next(ln for ln in output.splitlines() if label in ln)
+ return {code for code, _ in re.findall(rf"\x1b\[([0-9;]+)m([{SPARKS}])", line)}
+
+
+class TestRendering:
+ def test_idle_draws_flat_and_low_in_one_colour(self):
+ """An idle GPU is a flat low line in a single colour, not a rainbow of bands."""
+ job, metrics = make_run("idle")
+ assert set(bars(row(render(job, metrics), "gpu=0"))) == {0}
+ assert len(colours(render(job, metrics, color=True), "gpu=0")) == 1
+
+ def test_a_spike_survives_bucketing(self):
+ """One sample at 100% among 360 still draws tall; averaging would erase it."""
+ job, metrics = make_run("spike")
+ assert max(bars(row(render(job, metrics), "gpu=0"))) == len(SPARKS) - 1
+
+ def test_a_ramp_climbs_left_to_right(self):
+ """The server sends points newest first, so a missing reversal mirrors every chart
+ and nothing else on screen would give it away."""
+ job, metrics = make_run("ramp")
+ heights = bars(row(render(job, metrics), "gpu=0"))
+ assert heights == sorted(heights)
+ assert heights[0] < heights[-1]
+
+ def test_height_is_a_fraction_of_capacity_not_of_the_window(self):
+ """19% of capacity looks nearly empty. Rescaling to the window's own maximum would
+ draw a steady 154GB of 800GB as a full bar."""
+ job, metrics = make_run("low")
+ assert max(bars(row(render(job, metrics), "gpu=0"))) <= 1
+
+ def test_the_number_matches_the_last_bar(self):
+ """The printed value is the newest sample and the right-hand bar draws that same
+ sample, so a value that just dropped cannot show tall beside a 0."""
+ job, metrics = make_run("ramp")
+ gpu = row(render(job, metrics), "gpu=0")
+ assert "100%" in gpu
+ assert bars(gpu)[-1] == len(SPARKS) - 1
+
+ def test_no_data_is_not_zero(self):
+ """A device we have no metrics for reads differently from an idle one, and its row
+ is listed either way -- the device list comes from the offer."""
+ job, _ = make_run("idle", gpus=2)
+ missing = render(job, JobMetrics(metrics=[]))
+ assert "no data" in missing
+ assert not re.findall(rf"[{SPARKS}]", missing)
+ assert sum(1 for line in lines(missing) if "gpu=" in line) == 2
+ assert "1%" in render(*make_run("idle", gpus=2))
+
+
+class TestWindow:
+ @pytest.mark.parametrize("samples", [12, 360], ids=["two-minutes", "an-hour"])
+ @pytest.mark.parametrize("state", ["running", "terminated"])
+ def test_draws_only_what_was_measured(self, samples: int, state: str):
+ """A young run fills part of the row and the timeline stops with it. Drawn to the
+ full width it would claim a span nothing was measured over, and Rich would widen
+ the column to fit, pulling MEMORY out of line."""
+ job, metrics = make_run("ramp", samples=samples, state=state)
+ output = render(job, metrics, width=200)
+ drawn = len(bars(row(output, "cpu")))
+ assert drawn == min(samples, 80) # 80 is MAX_SPARK_WIDTH
+ axis = lines(output)[-1]
+ assert [len(segment) for segment in re.split(r"\s{3,}", axis.strip())] == [drawn, drawn]
+
+ @pytest.mark.parametrize("state,live", [("running", True), ("terminated", False)])
+ def test_a_finished_run_cannot_look_live(self, state: str, live: bool):
+ job, metrics = make_run("saturated", state=state)
+ axis = lines(render(job, metrics))[-1]
+ assert axis.endswith("now") == live
+ if not live:
+ assert ":" in axis # a real clock time, not an age
+
+
+@pytest.mark.parametrize("width", [80, 100, 140, 190, 240])
+def test_fits_every_terminal_width(width: int):
+ """Nothing wraps or gets truncated, on the widest realistic row: eight GPUs."""
+ job, metrics = make_run("saturated", gpus=8)
+ output = render(job, metrics, width=width)
+ assert max(len(line.rstrip()) for line in output.splitlines()) <= width
+ assert "…" not in output
+
+
+@pytest.mark.parametrize(
+ "bytes_value,decimal_places,expected",
+ [
+ # Test MB values with different decimal places
+ (512 * 1024 * 1024, 0, "512MB"), # exact MB, no decimals
+ (512 * 1024 * 1024, 2, "512MB"), # exact MB, with decimals
+ (512.5 * 1024 * 1024, 0, "512MB"), # decimal MB, no decimals
+ (512.5 * 1024 * 1024, 2, "512.5MB"), # decimal MB, 2 decimals
+ (512.5 * 1024 * 1024, 3, "512.5MB"), # decimal MB, 3 decimals
+ (999 * 1024 * 1024, 0, "999MB"), # just under 1GB, no decimals
+ (999 * 1024 * 1024, 2, "999MB"), # just under 1GB, with decimals
+ # Test GB values with different decimal places
+ (1.5 * 1024 * 1024 * 1024, 0, "2GB"), # decimal GB, no decimals
+ (1.5 * 1024 * 1024 * 1024, 2, "1.5GB"), # decimal GB, 2 decimals
+ (1.5 * 1024 * 1024 * 1024, 3, "1.5GB"), # decimal GB, 3 decimals
+ (2 * 1024 * 1024 * 1024, 0, "2GB"), # exact GB, no decimals
+ (2 * 1024 * 1024 * 1024, 2, "2GB"), # exact GB, with decimals
+ # Test edge cases
+ (0, 0, "0MB"), # zero bytes, no decimals
+ (0, 2, "0MB"), # zero bytes, with decimals
+ (1023 * 1024, 0, "1MB"), # just under 1MB, no decimals
+ (1023 * 1024, 2, "1MB"), # just under 1MB, with decimals
+ (1024 * 1024 * 1024 - 1, 0, "1024MB"), # just under 1GB, no decimals
+ (1024 * 1024 * 1024 - 1, 2, "1024MB"), # just under 1GB, with decimals
+ ],
+)
+def test_format_memory(bytes_value: int, decimal_places: int, expected: str):
+ result = format_memory(bytes_value, decimal_places)
+ assert result == expected